Understanding Pydantic AI through a case handling agent

Why Pydantic AI belongs in this series

#

The first two articles in this series explored Akgents itself: first the actor-based core, then the team layer that makes those actors durable. There is another foundation underneath the framework that deserves the same attention. Akgents uses Pydantic AI as the basis for its model-facing agent capabilities. If I want to understand what Akgents adds, I first need to understand the essentials it builds on.

That is why this episode steps down one layer. I built a small case-prioritisation application using Pydantic AI directly. It starts with the smallest useful agent, then adds typed dependencies, an async tool, multiple structured outcomes, validation with retries, an agent handoff, and observability. The sample now also exposes its case repository through MCP and uses it to explore Pydantic AI Capabilities. These concepts are small enough to see in one project, but they are also the ingredients that reappear when an agent framework grows around them.

The complete example is available in the Pydantic AI tryout repository. You can keep it open while reading or clone it and run the code yourself.

Start with the smallest agent

#

Pydantic AI’s Agent combines a model with the instructions, tools, dependencies, and output contract needed for a run. The smallest agent in my project only needs a model and one instruction:

from pydantic_ai import Agent

def create_friendly_agent():
    return Agent(
        name="Friendly Agent",
        description="A friendly agent that helps users with their tasks.",
        model="openai:gpt-5.6-luna",
        instructions="Always reply with a friendly message; you can never answer in bad language.",
    )

One detail is worth making explicit: an agent run is not automatically a chat session. I provide a prompt, the agent performs a run, and it returns an AgentRunResult. A conversation only emerges when I deliberately pass message history into a later run. That distinction matters because it keeps state and continuity under application control instead of hiding them behind the Agent object.

The friendly agent is useful as a first connection check, but it does not yet show why I would put Pydantic AI underneath an application. For that, I need a small domain problem.

A case agent with a real boundary

#

The next agent decides the priority of a support case. A priority ranges from 1 to 5, where 1 is the highest. Two small types describe the data that moves through the run. Case is the record the repository stores, and CaseResponse is the successful result the agent returns. Notice that they do not have to be the same kind of type: Case is a Pydantic BaseModel so the model can read and validate it, while CaseResponse is a plain dataclass because only application code constructs it:

from dataclasses import dataclass

from pydantic import BaseModel


class Case(BaseModel):
    case_id: str
    case_description: str
    case_creation_date: str
    case_creator: str
    case_priority: int = 0


@dataclass
class CaseResponse:
    case: Case
    message: str

The agent needs a case repository. For this sample, I use a simple in-memory repository. For demonstrating IO operations that take time, its methods are asynchronous and wait briefly to imitate a remote case system:

class CaseRepository:
    def __init__(self):
        # The repository in the sample project seeds a few more cases,
        # including case_1 (the default) and case_4 (already prioritized).
        self.cases = {
            "case_3": Case(
                case_id="case_3",
                case_description="The coffee machine broke down, no coffee for our software engineers.",
                case_creation_date="2023-09-17",
                case_creator="Bob Johnson",
            ),
        }

    async def get_case(self, case_id: str) -> Case:
        await asyncio.sleep(0.1)
        found_case = self.cases.get(case_id)
        if found_case is None:
            raise ValueError(f"Case with ID {case_id} not found.")
        return found_case.model_copy(deep=True)

    async def update_case_priority(self, case_id: str, priority: int) -> Case:
        await asyncio.sleep(0.1)
        found_case = self.cases.get(case_id)
        if found_case is None:
            raise ValueError(f"Case with ID {case_id} not found.")
        found_case.case_priority = priority
        return found_case.model_copy(deep=True)

The model should be allowed to reason about the case, but it should not be allowed to choose which case it is working on or how to connect to the case store. Those are application decisions. This is where dependencies enter the design.

Keep application context in dependencies

#

Pydantic AI’s dependency injection supplies data and services to instructions, tools, and output processing. In my sample, the dependency object contains the fixed case ID, the current user’s name, and the repository:

from dataclasses import dataclass


@dataclass
class CaseAgentDeps:
    case_id: str
    user_name: str
    case_repository: CaseRepository

I declare that type on the agent with deps_type=CaseAgentDeps. During a run, Pydantic AI exposes the actual instance through RunContext[CaseAgentDeps]. This gives the callbacks a typed path to application state without adding repository details or a changeable case ID to the prompt.

Dynamic instructions can use that same context:

@agent.instructions
def personalize(ctx: RunContext[CaseAgentDeps]):
    return (
        f"You are handling the case together with {ctx.deps.user_name}."
        f"Determine the priority with the case details, then finish the run by storing the priority."
        f"When the case must not be prioritized, finish with a {PriorityRejected.__name__} instead."
    )

