Quickstart

pip install "pyrula-workflows[ui]", then scaffold a runnable workflow:

pyrula init
pyrula dev --open

What it wrote

pyrula init writes app.py: a checkout workflow that reserves inventory, charges a card, suspends for two seconds, and sends a receipt. Every side effect goes through ctx.step, so a resumed run reads back what already happened instead of repeating it. The Quickstart walks through the console output line by line, including proof that the card is charged once, not twice.

Write it yourself

Prefer to see the pieces without the scaffold? Here’s a minimal workflow by hand. The body takes a context (ctx) and uses ctx for anything with a side effect, so it replays cleanly.

import asyncio
from pyrula.workflows import workflow, InlineWorkflowRunner, MemoryStore


@workflow(name="greet")
async def greet(ctx, name: str) -> str:
    # ctx.step runs the work once and replays the recorded result on recovery
    return await ctx.step("build", lambda: f"hello, {name}")


async def main() -> None:
    runner = InlineWorkflowRunner(store=MemoryStore())
    run_id = runner.submit("greet", params={"name": "ada"})
    for event in await runner.run("greet", run_id) or []:
        print(event.kind, event.payload)


asyncio.run(main())

submit enqueues a run and hands back its id. run executes it and appends lifecycle and step events to the run’s stream. MemoryStore keeps that stream in process, which is what you want for tests. Swap in a Valkey or Redis store to survive restarts.

The thing to internalize: side effects go through ctx.step. On a replay the step is skipped and its recorded result comes back, so the work doesn’t run twice. See Determinism for why that matters.

Options

@workflow takes a few caps. timeout bounds one execution attempt; the rest bound the run’s persisted event stream so a runaway workflow can’t grow it without limit.

@workflow(
    name="report",
    timeout=600.0,                  # wall-clock cap per attempt
    max_event_count_per_run=10_000,
    max_event_payload_bytes=1_000_000,
    max_attempt_wall_seconds=3600.0,    # also per attempt, NOT a whole-run budget
)
async def report(ctx):
    ...

Next: Runs as Event Streams.