Going to Production

Pydantic AI, Pydantic Evals and Pydantic AI Harness are open-source, MIT-licensed Python packages that run in your own process, and durable execution runs on an engine you operate (or that engine’s own cloud): there is no Pydantic-hosted agent runtime. Two hosted Pydantic services plug in with a couple of lines when you want them: Pydantic Logfire, which shows every run, model request and tool call as a trace next to the rest of your application, with a free tier (any OpenTelemetry backend works instead); and the Pydantic AI Gateway, one key for models across providers with spending caps, where calling providers directly works instead.

The rest of the page answers the questions teams ask before shipping an agent, each with the short answer and a link to the page that covers it in full, after one runnable service that puts the answers together.

Before you ship, you want to know…	Short answer	Install
What happens when the process crashes mid-run?	Run the agent on a durable execution engine	pydantic-ai-slim with one engine’s extra, such as [dbos] or [temporal]
How does a human approve a risky action, and the run resume days later?	Deferred tools end the run; a later run resumes it	pydantic-ai
How do I keep conversation state?	Store the run’s Conversation in your database	pydantic-ai
How do I bound cost and runaway loops?	UsageLimits per run, SpendLimits per day, tenant or month	pydantic-ai, and pydantic-ai-harness for SpendLimits
What happens when a provider is down?	A fallback model, transport retries, or Gateway failover	pydantic-ai, or the hosted Gateway
Where do API keys go?	Environment variables, or one Gateway key for every provider	pydantic-ai, or the hosted Gateway
How do I see what happened?	OpenTelemetry traces of every run, in Logfire or any OTel backend	pydantic-ai (includes logfire)
How do I test and evaluate changes?	Unit tests with TestModel, then evals on datasets and live traffic	pydantic-ai (includes pydantic-evals)
How do I limit what the agent can touch?	Per-caller tools, approval, and a sandbox for model-written code	pydantic-ai, and pydantic-ai-harness for sandboxes
How do I serve it to a frontend?	AG-UI or Vercel AI streams from any ASGI app	pydantic-ai
How do I deploy it?	As your own Python process: a web app, a worker or a function	Your infrastructure

An end-to-end service

One file with the pieces most production agents need: an HTTP API, a tool that waits for a person to approve it, durable execution on DBOS, and OpenTelemetry traces. It needs a model provider’s API key and nothing else: DBOS keeps its state in a SQLite file here (use Postgres in production), and traces go to Logfire when a Logfire token is available, from LOGFIRE_TOKEN or the credentials logfire auth saved locally, or to any OTel collector named by OTEL_EXPORTER_OTLP_ENDPOINT. To stay short, it leaves out what you add before exposing it: authentication and checks that the caller owns the run, deduplication of repeated requests, and usage limits.

pip

Terminal

pip install "pydantic-ai[dbos]" fastapi uvicorn

uv

Terminal

uv add "pydantic-ai[dbos]" fastapi uvicorn

production_service.py

from typing import Any

import logfire
from dbos import DBOS
from fastapi import FastAPI

from pydantic_ai import Agent, DeferredToolRequests
from pydantic_ai.durable_exec.dbos import DBOSDurability

logfire.configure(send_to_logfire='if-token-present')
logfire.instrument_pydantic_ai()

app = FastAPI()
DBOS(fastapi=app, config={'name': 'support', 'system_database_url': 'sqlite:///support.sqlite'})

agent = Agent(
    'anthropic:claude-sonnet-5-5',
    name='support',
    instructions='Help customers with their orders.',
    output_type=[str, DeferredToolRequests],
    capabilities=[DBOSDurability()],
)


@agent.tool_plain(requires_approval=True)
@DBOS.step()  # DBOSDurability doesn't wrap function tools, so mark side effects as steps
def issue_refund(order_id: str, amount: float) -> str:
    return f'Refunded {amount} on order {order_id}'  # call your payments API here


