Quickstart

Build your first Blazen workflow in Python

Installation

Preferred (using uv):

uv add blazen

Or with pip:

pip install blazen

Define Your Events

Blazen routes work through typed events. Subclass Event to declare the data each step expects:

from blazen import Event

class GreetEvent(Event):
    name: str

Subclassing Event automatically sets event_type to the class name ("GreetEvent"). Fields are declared as plain type annotations — Blazen uses them for validation and attribute access.

Define Your Steps

Create a simple greeter workflow with two steps: one to parse input and one to produce a greeting.

import asyncio
from blazen import Workflow, step, Event, StopEvent, Context

class GreetEvent(Event):
    name: str

@step
async def parse_input(ctx: Context, ev: Event):
    return GreetEvent(name=ev.name or "World")

@step
async def greet(ctx: Context, ev: GreetEvent):
    return StopEvent(result={"greeting": f"Hello, {ev.name}!"})

The @step decorator reads the type hint on the ev parameter to decide which events a step receives. parse_input accepts the base Event (so it handles the initial start event), while greet accepts GreetEvent specifically. No accepts= argument is needed.

Because GreetEvent has a typed name field, you access it directly as ev.name — no .to_dict() unpacking required.

Build and Run

async def main():
    wf = Workflow("greeter", [parse_input, greet])
    handler = await wf.run(name="Blazen")
    result = await handler.result()
    print(result.result)

asyncio.run(main())

Running this prints:

{'greeting': 'Hello, Blazen!'}

Using Context

Steps can share state through the Context object. Context access is synchronous — no await needed:

@step
async def parse_input(ctx: Context, ev: Event):
    ctx.set("request_count", ctx.get("request_count", 0) + 1)
    return GreetEvent(name=ev.name or "World")

@step
async def greet(ctx: Context, ev: GreetEvent):
    count = ctx.get("request_count", 0)
    return StopEvent(result={"greeting": f"Hello, {ev.name}!", "request_number": count})

ctx.set(key, value) stores a value and ctx.get(key) (or ctx.get(key, default)) retrieves it. Both are plain synchronous calls you can use anywhere inside a step.

ctx.set() accepts any Python value — not just JSON-serializable types. Bytes are stored as raw binary, Pydantic models and other complex objects are pickled automatically, and ctx.get() returns the original type transparently.

Key Concepts

  • Event subclasses — class MyEvent(Event): ... gives you typed fields, automatic event_type naming, and direct attribute access (ev.field).
  • Type-hint routing — @step inspects the ev parameter’s type hint to route events. A hint of Event receives the start event; a hint of GreetEvent receives only GreetEvent instances.
  • Context is synchronous — ctx.set("key", value) and ctx.get("key") do not require await.
  • Alternative syntax — Event("GreetEvent", name=value) still works for inline, one-off events when you don’t need a dedicated class.