Human-in-the-Loop

Build workflows that pause for human input in Python

Side-Effect Steps

A step can return None and use ctx.send_event() to manually route events:

class ReviewComplete(Event):
    pass

@step
async def review(ctx: Context, ev: Event):
    ctx.set("needs_approval", True)
    approved = simulate_human_review(ev)
    ctx.set("approved", approved)
    ctx.send_event(ReviewComplete())
    return None

Both ctx.send_event() and ctx.set() are synchronous.

Pause and Resume

handler = await wf.run(data="some input")
snapshot = handler.pause()

# Later...
handler = await Workflow.resume(snapshot)
result = await handler.result()