Building Governed Agents with PydanticAI: Typed, Validated, Safe

Introduction

Most AI agent frameworks treat type safety as optional. Your agent returns a string, you parse it with regex, and hope the LLM didn’t hallucinate a different format than expected. This works until it doesn’t—and when it fails in production, the error message tells you nothing about why.

PydanticAI takes a fundamentally different approach. Built by the team behind Pydantic (the validation layer used by the OpenAI SDK, Anthropic SDK, Google ADK, and most of the AI ecosystem), it brings end-to-end type safety to agents. Your IDE, type checker, and the LLM all agree on what your agent returns. Errors move from runtime to write-time.

This post shows how to build governed agents with PydanticAI—agents that are typed, validated, and safe by construction.

Why This Matters

Ungoverned agents fail in subtle, expensive ways:

• Invalid tool arguments: The LLM generates malformed parameters that crash your database calls
• Unstructured output: You parse free-text responses and miss edge cases in production
• Inconsistent behavior: Same prompt, different output format, broken downstream systems
• No validation pipeline: Errors surface in user-facing responses instead of during development

PydanticAI’s approach solves these by making validation part of the agent definition. When the agent returns a Sentiment object, you know it has a label (either “positive”, “negative”, or “neutral”) and a score (a float between -1 and 1). The LLM cannot return anything else—Pydantic validates it.

As the PydanticAI documentation states: “Your IDE, type checker, and coding agent all know what your agent returns, moving whole classes of errors from runtime to write-time.”

Core Concepts

Before diving into code, let’s establish the key PydanticAI primitives:

1. Agent: The core abstraction, parameterized by dependencies type and output type
2. Output Types: Pydantic models that define validated response structures
3. Tools: Functions the LLM can call, with typed arguments and return values
4. Capabilities: Reusable bundles of instructions, tools, and hooks
5. Dependency Injection: Type-safe passing of context into agents and tools

Building a Governed Agent: Step by Step

Step 1: Define Your Output Type

Start with what you want back. Define a Pydantic model for your agent’s output.

# models/output.py
from typing import Literal
from pydantic import BaseModel, Field


class CustomerSupportOutput(BaseModel):
    """Structured output for customer support agent."""
    response: str = Field(description="The response to send to the customer")
    action: Literal["reply", "escalate", "transfer", "close"] = Field(
        description="The action to take"
    )
    priority: Literal["low", "medium", "high", "critical"] = Field(
        description="Priority level of the request"
    )
    sentiment: float = Field(
        ge=-1.0, le=1.0,
        description="Customer sentiment score from -1 (negative) to 1 (positive)"
    )
    requires_human: bool = Field(
        description="Whether this requires human intervention"
    )
    category: str = Field(
        description="Issue category for routing"
    )


class FraudCheckOutput(BaseModel):
    """Structured output for fraud detection agent."""
    is_suspicious: bool
    confidence: float = Field(ge=0.0, le=1.0)
    risk_factors: list[str] = Field(default_factory=list)
    recommended_action: Literal["approve", "review", "block"]
    explanation: str


class DocumentAnalysisOutput(BaseModel):
    """Structured output for document analysis agent."""
    summary: str
    key_points: list[str]
    entities: list[dict[str, str]]
    sentiment: Literal["positive", "negative", "neutral"]
    word_count: int = Field(ge=0)
    language: str

Step 2: Create the Agent with Dependencies

PydanticAI agents accept typed dependencies—database connections, API clients, configuration—that are injected into tools and instructions.

# agents/support_agent.py
from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
from models.output import CustomerSupportOutput


@dataclass
class SupportDependencies:
    """Dependencies injected into the support agent."""
    customer_id: str
    db_connection: any  # Your database connection
    api_client: any  # Your API client
    max_response_length: int = 500


support_agent = Agent(
    "openai:gpt-4",
    deps_type=SupportDependencies,
    output_type=CustomerSupportOutput,
    instructions=(
        "You are a customer support agent for a SaaS company. "
        "Analyze the customer's message and provide a structured response. "
        "Always categorize the issue and assess priority. "
        "Escalate critical issues immediately."
    ),
)

Step 3: Define Typed Tools

Tools are Python functions that the LLM can call. PydanticAI automatically generates the tool schema from the function signature and validates arguments.

# tools/support_tools.py
from pydantic_ai import RunContext
from agents.support_agent import support_agent


@support_agent.tool
async def get_customer_history(
    ctx: RunContext[SupportDependencies],
    days_back: int = 30,
) -> dict:
    """Retrieve customer interaction history for the past N days.
    
    Args:
        days_back: Number of days to look back (default: 30)
    """
    customer_id = ctx.deps.customer_id
    db = ctx.deps.db_connection
    
    # Your database query here
    history = await db.query(
        "SELECT * FROM interactions "
        "WHERE customer_id = $1 AND created_at > NOW() - INTERVAL '%s days' "
        "ORDER BY created_at DESC",
        customer_id, days_back,
    )
    
    return {
        "total_interactions": len(history),
        "recent_issues": [h["subject"] for h in history[:5]],
        "satisfaction_trend": calculate_trend(history),
    }


