The type-safe way to build agents: structured output validated against a schema by default, dependency injection you can test, and a model-agnostic runtime from the team behind Pydantic. This is the long version, from first install to production.
The mental model
Pydantic AI is built by the team behind Pydantic, the validation library that most of Python's LLM tooling already runs on. It carries that same discipline into agents. You declare an Agent once, parameterized by two types: the dependencies it needs (deps_type) and the output it must return (output_type). The model can call tools and reason freely, but what crosses back into your code is validated against a schema you own. If the model returns something that does not fit, it is asked to retry, not handed to your app.
If you have used FastAPI, this will feel familiar: the framework leans on type hints and Pydantic models so your editor, your type checker, and the runtime all agree on the shape of things. Its dependency-injection style also makes agents testable, so you can swap the real model for a mock without rewriting the agent.
flowchart LR
U(["user prompt"]) --> A["Agent"]
A --> M["LLM model"]
M -->|"tool call"| T["your tool fn"]
T --> M
M -->|"final text"| V["validate output_type"]
V -->|"ok"| O(["typed result"])
V -->|"schema error, ModelRetry"| M
classDef code fill:#D6F5E3,stroke:#1B7F46,color:#09244B;
classDef model fill:#EEE6FF,stroke:#6D28D9,color:#09244B;
class T,V code;
class A,M,O model;
The model is allowed to talk in prose and to ask for tools, but the door back into your code is guarded by a schema. Nothing untyped gets through. A validation failure loops back to the model instead of leaking downstream.
Install & setup
The full pydantic-ai package bundles the OpenAI, Anthropic, and Google adapters. If you want a lean install, use pydantic-ai-slim with only the extras you need. The model is chosen by a string like provider:model, so switching providers is a one-line change.
# everything, all providers
pip install pydantic-ai
# or lean: pick your providers (and logfire for tracing)
pip install "pydantic-ai-slim[openai,anthropic,google,logfire]"
# set the key for whichever provider you use
export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."The same agent runs against any provider. Below, openai:gpt-5.2, anthropic:claude-sonnet-4-5, and google:gemini-3-flash-preview are interchangeable in the first argument.
Hello world
The smallest useful agent: a model, an instruction, a prompt. With no output_type, the result is a plain string. run_sync is the blocking entry point; result.output is the answer.
from pydantic_ai import Agent
agent = Agent(
"openai:gpt-5.2",
instructions="Be concise, reply with one sentence.",
)
result = agent.run_sync('Where does "hello world" come from?')
print(result.output)
# -> a str. result.usage() has token counts; result.all_messages() the transcript.instructions is the modern system prompt; more on the difference below
The headline feature · structured output
This is why the framework exists. Pass a BaseModel as output_type and the agent guarantees the return value is an instance of that model, already parsed and validated. No json.loads, no defensive try/except, no fields that are secretly None because the model wrote prose. The Agent is generic, Agent[Deps, Output], so your editor knows result.output is a SupportOutput.
from pydantic import BaseModel, Field
from pydantic_ai import Agent
class SupportOutput(BaseModel):
advice: str
block_card: bool = Field(description="whether to freeze the card now")
risk: int = Field(ge=0, le=10, description="risk score 0 to 10")
agent = Agent("openai:gpt-5.2", output_type=SupportOutput)
result = agent.run_sync("my card was charged twice at a shop I never visited")
print(result.output.block_card) # a real bool, type-checked
print(result.output.risk) # guaranteed 0..10 by the schemaoutput_type is not limited to a single model. It accepts scalars (bool, int), lists, TypedDict, plain functions (output functions the model calls), and unions of several types so the model can pick the right shape. Field descriptions and constraints are sent to the model as part of the schema, so they double as instructions.
The model only knows about a field if it is in the schema. Vague field names produce vague extractions. Name fields precisely and add Field(description=...) and constraints (ge, le, max_length) so the model is nudged toward valid values and a violation is caught at the boundary.
Dependency injection
Real agents need a database handle, an HTTP client, the current user id, config. Rather than reaching for globals, you declare a deps_type (usually a @dataclass) and pass an instance at run time. Inside tools, validators, and dynamic prompts, those dependencies arrive through RunContext[Deps] as ctx.deps. Because the dependencies are injected, a test can pass a fake DB and assert behaviour without touching the network.
from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
@dataclass
class SupportDeps:
customer_id: int
db: DatabaseConn # your own connection object
agent = Agent(
"openai:gpt-5.2",
deps_type=SupportDeps,
output_type=SupportOutput,
)
# pass a concrete instance per run; RunContext carries it everywhere
deps = SupportDeps(customer_id=42, db=DatabaseConn())
result = agent.run_sync("what is my balance?", deps=deps)
flowchart TB
D["deps: customer_id, db, http client"] --> C["RunContext"]
C --> T["agent.tool"]
C --> S["dynamic system_prompt"]
C --> V["output_validator"]
T --> M["LLM model"]
S --> M
M --> V
V --> O(["typed output"])
classDef code fill:#D6F5E3,stroke:#1B7F46,color:#09244B;
classDef model fill:#EEE6FF,stroke:#6D28D9,color:#09244B;
class D,T,S,V code;
class M,O model;
Tools
A tool is a Python function the model can request. Pydantic AI reads the signature and docstring to build the JSON schema the model sees, so typed parameters and a clear docstring are the whole contract. There are two flavours. Use @agent.tool when the tool needs the run context (dependencies, retry count); use @agent.tool_plain when it does not.
import random
from pydantic_ai import Agent, RunContext
agent = Agent("openai:gpt-5.2", deps_type=SupportDeps)
@agent.tool # takes RunContext -> can read deps
async def customer_balance(ctx: RunContext[SupportDeps]) -> float:
"""Return the current account balance for the customer."""
return await ctx.deps.db.balance(id=ctx.deps.customer_id)
@agent.tool_plain # no context needed
def roll_die() -> str:
"""Roll a six-sided die and return the number."""
return str(random.randint(1, 6))You can also register tools without decorators by passing them to the constructor: Agent(..., tools=[roll_die, customer_balance]), or use Tool(fn, takes_ctx=True) for finer control. Parameter descriptions are lifted from Google, NumPy, or Sphinx-style docstrings automatically.
The model proposes a tool call; your function runs it. For anything with a side effect (refunds, deletes, emails) validate the arguments and check authorization inside the tool, or raise ModelRetry to bounce a bad call back. Registering a tool is not the same as approving every call to it.
Dynamic system prompts & instructions
Static text goes in the instructions= (or legacy system_prompt=) argument. When the prompt depends on runtime state, register a function. It runs just before each request and can read ctx.deps. You can stack several; they concatenate.
from datetime import date
from pydantic_ai import Agent, RunContext
agent = Agent("openai:gpt-5.2", deps_type=SupportDeps)
@agent.instructions # re-evaluated every run
def add_customer(ctx: RunContext[SupportDeps]) -> str:
return f"You are helping customer #{ctx.deps.customer_id}."
@agent.instructions
def add_date() -> str: # RunContext is optional
return f"Today is {date.today()}."Prefer instructions. The difference bites in multi-turn chats: when you pass message_history, old system_prompt text stays embedded in the history, while instructions are dropped and only the current agent instructions are sent. Use system_prompt only when you deliberately want prior system text to persist across agents.
Running
Three entry points, one agent. run is the native async coroutine; run_sync wraps it for scripts and notebooks; run_stream is an async context manager for progressive output. All of them accept deps, message_history, model_settings, and usage_limits, and you can override the model per call.
# async, the real thing
result = await agent.run("what is my balance?", deps=deps)
print(result.output)
# sync wrapper for scripts
result = agent.run_sync("what is my balance?", deps=deps)
# override the model and cap spend for one call
from pydantic_ai.usage import UsageLimits
result = agent.run_sync(
"summarize",
model="anthropic:claude-sonnet-4-5",
usage_limits=UsageLimits(request_limit=3),
)| Method | Shape | Use it for |
|---|---|---|
run | async coroutine | web servers, async apps |
run_sync | blocking call | scripts, notebooks, tests |
run_stream | async context manager | token or partial-output streaming |
Streaming
Open run_stream as an async context manager. For a chat UI, stream_text() yields text as it generates. The interesting part is streaming structured output: stream_output() yields the partial Pydantic object, re-validated at each step, so you can render fields the moment they arrive without waiting for the full response.
# plain text, token by token
async with agent.run_stream("tell me a story") as response:
async for chunk in response.stream_text():
print(chunk, end="")
# structured output, validated as it fills in
async with agent.run_stream(user_input) as response:
async for profile in response.stream_output():
print(profile)
# -> SupportOutput(advice='...', block_card=None, ...) partial
# -> then fields fill in until the object is completeFor fine-grained control, iterate stream_response(debounce_by=0.01) to get raw ModelResponse chunks and call validate_response_output(msg, allow_partial=...) yourself, swallowing the ValidationError on still-incomplete objects.
Message history
Each run returns its transcript. To continue a conversation, feed the previous messages into the next run via message_history. When history is present, the model uses it for context and the fresh instructions are applied on top; you do not resend the system setup.
r1 = agent.run_sync("Who was Ada Lovelace?")
r2 = agent.run_sync(
"What is she best known for?",
message_history=r1.new_messages(), # just this run's messages
)
print(r2.output)
# r1.all_messages() -> full transcript incl. system/instructions
# r1.new_messages() -> only messages produced by this runBecause messages are plain serializable objects, you persist them wherever you like (a DB row, Redis) and rehydrate them on the next turn. That is your conversation store, and it stays under your control.
Output validators & model retry
Schema validation handles shape. When a check needs a side effect (does this SQL actually run? does this order id exist?), register an @agent.output_validator. Raise ModelRetry and the framework sends the error back to the model and asks it to try again, up to the retry budget. The same ModelRetry works inside tools when arguments are wrong.
from pydantic_ai import Agent, ModelRetry, RunContext
agent = Agent(
"openai:gpt-5.2",
deps_type=SupportDeps,
output_type=SqlQuery,
retries=2, # how many times the model may retry
)
@agent.output_validator
async def validate_sql(ctx: RunContext[SupportDeps], output: SqlQuery) -> SqlQuery:
try:
await ctx.deps.db.execute(f"EXPLAIN {output.sql}")
except QueryError as e:
raise ModelRetry(f"invalid query: {e}") from e
return output
flowchart LR
M["LLM model"] --> P["parse to output_type"]
P -->|"schema ok"| V["output_validator"]
P -->|"schema fail"| M
V -->|"valid"| O(["typed result"])
V -->|"ModelRetry"| M
classDef code fill:#D6F5E3,stroke:#1B7F46,color:#09244B;
classDef model fill:#EEE6FF,stroke:#6D28D9,color:#09244B;
class P,V code;
class M,O model;
When streaming, check ctx.partial_output and skip expensive side effects on intermediate values so you only run the DB check once the object is complete.
Testing
This is where the dependency-injection design pays off. Agent.override replaces the model (and deps or toolsets) inside a with block, so tests run offline and deterministically. TestModel auto-generates schema-valid data and calls every tool once; FunctionModel lets you script exact responses; capture_run_messages lets you assert on the full exchange. A global flag stops any accidental real call.
from pydantic_ai import models, capture_run_messages
from pydantic_ai.models.test import TestModel
from pydantic_ai.models.function import FunctionModel, AgentInfo
models.ALLOW_MODEL_REQUESTS = False # fail loudly if a real call slips through
def test_blocks_card_on_fraud():
with agent.override(model=TestModel()):
result = agent.run_sync("unknown charge", deps=fake_deps)
assert isinstance(result.output, SupportOutput)
# script exact model behaviour
def handler(messages, info: AgentInfo):
return ModelResponse(parts=[TextPart("canned reply")])
with agent.override(model=FunctionModel(handler)):
with capture_run_messages() as msgs:
agent.run_sync("hi", deps=fake_deps)
# assert on msgs: the exact requests and responses exchangedObservability · Pydantic Logfire
Logfire is the observability product from the same team, built on OpenTelemetry. Add two calls at startup and every run emits a trace: a span per model call and per tool execution, with prompts, tool arguments, token usage, and latency. Naming the agent keeps traces legible when you run several.
import logfire
from pydantic_ai import Agent
logfire.configure() # after `logfire auth` + `logfire projects new`
logfire.instrument_pydantic_ai() # trace every agent run
agent = Agent(
"openai:gpt-5.2",
name="support_agent", # shows up as the span name
instructions="Be concise.",
)
agent.run_sync("where does hello world come from?")Because it is plain OpenTelemetry under the hood, you can point the same spans at any OTel-compatible backend if you do not want to use Logfire itself.
Putting it together
Deps carry the customer id and DB. A tool reads the balance. A dynamic instruction personalizes the prompt. The output is a validated model, and an output validator enforces a business rule the schema alone cannot. One fuzzy step, typed code on both sides.
from dataclasses import dataclass
from pydantic import BaseModel, Field
from pydantic_ai import Agent, ModelRetry, RunContext
@dataclass
class SupportDeps:
customer_id: int
db: DatabaseConn
class SupportOutput(BaseModel):
advice: str
block_card: bool
risk: int = Field(ge=0, le=10)
support_agent = Agent(
"openai:gpt-5.2",
deps_type=SupportDeps,
output_type=SupportOutput,
retries=2,
instructions="You are a bank support agent. Judge risk and advise.",
)
@support_agent.instructions
def who(ctx: RunContext[SupportDeps]) -> str:
return f"You are helping customer #{ctx.deps.customer_id}."
@support_agent.tool
async def balance(ctx: RunContext[SupportDeps]) -> float:
"""Current account balance for the customer."""
return await ctx.deps.db.balance(id=ctx.deps.customer_id)
@support_agent.output_validator
def sane_risk(ctx: RunContext[SupportDeps], out: SupportOutput) -> SupportOutput:
if out.block_card and out.risk < 5:
raise ModelRetry("do not block a card below risk 5")
return out
deps = SupportDeps(customer_id=42, db=DatabaseConn())
result = support_agent.run_sync("a charge I never made just appeared", deps=deps)
print(result.output.block_card, result.output.risk)Production
agent.run (async) inside FastAPI or any async server; the agent object is cheap to define once at module load and reuse across requests.request_limit, token limits) on every run so a misbehaving model cannot rack up cost or spin forever.result.all_messages() as your conversation store and rehydrate via message_history; the messages are serializable.instrument_pydantic_ai) or any OpenTelemetry backend for per-run prompt, tool, token, and latency visibility.models.ALLOW_MODEL_REQUESTS = False plus Agent.override so CI never makes a real API call.retries modest (1 to 2). A validator that keeps raising ModelRetry against a model that cannot satisfy it burns tokens on every attempt.Gotchas
The run returns an AgentRunResult, not the answer. Your validated model is result.output. Forgetting .output hands the wrapper to your code and the type checker will not always catch it in dynamically typed call sites.
With message_history, fresh instructions are re-sent but prior instructions are not embedded in history, while system_prompt text is. Mixing the two, or expecting instructions to persist across turns, produces prompts that quietly change shape. Pick instructions and be consistent.
An output_validator that raises ModelRetry on a rule the model structurally cannot meet will retry until the budget runs out, then raise UnexpectedModelBehavior, having paid for every attempt. Keep retries low and make retry messages specific enough for the model to actually fix the output.
An async def tool needs an event loop. run_sync creates one, but calling it from inside an already-running loop (a notebook cell, an async web handler) raises. Use await agent.run(...) there instead.
Reach for it when
Skip it when
The principle it teaches
Parse and validate at the boundary. Pydantic AI bakes "never trust raw model output" into the type system: the output_type is the contract, and a schema violation loops back to the model instead of leaking downstream. In this course's TypeScript stack, your Zod extract() helper is exactly the same idea, one language over.
Instructor does structured output and nothing else: one call, typed result, no tools, no agent loop. Pydantic AI is Instructor plus the agent runtime: tools, dependency injection, retries, streaming, and message history around that same validated core. The OpenAI Agents SDK covers similar agent-loop ground but leans on OpenAI conventions; Pydantic AI is model-agnostic and puts the typed boundary first. Reach for Instructor when you only need the extract, Pydantic AI when the extract is one step inside a larger typed agent.
A Week 1 tool, tied directly to the boundary lesson from Sessions 1 and 2. It is the Python embodiment of the rule you enforce in TypeScript: nothing untyped crosses from the model into your code. When your Lab validation step wants to become an agent that can also call a tool or two, this is the smallest step up that keeps the boundary type-safe.