The distinction between static and dynamic instructions is useful. Stable behaviour belongs in the agent definition; run-specific context can be calculated from the dependencies. The Pydantic AI documentation also explains that dynamic @agent.instructions functions are reevaluated for every run, which is exactly what I want for a case ID supplied at runtime.

Let a tool retrieve the case

#

The model needs case details before it can prioritise anything. A function tool gives it that capability:

@agent.tool()
async def get_case_details(ctx: RunContext[CaseAgentDeps]) -> Case:
    return await ctx.deps.case_repository.get_case(ctx.deps.case_id)

The @agent.tool decorator is the right fit because the function needs RunContext. A tool that does not need the context can use @agent.tool_plain instead. Pydantic AI reads the signature and docstring to build the tool definition presented to the model, then validates the arguments before calling the Python function. The function tools documentation covers the different registration styles.

This tool is deliberately asynchronous. A real repository call would spend most of its time waiting for a network or database response, so async def and await let the event loop do other work in the meantime. Pydantic AI can also run synchronous tools, but those are offloaded to worker threads.

There is an important boundary here. The model can decide when it needs the case, but it cannot provide another case_id: the tool has no such argument. It always reads ctx.deps.case_id. A typed dependency is therefore more than a convenience; it helps shape what the model is and is not allowed to control.

Call the agent with its dependencies

#

The application supplies a user prompt and a dependency instance when it starts the run:

result = await agent.run(
    user_prompt=(
        "Your goal is to prioritise the case, but only if it is not yet prioritized."
        "You can determine a case is prioritized by checking if case_priority is greater than 0."
        "A case can have a priority of 1 to 5, where 1 is highest priority."
        "Explain on what basis you prioritize the case."
        "Use your tools to get the case details and determine if it is prioritized."
        "Finish by storing the priority you determined, or by rejecting the prioritization."
    ),
    deps=CaseAgentDeps(
        case_id=case_id,
        user_name="Jettro",
        case_repository=CaseRepository(),
    ),
)

The prompt describes the goal and the decision policy. The dependency object carries the case_id to handle. The name of the current user is fixed, but should be read from an authentication context. The object also contains an instance of the CaseRepository.

Turn the final action into an output function

#

My first version exposed both reading and storing as ordinary tools, then asked the model for a CaseResponse afterwards. That worked, but it left an awkward gap: the model could call the storage tool with an invalid priority, and after the write it still had to invent a final response that matched the updated data.

Pydantic AI’s output functions fit this operation better. The model calls an output function with validated arguments, the function performs the final processing, its return value becomes the result, and the run ends. Unlike the return value of a normal tool, that result is not passed back to the model.

async def case_to_prioritize(
    ctx: RunContext[CaseAgentDeps], new_priority: int
) -> Case:
    if not 1 <= new_priority <= 5:
        raise ModelRetry(
            f"A priority of {new_priority} is not allowed, pick a number from 1 to 5."
        )

    case = await ctx.deps.case_repository.get_case(ctx.deps.case_id)
    if case.case_priority > 0:
        raise ModelRetry(
            f"Case {case.case_id} already has priority {case.case_priority}, "
            f"return a {PriorityRejected.__name__} instead."
        )
    return case


async def store_priority(
    ctx: RunContext[CaseAgentDeps], new_priority: int, motivation: str
) -> CaseResponse:
    """Store the priority you determined for the case and finish the run."""
    case = await case_to_prioritize(ctx, new_priority)
    updated_case = await ctx.deps.case_repository.update_case_priority(
        case.case_id, new_priority
    )
    return CaseResponse(case=updated_case, message=motivation)

The write and the response now happen in one final operation. The function signature gives new_priority and motivation a typed home, while the docstring tells the model when to choose this output.

ModelRetry is the feedback channel when the model’s proposed arguments are not acceptable. The message is sent back to the model so it can try again, bounded by a retry budget. That is a much better fit than either silently accepting priority 9 or turning a correctable model mistake into an application failure.

Model valid alternatives as different output types

#

Not every case should be updated. If it already has a priority, rejection is a valid business outcome rather than an exception. I represent it explicitly:

class PriorityRejected(BaseModel):
    """Use me when the case must not be prioritized, for example because it already has a priority."""

    reason: str

The agent gets both possible outcomes:

agent = Agent[CaseAgentDeps, CaseResponse | PriorityRejected](
    model="openai:gpt-5.6-luna",
    deps_type=CaseAgentDeps,
    output_type=[
        ToolOutput(store_priority, name="store_priority", max_retries=2),
        PriorityRejected,
    ],
    retries={"output": 2},
)