@support_agent.tool
async def check_subscription_status(
    ctx: RunContext[SupportDependencies],
) -> dict:
    """Check the customer's current subscription and billing status."""
    customer_id = ctx.deps.customer_id
    api = ctx.deps.api_client
    
    subscription = await api.get(f"/customers/{customer_id}/subscription")
    
    return {
        "plan": subscription["plan"],
        "status": subscription["status"],
        "next_billing_date": subscription["next_billing_date"],
        "overdue": subscription.get("overdue_amount", 0) > 0,
    }


@support_agent.tool
async def create_support_ticket(
    ctx: RunContext[SupportDependencies],
    title: str,
    description: str,
    priority: str,
) -> str:
    """Create a support ticket for issues requiring follow-up.
    
    Args:
        title: Brief title of the issue
        description: Detailed description of the problem
        priority: Priority level (low, medium, high, critical)
    """
    api = ctx.deps.api_client
    
    ticket = await api.post("/tickets", json={
        "customer_id": ctx.deps.customer_id,
        "title": title,
        "description": description,
        "priority": priority,
        "source": "ai_agent",
    })
    
    return f"Ticket created: {ticket['id']}"


@support_agent.tool
async def escalate_to_human(
    ctx: RunContext[SupportDependencies],
    reason: str,
    urgency: str,
) -> str:
    """Escalate the conversation to a human agent.
    
    Args:
        reason: Reason for escalation
        urgency: How urgent the escalation is (low, medium, high, critical)
    """
    api = ctx.deps.api_client
    
    await api.post("/escalations", json={
        "customer_id": ctx.deps.customer_id,
        "reason": reason,
        "urgency": urgency,
        "context": "AI agent escalation",
    })
    
    return "Escalated to human agent. Please hold."

Step 4: Dynamic Instructions

Instructions can be dynamic, using dependency injection to personalize the agent’s behavior.

# agents/support_agent.py (continued)
from pydantic_ai import Agent, RunContext


@support_agent.instructions
async def add_customer_context(ctx: RunContext[SupportDependencies]) -> str:
    """Add customer context to the agent's instructions."""
    db = ctx.deps.db_connection
    customer = await db.get_customer(ctx.deps.customer_id)
    
    return (
        f"Customer: {customer['name']} ({customer['email']})\n"
        f"Plan: {customer['plan']}\n"
        f"Tenure: {customer['tenure_months']} months\n"
        f"Previous issues: {customer['issue_count']}\n"
        f"Lifetime value: ${customer['ltv']:.2f}"
    )

Step 5: Using Capabilities for Reusable Components

Capabilities bundle related tools and instructions into reusable units.

# capabilities/billing.py
from pydantic_ai import Capability, RunContext


billing_capability = Capability[SupportDependencies](
    id="billing",
    description="Tools for checking billing status, processing refunds, and managing subscriptions.",
)


@billing_capability.tool
async def check_payment_history(
    ctx: RunContext[SupportDependencies],
    months: int = 6,
) -> list[dict]:
    """Get payment history for the specified number of months.
    
    Args:
        months: Number of months of history to retrieve
    """
    customer_id = ctx.deps.customer_id
    db = ctx.deps.db_connection
    
    payments = await db.query(
        "SELECT * FROM payments "
        "WHERE customer_id = $1 "
        "ORDER BY created_at DESC LIMIT $2",
        customer_id, months * 10,
    )
    
    return [
        {
            "date": p["created_at"],
            "amount": p["amount"],
            "status": p["status"],
        }
        for p in payments
    ]


@billing_capability.tool
async def process_refund(
    ctx: RunContext[SupportDependencies],
    payment_id: str,
    amount: float,
    reason: str,
) -> dict:
    """Process a refund for a specific payment.
    
    Args:
        payment_id: The payment ID to refund
        amount: Refund amount in dollars
        reason: Reason for the refund
    """
    api = ctx.deps.api_client
    
    refund = await api.post("/refunds", json={
        "payment_id": payment_id,
        "amount": amount,
        "reason": reason,
        "customer_id": ctx.deps.customer_id,
    })
    
    return {"refund_id": refund["id"], "status": refund["status"]}


# Create agent with capabilities
support_agent_with_billing = Agent(
    "openai:gpt-4",
    deps_type=SupportDependencies,
    output_type=CustomerSupportOutput,
    capabilities=[billing_capability],
)

Step 6: Running the Agent

# main.py
import asyncio
from agents.support_agent import support_agent
from agents.support_agent import SupportDependencies
from models.output import CustomerSupportOutput


