Tools

A tool is an async function the model can call. The first parameter is a ToolContext; the rest are the arguments the model fills in. The signature and docstring become the schema.

from pyrula.agents.decorators.tool import tool
from pyrula.agents.context.tool_context import ToolContext

@tool(retry=0, timeout=10.0)
async def search(ctx: ToolContext, query: str) -> str:
    return await backend.search(query)

A string result goes to the model as is. Any other JSON-serializable result is sent as compact JSON with sorted keys, so replays and forks build the same model input. A result that isn’t serializable raises an error naming the tool.

Unknown tool names

If the model calls a tool the agent didn’t offer, nothing runs and the turn goes on. The model gets a recoverable error, unknown tool 'x'; available: a, b (name cut to 64 characters, at most 20 names listed). The call is recorded as a tool:result with status: "error" and no tool:start, and it counts toward max_tool_calls.

Replaying a recorded call to a tool your code no longer offers, or offers with another schema or description, fails with replay_input_changed. Pydantic AI and LangGraph ToolNode handle unknown names in their own loop.

Options

Option	What it does
retry	Default 0. An int (retries, no wait) or a RetryPolicy from pyrula.workflows. Fatal and Poison are never retried. A sync tool that times out is not retried, because its thread cannot be stopped; use async def to retry on timeout.
timeout	Default 30.0. Per-call wall-clock cap. A sync tool’s thread keeps running past it; the late result is discarded.
cache_ttl	Cache the result in the store, keyed by tool name and arguments, for this many seconds.
writes	Marks side effects. Refused in every branch and case run unless allowed.
read_only	Free of side effects, so a branch may run it live. Cannot combine with writes.
requires_approval	True, or a function of the call’s arguments returning whether to ask. See Approval.
approval_timeout	Seconds or a timedelta before an unanswered call is denied. Default: wait indefinitely. Embedded mode refuses it (What embedded mode cannot run).
name / description / schema	Override what the model sees. name is [a-zA-Z0-9_-]+; schema replaces the JSON Schema derived from the signature.

Approval

The run parks before the call starts and resumes when someone answers, from the dashboard or the resume endpoint.

@tool(requires_approval=lambda args: args["cents"] > 10_000, approval_timeout=3600)
async def refund(ctx: ToolContext, cents: int) -> str:
    return await payments.refund(cents, key=ctx.idempotency_key)

The predicate must be pure, because a resumed run evaluates it again; if it raises, the call is asked about. For MCP servers, use MCPServer(requires_approval=("delete_issue",)) (MCP).

The question is asked before tool:start is written, one gated call at a time. It is an ordinary interrupt: run:interrupted carries reason: "tool_approval", tool_name, tool_use_id and tool_input. A reviewer answers approve or reject (with an optional reason).

Only approve runs the tool. A rejection, an expired approval_timeout or any other answer denies it: no tool:start, a tool:result with status: "error" and approval: "rejected" or "expired", and the model is told the call wasn’t approved.

On replay the answer is recorded against the call’s position and arguments; if the arguments at that position changed, the resume fails with replay_input_changed. The question is keyed by the tool and the call’s position (tool_approval:refund#0, or tool_approval:refund#id=order-9 for a call made with id=), and a call from agent code keeps its position after a denial, so a call that starts in another order on replay never takes it. A crash between call start and result still reruns the tool, so keep it idempotent.

Calling a gated tool from agent code (await refund(ctx, cents=25_000)) asks the same question. A denial raises ToolCallRejected (tool_name, reason, expired). Inside another tool, a workflow body or a ctx.step, it raises NestedDurableOperationError whenever its condition holds. With no run around it (a hand-built ToolContext in a unit test) it just runs.

Arguments ship at the io capture level and above (What leaves your runtime).

Limits:

• A Kafka agent can’t wait for a person. @kafka_agent refuses to bind when an MCP server sets requires_approval, and a turn offering a tool declared requires_approval=True fails with ToolRegistrationError. build_mcp_server refuses such a tool the same way.
• Tools that bypass Pyrula’s tool loop (LangGraph ToolNode, compat-wrapped clients) and sub-agents used as tools are not gated; use ctx.interrupt, or gate the tools inside the sub-agent.
• An approval can’t change arguments.

Branches

A branch re-runs part of a recorded run with something changed (model, prompt, tool result). A tool runs live in a branch only if declared read_only=True.

@tool(read_only=True)
async def lookup_order(ctx: ToolContext, order_id: str) -> dict:
    return await orders.get(order_id)

Any other tool replays its recorded result if the original run made that call. If the branch needs a call the original never made, it stops with write_tool_blocked naming the tool; Pyrula never invents a result. The error detail carries a reason (write, unmarked, step or effect), withheld at capture=none. Declare read_only=True or fork after the recorded result. writes=True tools are refused in every branch. MCP tools take read_only from the server’s readOnlyHint (MCP).

The guard covers the model’s tool loop (sub-agents included), MCP tools, direct calls, and calls from inside another tool, each checked against its own declaration. The refusal ends the branch even if your code catches the exception.

Also refused with write_tool_blocked: ctx.continue_as_new, ctx.spawn_workflow, ctx.run_child, ctx.send_signal, ctx.memory.put and ctx.memory.delete. A run_child or send_signal result the original already recorded is replayed. ctx.memory.get, list and search are allowed.

ctx.step and ctx.once follow the same rule. An unrecorded step runs only with effect="read" or effect="pure"; the default effect="write" and every ctx.once are refused as ctx.step:<step id>. Workflow (non-agent) branches don’t apply this rule.

If the original run is deleted before a sleeping or paused branch resumes, the branch ends with branch_unresumable (reason parent_missing).

Not guarded: ctx.stream / ctx.commit, and effects your own code performs outside a Pyrula tool or step (including tool calls that compat or LangGraph code runs itself). Put those in a @tool(writes=True) or a default ctx.step.

Case runs use the same rule: only read_only=True tools (and sub-agents) count as reads. A model-called tool needs the allow token tool:<name>, a directly called one step:<name>. See Live calls and effects.

Idempotency

A tool can run again on recovery if it was interrupted before its result was recorded. ctx.idempotency_key is stable across retries and reruns of the same call; use it as a dedup key.

@tool
async def charge(ctx: ToolContext, amount_cents: int) -> str:
    return await payments.charge(amount_cents, key=ctx.idempotency_key)

The key is built from the turn and the call’s position, not the provider’s tool call id. It is stable only while step_boundary_durability is on (the default); with it off, a regenerated batch may order calls differently, so dedupe on a business key (the order id). A direct call keys on a per-name counter or its id=; pass a distinct id= when parallel sub-agents make the same direct call. Two calls of one name that are running together under one id= raise ConcurrentCallIdError (pyrula.contracts.errors).

ToolContext

ctx.http (an httpx.AsyncClient), ctx.deps (the app’s deps), ctx.turn_id, ctx.cancelled (poll in long loops), await ctx.emit_status("...") (ephemeral status event), and ctx.cache:

hit = await ctx.cache.get("rates")             # store-backed, namespaced per tool
await ctx.cache.set("rates", rates, ttl=300)