# How to checkpoint Pydantic AI agents

To make a Pydantic AI agent durable with Tidebase, wrap each `agent.run(...)` in a checkpointed step using the async SDK (`tidebase.aio`), and hand the Tidebase run context to tools through the agent's deps so external writes get their own checkpoints. Re-invoking with the same `run_id` after a crash replays finished steps from Postgres.

```python
from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
from tidebase.aio import AsyncTidebase, AsyncRunContext

tide = AsyncTidebase()

@dataclass
class Deps:
    tide_run: AsyncRunContext

agent = Agent("openai:gpt-4.1-mini", deps_type=Deps)

@agent.tool
async def create_ticket(ctx: RunContext[Deps], title: str, body: str) -> str:
    run = ctx.deps.tide_run
    return await run.step(
        f"create-ticket:{title}",
        lambda: ticket_api.create(title, body),
        input={"title": title, "body": body},
        side_effects=["ticketing-api"],
        idempotency_key=f"ticket-{run.run_id}-{title}",
    )

async def workflow(run, input):
    result = await run.step(
        "triage",
        lambda: agent.run(input["report"], deps=Deps(tide_run=run)),
        input={"report": input["report"]},
    )
    await run.state_set({"phase": "done"})
    return result.output

await tide.run("bug-triage", workflow, run_id=run_id, input={"report": report})
```

Tidebase is an open-source checkpoint layer for AI agents: wrap your steps, and failed runs resume from the last safe point — in your own Postgres, without moving execution into a new runtime.

One wrinkle to know up front: the outer `triage` step checkpoints the agent run's **result**. Pydantic AI's `AgentRunResult` carries a typed `output`; what lands in the checkpoint is its JSON serialization, so on replay you get the data back, not a live result object. If your output type is a Pydantic model, return `result.output.model_dump()` from the step (and re-validate on read) rather than checkpointing the result wholesale.

## Deps are the natural seam

Pydantic AI already threads a typed dependencies object into every tool via `RunContext.deps` — exactly where the Tidebase run context belongs. No globals, no monkey-patching: tools that touch the outside world call `run.step(...)` with content-keyed names, `side_effects`, and an `idempotency_key`. If the process dies mid-`agent.run`, the outer step has no checkpoint, so the agent re-runs on resume — and the inner tool checkpoints are what stop the re-run from double-firing a write that already happened. A side-effecting step that fails without an idempotency key parks as `manual_review` per [the replay contract](../replay-contract-is-it-safe-to-rerun.md).

## Gates for the dangerous tools

A tool that deploys, sends, or spends can park on a durable approval first:

```python
@agent.tool
async def deploy(ctx: RunContext[Deps], service: str) -> str:
    run = ctx.deps.tide_run
    decision = await run.gate(f"approve-deploy:{service}", f"Deploy {service}?")
    if not decision.approved:
        return "Deploy denied by operator."
    return await run.step(
        f"deploy:{service}",
        lambda: do_deploy(service),
        input={"service": service},
        side_effects=["deploy"],
        idempotency_key=f"deploy-{run.run_id}-{service}",
    )
```

The gate blocks until resolved — fine inside your own process, and the decision is exactly-once, durable, and recorded with the actor. See [human approval gates](../human-approval-gates-for-ai-agents.md).

## Recording usage

`agent.run` results expose token usage via `result.usage()`. Record it inside the step so replays don't double-count:

```python
usage = result.usage()
await run.usage_record(
    kind="llm", provider="openai", model="gpt-4.1-mini",
    input_tokens=usage.input_tokens, output_tokens=usage.output_tokens,
)
```

## The honest tradeoffs

- **Tidebase does not execute your code.** After a crash, a Tidebase queue worker, recovery webhook, or your own retry re-invokes the workflow; Tidebase guarantees the re-invocation is safe, not that it happens.
- **A replayed step returns the recorded output** — it does not re-run the model. That's the feature, but a replayed answer reflects the first run.
- **Alpha, opt-in auth.** Self-hosted alpha — set `TIDEBASE_API_KEY` before exposing the server beyond localhost.

Repo: <https://github.com/BlueprintLabIO/tidebase> · See also: [How to checkpoint OpenAI Agents SDK runs](openai-agents-sdk.md)