async def handle_support_request(
    customer_id: str,
    message: str,
) -> CustomerSupportOutput:
    """Handle a customer support request."""
    
    # Create dependencies
    deps = SupportDependencies(
        customer_id=customer_id,
        db_connection=your_db_connection,
        api_client=your_api_client,
    )
    
    # Run the agent
    result = await support_agent.run(message, deps=deps)
    
    # result.output is typed as CustomerSupportOutput
    output = result.output
    
    print(f"Response: {output.response}")
    print(f"Action: {output.action}")
    print(f"Priority: {output.priority}")
    print(f"Sentiment: {output.sentiment}")
    print(f"Requires human: {output.requires_human}")
    
    # Type-safe access—your IDE knows all fields
    if output.action == "escalate":
        print("Escalating to human agent...")
    
    return output


# Run it
async def main():
    result = await handle_support_request(
        customer_id="cust_12345",
        message="I've been charged twice for my subscription this month. "
                "This is the third time this has happened. I'm very frustrated.",
    )


if __name__ == "__main__":
    asyncio.run(main())

Testing with Type Safety

PydanticAI’s test model runs offline without calling an LLM, making unit tests fast and deterministic.

# tests/test_support_agent.py
import pytest
from pydantic_ai import Agent
from pydantic_ai.testing import TestModel
from agents.support_agent import support_agent
from models.output import CustomerSupportOutput


@pytest.fixture
def test_agent():
    """Create an agent with test model for offline testing."""
    test_model = TestModel()
    return support_agent.override(model=test_model)


async def test_support_agent_escalation(test_agent):
    """Test that high-priority issues trigger escalation."""
    result = await test_agent.run(
        "I want to cancel my account immediately!",
        deps=SupportDependencies(
            customer_id="test_123",
            db_connection=mock_db,
            api_client=mock_api,
        ),
    )
    
    # Output is typed—no string parsing needed
    assert isinstance(result.output, CustomerSupportOutput)
    assert result.output.requires_human is True
    assert result.output.priority in ["high", "critical"]
    assert result.output.action in ["escalate", "transfer"]


async def test_support_agent_fraud_detection():
    """Test fraud detection agent with structured output."""
    result = await fraud_agent.run(
        "Someone accessed my account from a different country",
        deps=FraudDependencies(customer_id="test_456"),
    )
    
    assert result.output.is_suspicious is True
    assert result.output.confidence > 0.5
    assert len(result.output.risk_factors) > 0
    assert result.output.recommended_action == "review"

Observability with Pydantic Logfire

PydanticAI is OpenTelemetry-native. One line of setup lights up tracing and debugging.

# observability/setup.py
import logfire
from pydantic_ai import Agent


def setup_observability():
    """Configure Pydantic Logfire for agent observability."""
    logfire.configure()
    logfire.instrument_pydantic_ai()  # Instruments all agents


def create_traced_agent():
    """Create an agent with built-in tracing."""
    setup_observability()
    
    agent = Agent(
        "openai:gpt-4",
        deps_type=SupportDependencies,
        output_type=CustomerSupportOutput,
        instructions="You are a customer support agent.",
    )
    
    return agent

Best Practices

1. Define Output Types First: Start with your Pydantic models before writing agent logic. This clarifies requirements and catches design issues early.

2. Use Dependency Injection: Never hardcode connections or configuration. Typed dependencies make testing trivial and agents composable.

3. Validate at Boundaries: Let Pydantic handle LLM output validation. Don’t add manual parsing—trust the framework.

4. Leverage Capabilities: Group related tools and instructions into capabilities. They’re reusable across agents and easier to test in isolation.

5. Instrument Everything: OpenTelemetry tracing is one line of code. Don’t deploy without it.

Common Pitfalls

• Overly complex output types: Start simple. Add fields only when you need them. Complex schemas confuse LLMs.
• Missing tool docstrings: PydanticAI uses docstrings for tool descriptions. Missing docstrings mean the LLM doesn’t know what the tool does.
• Ignoring validation errors: When Pydantic rejects LLM output, the agent retries. Monitor retry rates—they indicate prompt or schema issues.
• Skipping type annotations: Every tool argument needs a type annotation. Untyped arguments generate poor schemas.

Conclusion

Governed agents aren’t a luxury—they’re a requirement for production deployment. PydanticAI makes governance natural by making type safety part of the agent definition. When your IDE, type checker, and LLM all agree on what your agent returns, entire classes of production errors disappear.

The key insight: validation isn’t overhead. It’s the thing that makes your agent reliable. Every Pydantic model you define is a contract between your agent and the rest of your system.

Next steps:

• Start with the PydanticAI quickstart
• Define output types for your current agent’s responses
• Add typed tools with proper docstrings
• Set up Pydantic Logfire for observability
• Explore the PydanticAI Harness for advanced capabilities