Skip to main content

@workflow Decorator

The @workflow decorator defines a multi-step execution flow. A workflow orchestrates tools, domain calls, and other workflows — and its tool calls are the points at which the run is checkpointed.

Basic usage​

from waxell_runtime import agent, workflow, tool


@agent(name="data_processor", description="Validates, transforms, and stores records")
class DataProcessor:

    @tool
    async def validate(self, ctx, record: dict) -> dict:
        ...

    @tool
    async def transform(self, ctx, data: dict) -> dict:
        ...

    @tool
    async def store(self, ctx, data: dict) -> dict:
        return await ctx.domain("warehouse", "upsert", row=data)

    @workflow("process_data")
    async def process_data(self, ctx, record: dict):
        validated = await ctx.tool("validate", record=record)
        transformed = await ctx.tool("transform", data=validated)
        return await ctx.tool("store", data=transformed)

Three rules that account for most first-time errors:

  • Everything is async. ctx.tool, ctx.domain, and ctx.llm.generate are all awaited.
  • Steps are called by name. await ctx.tool("validate", ...), not self.validate(...) — calling the method directly bypasses the runtime, so the step isn't checkpointed, traced, or governed.
  • The name is optional. @workflow, @workflow("custom_name"), and @workflow(description=...) are all valid; a bare @workflow uses the function name.

What ctx gives you​

AttributeWhat it is
ctx.inputsThe workflow's inputs, as a dict
ctx.tool(name, **kwargs)Call one of the agent's tools — a checkpoint
ctx.domain(domain, action, **kwargs)Call back into your application
ctx.llm.generate(prompt=..., output_format=...)Call a model
ctx.memory / ctx.scratchpadWorking memory
ctx.metadataRun metadata
ctx.workflow_id / ctx.parent_workflow_idIdentity and lineage

Note that ctx.inputs is plural. The full surface is documented in Execution Context.

Durability​

Checkpointing is automatic. There's no flag to set and no separate durable-workflow type — a workflow's tool calls are its checkpoints.

@workflow("long_running")
async def long_running(self, ctx, job_id: str):
    # Each ctx.tool(...) is a checkpoint boundary.
    step1 = await ctx.tool("expensive_operation", job_id=job_id)

    # If the process dies here, resuming replays the line above from its
    # saved result rather than re-running it.
    return await ctx.tool("another_operation", data=step1)

Work done between tool calls is not checkpointed. If a computation is expensive enough that you'd hate to repeat it, put it in a @tool.

For run states, resume semantics, and where checkpoints are stored, see Durable Execution.

Next steps​