Prismix
PydanticAI Pydantic Python 9 min read

PydanticAI Guide 2025: Type-Safe Python AI Agent Framework

PydanticAI is the agent framework from the Pydantic team — the same people who built Python's most popular data validation library. Instead of string outputs and hidden magic, PydanticAI agents return validated Pydantic models, use plain Python decorators for tools, and inject dependencies through a typed RunContext. It supports OpenAI, Anthropic, Gemini, Groq, Mistral, and Ollama.

What Is PydanticAI?

PydanticAI was created by Samuel Colvin (author of Pydantic) with a single goal: bring Pydantic's type-safety philosophy to AI application development. While frameworks like LangChain introduced their own DSLs and chain abstractions, PydanticAI keeps everything in plain Python. An Agent is just a Python object. Tools are decorated functions. Results are Pydantic models validated at runtime.

Key design principles: strict Python types end-to-end, no custom DSL, production-grade error handling, built-in dependency injection via RunContext, and first-class support for async. Built on Pydantic v2.

Install PydanticAI

pip install pydantic-ai

To install with a specific provider's extra dependencies: pip install "pydantic-ai[anthropic]" or pip install "pydantic-ai[groq]".

Install and First Agent

Create an Agent with a model string and an optional system_prompt. Call agent.run_sync() for synchronous code or await agent.run() in async contexts:

first_agent.py

from pydantic_ai import Agent

agent = Agent(
    model="openai:gpt-4o",
    system_prompt="You are a concise technical assistant.",
)

result = agent.run_sync("What is the difference between a list and a tuple in Python?")
print(result.data)
# -> "Lists are mutable; tuples are immutable. ..."

The model string follows the pattern provider:model-name. Supported prefixes: openai:, anthropic:, google-gla:, groq:, mistral:, ollama:.

Structured Output with result_type

The killer feature of PydanticAI: set result_type to a Pydantic model and the agent returns a validated instance — not a raw string. The LLM is automatically instructed to return JSON matching the schema, and Pydantic validates it at runtime:

structured_output.py

from pydantic import BaseModel
from pydantic_ai import Agent

class CodeReview(BaseModel):
    summary: str
    issues: list[str]
    severity: str  # "low" | "medium" | "high"
    approve: bool

agent = Agent(
    model="anthropic:claude-3-5-sonnet-latest",
    result_type=CodeReview,
    system_prompt="You are a senior code reviewer.",
)

result = agent.run_sync(
    "Review this function: def add(a, b): return a + b"
)

review = result.data  # type: CodeReview
print(review.approve)   # True
print(review.severity)  # "low"
print(review.issues)    # []

If the LLM returns invalid JSON or the schema doesn't validate, PydanticAI automatically retries with the validation error included in the next prompt — up to a configurable retries count.

Tools / Function Calling with @agent.tool

Decorate a function with @agent.tool to register it as a callable tool. The first argument must be RunContext (for dependency access); the remaining arguments become the tool's JSON schema parameters. Both sync and async tools are supported:

tools.py

import httpx
from pydantic_ai import Agent, RunContext

agent = Agent(
    model="openai:gpt-4o",
    system_prompt="You can look up current weather.",
)