In the default tool-output mode, Pydantic AI registers the options as output tools. ToolOutput lets me give the function a stable tool name and a specific retry limit. The explicit generic type on Agent helps static type checkers understand the combination of an async output function and a second structured type.

The caller no longer has to inspect a message to discover what happened. It can branch on the type:

match result.output:
    case CaseResponse() as response:
        print(f"Priority {response.case.case_priority}: {response.message}")
    case PriorityRejected() as rejected:
        print(f"Not prioritized: {rejected.reason}")

That separation identifies three kinds of failures the application runs into:

1. A model mistake that can be corrected raises ModelRetry and consumes retry budget.
2. A valid negative outcome is data, such as PriorityRejected.
3. A broken run remains an exception for application code to handle.

This is a useful design rule beyond this example: expected domain outcomes belong in the type system, not in exception handling and not hidden inside free-form prose.

Pydantic AI supports other output transport modes too. NativeOutput asks a compatible provider to enforce the schema, while PromptedOutput adds the schema to the instructions for a model without tool calling. Those modes change how the output reaches the application; they do not change the domain types themselves. For this example, the default tool-output mode is the clearest fit.

Use an output function as a handoff point

#

An output function can do more than validate or store data. It can also start another agent. I added a small message agent that turns the internal prioritisation motivation into a friendly response for the reporter:

message_agent = Agent(
    model="openai:gpt-5.6-luna",
    output_type=str,
    defer_model_check=True,
    instructions=(
        "You turn an internal case note into a short, friendly message "
        "for the reporter of the case."
    ),
)

The defer_model_check=True flag tells Pydantic AI not to validate the model identifier when the agent is constructed, but to wait until the first run. That keeps import-time cheap and avoids a failure when the model string is only resolved later.

The handoff variant stores the case and then runs that second agent:

async def hand_off_to_message_agent(
    ctx: RunContext[CaseAgentDeps], new_priority: int, motivation: str
) -> CaseResponse:
    """Store the priority and let the message agent phrase the reply to the reporter."""
    case = await case_to_prioritize(ctx, new_priority)
    updated_case = await ctx.deps.case_repository.update_case_priority(
        case.case_id, new_priority
    )

    messages = ctx.messages[:-1]
    try:
        result = await message_agent.run(motivation, message_history=messages)
    except UnexpectedModelBehavior as exc:
        if (cause := exc.__cause__) and isinstance(cause, ModelRetry):
            raise ModelRetry(f"The message agent failed: {cause.message}") from exc
        raise

    return CaseResponse(case=updated_case, message=result.output)

ctx.messages[:-1] removes the output-tool call before the history is passed on. That call is a control message for the outer agent, not part of the conversation the message agent should receive. If the inner run exhausts a retryable failure, the wrapper translates it back into ModelRetry so the outer agent gets another bounded attempt.

Both store_priority and hand_off_to_message_agent share the same signature, so a single factory can build either version of the agent. The output function becomes a parameter, which keeps the agent configuration in one place and makes the handoff a drop-in replacement for the plain store:

def build_case_agent(
    output_function: Callable[[RunContext[CaseAgentDeps], int, str], Awaitable[CaseResponse]]
) -> Agent[CaseAgentDeps, CaseResponse | PriorityRejected]:
    agent = Agent[CaseAgentDeps, CaseResponse | PriorityRejected](
        model="openai:gpt-5.6-luna",
        deps_type=CaseAgentDeps,
        output_type=[ToolOutput(output_function, name="store_priority", max_retries=2), PriorityRejected],
        retries={"output": 2},
    )
    # register @agent.instructions and @agent.tool here
    return agent


def create_case_agent():
    return build_case_agent(store_priority)


def create_case_agent_with_hand_off():
    return build_case_agent(hand_off_to_message_agent)

The output_type I showed earlier is exactly what this factory produces; passing a different output function is all it takes to switch from storing the priority to handing the reply off to the message agent.

This is a compact handoff, not yet a durable team. That distinction connects directly to the wider series: Pydantic AI supplies the model interaction and typed run mechanics, while Akgents adds the actor lifecycle, team structure, messaging, and persistence needed for longer-lived collaboration.

Bundle agent behaviour in Capabilities

#

The agent constructor has gradually accumulated instructions, dependencies, tools, output handling, and model configuration. Those individual arguments remain useful, but they are not always the best unit for reuse. Pydantic AI defines a Capability as a reusable, composable unit of agent behaviour. A Capability can contribute tools or toolsets, instructions, lifecycle hooks, model settings, and even model selection.

