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.

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.

Gates for the dangerous tools

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

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

Recording usage

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

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