Coming from Pydantic AI

PyrulaDurability plugs into Pydantic AI’s durability builder alongside the Temporal, DBOS and Prefect backends (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.

Scaffold

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).

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, 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 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)	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:

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())