Events

Create and route custom events in Python

Defining Custom Events

Subclass Event to define typed, self-documenting events. The event_type is automatically set to the class name, and fields are type-annotated for clarity but stored as JSON internally.

from blazen import Event

class AnalyzeEvent(Event):
    text: str
    score: float

ev = AnalyzeEvent(text="hello", score=0.9)
print(ev.event_type)  # "AnalyzeEvent"
print(ev.text)        # "hello"
print(ev.to_dict())   # {"text": "hello", "score": 0.9}

Built-in Events

StartEvent and StopEvent are provided by the framework. Every workflow begins with a StartEvent and terminates when a StopEvent is returned.

from blazen import StartEvent, StopEvent

start = StartEvent(message="hello")      # event_type: "blazen::StartEvent"
stop = StopEvent(result={"answer": 42})  # event_type: "blazen::StopEvent"

Event Routing via Type Hints

The type hint on the ev parameter controls which events a step accepts. No explicit accepts list is needed.

• ev: Event (or no type hint) — the step handles StartEvent by default.
• ev: SomeCustomEvent — the step automatically accepts only SomeCustomEvent.

from blazen import Event, StartEvent, StopEvent, Context, step

class AnalyzeEvent(Event):
    text: str

@step
async def first(ctx: Context, ev: Event):
    # ev: Event -> accepts StartEvent
    return AnalyzeEvent(text=ev.message)

@step
async def second(ctx: Context, ev: AnalyzeEvent):
    # ev: AnalyzeEvent -> automatically accepts=["AnalyzeEvent"]
    return StopEvent(result={"text": ev.text})

Fan-out (Multiple Events)

Return a list to dispatch multiple events simultaneously. Each event routes to its matching step independently.

from blazen import Event, Context, step

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")]

Side-Effect Steps

Return None and use ctx.send_event() when a step needs to perform a side effect without directly producing a next event in the return value.

from blazen import Event, Context, step

class ContinueEvent(Event):
    pass

@step
async def side_effect(ctx: Context, ev: Event):
    ctx.set("processed", True)
    ctx.send_event(ContinueEvent())
    return None

Alternative: Inline Events

For quick prototyping you can still create events inline without a subclass:

ev = Event("AnalyzeEvent", text="hello", score=0.9)

This is equivalent to defining a one-off event class. Prefer subclasses for production code since they give you type safety and self-documenting fields.