Steps

@step makes a function an atomic workflow operation. The result is recorded and returned on replay instead of calling the function again.

from pyrula.workflows import StepContext, step, workflow


@step(timeout=10.0)
def reserve(ctx: StepContext, order_id: str) -> dict[str, str]:
    return inventory.reserve(order_id, idempotency_key=ctx.idempotency_key)


@workflow(name="checkout")
async def checkout(ctx, order_id: str) -> dict[str, str]:
    return await reserve(order_id)

The decorated function is awaited, sync or async. @step on a method raises TypeError.

Step context

A first parameter named ctx: StepContext receives run data and is stripped from the caller’s signature.

• run_id, run_name, step_id, call_index
• attempt, starting at one and increasing per retry
• metadata, deps, principal from the run context
• idempotency_key, stable across retries and recovery
• cancelled, the current cancellation state

Steps run at least once: a crash after the external write but before step:done is recorded reruns the step, so pass ctx.idempotency_key to systems that take one. Use the workflow’s ctx.once when the key must hold across runs.

Results

Results are stored as JSON. Core types like Ok, Err, Some use their plain JSON form, and an unannotated step returns that JSON. A return annotation restores the declared type on first run and on replay (checked at decoration). An Err or Failure keeps its exception class when it’s safe to rebuild, else reads back as a RecordedError. A result that can’t be restored fails the step.

@step
def load_order(order_id: str) -> Order:
    return Order.model_validate(api.get(order_id))

Changing an annotation can invalidate recorded results; use a new step name or version the workflow while old runs are in flight.

Options

retry= is the retry policy (Errors and retries), timeout= bounds each attempt (positive seconds), effect= records "read", "write" (default) or "pure". The name defaults to the function name; pin name= before renaming a function that long-lived workflows use. Names can’t contain ::.

A timeout= or a worker’s shutdown abandons a sync step but cannot stop its thread, so the thread may still be running while another worker runs the step again (The window where a step can run twice).

from pyrula.workflows import RetryPolicy


@step(name="charge-card-v1", retry=RetryPolicy(max_attempts=3), timeout=15.0)
async def charge(order: Order) -> Charge:
    return await payments.charge(order)

Atomic boundaries

A step can’t start another durable step, emit a durable event or call any journaled operation: NestedDurableOperationError is raised before the inner write. OpenAI, Anthropic and LangChain drop-in clients inside a step run as plain provider calls.

inspect.unwrap returns the body, for a test without a run:

from inspect import unwrap

result = unwrap(load_order)("order-123")

ctx.step is for ids computed at runtime (one per input item). @tool exposes a function to an agent’s model; a step is application code called directly.