That makes a Capability a larger building block than a function tool. A tool gives the model one operation. A Capability can package the behaviour and runtime integration for a complete concern, such as memory, an approval workflow, web search, or access to an MCP server. The agent only needs to include that building block in its capabilities list.

MCP is a useful example because it shows both the lower-level mechanism and the higher-level abstraction. The Model Context Protocol gives AI applications a standard way to connect to external tools and services. Instead of defining get_case_details directly on the agent, I moved the repository boundary into a FastMCP server.

Expose the case repository through MCP

#

The new case_server.py wraps the same CaseRepository used earlier in this article. It exposes three tools, two resource patterns, and one prompt template:

• get_case, update_case_priority, and list_cases are callable tools;
• cases://all and cases://{case_id} expose read-only JSON resources;
• prioritize_case creates a reusable prompt for one case.

The essential server code remains small:

from mcp.server.fastmcp import FastMCP


def create_case_server(repository: CaseRepository | None = None) -> FastMCP:
    repo = repository if repository is not None else CaseRepository()
    server = FastMCP(
        name="Case Repository Server",
        instructions=(
            "MCP server providing access to support cases and "
            "case prioritization tools and resources."
        ),
    )

    @server.tool()
    async def get_case(case_id: str) -> Case:
        """Retrieve details of a support case by its unique case_id."""
        return await repo.get_case(case_id)

    @server.tool()
    async def update_case_priority(case_id: str, priority: int) -> Case:
        """Update the priority of a case, where 1 is highest and 5 is lowest."""
        if not 1 <= priority <= 5:
            raise ValueError(
                f"A priority of {priority} is not allowed, must be between 1 and 5."
            )
        return await repo.update_case_priority(case_id, priority)

    return server

The repository is no longer an implementation detail of one agent. Any MCP client can connect to the server contract. The two agent examples currently consume its tools; the resources and prompt show that the server can expose more than callable functions, even though this sample does not yet feed those primitives into the agent run.

The dependency list reflects that boundary. The project now installs the mcp extra for pydantic-ai-slim, FastMCP for the server, and pins the underlying MCP SDK below its next major version. uv sync --all-groups from the run section installs the complete locked environment.

Connect the server as a toolset

#

The first implementation uses MCPToolset directly. A toolset is a collection of tools that can be attached to an agent in one operation:

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset

from pydantic_ai_tryout.case_server import create_case_server


def create_case_mcp_agent(
    server: FastMCP | None = None,
    model: str = "openai:gpt-5.6-luna",
) -> Agent:
    mcp_server = server if server is not None else create_case_server()
    toolset = MCPToolset(mcp_server)

    return Agent(
        model=model,
        toolsets=[toolset],
        instructions=(
            "You are a case prioritization assistant. "
            "Use the MCP tools to inspect support cases and update their priority when needed. "
            "If a case already has a priority > 0, do not modify it."
        ),
    )

This keeps the MCP integration explicit. Direct MCPToolset is useful when I need lower-level control over the client lifecycle, want to share the same MCP server between several agents, or need advanced transport configuration that does not fit the Capability abstraction.

Make MCP a Capability

#

For the usual agent configuration, Pydantic AI recommends its built-in MCP Capability. The sample wraps the in-process FastMCP server and passes that one object to the agent:

from pydantic_ai import Agent
from pydantic_ai.capabilities import MCP

from pydantic_ai_tryout.case_server import create_case_server


def create_case_mcp_capability(
    server: FastMCP | None = None,
    *,
    url: str | None = None,
) -> MCP:
    if url is not None:
        return MCP(url=url, native=False)

    mcp_server = server if server is not None else create_case_server()
    return MCP(local=mcp_server, native=False)


def create_case_capability_agent(
    server: FastMCP | None = None,
    capability: MCP | None = None,
    model: str = "openai:gpt-5.6-luna",
) -> Agent:
    mcp_capability = (
        capability
        if capability is not None
        else create_case_mcp_capability(server=server)
    )

    return Agent(
        model=model,
        capabilities=[mcp_capability],
        instructions=(
            "You are a case prioritization assistant. "
            "Use the MCP tools to inspect support cases and update their priority when needed. "
            "If a case already has a priority > 0, do not modify it."
        ),
    )

Both examples give the model the same case tools. The difference is the level at which I describe the integration. MCPToolset says attach this MCP client as a collection of tools. capabilities=[MCP(...)] says this agent has the ability to work with this MCP server. The second form gives Pydantic AI one place to coordinate the MCP toolset, transport, and lifecycle, and it can adapt between local and provider-native execution.

