Build a Pydantic AI agent with typed tools and DI

Introduction

When you wire a raw LLM directly into a FastAPI handler, every tool call becomes a runtime gamble — the model can hallucinate a string where you expected an integer, drop a required field, or return free-form prose where you needed structured JSON, and the silent versions of those failures (a malformed project_id that quietly scopes a query to the wrong tenant) are the most expensive. By the end of this lesson you will be able to create a Pydantic AI agent with typed dependencies and validated tool calls — defining a TaskResult model, wiring a request-scoped TaskDeps dataclass into the agent, registering tools that receive injected context, and returning guaranteed-shape output to a FastAPI handler.

Key Terminology

• Agent: the pydantic_ai.Agent instance that binds a model, system prompt, dependency type, and result_type into a runnable agentic loop.
• result_type: the Pydantic model the agent's final answer must parse into; on parse failure the framework raises ValidationError and may retry.
• RunContext[TaskDeps]: the typed parameter Pydantic AI injects into each tool call, giving tools access to the per-request dependencies (DB session, user ID, limits).

Concepts

Key Design Decisions for Production Agents

• result_type vs. free-form text: Always declare a result_type for agents that feed structured data into downstream systems (databases, APIs, other agents). Use free-form text (omit result_type) only for chat-style agents where the output is displayed directly to humans.

• Tool error handling: Return error dictionaries from tools instead of raising exceptions. The LLM can reason about {"error": "Task not found"} and try a different approach, but a Python exception kills the agentic loop unless you add custom exception handlers.

• Dependency scoping: Create a new TaskDeps instance per request, never per application. The AsyncSession must be request-scoped to prevent connection leaks, and the user_id must reflect the authenticated user from the current request, not a cached value.

• Message history serialization: Pydantic AI's message objects are serializable via result.new_messages() after each run. Store these in your session backend (Redis, database) and pass them back via message_history on subsequent requests. This avoids re-processing the full conversation with every turn while preserving tool call results the model has already seen.

• Iteration limits: Set max_tool_calls on the TaskDeps or pass model_settings={"max_tokens": 4096} to prevent runaway agents. In production, an agent that exceeds 10 tool calls per request is almost always stuck in a loop—fail fast and return an error rather than burning tokens.

These patterns compose directly with the MCP protocol covered earlier in this chapter. You can register an MCP-discovered tool as a Pydantic AI tool by wrapping the MCP call_tool invocation in a typed function with a RunContext parameter, giving you MCP's dynamic discovery with Pydantic AI's type safety and streaming.

Code Walkthrough

Architecture of a Pydantic AI Agent

Before writing code, you need to understand how Pydantic AI's component model maps to the agentic loop patterns covered earlier in this chapter. The agent sits at the center of a typed pipeline: dependencies flow in, tools execute with validated inputs, and a structured result flows out.

This flowchart maps the full request lifecycle of a Pydantic AI agent served behind a FastAPI handler. A client request triggers Agent.run or Agent.run_stream, entering an LLM decision loop where tool calls pass through Pydantic validation before execution—invalid arguments route back to the LLM for self-correction. The loop continues until the LLM produces a final answer, validated against result_type and returned as a TaskResult. The optional SSE stream path delivers intermediate steps to the client in real time.

Loading diagram...

This diagram captures a critical detail: validation occurs at three points—tool argument ingress, tool return egress, and final result emission. The LLM receives validation errors as tool-call responses, which means the model can self-correct malformed arguments without your application code handling retries. The SSE stream path runs in parallel, emitting each tool call and partial result as the agent iterates through its observe-think-act cycle.

Defining Types and Wiring Tools into the Agent

Every Pydantic AI agent declares a result_type that constrains what the model can return as its final answer. This is not a suggestion—if the model's output cannot be parsed into the declared type, the framework raises a ValidationError and optionally retries. For a task management agent, you need a TaskResult model that captures structured output and a TaskDeps dataclass that carries request-scoped services like database sessions and authenticated user context. With those contracts in place, you construct the agent and register tools that receive the dependency object automatically via a RunContext[TaskDeps] parameter—Pydantic AI's dependency injection mechanism, replacing the service locator or global singleton patterns you might use elsewhere.

The following code does both halves of the wiring in one pass. It defines the TaskResult Pydantic model (task summary, priority, owner, sub-tasks) and the TaskDeps dataclass (AsyncSession, user_id, project_id, iteration limit). It then instantiates pydantic_ai.Agent with the openai:gpt-4o model, declares result_type=TaskResult and deps_type=TaskDeps, and registers three tools — list_tasks, get_task_detail, assign_task — each receiving a RunContext[TaskDeps] as its first parameter so it can reach the DB session and user context without global state.

