Skip to main content

Execution Context

Every decorated method receives ctx as its second argument, after self. It's how your agent reaches everything outside itself: your application, the LLM, tools, memory, secrets — and how it pauses for a human.

from waxell_runtime import agent, workflow


@agent(name="report_builder", description="Builds a report")
class ReportBuilder:

    @workflow("build")
    async def build(self, ctx, quarter: str):
        data = await ctx.domain("finance", "quarterly", quarter=quarter)
        return await ctx.llm.generate(prompt=f"Summarize: {data}")
Everything that reaches outside the agent is async

ctx.domain, ctx.tool, ctx.llm.generate, ctx.spawn, ctx.ask_user and friends are coroutines. Forget the await and you get a coroutine object instead of your result — Python only warns, so it fails quietly.

For editor autocomplete, type-hint it:

from waxell_runtime import WorkflowContext, ToolContext

@workflow("build")
async def build(self, ctx: WorkflowContext, quarter: str): ...

@tool
async def fetch(self, ctx: ToolContext, id: str) -> dict: ...

Reaching your application​

ctx.domain() is the main one. It calls a named action on a domain you've registered; your service answers over a shared-secret callback — so the agent never holds your credentials.

company = await ctx.domain("company", "get", company_id="acme")

See Domains.

Calling a tool​

ctx.tool() invokes another @tool on this agent. Each call is a durable checkpoint — see Durable Execution.

data = await ctx.tool("fetch_company", domain="acme.com")

The LLM​

Routed, cost-attributed and policy-checked. You never hold a provider key.

summary = await ctx.llm.generate(
    prompt="Summarize this contract",
    task="contract_summary",     # names the task for routing + cost analytics
    max_tokens=2000,
)

Secrets​

key = ctx.secrets.get("STRIPE_API_KEY")

Resolved from the tenant's configured secret provider. Don't read os.environ — it works locally and fails in deployment.

Memory​

await ctx.memory.set("last_seen", value)
value = await ctx.memory.get("last_seen")

ctx.scratchpad is the short-lived working tier for a single run. See Working Memory.

Pausing for something external​

These suspend the run without holding compute. It resumes when the thing arrives.

answer = await ctx.ask_user("Approve this $40k discount?")
event  = await ctx.wait_for_signal("po_received", timeout_seconds=86400)
await ctx.sleep(3600)

A paused run shows as PAUSED with a reason — see Durable Execution.

Fanning out​

child   = await ctx.spawn("enrichment_agent", inputs={"lead_id": lead_id})
results = await ctx.spawn_many("scorer", [{"id": i} for i in ids])

Children are governed and traced like any other run, and appear under the parent in the lineage graph.

Identity and provenance​

AttributeWhat it is
ctx.inputsThe inputs this run was started with
ctx.workflow_idThis run's id
ctx.parent_workflow_idThe run that spawned this one, if any
ctx.root_workflow_idThe top of the spawn tree
ctx.metadataFree-form metadata carried with the run
ctx.sub_user_id / ctx.sub_user_emailThe end user this run is acting as, when it carries a signed sub-user identity

sub_user_* is what makes per-end-user auth work: the agent acts as a specific person, and your domain endpoint can verify that cryptographically.

Observability​

ctx.log_step("validated_input", {"rows": len(rows)})   # note: not async
await ctx.checkpoint()                                 # force a save point

ctx.budget exposes the run's spend against its cap.