@DBOS.workflow()
async def support_run(prompt: str) -> str:
    try:
        await DBOS.set_event_async('state', {'status': 'running'})
        result = await agent.run(prompt)
        while isinstance(output := result.output, DeferredToolRequests):
            pending = {call.tool_call_id: call.args_as_dict() for call in output.approvals}
            await DBOS.set_event_async('state', {'status': 'awaiting_approval', 'pending': pending})
            approvals: dict[str, bool] | None = await DBOS.recv_async('approvals', timeout_seconds=7 * 24 * 3600)
            if approvals is None:  # nobody answered within a week
                approvals = {call_id: False for call_id in pending}
            await DBOS.set_event_async('state', {'status': 'running'})
            result = await agent.run(
                conversation=result.conversation,
                deferred_tool_results=output.build_results(approvals=approvals),
            )
        await DBOS.set_event_async('state', {'status': 'done', 'output': output})
        return output
    except Exception as e:  # so GET /runs/{run_id} doesn't keep reporting 'running'
        await DBOS.set_event_async('state', {'status': 'failed', 'error': str(e)})
        raise


@app.post('/runs')
async def start_run(prompt: str) -> dict[str, str]:
    handle = await DBOS.start_workflow_async(support_run, prompt)
    return {'run_id': handle.get_workflow_id()}


@app.get('/runs/{run_id}')
async def get_run(run_id: str) -> dict[str, Any] | None:
    return await DBOS.get_event_async(run_id, 'state', timeout_seconds=0)  # None until the workflow starts


@app.post('/runs/{run_id}/approvals')
async def approve(run_id: str, approvals: dict[str, bool]) -> None:
    await DBOS.send_async(run_id, approvals, topic='approvals')

Run it with uvicorn production_service:app, start a run with POST /runs?prompt=..., read its status and any pending calls with GET /runs/{run_id}, and approve or deny the pending refund with POST /runs/{run_id}/approvals and a body like {"<tool_call_id>": true}. Kill the server at any point and start it again: DBOS recovers the workflow, replays the model requests and the issue_refund step that already finished from its database, and goes back to waiting for the approval if that is where it was. The HTTP requests only start, read and signal the workflow, so the run does not depend on any of them staying open. The sections below explain each piece and the alternatives.

What happens when the process crashes mid-run?

Without durable execution, a run in flight dies with the process: you start it again, or continue it from the last StepPersistence snapshot. With it, an engine journals each model request and each tool call it wraps or you mark, and a restarted run replays completed steps from the journal and retries the one that was in flight, without paying for completed model requests again. Which tool calls an engine wraps for you differs (DBOS leaves function tools to you, as in the service above). The engines that ship in Pydantic AI and the Harness attach to an ordinary agent as a capability and take effect when the run executes inside the engine’s workflow or handler; Restate wraps the agent in a class of its own instead, and Airflow in an operator.

A step that was in flight when the process died runs again from its start, so make a tool with side effects safe to repeat, for example by passing an idempotency key to the API it calls. Model requests, tool calls and output validation each retry on their own layer before a run fails with an exception such as UnexpectedModelBehavior; see Model errors.

The service above uses DBOS because it runs in your process against Postgres or SQLite, with no server to operate; Temporal and Restate run a server that your workers connect to, Prefect and Airflow fit when you already orchestrate with them, and Absurd and pgtask (Harness) run on Postgres alone. Durable Execution lists every engine and what each one wraps for you; an agent takes one engine at a time.

What you get compared to a checkpointer

In frameworks built around a checkpointer, one mechanism saves state after each step and covers crash recovery, pausing for a human and resuming a thread. Pydantic AI splits those jobs:

• Crash recovery inside a run is durable execution, at the granularity of one model request or journaled tool call.
• Pausing for a human needs no live process: deferred tools end the run and a later run picks it up.
• Continuing between runs is the stored Conversation. StepPersistence (Harness) adds message-history snapshots at each settled tool-call boundary, to continue or fork a run from, and a tool-effect ledger that records whether each tool call started, returned or raised: a call that never finished has an unknown outcome, so check the external system before running it again. Snapshots hold messages only, not the usage and pending deferred calls a Conversation carries, so they complement storing it rather than replace it.

Persistence lays out which piece answers which question.

How does a human approve a risky action, and the run resume days later?

