Try Live
Add Docs
Rankings
Pricing
Docs
Install
Theme
Install
Docs
Pricing
More...
More...
Try Live
Rankings
Enterprise
Create API Key
Add Docs
Pydantic AI
https://github.com/pydantic/pydantic-ai
Admin
Pydantic AI is a Python agent framework designed to simplify building production-grade applications
...
Tokens:
307,660
Snippets:
1,856
Trust Score:
9.6
Update:
6 days ago
Context
Skills
Chat
Benchmark
81.4
Suggestions
Latest
Show doc for...
Code
Info
Show Results
Context Summary (auto-generated)
Raw
Copy
Link
# Pydantic AI Pydantic AI is a Python agent framework designed to help build production-grade applications with Generative AI. Built by the creators of Pydantic, it brings FastAPI-style ergonomic design to GenAI development, offering type-safe interactions with LLMs, model-agnostic architecture supporting virtually every major provider (OpenAI, Anthropic, Gemini, Bedrock, Groq, Mistral, and many more), and seamless integration with Pydantic Logfire for observability. The framework emphasizes structured outputs, dependency injection for tools and system prompts, and comprehensive streaming support. The core abstraction is the `Agent` class, which orchestrates interactions with LLMs by combining instructions, function tools, structured output types, and dependency injection. Agents can run synchronously or asynchronously, support streaming responses, and maintain conversation history across multiple runs. The framework also includes Pydantic Graph for complex state-machine workflows, Pydantic Evals for systematic testing of AI systems, and MCP (Model Context Protocol) integration for connecting to external tools and services. ## Creating a Basic Agent The Agent class is the primary interface for interacting with LLMs. Agents are configured with a model, optional instructions, and can be run synchronously or asynchronously to generate responses. ```python from pydantic_ai import Agent # Create a simple agent with static instructions agent = Agent( 'openai:gpt-5.2', instructions='Be concise, reply with one sentence.', ) # Run synchronously result = agent.run_sync('Where does "hello world" come from?') print(result.output) # Output: The first known use of "hello, world" was in a 1974 textbook about the C programming language. # Run asynchronously async def main(): result = await agent.run('What is the capital of France?') print(result.output) # Output: The capital of France is Paris. ``` ## Structured Output with Pydantic Models Agents can be configured to return structured data using Pydantic models as the output_type. The LLM is forced to return data matching the specified schema, with automatic validation. ```python from pydantic import BaseModel, Field from pydantic_ai import Agent class CityLocation(BaseModel): city: str country: str class SupportOutput(BaseModel): support_advice: str = Field(description='Advice returned to the customer') block_card: bool = Field(description="Whether to block the customer's card") risk: int = Field(description='Risk level of query', ge=0, le=10) # Agent with structured output agent = Agent('google-gla:gemini-3-flash-preview', output_type=CityLocation) result = agent.run_sync('Where were the olympics held in 2012?') print(result.output) # Output: city='London' country='United Kingdom' print(result.usage()) # Output: RunUsage(input_tokens=57, output_tokens=8, requests=1) # Multiple output types (union) agent = Agent( 'openai:gpt-5-mini', output_type=[CityLocation, str], instructions='Extract city location if provided, otherwise respond with text.', ) result = agent.run_sync('Tell me about Paris') print(result.output) # Output: city='Paris' country='France' ``` ## Function Tools with @agent.tool Decorator Tools provide a mechanism for models to perform actions and retrieve information. Use @agent.tool for tools that need context access, or @agent.tool_plain for standalone tools. ```python import random from pydantic_ai import Agent, RunContext agent = Agent( 'google-gla:gemini-3-flash-preview', deps_type=str, # Player's name as dependency instructions=( "You're a dice game, roll the die and see if the number " "matches the user's guess. Use the player's name in the response." ), ) @agent.tool_plain def roll_dice() -> str: """Roll a six-sided die and return the result.""" return str(random.randint(1, 6)) @agent.tool def get_player_name(ctx: RunContext[str]) -> str: """Get the player's name.""" return ctx.deps # Run the agent with dependencies result = agent.run_sync('My guess is 4', deps='Anne') print(result.output) # Output: Congratulations Anne, you guessed correctly! You're a winner! ``` ## Dependency Injection System Dependencies allow passing data, connections, and services to system prompts, tools, and validators. Define dependencies using dataclasses and access them via RunContext. ```python from dataclasses import dataclass import httpx from pydantic_ai import Agent, RunContext @dataclass class MyDeps: api_key: str http_client: httpx.AsyncClient agent = Agent( 'openai:gpt-5.2', deps_type=MyDeps, ) @agent.system_prompt async def get_system_prompt(ctx: RunContext[MyDeps]) -> str: response = await ctx.deps.http_client.get( 'https://example.com', headers={'Authorization': f'Bearer {ctx.deps.api_key}'}, ) response.raise_for_status() return f'System context: {response.text}' @agent.tool async def fetch_data(ctx: RunContext[MyDeps], query: str) -> str: """Fetch data from external API.""" response = await ctx.deps.http_client.get( 'https://api.example.com/data', params={'q': query}, headers={'Authorization': f'Bearer {ctx.deps.api_key}'}, ) return response.text async def main(): async with httpx.AsyncClient() as client: deps = MyDeps('api-key-here', client) result = await agent.run('Search for Python tutorials', deps=deps) print(result.output) ``` ## Streaming Responses Agents support streaming text and structured output for real-time response handling. Use run_stream() for streaming with context manager. ```python from pydantic_ai import Agent from typing_extensions import NotRequired, TypedDict agent = Agent('google-gla:gemini-3-flash-preview') # Stream text output async def stream_text(): async with agent.run_stream('Tell me a story about a cat') as result: async for text in result.stream_text(): print(text) # Output (incremental): # Once upon a # Once upon a time, there was # Once upon a time, there was a curious cat... # Stream structured output class UserProfile(TypedDict): name: str dob: NotRequired[str] bio: NotRequired[str] profile_agent = Agent( 'openai:gpt-5.2', output_type=UserProfile, instructions='Extract a user profile from the input', ) async def stream_structured(): user_input = 'My name is Ben, born January 28th 1990, I like hiking.' async with profile_agent.run_stream(user_input) as result: async for profile in result.stream_output(): print(profile) # Output (incremental): # {'name': 'Ben'} # {'name': 'Ben', 'dob': '1990-01-28'} # {'name': 'Ben', 'dob': '1990-01-28', 'bio': 'Likes hiking'} ``` ## Message History and Conversations Maintain conversation context across multiple agent runs by passing message history. Messages are model-agnostic and can be serialized for persistence. ```python from pydantic_ai import Agent, ModelMessagesTypeAdapter from pydantic_core import to_jsonable_python agent = Agent('openai:gpt-5.2', instructions='Be a helpful assistant.') # First run result1 = agent.run_sync('Who was Albert Einstein?') print(result1.output) # Output: Albert Einstein was a German-born theoretical physicist. # Continue conversation with message history result2 = agent.run_sync( 'What was his most famous equation?', message_history=result1.new_messages(), ) print(result2.output) # Output: Albert Einstein's most famous equation is E = mc^2. # Serialize messages for storage history = result2.all_messages() json_history = to_jsonable_python(history) # Restore messages from storage restored_history = ModelMessagesTypeAdapter.validate_python(json_history) result3 = agent.run_sync('Tell me more about it', message_history=restored_history) ``` ## Dynamic Instructions with Decorators Use the @agent.instructions decorator to create dynamic instructions that can access dependencies and runtime context. ```python from datetime import date from pydantic_ai import Agent, RunContext agent = Agent( 'openai:gpt-5.2', deps_type=str, instructions="Use the customer's name while replying to them.", ) @agent.instructions def add_customer_name(ctx: RunContext[str]) -> str: return f"The customer's name is {ctx.deps}." @agent.instructions def add_date() -> str: return f'The date is {date.today()}.' result = agent.run_sync('What is the date?', deps='Frank') print(result.output) # Output: Hello Frank, the date today is 2032-01-02. ``` ## Model Retry and Error Handling Use ModelRetry to ask the model to retry with different parameters. Configure retry limits at agent, tool, or output level. ```python from pydantic_ai import Agent, RunContext, ModelRetry from pydantic_ai.exceptions import UnexpectedModelBehavior, UsageLimitExceeded from pydantic_ai.usage import UsageLimits agent = Agent( 'openai:gpt-5.2', retries=3, # Global retry limit ) @agent.tool(retries=2) # Tool-specific retry limit def get_user_by_name(ctx: RunContext, name: str) -> int: """Get a user's ID from their full name.""" user_id = lookup_user(name) # Your lookup function if user_id is None: raise ModelRetry( f'No user found with name {name!r}, provide their full name' ) return user_id # Usage limits to prevent runaway costs try: result = agent.run_sync( 'Find user John', usage_limits=UsageLimits( response_tokens_limit=100, request_limit=5, tool_calls_limit=10, ), ) except UsageLimitExceeded as e: print(f'Limit exceeded: {e}') except UnexpectedModelBehavior as e: print(f'Model error: {e}') ``` ## Output Functions Output functions let the model call a function as its final action. The function result becomes the agent's output without being sent back to the model. ```python from pydantic import BaseModel from pydantic_ai import Agent, ModelRetry, RunContext class Row(BaseModel): name: str country: str def run_sql_query(query: str) -> list[Row]: """Run a SQL query on the database.""" if 'SELECT' not in query.upper(): raise ModelRetry('Only SELECT queries are supported') # Execute query and return results return [Row(name='Paris', country='France')] sql_agent = Agent[None, list[Row]]( 'openai:gpt-5.2', output_type=run_sql_query, # Function as output type instructions='You are a SQL agent that runs queries on a database.', ) result = sql_agent.run_sync('Get all capital cities') print(result.output) # Output: [Row(name='Paris', country='France')] ``` ## Output Validators Add custom validation logic for agent outputs using the @agent.output_validator decorator. Validators can be async and raise ModelRetry. ```python from pydantic import BaseModel from pydantic_ai import Agent, RunContext, ModelRetry class SQLQuery(BaseModel): query: str agent = Agent('openai:gpt-5.2', output_type=SQLQuery) @agent.output_validator async def validate_sql(ctx: RunContext, output: SQLQuery) -> SQLQuery: try: # Validate query syntax await execute_explain(output.query) except QueryError as e: raise ModelRetry(f'Invalid query: {e}') return output result = agent.run_sync('Generate a query to get all users') print(result.output.query) ``` ## Multiple Model Providers Pydantic AI supports multiple model providers with a consistent interface. Switch providers by changing the model string or using model classes directly. ```python from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIChatModel from pydantic_ai.models.anthropic import AnthropicModel from pydantic_ai.models.google import GoogleModel from pydantic_ai.models.fallback import FallbackModel # Using model strings (auto-detection) openai_agent = Agent('openai:gpt-5.2') anthropic_agent = Agent('anthropic:claude-sonnet-4-5') gemini_agent = Agent('google-gla:gemini-3-flash-preview') groq_agent = Agent('groq:llama-4-scout-17b-16e-instruct') # Using model classes for more control model = OpenAIChatModel('gpt-5.2', api_key='your-key') agent = Agent(model) # Fallback model for reliability fallback = FallbackModel( OpenAIChatModel('gpt-5.2'), AnthropicModel('claude-sonnet-4-5'), ) reliable_agent = Agent(fallback) result = reliable_agent.run_sync('Hello') # Falls back if first model fails ``` ## MCP (Model Context Protocol) Integration Connect agents to MCP servers for external tools and services. Pydantic AI can act as an MCP client or be used within MCP servers. ```python from pydantic_ai import Agent from pydantic_ai.mcp import MCPServer # Connect to an MCP server mcp_server = MCPServer('http://localhost:8080') agent = Agent( 'openai:gpt-5.2', toolsets=[mcp_server], # Add MCP tools to agent ) result = agent.run_sync('Use the external tool to fetch data') print(result.output) ``` ## Pydantic Evals for Testing Pydantic Evals provides a framework for systematically testing AI systems with datasets, evaluators, and experiment reports. ```python from pydantic_evals import Case, Dataset from pydantic_evals.evaluators import Evaluator, EvaluatorContext, IsInstance # Define test cases case1 = Case( name='capital_question', inputs='What is the capital of France?', expected_output='Paris', metadata={'difficulty': 'easy'}, ) # Custom evaluator class MyEvaluator(Evaluator[str, str]): def evaluate(self, ctx: EvaluatorContext[str, str]) -> float: if ctx.output == ctx.expected_output: return 1.0 elif ctx.expected_output.lower() in ctx.output.lower(): return 0.8 return 0.0 # Create dataset with evaluators dataset = Dataset( cases=[case1], evaluators=[IsInstance(type_name='str'), MyEvaluator()], ) # Define the task to evaluate async def answer_question(question: str) -> str: agent = Agent('openai:gpt-5-mini') result = await agent.run(question) return result.output # Run evaluation report = dataset.evaluate_sync(answer_question) report.print(include_input=True, include_output=True) ``` ## Pydantic Graph for Complex Workflows Use pydantic-graph for state-machine workflows where nodes and edges are defined using type hints. ```python from __future__ import annotations from dataclasses import dataclass from pydantic_graph import BaseNode, End, Graph, GraphRunContext @dataclass class DivisibleBy5(BaseNode[None, None, int]): foo: int async def run(self, ctx: GraphRunContext) -> Increment | End[int]: if self.foo % 5 == 0: return End(self.foo) return Increment(self.foo) @dataclass class Increment(BaseNode): foo: int async def run(self, ctx: GraphRunContext) -> DivisibleBy5: return DivisibleBy5(self.foo + 1) # Create and run the graph graph = Graph(nodes=[DivisibleBy5, Increment]) result = graph.run_sync(DivisibleBy5(4)) print(result.output) # Output: 5 # Generate mermaid diagram print(graph.mermaid_code(start_node=DivisibleBy5)) ``` ## Logfire Integration for Observability Integrate with Pydantic Logfire for real-time debugging, tracing, and monitoring of agent behavior. ```python import logfire from pydantic_ai import Agent # Configure Logfire logfire.configure() logfire.instrument_pydantic_ai() # All agent runs are now traced agent = Agent('openai:gpt-5.2') result = agent.run_sync('What is 2 + 2?') # Traces show: # - Messages exchanged with the model # - Tool calls with arguments and return values # - Token usage per request # - Latency for each operation # - Any errors with full context ``` ## Complete Bank Support Agent Example A complete example demonstrating dynamic instructions, structured output, tools, and dependency injection. ```python import sqlite3 from dataclasses import dataclass from pydantic import BaseModel from pydantic_ai import Agent, RunContext @dataclass class DatabaseConn: conn: sqlite3.Connection async def customer_name(self, id: int) -> str | None: cur = self.conn.cursor() res = cur.execute('SELECT name FROM customers WHERE id=?', (id,)) row = res.fetchone() return row[0] if row else None async def customer_balance(self, id: int) -> float: cur = self.conn.cursor() res = cur.execute('SELECT balance FROM customers WHERE id=?', (id,)) row = res.fetchone() if row: return row[0] raise ValueError('Customer not found') @dataclass class SupportDeps: customer_id: int db: DatabaseConn class SupportOutput(BaseModel): support_advice: str block_card: bool risk: int support_agent = Agent( 'openai:gpt-5.2', deps_type=SupportDeps, output_type=SupportOutput, instructions=( 'You are a support agent in our bank. Give the customer ' 'support and judge the risk level of their query.' ), ) @support_agent.instructions async def add_customer_name(ctx: RunContext[SupportDeps]) -> str: name = await ctx.deps.db.customer_name(ctx.deps.customer_id) return f"The customer's name is {name!r}" @support_agent.tool async def customer_balance(ctx: RunContext[SupportDeps]) -> str: """Returns the customer's current account balance.""" balance = await ctx.deps.db.customer_balance(ctx.deps.customer_id) return f'${balance:.2f}' # Usage with sqlite3.connect(':memory:') as conn: conn.execute('CREATE TABLE customers(id, name, balance)') conn.execute("INSERT INTO customers VALUES (123, 'John', 123.45)") conn.commit() deps = SupportDeps(customer_id=123, db=DatabaseConn(conn)) result = support_agent.run_sync('What is my balance?', deps=deps) print(result.output) # Output: support_advice='Hello John, your balance is $123.45.' block_card=False risk=1 result = support_agent.run_sync('I just lost my card!', deps=deps) print(result.output) # Output: support_advice="I'm sorry John, blocking your card now." block_card=True risk=8 ``` ## Summary Pydantic AI provides a comprehensive framework for building production-grade AI applications with strong typing, model-agnostic design, and enterprise-ready features. The core Agent class serves as the primary interface, supporting synchronous and asynchronous execution, streaming responses, and structured outputs validated by Pydantic. The dependency injection system enables clean separation of concerns, while tools and dynamic instructions allow agents to interact with external services and adapt to runtime context. Key integration patterns include: using message history for multi-turn conversations, implementing fallback models for reliability, connecting to MCP servers for external tools, and leveraging Pydantic Evals for systematic testing. For complex workflows, pydantic-graph provides a type-safe state machine abstraction. Throughout all these patterns, Logfire integration enables comprehensive observability for debugging and monitoring agent behavior in production environments.