Agent[DepType, OutputType]
  ├── instructions / system_prompt
  ├── @agent.tool / @agent.tool_plain  →  functions LLM calls
  ├── deps_type  →  injected via RunContext[DepType]
  └── output_type  →  validated Pydantic model or primitive

pip install pydantic-ai                # base install
pip install 'pydantic-ai[openai]'      # with OpenAI support
pip install 'pydantic-ai[anthropic]'   # with Anthropic support
pip install 'pydantic-ai[logfire]'     # with observability

export OPENAI_API_KEY=sk-...
export ANTHROPIC_API_KEY=sk-ant-...

from pydantic_ai import Agent

undefined

undefined

---

from pydantic_ai import Agent, RunContext

agent = Agent('openai:gpt-4o', deps_type=str)

@agent.tool                          # has RunContext access (for deps)
async def search_db(ctx: RunContext[str], query: str) -> list[str]:
    """Search the database for records matching the query."""
    db_conn = ctx.deps  # access injected dependency
    return await db_conn.search(query)

@agent.tool_plain                    # no RunContext needed
def get_current_time() -> str:
    """Return the current UTC time."""
    from datetime import datetime, UTC
    return datetime.now(UTC).isoformat()

from pydantic_ai.exceptions import ModelRetry

@agent.tool_plain
def divide(a: float, b: float) -> float:
    """Divide a by b."""
    if b == 0:
        raise ModelRetry('Cannot divide by zero — provide a non-zero b.')
    return a / b

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

@dataclass
class AppDeps:
    http_client: httpx.AsyncClient
    user_id: int
    api_key: str

agent = Agent('openai:gpt-4o', deps_type=AppDeps)

@agent.tool
async def fetch_user_data(ctx: RunContext[AppDeps]) -> dict:
    """Fetch the current user's profile."""
    resp = await ctx.deps.http_client.get(
        f'/users/{ctx.deps.user_id}',
        headers={'Authorization': f'Bearer {ctx.deps.api_key}'}
    )
    return resp.json()

undefined

@agent.instructions
async def personalized_instructions(ctx: RunContext[AppDeps]) -> str:
    user = await ctx.deps.db.get_user(ctx.deps.user_id)
    return f"The user's name is {user.name!r}. Personalize all responses."

from pydantic import BaseModel, Field
from pydantic_ai import Agent

class MovieReview(BaseModel):
    title: str
    rating: int = Field(ge=1, le=10, description='Rating from 1-10')
    summary: str
    recommend: bool

agent = Agent('openai:gpt-4o', output_type=MovieReview)
result = agent.run_sync('Review the movie Inception')
review = result.output  # typed as MovieReview
print(review.rating)    # int, validated by Pydantic

from typing import Union

class SuccessResult(BaseModel):
    data: dict

class ErrorResult(BaseModel):
    error_message: str
    retry_after: int

agent = Agent('openai:gpt-4o', output_type=Union[SuccessResult, ErrorResult])

summarizer = Agent('openai:gpt-4o-mini', instructions='Summarize text concisely.')
analyst = Agent('openai:gpt-4o', instructions='Analyze data and provide insights.')

@analyst.tool_plain
async def summarize(text: str) -> str:
    """Summarize a long piece of text before analysis."""
    result = await summarizer.run(text)
    return result.output

from pydantic_ai.messages import ModelMessagesTypeAdapter

result1 = await agent.run('Hello, my name is Alice.')
result2 = await agent.run('What is my name?', message_history=result1.new_messages())
print(result2.output)  # "Your name is Alice."

from pydantic import BaseModel

class Analysis(BaseModel):
    summary: str
    key_points: list[str]
    confidence: float

agent = Agent('openai:gpt-4o', output_type=Analysis)

async with agent.run_stream('Analyze quantum computing trends') as resp:
    async for partial in resp.stream():
        print(partial)  # partial Analysis objects as they arrive
    final = await resp.get_output()

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio

---

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

agent = Agent('openai:gpt-4o', output_type=bool)

def test_agent_response():
    with agent.override(model=TestModel(custom_output_text='true')):
        result = agent.run_sync('Is Paris the capital of France?')
    assert result.output is True

from pydantic_ai.models.function import FunctionModel, ModelRequest

def my_model(messages: list[ModelRequest], info) -> str:
    last = messages[-1].parts[-1].content
    return f'Echo: {last}'

with agent.override(model=FunctionModel(my_model)):
    result = agent.run_sync('test')

from pydantic_ai import Agent
from pydantic_ai.exceptions import ModelRetry, UnexpectedModelBehavior

agent = Agent('openai:gpt-4o', retries=3)  # retry output validation up to 3x

@agent.tool_plain
def parse_number(text: str) -> int:
    """Parse a number from user input."""
    try:
        return int(text.strip())
    except ValueError:
        raise ModelRetry(f'Could not parse {text!r} as an integer. Please provide a valid number.')

---