# Coming from Pydantic AI

> Keep your Pydantic AI agent and make it durable on Pyrula with PyrulaDurability: a concept map, the scaffold and an executed example.

Source: https://docs.pyrula.io/getting-started/coming-from/pydantic-ai/

`PyrulaDurability` plugs into Pydantic AI's durability builder alongside the Temporal, DBOS and
Prefect backends ([durable execution backends](https://pydantic.dev/docs/ai/capabilities/durable_execution/backends/)).
Model and tool calls are journaled and replay after a crash. Your `Agent`, tools, output type and
tests stay as they are; the agent's `run()` moves into a Pyrula `@workflow` (or `@agent`) body that
a worker executes. Detail is in [Pydantic AI](https://docs.pyrula.io/agents/integrations/pydantic-ai.md).

## Scaffold

```bash
pip install "pyrula[agents,pydantic-ai]"
pyrula init --pydantic-ai
python app.py
```

It runs offline; `pyrula dev app.py --open` shows each call
([Start from a scaffold](https://docs.pyrula.io/getting-started/existing-app.md#start-from-a-scaffold)).

## Concept map

| Pydantic AI | Pyrula | Change |
|---|---|---|
| `Agent(model, ...)` | The same `Agent`, plus `capabilities=[PyrulaDurability(name=...)]` | Direct swap |
| `DBOSDurability()` | `PyrulaDurability(name=...)` | Direct swap |
| `@DBOS.workflow` calling `agent.run()` | `@workflow` with an `async def` body | Direct swap |
| `agent.run_sync()` | `await agent.run()` | Small rewrite |
| Calling the agent from Django or a task | Submit with the blocking [`WorkflowClient`](https://docs.pyrula.io/reference/workflows-sync.md#workflowclient), read `.result()` | Small rewrite |
| Agent `name=` | `PyrulaDurability(name=...)` keys the journal; renaming re-runs calls live | Operational change |
| `@agent.tool`, `@agent.tool_plain` | Unchanged | Direct swap |
| `FunctionToolset(id=...)` | Unchanged; keep the `id` fixed | Direct swap |
| MCP toolsets | Unchanged | Direct swap |
| `agent.run(toolsets=[...])` | Refused in a run; give toolsets to the `Agent` | Small rewrite |
| Parallel tool calls | Run one at a time | Operational change |
| `ctx.tool_call_id` | Stable across a resume; use it as an idempotency key | Direct swap |
| A tool's side effect | `Tool(fn, metadata={"pyrula": {"effect": "read"}})`; undeclared counts as `write` | Small rewrite |
| `requires_approval`, `DeferredToolRequests` | Work, but don't pause durably; use `ctx.interrupt(...)` or a native `@tool(requires_approval=True)` | Conceptual redesign |
| `UsageLimits` | Applies; [`AgentLimits`](https://docs.pyrula.io/reference/agents.md#agentlimits) adds to it in an `@agent` | Direct swap |
| Token and cost accounting | Metered in an `@agent`; journaled only in a `@workflow` | Operational change |
| `run_stream()`, `event_stream_handler` | One journaled call; replay returns the whole response | Operational change |
| Instructions from a clock | A resume raises `ReplayInputChanged`; pass the value in the prompt | Small rewrite |
| Retry loop around `agent.run()` | A caught failure replays as the same exception | Direct swap |
| `TestModel`, `FunctionModel` | Unchanged; `WorkflowTestEnv` or `AgentRunner` | Direct swap |
| `output_type` | Unchanged | Direct swap |
| Logfire, OpenTelemetry | `step:pending` and `step:done` in `pyrula dev` and Pyrula Cloud | Operational change |
| The process calling `agent.run()` | A worker (`pyrula worker`, `WorkflowWorker`) on a store | Operational change |
| Rewriting natively | `@agent` with `@tool`: approval deadlines, `@tool(writes=...)`, per-tool limits ([Agents](https://docs.pyrula.io/agents.md)) | Conceptual redesign |

## Example

The worker dies as the second model request starts. On resume the first request and the refund
come back from the journal, and only the second request is made:

```python
import asyncio

from pydantic_ai import Agent, RunContext
from pydantic_ai.messages import ModelMessage, ModelResponse, TextPart, ToolCallPart
from pydantic_ai.models.function import AgentInfo, FunctionModel
from pyrula.agents import AgentContext, agent
from pyrula.agents.integrations import PyrulaDurability
from pyrula.agents.testing import AgentRunner

model_calls: list[int] = []
refunds: list[str] = []


def scripted(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
    """Stands in for a provider: asks for a refund, then answers."""
    model_calls.append(len(messages))
    if len(messages) == 1:
        return ModelResponse(parts=[ToolCallPart("refund", {"order": "A-1"})])
    return ModelResponse(parts=[TextPart("refunded A-1")])


support = Agent(
    FunctionModel(scripted),  # your 'openai:...' or 'anthropic:...' model here
    name="support",
    capabilities=[PyrulaDurability(name="support")],
)


@support.tool
def refund(ctx: RunContext[None], order: str) -> str:
    """Refund an order."""
    # tool_call_id is the same after a resume: give it to the provider as an idempotency key.
    refunds.append(f"{order}:{ctx.tool_call_id}")
    return f"refunded {order}"


@agent(name="docs_from_pydantic_ai_support", timeout=60)
async def answer(ctx: AgentContext, message: str) -> str:
    return (await support.run(message)).output


async def main() -> None:
    runner = AgentRunner(answer)
    # The worker dies as the second model request starts.
    runner.inject_crash_at("llm_call", nth=2, when="pending")
    await runner.run(message="refund order A-1")
    events = await runner.run(message="refund order A-1")

    assert events[-1].payload["result"] == "refunded A-1"
    assert model_calls == [1, 3]  # the first request was not made again
    assert len(refunds) == 1  # nor was the refund


asyncio.run(main())
```