aiagent.org aiagent.org Pydantic AI
Type-Safe Production Systems

Pydantic AI Framework

Build robust production agents in Python using structural data models. Learn type validation, dynamic prompts, output streaming, dependency injection, and clean agentic design configurations.

Core Concepts Setup & Installation Agent Code Example FAQ

Core Concepts

Key features that make Pydantic AI a robust choice for enterprise applications.

Type Validation

Agent inputs, outputs, and system tools are strictly defined using Pydantic schemas, raising validation errors immediately if models return invalid formats.

Dynamic Prompts

Inject prompts dynamically based on runtime context using `@agent.system_prompt`, allowing agent behaviors to adapt to authenticated user roles or state databases.

Structured Streaming

Stream agent responses in real time. Supports streaming plain text or structured Pydantic objects as they are built chunk-by-chunk by the model.

Dependency Injection

Provide external dependencies (like active database clients, search engines, or API keys) dynamically to tools based on conversation contexts.

Conversation Memory

Persist and pass multi-turn chat history objects directly inside execution calls, allowing the agent to remember context across API thresholds.

Model Agnostic

Supports OpenAI, Anthropic, Gemini, or open-weights local models (via Ollama) with a single unified tool-calling interface.

Setup & Installation

Get started building with Pydantic AI in your Python environment.

Install Framework Package

# Install the core pydantic-ai package
pip install pydantic-ai

# Install dependencies for specific models (e.g. OpenAI or Gemini)
pip install "pydantic-ai[openai]"

Production Agent Example

Defining schemas, injecting dynamic prompts, streaming text completions, and managing conversation history.

from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
import httpx

# Define strict validation response schema
class AgentOutput(BaseModel):
    answer: str = Field(description="Direct response to user query")
    confidence_score: float = Field(description="Floating score between 0.0 and 1.0")
    sources: list[str] = Field(default_factory=list)

# Declare active HTTP client as dependency state
class Dependencies:
    def __init__(self, api_key: str, user_role: str):
        self.client = httpx.AsyncClient(headers={"Authorization": f"Bearer {api_key}"})
        self.user_role = user_role

# Configure structured Agent
agent = Agent(
    'openai:gpt-4o',
    result_type=AgentOutput,
    deps_type=Dependencies
)

# Register dynamic system prompt generator
@agent.system_prompt
def get_system_prompt(ctx: RunContext[Dependencies]) -> str:
    return f"You are a support bot. The user has {ctx.deps.user_role} role. Answer accordingly."

# Register a tool that consumes injected dependencies
@agent.tool
async def query_knowledge_base(ctx: RunContext[Dependencies], query: str) -> str:
    response = await ctx.deps.client.get(f"https://kb.internal/search?q={query}")
    return response.json().get("snippet", "No sources found")

# Execute execution loop with streaming response and history
async def main():
    deps = Dependencies(api_key="internal_sec_token", user_role="Admin")

    # Run agent, streaming outputs as they are received
    async with agent.run_stream("Why did deploy task tx_992 fail?", deps=deps) as response:
        print("Streaming response chunks:")
        async for message in response.stream_text():
            print(message, end="", flush=True)

    # Access completed validated structured result
    final_result = await response.get_data()
    print(f"\nSources validated: {final_result.sources}")

    # Capture message history for next turn
    history = response.new_messages()
    next_turn = await agent.run("Can you fix it?", deps=deps, message_history=history)
    print(f"Response to follow-up: {next_turn.data.answer}")

Frequently Asked Questions

How does Pydantic AI compare to LangChain or CrewAI?▶
Unlike LangChain which relies on complex chain configurations and prebuilt tool adapters, Pydantic AI takes a minimalist approach. It leverages native Python decorators, dependency injection scopes, and standard Pydantic validation schemas to minimize runtime library overhead.
What happens if the model returns JSON that fails validation?▶
Pydantic AI catches validation errors raised by parsing schemas, automatically formatting the validation error message and passing it back to the LLM as a system instruction to re-attempt output generation with corrected structures (with configured retry limits).
How do I handle dynamic system prompts in Pydantic AI?▶
Instead of static configuration strings, use the `@agent.system_prompt` decorator on a function. This function has access to a `RunContext[Dependencies]` parameter, allowing it to extract session details, auth levels, or state objects to construct the prompt dynamically for each LLM query call.
Can I stream agent responses in real time?▶
Yes. Use `agent.run_stream()` which returns an async context manager. Within it, you can iterate over text chunks using `async for chunk in response.stream_text()`, or structured data segments using `async for obj in response.stream_data()`.
How do I mock models in Pydantic AI for testing?▶
You can import `TestModel` from `pydantic_ai.models.test`. This allows you to construct a dummy model returning predefined strings or structured JSON chunks, which you pass directly when executing `agent.run(..., model=test_model)` inside your unit tests.
How does Pydantic AI manage conversation memory?▶
Pydantic AI models memory using standard lists of message schemas. After executing `agent.run()`, collect messages via `result.new_messages()` and pass them back into subsequent runs using `agent.run(query, message_history=history)`.