Branching

Conditional routing and fan-out in Python

Conditional Branching

Define typed event subclasses and return them based on conditions:

class PositiveEvent(Event):
    text: str
class NegativeEvent(Event):
    text: str

@step
async def classify(ctx: Context, ev: Event):
    if "good" in ev.text.lower():
        return PositiveEvent(text=ev.text)
    else:
        return NegativeEvent(text=ev.text)

Handling Branches

Each handler declares the specific event subclass it accepts via the type annotation:

@step
async def handle_positive(ctx: Context, ev: PositiveEvent):
    return StopEvent(result={"sentiment": "positive", "text": ev.text})

@step
async def handle_negative(ctx: Context, ev: NegativeEvent):
    return StopEvent(result={"sentiment": "negative", "text": ev.text})

Fan-out

Define branch event subclasses and return a list to dispatch to multiple branches simultaneously:

class BranchA(Event):
    value: str
class BranchB(Event):
    value: str

@step
async def fan_out(ctx: Context, ev: Event):
    return [BranchA(value="a"), BranchB(value="b")]

Both branches execute concurrently. First StopEvent wins.