Context

AgentContext extends the workflow context, so every durable primitive is available beside the agent operations. Extra on top: ctx.llm, ctx.call_agent, ctx.gather_subagents, ctx.mcp_tools, ctx.memory, ctx.thread, ctx.turn_id, ctx.cancelled.

Inherited primitives

Use these for anything non-deterministic or side-effecting: @step, ctx.step, ctx.now, ctx.uuid, ctx.random, ctx.emit, ctx.emit_status, ctx.sleep, ctx.sleep_until, ctx.wait_for_signal, ctx.interrupt, ctx.checkpoint. See Steps, Timers, Signals and Human-in-the-Loop.

continue_as_new

ctx.continue_as_new(params=...) ends the run and starts a successor in the same thread, carrying params, to bound history growth. Consumers follow the chain, the chain has no length limit, and the thread keeps its root turn id for status, cancel and signals.

ctx.continue_suggested turns True at 80% of AgentLimits.max_event_count_per_turn (50,000 by default; None sets none, though a store that trims a run’s stream still lowers it), so you can hand off at a tool-loop boundary:

if ctx.continue_suggested:
    await ctx.continue_as_new(params={"summary": running_summary})

Kafka-driven and branch/stream executions raise ContinueAsNewNotSupportedError.

Sub-agents

call_agent runs another agent and records the result, so the child isn’t re-run on replay.

summary = await ctx.call_agent(summarizer, text=document, id="summarize")

gather_subagents runs several concurrently as child runs that a worker claims, so it needs pyrula worker, not embedded mode (why). ctx.gather is for plain awaitables. AgentLimits.max_agent_depth and max_parallel_subagents cap them.

results = await ctx.gather_subagents(
    (classifier, {"text": a}),
    (classifier, {"text": b}),
    id="classify-batch",
)

Workflows

spawn_workflow submits and returns the run id; wait_for_workflow polls (timeout= optional). Also refused in embedded mode. To await a child durably, use ctx.run_child.

run_id = await ctx.spawn_workflow("nightly_report", params={"date": today})
result = await ctx.wait_for_workflow("nightly_report", run_id)

Memory and threads

ctx.memory is cross-turn state keyed by a namespace tuple. Values are dicts. It needs a store (UnboundContextError without one) and defaults to a key-value backend over the run store.

await ctx.memory.put(("user", user_id), "preferences", {"tone": "concise"})
prefs = await ctx.memory.get(("user", user_id), "preferences")   # {"tone": "concise"} or None
items = await ctx.memory.search(("user", user_id), "tone", limit=5)

Also delete and list(prefix, key_prefix=, limit=). list and search refuse an empty prefix. Writes are refused in a branch. put(..., source="user") marks who asserted the fact; default "agent".

ctx.thread is the conversation thread, or None for a turn submitted off-thread: await ctx.thread.last_result(), await ctx.thread.history(limit=50). ctx.turn_id identifies the current turn.