Code snippetpython
1from dataclasses import dataclass
2from pydantic import BaseModel, Field
3from pydantic_ai import Agent, RunContext
4from sqlalchemy import text
5from sqlalchemy.ext.asyncio import AsyncSession
6
7class SubTask(BaseModel):
8    title: str = Field(description="Short actionable title for the sub-task")
9    effort_hours: float = Field(ge=0.5, le=200, description="Estimated effort in hours")
10    status: str = Field(default="pending", pattern="^(pending|in_progress|done)$")
11
12class TaskResult(BaseModel):
13    summary: str = Field(min_length=10, max_length=500, description="Task summary")
14    priority: str = Field(pattern="^(low|medium|high|critical)$")
15    owner: str = Field(description="Username of the assigned owner")
16    sub_tasks: list[SubTask] = Field(default_factory=list, max_length=20)
17    confidence: float = Field(ge=0.0, le=1.0, description="Agent confidence score")
18
19@dataclass
20class TaskDeps:
21    db: AsyncSession
22    user_id: str
23    project_id: int
24    max_tool_calls: int = 10
25
26task_agent = Agent(
27    "openai:gpt-4o",
28    result_type=TaskResult,
29    deps_type=TaskDeps,
30    system_prompt=(
31        "You are a project task manager. Analyze the user's request, "
32        "query existing tasks, and return a structured TaskResult. "
33        "Always verify task ownership before making changes."
34    ),
35)
36
37@task_agent.tool
38async def list_tasks(ctx: RunContext[TaskDeps], status_filter: str = "all") -> list[dict]:
39    """List tasks for the current project, optionally filtered by status."""
40    query = "SELECT id, title, status, owner FROM tasks WHERE project_id = :pid"
41    params = {"pid": ctx.deps.project_id}
42    if status_filter != "all":
43        query += " AND status = :status"
44        params["status"] = status_filter
45    result = await ctx.deps.db.execute(text(query), params)
46    rows = result.fetchall()
47    return [{"id": r.id, "title": r.title, "status": r.status, "owner": r.owner} for r in rows]
48
49@task_agent.tool
50async def get_task_detail(ctx: RunContext[TaskDeps], task_id: int) -> dict:
51    """Fetch full details for a single task by ID."""
52    result = await ctx.deps.db.execute(
53        text("SELECT * FROM tasks WHERE id = :tid AND project_id = :pid"),
54        {"tid": task_id, "pid": ctx.deps.project_id},
55    )
56    row = result.fetchone()
57    if row is None:
58        return {"error": f"Task {task_id} not found in project {ctx.deps.project_id}"}
59    return dict(row._mapping)
60
61@task_agent.tool
62async def assign_task(ctx: RunContext[TaskDeps], task_id: int, new_owner: str) -> dict:
63    """Reassign a task to a new owner. Only the current owner or admins can reassign."""
64    row = await ctx.deps.db.execute(
65        text("SELECT owner FROM tasks WHERE id = :tid AND project_id = :pid"),
66        {"tid": task_id, "pid": ctx.deps.project_id},
67    )
68    task = row.fetchone()
69    if task is None:
70        return {"error": "Task not found"}
71    if task.owner != ctx.deps.user_id:
72        return {"error": "Permission denied: only the current owner can reassign"}
73    await ctx.deps.db.execute(
74        text("UPDATE tasks SET owner = :owner WHERE id = :tid"),
75        {"owner": new_owner, "tid": task_id},
76    )
77    await ctx.deps.db.commit()
78    return {"success": True, "task_id": task_id, "new_owner": new_owner}

• Imports: Bring in dataclass for lightweight dependency containers, BaseModel/Field for validated schemas, Agent/RunContext from pydantic_ai, and SQLAlchemy's AsyncSession/text for parameterized async queries.
• SubTask and TaskResult: Define constrained fields the LLM sees in the JSON schema—effort_hours rejects unreasonable estimates via ge=0.5/le=200, status/priority use regex patterns instead of Enums to keep the schema simple, summary enforces a minimum length to prevent trivially short answers, and sub_tasks caps at 20 entries to prevent runaway generation.
• TaskDeps: A plain dataclass (not a Pydantic model) because dependencies are constructed by your application code, never parsed from LLM output. max_tool_calls gives you a per-request iteration limit without hardcoding.
• Agent(...) instantiation: The model string "openai:gpt-4o" uses Pydantic AI's provider prefix syntax ("anthropic:claude-sonnet-4-20250514", "gemini:gemini-1.5-pro" also work). result_type=TaskResult triggers final-output validation; deps_type=TaskDeps declares the dependency contract injected into every tool.
• list_tasks: Parameterized query scoped to ctx.deps.project_id. The status_filter default of "all" appears in the schema so the model knows it's optional. Returning list[dict] is fine—Pydantic AI serializes it to JSON for the model.
• get_task_detail: When the row is missing, returns an error dictionary rather than raising. This lets the LLM reason about the missing ID and try list_tasks to recover, instead of crashing the agentic loop.
• assign_task: Enforces authorization by checking task.owner != ctx.deps.user_id—a security boundary the LLM cannot bypass regardless of how it phrases the tool call. Commits the transaction explicitly so uncommitted changes don't accumulate across tool invocations.