Mark the tool as needing approval. When the model calls it, the run ends with a DeferredToolRequests output listing the pending calls, and nothing stays in memory while you wait. Store the run’s Conversation, which keeps the pending calls as deferred_tool_requests, ask a person, and when they answer, minutes or weeks later, continue the conversation with their decision:

production_approval.py

from pydantic_ai import Agent, ConversationTypeAdapter, DeferredToolRequests

agent = Agent('anthropic:claude-sonnet-5-5', output_type=[str, DeferredToolRequests])


@agent.tool_plain(requires_approval=True)
def issue_refund(order_id: str, amount: float) -> str:
    return f'Refunded {amount} on order {order_id}'


async def start(prompt: str) -> bytes:
    result = await agent.run(prompt)
    if isinstance(result.output, DeferredToolRequests):
        ...  # ask a person about each call in result.output.approvals
    return ConversationTypeAdapter.dump_json(result.conversation)  # store under result.conversation_id


async def resume(
    stored: bytes, approvals: dict[str, bool]
) -> tuple[bytes, str | DeferredToolRequests]:
    conversation = ConversationTypeAdapter.validate_json(stored)
    assert conversation.deferred_tool_requests is not None
    result = await agent.run(
        conversation=conversation,
        deferred_tool_results=conversation.deferred_tool_requests.build_results(approvals=approvals),
    )
    return ConversationTypeAdapter.dump_json(result.conversation), result.output  # store it again

Store the returned conversation after every continuation: the model can ask for another approval, and the conversation is what answers it.

The same mechanism hands a call to a frontend or a background worker to execute, and an approval can come back with edited arguments or a denial message the model sees. The UI adapters and the CLI carry approvals; under a durable engine the workflow can wait for the decision itself, as support_run does in the service above; and when the approver is in the same process, a HandleDeferredToolCalls handler resolves calls inline without ending the run. All of it is on Deferred Tools.

For an agent with nobody watching, ToolCallJudge (Harness) asks a second model whether a call may run and escalates the calls it is unsure about to a person; ApprovalReviewer (Harness) lets a second model approve the low-risk calls and leaves the rest to the person.

Approval guards against the model, not the client

An approval submitted by a client is trusted as given. Authenticate the endpoint, and check authorization inside the tool for sensitive actions; see the trust boundary for client-supplied history.

How do I keep conversation state?

A conversation’s state is its Conversation: the messages, the running usage that UsageLimits keep counting across turns and approvals, the conversation_id that correlates its runs, and any pending deferred calls. Store result.conversation in your own database, serialized with ConversationTypeAdapter and keyed by its conversation_id (a jsonb column is enough), and pass it back as conversation= on the next run.

A conversation ID is a key, not a credential: check that the caller may access the conversation before you load it, and serialize concurrent updates to one conversation, for example with a version column or a per-conversation lock, so two instances don’t continue the same stored version. Persistence covers an append-only variant, StepPersistence, and Memory (Harness) for what an agent should remember about a user across conversations; compaction keeps a long conversation inside the context window.

How do I bound cost and runaway loops?

UsageLimits caps one run’s agent-loop requests, successful tool calls, tokens and cost in USD, and raises UsageLimitExceeded once a limit is hit. request_limit counts turns of the agent loop, not every attempt sent to a provider, and tokens and cost are checked after a response arrives, so the request that crosses a limit has already been paid for:

production_limits.py

from decimal import Decimal

from pydantic_ai import Agent, UsageLimits

agent = Agent('openai:gpt-6.1-sol', tool_timeout=30)

limits = UsageLimits(request_limit=25, tool_calls_limit=50, cost_limit=Decimal('0.50'))


async def handle(prompt: str) -> str:
    result = await agent.run(prompt, usage_limits=limits)
    return result.output

Cost is best-effort: a model genai-prices has no pricing data for isn’t counted, so keep request_limit alongside cost_limit and keep model prices up to date; see usage limits.

