Skip to main content

Multi-Step Workflows

This tutorial covers branching, composition, and error handling in a workflow. It assumes you've read @workflow.

Two things to keep in front of you throughout:

  • Everything is async — ctx.tool, ctx.domain, ctx.llm.generate.
  • Steps are invoked by name — await ctx.tool("validate", ...). Calling self.validate(...) bypasses the runtime, so the step isn't checkpointed, traced, or governed.

Branching​

Branch on the result of a step like ordinary Python — the control flow is yours:

from waxell_runtime import agent, workflow, tool


@agent(name="support_router", description="Routes support tickets")
class SupportRouter:

    @tool
    async def classify(self, ctx, message: str) -> dict:
        """Classify a ticket into billing, technical, or general."""
        return await ctx.llm.generate(
            prompt=(
                "Classify this ticket as billing, technical, or general. "
                f'Reply with JSON {{"category": ...}}.\n\n{message}'
            ),
            output_format="json",
            task="classification",
        )

    @tool
    async def handle_billing(self, ctx, message: str) -> dict:
        return await ctx.domain("billing", "open_case", summary=message)

    @tool
    async def handle_technical(self, ctx, message: str) -> dict:
        return await ctx.domain("support", "open_ticket", summary=message)

    @tool
    async def handle_general(self, ctx, message: str) -> dict:
        return await ctx.domain("support", "acknowledge", summary=message)

    @workflow("handle_ticket")
    async def handle_ticket(self, ctx, message: str):
        classified = await ctx.tool("classify", message=message)
        category = classified.get("category")

        if category == "billing":
            return await ctx.tool("handle_billing", message=message)
        if category == "technical":
            return await ctx.tool("handle_technical", message=message)
        return await ctx.tool("handle_general", message=message)

Note the shape: the model decides the category, and ordinary code decides what to do about it. That split is deliberate — see where to put logic.

Composition​

Sequence steps by awaiting them in order. Each is a checkpoint, so a crash halfway through resumes rather than restarts:

@agent(name="order_processor", description="Validates, charges, and fulfils orders")
class OrderProcessor:

    @tool
    async def validate_order(self, ctx, order: dict) -> dict:
        return await ctx.domain("orders", "validate", order=order)

    @tool
    async def process_payment(self, ctx, order: dict) -> dict:
        return await ctx.domain("payments", "charge", order_id=order["id"])

    @tool
    async def fulfill_order(self, ctx, order: dict, payment: dict) -> dict:
        return await ctx.domain(
            "fulfillment", "ship", order_id=order["id"], payment_id=payment["id"]
        )

    @workflow("process_order")
    async def process_order(self, ctx, order: dict):
        validated = await ctx.tool("validate_order", order=order)
        payment = await ctx.tool("process_payment", order=validated)
        return await ctx.tool("fulfill_order", order=validated, payment=payment)

Passing large results between steps​

If a step returns something big, pass a scratchpad handle instead of the payload so the bytes never enter the model's conversation:

await ctx.domain("market_data", "fetch_prices", tickers=["AAPL"])
result = await ctx.domain(
    "analytics", "calculate_performance",
    prices="$ref:market_data_fetch_prices.1",
)

See Working Memory for how handles are named.

Delegating to another agent​

For work that belongs to a different agent, spawn a child run rather than inlining it. The child gets its own trace, its own budget, and its result comes back to the parent:

@workflow("research_and_write")
async def research_and_write(self, ctx, topic: str):
    findings = await ctx.spawn(
        "research_agent",
        inputs={"topic": topic},
        budget_usd=2.00,
        timeout_seconds=300,
    )
    return await ctx.tool("draft", findings=findings)

ctx.spawn_many(...) runs several children concurrently. Budgets debit from the parent's ledger, so a runaway child can't outspend the parent's cap.

Error handling​

A tool that raises marks its span failed and records an ExecutionIncident. You can catch it and route around the failure:

@workflow("resilient")
async def resilient(self, ctx, payload: dict):
    try:
        return await ctx.tool("risky_operation", payload=payload)
    except Exception as exc:
        return await ctx.tool("handle_error", error=str(exc))

Two things happen whether or not you catch:

  • The failure is still recorded on the span tree and as an incident. Catching changes your control flow, not the audit trail.
  • A policy block is not an exception. A run stopped by governance finishes in the BLOCKED state with a recorded decision — there's nothing to except.

If the process itself dies (OOM, SIGTERM, host loss), a reconciler finalises the run as INTERRUPTED rather than leaving it hanging. Both BLOCKED and INTERRUPTED are terminal — code polling only for COMPLETED/FAILED waits forever. See run states.

Next steps​