Pydantic AI

PyrulaDurability is a Pydantic AI capability. Attach it to an agent and run the agent inside a Pyrula @workflow, and every model request and tool call is recorded in the run’s journal. After a crash the run resumes from the top, and each call that had finished is answered from the journal.

pip install "pyrula[agents,pydantic-ai]"

pyrula init --pydantic-ai writes a runnable app in this shape; see Start from a scaffold.

Use it

Give the agent a name, give every toolset an id, and run the agent in a workflow body.

import asyncio

from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel
from pydantic_ai.toolsets import FunctionToolset
from pyrula.contracts.events import EventKind
from pyrula.workflows import workflow
from pyrula.workflows.testing import WorkflowTestEnv

from pyrula.agents.integrations import PyrulaDurability

charged: list[str] = []

tools = FunctionToolset(id="support_tools")


@tools.tool_plain
def charge(order: str) -> str:
    """Charge an order."""
    charged.append(order)
    return f"charged {order}"


support = Agent(
    TestModel(call_tools=["charge"]),  # swap for 'openai:...' or 'anthropic:...' and set its key
    name="support",
    toolsets=[tools],
    capabilities=[PyrulaDurability(name="support")],
)


@workflow(name="docs_pai_support")
async def answer(ctx, question: str) -> str:
    return (await support.run(question)).output


async def main() -> None:
    env = WorkflowTestEnv()
    run = await env.client.submit("docs_pai_support", {"question": "refund order A-1"})
    await run.result()
    assert len(charged) == 1
    names = [
        e.payload["step_id"]
        for e in env.store.load_run("docs_pai_support", run.run_id)
        if e.kind == EventKind.STEP_DONE
    ]
    assert names == [
        "call:pydantic_ai.support__model.request:0",
        "call:pydantic_ai.support__function_toolset__support_tools.call_tool:charge:0",
        "call:pydantic_ai.support__model.request:1",
    ]


asyncio.run(main())

Outside a Pyrula run the capability does nothing: the agent runs as plain Pydantic AI, with a one-time warning. Model calls are written as llm:done events in a @workflow and an @agent alike; in an @agent they are also metered and subject to AgentLimits.

Caution

Spend caps need an @agent. In a @workflow the calls are journaled and each writes an llm:done with its usage, but nothing holds AgentLimits, as with the drop-in clients: only a case run’s cost cap acts on that spend.

Pin the names

Each journaled call is keyed by the agent’s name, the kind of call, the toolset id and the tool name, for example support__function_toolset__support_tools.call_tool:charge. The prompt is not in the key. Pass name= to PyrulaDurability and an explicit id= to every FunctionToolset.

• Renaming the agent or a toolset renames every call, and the whole agent run executes live again on the next resume.
• Renaming one tool gives it a new key, so on resume its call runs once more.
• A # or @ in a tool name raises ValueError.
• Instructions and system prompts count as the call’s input. A prompt built from the date or any other value that changes between the attempt and the resume raises ReplayInputChanged. Pass such values in the user prompt, or journal them with ctx.step first.
• Adding or removing an event_stream_handler, or switching between run and run_stream, renames the model calls, so a resume runs them live again.

What replays

Call	Journaled as	Declared effect
Model request, streamed request, compaction	llm_call	read
Tool call	tool_call	write, or what the tool declares
Tool argument validation, MCP tool and instruction discovery	tool_call	read
Capability operations	tool_call	write
The run’s start time, for timeout=	direct_call	read

A call that raised is journaled too. A body that catches the failure of agent.run() and retries meets the same exception on replay, and the call is not made again. A failure that ended the run is made again when the run is retried.

A resume replays a call only if what it asked is the same. The fingerprint covers the messages, the tool names and schemas, the output mode, and a tool call’s name and arguments. It leaves out timestamps, ids, usage, provider details and model settings. A changed prompt, history, tool list or tool argument raises ReplayInputChanged out of agent.run(); a changed temperature does not.

Declare a tool’s side effect in its metadata so what-if runs know which tools may run live:

from pydantic_ai.tools import Tool

lookup = Tool(find_order, metadata={"pyrula": {"effect": "read"}})

An undeclared tool is write: a what-if run serves its recorded result and refuses a live call.

Under a bare asyncio.gather, two agent runs can swap Pydantic AI run ids on replay. Fan out with ctx.gather instead.

What does not apply

• Approval and per-tool limits. requires_approval, @tool(writes=...), human-in-the-loop and per-tool AgentLimits belong to a Pyrula @tool. Pydantic AI’s own UsageLimits and deferred-tool flow still work, but a deferred tool ends the Pydantic AI run instead of pausing durably.
• Toolsets passed at run time. agent.run(toolsets=[...]) is refused inside a run. Give the toolsets to the agent.
• Parallel tool calls. Inside a run the model’s tool calls run one at a time.
• Nesting. A call inside a Pydantic AI tool (ctx.step, a drop-in client, another agent) is not journaled separately. Put side effects in the tool, or in a @step it calls.

Streaming and size

A streamed request is one journaled call. Inside a run nothing streams live: the model call finishes first, then event_stream_handler gets the events (without text or tool-call deltas) and run_stream hands over the whole response. On replay the handler runs again, so a side effect in the handler repeats.

A model response must fit the run’s event limit, 1 MB by default (@workflow(max_event_payload_bytes=...) raises it). A larger one fails the run after the call was made.

What Cloud sees

Each call is a step:pending and step:done pair (step:error when it raised) with the step id call:pydantic_ai.<name>:<index>. Model and tool results leave the process only at the io and full capture levels, and Cloud cuts a payload at 64 KiB. Token counts and cost appear for runs inside an @agent.

At-least-once

A call that is cut off after its effect and before its record is written runs again on resume. A tool with an external effect needs an idempotency key of its own; Pydantic AI gives it ctx.tool_call_id, which is stable across a resume.

Versions

The extra installs pydantic-ai-slim>=2.62,<3. 2.62 is the first release that exports the durable-execution names the adapter imports.