Streaming Multi-Turn Conversations with Tool Call Visibility

Production agents need to stream intermediate steps to the frontend so users see tool calls happening in real time rather than waiting for a final answer. Pydantic AI's run_stream method returns an async context manager that yields partial results, including tool call names and arguments, as the agent iterates through its observe-think-act cycle. Integrating this with FastAPI's StreamingResponse and Server-Sent Events gives you a complete real-time pipeline.

The following code defines a FastAPI endpoint that accepts a user message, constructs the TaskDeps dependency object from the request context, and runs the agent in streaming mode. It uses an async generator to yield SSE-formatted events for each tool call and the final TaskResult. The message_history parameter enables multi-turn conversations by passing previous messages back into the agent, so the model retains context across requests without re-processing the entire conversation.

1from fastapi import FastAPI, Depends
2from fastapi.responses import StreamingResponse
3from pydantic_ai.messages import ToolCallPart, TextPart
4import json
5
6app = FastAPI()
7
8async def stream_agent_response(user_msg: str, deps: TaskDeps, history: list | None):
9    async with task_agent.run_stream(
10        user_msg,
11        deps=deps,
12        message_history=history,
13    ) as stream:
14        async for message in stream.stream_structured():
15            for part in message.parts:
16                if isinstance(part, ToolCallPart):
17                    event_data = {
18                        "type": "tool_call",
19                        "tool": part.tool_name,
20                        "args": part.args_as_dict(),
21                    }
22                    yield f"event: tool_call\ndata: {json.dumps(event_data)}\n\n"
23                elif isinstance(part, TextPart):
24                    yield f"event: text\ndata: {json.dumps({'chunk': part.content})}\n\n"
25
26        result = stream.result()
27        yield f"event: result\ndata: {result.data.model_dump_json()}\n\n"
28        yield f"event: usage\ndata: {json.dumps({'tokens': result.usage().total_tokens})}\n\n"
29
30@app.post("/agent/task")
31async def run_task_agent(
32    request: dict,
33    db: AsyncSession = Depends(get_db_session),
34):
35    deps = TaskDeps(db=db, user_id=request["user_id"], project_id=request["project_id"])
36    history = request.get("message_history")
37    return StreamingResponse(
38        stream_agent_response(request["message"], deps, history),
39        media_type="text/event-stream",
40    )

Do's and Don'ts

Do's

1. ✓Do declare result_type=TaskResult on the pydantic_ai.Agent constructor — if the model's output cannot be parsed into the declared Pydantic model, the framework raises ValidationError and optionally retries, which is what prevents the LLM from returning free-form prose or a silently malformed project_id where a structured TaskResult is required.
2. ✓Do accept a RunContext[TaskDeps] as the first parameter of every registered tool — this is Pydantic AI's dependency injection mechanism and the only way list_tasks, get_task_detail, and assign_task can reach the request-scoped AsyncSession, user_id, and project_id without module-level globals or a service locator that leaks across concurrent requests.
3. ✓Do use Agent.run_stream instead of Agent.run when your FastAPI handler needs SSE visibility — streaming emits each tool call and partial result as an intermediate step so clients observe the observe-think-act cycle in real time, rather than blocking until the final TaskResult is validated and returned.

Don'ts

1. ✗Don't carry request-scoped services like AsyncSession or user_id in module-level globals instead of TaskDeps — RunContext[TaskDeps] exists precisely to scope those values to a single agent invocation, and bypassing it risks cross-request contamination when concurrent FastAPI handlers share the same session or user context.
2. ✗Don't omit Field constraints on TaskResult and nested models like SubTask — without bounds like ge=0.5, le=200, or pattern="^(pending|in_progress|done)$", the three-point validation (tool argument ingress, tool return egress, final result emission) degrades to a shape-only check that lets an out-of-range effort_hours or an unrecognized status string pass silently through to the database.
3. ✗Don't add your own retry loop around Agent.run for tool-call validation failures — the framework already returns ValidationError back to the LLM as a tool-call response so it can self-correct malformed arguments in the next reasoning step; wrapping Agent.run in external retries duplicates that logic and can re-invoke the model with a fresh context that discards the self-correction signal.