The repository deliberately uses native=False. Tool calls are handled by the Python process and sent to either the in-process FastMCP server or the configured remote URL. This keeps credentials, hooks, and Logfire tracing on the application side, and it works with a server that only exists locally.

With an HTTP-accessible MCP server, MCP(url=..., native=True) can opt into a model provider’s native MCP support. The provider then connects to that URL itself. Pydantic AI can retain local execution as a fallback when the selected model does not support native MCP. A provider cannot reach an in-process server, localhost, or a local stdio process, so native execution requires a server endpoint that is reachable from the provider. Authentication tokens and headers belong in runtime configuration or a secret store, never in the committed agent definition.

This is the Capability idea in a concrete form: the business agent does not need to know how the MCP tools become available. It declares the ability it needs, while the Capability owns the runtime integration. That separation is especially valuable when the same behaviour must move between agents, model providers, and deployment environments.

The existing make run command still starts the original dependency-and-tool implementation. The MCP variants are separate agent factories in case_mcp_agent.py and case_capability_agent.py; the sample does not yet expose a command-line switch for choosing between them.

Observe the run with Logfire

#

Model calls, tool calls, retries, and handoffs are difficult to reconstruct from a final print statement. Pydantic AI includes optional Logfire instrumentation that records the run as a trace with child spans for model and tool activity.

The setup in main.py is short:

import logfire

logfire.configure()
logfire.instrument_system_metrics()
logfire.instrument_pydantic_ai()

I keep the Logfire write token outside the code and load it from .env. The screenshot below was captured from an earlier iteration in which storing the priority was still an ordinary tool. That makes it useful for seeing the model/tool exchange, but the current repository finishes through the store_priority output function described above.

The second trace comes from the current implementation. It shows the same instructions and prompt, the get_case_details tool returning case_3 with case_priority still 0, and then the model calling the store_priority output tool with new_priority and a motivation. That output-tool call is what ends the run, so the span closes with Final result processed. instead of another model turn.

The returned AgentRunResult also exposes usage data without opening Logfire:

usage = result.usage
print(f"Cost: {usage.cost}")
print(f"input tokens: {usage.input_tokens}")
print(f"output tokens: {usage.output_tokens}")
print(f"Requests: {usage.requests}")
print(f"tool calls: {usage.tool_calls}")

The provider supplies the usage information and Pydantic AI aggregates it across the requests in the run. I find both views useful: the counters show the size and cost of a run, while the trace explains why those requests and tool calls happened.

Run the complete example

#

The project requires Python 3.12 or newer and uses uv for its environment and locked dependencies:

git clone https://github.com/jettro/Pydantic-ai-tryout.git
cd Pydantic-ai-tryout
uv sync --all-groups

Create a local .env file with your model-provider credential. Add a Logfire token only when you want to send the trace to Logfire:

OPENAI_API_KEY=your-openai-api-key
LOGFIRE_TOKEN=your-logfire-write-token

Do not commit this file or put either credential in the source. Run the default case (case_1) or select one of the other cases from the repository:

make run
make run case_3
make run case_4

case_3 starts without a priority and should finish through CaseResponse. case_4 already has priority 2 and gives the agent a reason to choose PriorityRejected. These commands call a real model, so they use tokens and incur the corresponding provider cost.

The Makefile also collects the repeating environment commands:

sync: ## Install the project and the dev dependencies in .venv
	uv sync --all-groups

run: ## Run main.py, calls the case agent with the real model
	uv run python main.py $(CASE)

Run make without arguments to see all available targets. The remote repository contains the complete classes and the latest runnable version, so use that as the source of truth when experimenting.

What I take back to Akgents

#

I learned a lot from this small agent that I want to use while understanding the higher-level framework. An agent is not just a prompt wrapped around a model. Its useful boundaries are visible in code:

• dependencies carry trusted application context and services;
• tools expose controlled capabilities whose results return to the model;
• output functions validate or perform the final action and then end the run;
• multiple output types turn domain alternatives into explicit contracts;
• ModelRetry separates correctable model mistakes from valid rejection and broken execution;
• a handoff can be expressed as output processing, while observability keeps the nested work visible.
• Capabilities package reusable agent behaviour, with MCP showing how one integration can adapt between a local toolset and provider-native execution.

Pydantic AI provides these typed mechanics for one run. Akgents builds on that foundation to organise agents into actor-based teams with their own lifecycle and durability. Understanding this lower layer makes the next Akgents concepts less magical: I can see which responsibility comes from the model agent and which responsibility belongs to the framework around it.