For budgets longer than one run, such as a daily, monthly or per-tenant ceiling, SpendLimits (Harness) refuses the next request once a budget is spent. It is a brake, not a reservation: the request that crosses the line completes and concurrent runs can each pass the check, so spend can overshoot; budgets that span worker processes or restarts need a shared store instead of the default in-memory one; and an unpriced response counts as free unless you set on_unpriced='raise'. The Gateway enforces daily, weekly and monthly spending caps per project, user and API key before a request reaches the provider.

Timeouts bound each model request, tool call and hook and a whole run (the tool_timeout=30 above is one), cancellation stops a run from a stop button or a timer, max_concurrency caps how many runs an agent executes at once, and ConcurrencyLimitedModel caps concurrent requests to a model within one process (concurrency, not requests per minute, so still handle the provider’s rate-limit responses).

The Caching capability turns on prompt caching, so a multi-turn run stops paying full price for its instructions, tools and history on every request; prompt-cache health on each request span shows when a change breaks the cache, and token and cost metrics break spend down by provider and model.

What happens when a provider is down?

FallbackModel tries the next model when a request fails with an API error, or when a non-streamed response fails a check you define (streamed responses, as a frontend gets, are accepted as they are); see Fallback Model:

production_fallback.py

from pydantic_ai import Agent
from pydantic_ai.models.fallback import FallbackModel

agent = Agent(FallbackModel('anthropic:claude-sonnet-5-5', 'openai:gpt-6.1-sol'))

Provider SDK clients already retry some failures on their own. For rate limits and 5xx responses with backoff that respects Retry-After, install a retrying transport on the HTTP client; nothing retries at that layer by default. Retries explains how the layers multiply. The Gateway can do failover and load balancing between providers serving the same model on its side, configured as gateway endpoints.

Where do API keys go?

Each provider reads its key from an environment variable, such as OPENAI_API_KEY or ANTHROPIC_API_KEY (each model page names its own), or takes api_key= on its provider class, so you load keys from your secret manager like any other credential. With the Gateway, the application holds one PYDANTIC_AI_GATEWAY_API_KEY for every provider, and each Gateway key can carry its own spending cap, so you can issue one per service or environment; see the Gateway quick start.

How do I see what happened?

Pydantic AI emits OpenTelemetry traces of every run: each model request with its tokens and cost, each tool call with its arguments and result, and the run around them. Send them to Logfire or to any OpenTelemetry backend:

production_tracing.py

import logfire

logfire.configure(send_to_logfire='if-token-present')
logfire.instrument_pydantic_ai()

With Logfire, getting there takes logfire auth and logfire projects new (or logfire projects use for an existing project) on your machine and a write token in production; a coding agent can do it for you from pydantic.dev/ai-setup.md, and the free tier covers getting started. The same SDK instruments FastAPI and the HTTP requests to the provider, so a run shows up inside the request that started it, with the raw prompt it sent. Using Logfire walks through it.

Without the Logfire SDK, Agent.instrument_all() or the Instrumentation capability does the same against your own tracer provider; see OTel without Logfire. Every run carries its run_id and conversation_id, and run metadata such as a tenant or user ID lands on the run span, so you can filter traces by it.

Telemetry includes prompts, completions and tool arguments by default. To keep them out of Pydantic AI’s spans, set include_content=False on InstrumentationSettings, and include_binary_content=False leaves out the bytes of images, audio and documents. Neither touches run metadata, your own spans or captured HTTP request bodies, so configure those separately.

How do I test and evaluate changes?

Unit tests swap the model for TestModel, which calls your tools and returns data that matches your output type, or FunctionModel, which answers with a function you write. Agent.override swaps it in without touching your application code, and ALLOW_MODEL_REQUESTS = False makes any real model request fail the test:

production_testing.py

from pydantic_ai import Agent, models
from pydantic_ai.models.test import TestModel

models.ALLOW_MODEL_REQUESTS = False

agent = Agent(
    'anthropic:claude-sonnet-5-5',
    instructions='Answer support questions.',
    defer_model_check=True,  # build the model on first use, so tests need no API key
)


def test_support_agent():
    with agent.override(model=TestModel()):
        result = agent.run_sync('Where is my order?')
    assert result.output == 'success (no tool calls)'

Testing covers pytest fixtures, asserting on the messages a run exchanged, and FunctionModel.

