# Pydantic AI

> Make a Pydantic AI agent durable with PyrulaDurability: its model and tool calls are journaled in a Pyrula run and replayed after a restart.

Source: https://docs.pyrula.io/agents/integrations/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.

```bash
pip install "pyrula[agents,pydantic-ai]"
```
`pyrula init --pydantic-ai` writes a runnable app in this shape; see [Start from a scaffold](https://docs.pyrula.io/getting-started/existing-app.md#start-from-a-scaffold).

## Use it

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

```python
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](https://docs.pyrula.io/agents/integrations/durable-vs-not.md): 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:

```python
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.