Steps

@step turns a function into an atomic workflow operation. Pyrula records its result and returns that result 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)

A decorated sync function still returns an awaitable. Async step functions use the same call form.

Step context

Declare ctx: StepContext as the first parameter when the step needs run data. Pyrula injects it and removes it from the caller-facing signature.

• run_id and run_name identify the current run.
• step_id and call_index identify this call in the run.
• attempt starts at one and increases for each retry.
• metadata, deps, and principal come from the run context.
• idempotency_key stays stable across retries and recovery.
• cancelled reports the current cancellation state.

Pass ctx.idempotency_key to external systems that support idempotent writes. Step execution is at least once: a process can fail after the external write but before Pyrula records step:done. Use the workflow context’s ctx.once when the deduplication key must also apply across separate runs.

Results

Pyrula stores step results as JSON. A return annotation restores the declared type when the result is replayed. Err and Failure preserve the message and restore the error as RuntimeError.

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

Changing the annotation can make an existing recorded result invalid. Deploy a new step name or version the workflow when runs using the old result remain active.

Options

retry= sets the retry policy, timeout= limits each attempt, and effect= records whether the step is "read", "write", or "pure". The default step name is the function name. Set name= before renaming a function used by long-lived workflows.

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 cannot start another durable step, emit a durable event, or invoke another journaled operation. Pyrula raises NestedDurableOperationError before writing the inner operation. OpenAI, Anthropic, and LangChain drop-in clients called inside a step run as ordinary provider calls; the outer step owns the recorded result.

Use inspect.unwrap to test the function body without a run:

from inspect import unwrap

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

Use ctx.step when the step ID is computed at runtime, such as one ID per input item. Use @tool for functions exposed to an agent’s model. A tool has model-facing schema, tool-call events, and agent limits; a step is application code called directly by a workflow or agent.