@agent.tool
async def get_weather(ctx: RunContext[None], city: str) -> str:
    """Return current temperature for a city."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"https://wttr.in/{city}?format=%t"
        )
        return resp.text.strip()

import asyncio
result = asyncio.run(agent.run("What's the weather in Kyiv?"))
print(result.data)  # "Current temperature in Kyiv is 18°C"

Use @agent.tool_plain when you don't need RunContext — the first argument then becomes the first tool parameter directly.

Dependency Injection via RunContext

Set deps_type on the Agent to declare what dependencies tools receive. Pass the actual dependency at call time via the deps parameter. This eliminates hidden globals and makes testing trivial — just pass a mock:

dependency_injection.py

from dataclasses import dataclass
import asyncpg
from pydantic_ai import Agent, RunContext

@dataclass
class Deps:
    db: asyncpg.Connection
    user_id: int

agent = Agent(
    model="openai:gpt-4o",
    deps_type=Deps,
    system_prompt="Answer questions about the user's orders.",
)

@agent.tool
async def get_orders(ctx: RunContext[Deps]) -> list[dict]:
    """Fetch recent orders for the current user."""
    rows = await ctx.deps.db.fetch(
        "SELECT id, total FROM orders WHERE user_id = $1",
        ctx.deps.user_id,
    )
    return [dict(r) for r in rows]

# Inject real deps at call time
async def main(conn, uid):
    deps = Deps(db=conn, user_id=uid)
    result = await agent.run("Show my last 3 orders", deps=deps)
    print(result.data)

Streaming Responses

Use agent.run_stream() as an async context manager to iterate over text tokens in real time. This is useful for chat UIs where you want to display tokens as they arrive:

streaming.py

import asyncio
from pydantic_ai import Agent

agent = Agent(
    model="groq:llama-3.3-70b-versatile",
    system_prompt="You are a helpful assistant.",
)

async def stream_response():
    async with agent.run_stream("Explain async/await in Python") as stream:
        async for text in stream.stream_text(delta=True):
            print(text, end="", flush=True)
        print()  # newline at end

asyncio.run(stream_response())

Pass delta=True to receive incremental chunks; omit it to receive the accumulated text so far on each iteration. The final validated result is available as await stream.get_data() after the loop.

Multi-Turn Conversation

PydanticAI tracks message history on each RunResult. Pass message_history=result.new_messages() to the next call to continue the conversation with full context:

multi_turn.py

from pydantic_ai import Agent

agent = Agent(
    model="anthropic:claude-3-5-haiku-latest",
    system_prompt="You are a Python tutor.",
)

# Turn 1
result1 = agent.run_sync("What is a list comprehension?")
print(result1.data)

# Turn 2 — passes history so the agent remembers turn 1
result2 = agent.run_sync(
    "Can you show me an example?",
    message_history=result1.new_messages(),
)
print(result2.data)

# Turn 3 — full 3-turn context
result3 = agent.run_sync(
    "Now show me a nested one.",
    message_history=result2.new_messages(),
)
print(result3.data)

result.new_messages() returns only the messages added in that run. result.all_messages() returns the complete history including the original input.

PydanticAI vs LangChain vs LlamaIndex vs Instructor

Framework Type safety Agent loops Streaming Ecosystem Complexity
PydanticAI Excellent (Pydantic v2) Built-in Yes Growing Low
LangChain Weak (string-heavy) LangGraph Yes Huge High
LlamaIndex Moderate Yes (AgentRunner) Yes Large (RAG focus) Medium
Instructor Excellent (Pydantic) None (extraction only) Partial Small Very low

Choose PydanticAI when you want full agent loops with type-safe output and clean dependency injection. Choose Instructor when you only need structured extraction from a single LLM call. Choose LangChain when you need its broad integration ecosystem.

Testing with TestModel and FunctionModel

PydanticAI ships two test-only models that avoid any network calls. TestModel returns a canned response. FunctionModel lets you write a Python function that decides the response, enabling you to verify which tools the agent calls and what prompts it sends:

test_agent.py

import pytest
from pydantic import BaseModel
from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel, FunctionModel
from pydantic_ai.messages import ModelMessage, ModelResponse, TextPart

class Answer(BaseModel):
    text: str
    confidence: float

agent = Agent(result_type=Answer)

@agent.tool_plain
def get_context(topic: str) -> str:
    """Look up context for a topic."""
    return f"Context about {topic}"

# --- Test 1: TestModel returns a fixed canned response ---
def test_agent_with_test_model():
    with agent.override(model=TestModel()):
        result = agent.run_sync("Tell me about Python.")
    # TestModel returns a valid JSON matching result_type schema
    assert isinstance(result.data, Answer)

# --- Test 2: FunctionModel — verify tool was called ---
def test_agent_calls_tool():
    called_with = []

    def my_model(messages: list[ModelMessage], info) -> ModelResponse:
        # First call: instruct the agent to call get_context
        if not called_with:
            return ModelResponse(parts=[
                # Simulate a tool call
            ])
        # Second call: return final answer
        return ModelResponse(parts=[TextPart('{"text": "ok", "confidence": 0.9}')])

    with agent.override(model=FunctionModel(my_model)):
        result = agent.run_sync("Tell me about Python.")
    assert result.data.confidence == pytest.approx(0.9)

Use agent.override(model=TestModel()) as a context manager to swap the model for the duration of a test without modifying the agent definition.

Monitor Your AI APIs in Real Time

PydanticAI agents call OpenAI, Anthropic, and Groq under the hood. Prismix monitors all of them in real time — so when an API goes down or degrades, you know before your users do.

Monitor AI API Status →