Evals check what unit tests can’t: whether the model’s answers are good. Pydantic Evals runs your agent against a dataset of cases in code and scores the results with your own evaluators, built-in checks or an LLM judge, and trajectory evaluators grade the agent’s tool calls, not just its final answer.

To roll out a prompt or model change, run the same dataset against it and compare the reports; sending experiments to Logfire keeps every run side by side with the trace of each case. ManagedPrompt (Harness) backs instructions with a Logfire-managed prompt that you version, label and roll out from the Logfire UI without a redeploy, and an agent spec keeps the model, instructions and capabilities in a YAML or JSON file instead of code.

Online evaluation attaches the same evaluators to production traffic, scoring in the background and emitting the results as OpenTelemetry events. That background work lives in your process, not on a job queue: sample and cap the concurrency of expensive evaluators such as an LLM judge, and await wait_for_evaluations() on shutdown so pending ones finish. To steer a long run while it happens, TrajectoryJudge (Harness) reviews it every few model requests.

How do I limit what the agent can touch?

• Authorize tools and conversation access: give the agent only the tools the current caller may use, by building the toolset per run or filtering it against the user in your dependencies; check that the caller may access the conversation it continues; and treat a history sent by a client as untrusted input.
• Require approval for sensitive actions: deferred tools put a person, or a second model with ToolCallJudge, in front of the call.
• Isolate model-written code: run shell commands and generated code in a sandbox, such as bubblewrap locally or E2B or Modal remotely. FileSystem path patterns guard what the model’s file tools write, but they are checks, not isolation, and don’t bound shell commands.

Pydantic AI Harness adds more, such as guardrails with secret and PII detectors and a prompt injection defender for tool results.

How do I serve it to a frontend?

A UI adapter turns a request from your frontend into an agent run and streams its text, thinking and tool calls back over the AG-UI or Vercel AI protocol, from FastAPI or any Starlette app:

production_frontend.py

from fastapi import FastAPI
from starlette.requests import Request
from starlette.responses import Response

from pydantic_ai import Agent
from pydantic_ai.ui.vercel_ai import VercelAIAdapter

agent = Agent('anthropic:claude-sonnet-5-5')
app = FastAPI()


@app.post('/chat')
async def chat(request: Request) -> Response:
    return await VercelAIAdapter.dispatch_request(request, agent=agent)

Approvals for deferred tools travel over the same stream. The adapter endpoint is not an authentication boundary: run it inside your own authenticated route, as the trust model for client-submitted messages explains. For a durable run or a background job, the endpoint starts the run elsewhere and streams its events back. Interfaces lists every other way to serve an agent; the built-in web chat UI is for local development, not production.

How do I deploy it?

An agent is a Python object in your process, so you deploy it the way you deploy the rest of your Python: there is nothing Pydantic-specific to host. Typical shapes:

• A web service: run the agent inside your FastAPI, Starlette, Django or Flask handler, as above, and scale it like any other web app. A conversation’s state lives in what you store or in the durable engine, not in the process, so any instance can continue any conversation.
• A worker: start runs from a queue or schedule, on a durable engine’s workers such as Temporal, DBOS queues or Prefect deployments, or on Render Workflows (Harness), which run model requests and supported tool calls as separate tasks.
• A serverless function: run on AWS Lambda durable functions so a timed-out or retried invocation resumes instead of starting over.
• A voice agent: see realtime deployment.
• An agent on a repository: GitHub Agentic Workflows run an agent on issues, pull requests or a schedule.

In a long-running server, runs are async and share one event loop: max_concurrency caps how many run at once (with a queue limit for backpressure), UseThreadExecutor gives sync tools a bounded thread pool, and a slow tool can run as a background tool (Harness) while the agent keeps working.

Pin your versions: Pydantic AI follows its version policy, and Pydantic AI Harness is on 0.x releases whose API may change between minor versions. With a durable engine, a deploy also changes what in-flight runs replay against: adding a tool or changing a type the engine stored can break a run started under the old code, so let those runs drain on the old version or follow the engine’s versioning guidance, such as Temporal’s or AWS Lambda’s.