Chapter 17

Week 5.2: PydanticAI

Notebook.venv26 cells

Welcome to Week 5 - Agent Frameworks

Day 2: Pydantic AI

Same week, same five steps, a new framework. The whole idea this week is that once you know one agent framework, you basically know them all, so every day we build the same agent the same five steps and watch the idioms rhyme:

Create the agent - give it a model and a system prompt.
Run it - send a message, get a reply.
Add tools - plain typed functions the agent can call.
Add MCP - connect a tool server someone else wrote, wired the same way every time.
Put it in a loop with a goal - hand it an objective and let it work, step after step, until the job is done.

Steps 1 and 2 are still just an LLM call. Tools and MCP give the agent things it can do. Step 5 is where it becomes an agent: the framework runs the loop on its own, choosing a tool, reading the result, choosing again, until the goal is met.

The running project is the same SQLite todo board from Day 1. A worker picks one goal off the board, plans its own steps, does the work with its agent loop, and ticks each step off. The board code (board.py) is the same file, byte for byte; only the framework around it changes.

This is the second framework of the day: Pydantic AI, from the team behind Pydantic, the validation library under FastAPI and most of the modern Python data stack. It brings that same types-everywhere feel to agents: you declare what a tool takes and what an agent returns, and the framework validates and types the whole flow. It is famous for the cleanest typed-output story in the field and for one-line observability through Logfire. Notice how little changes from Strands; that is the point.

Setup

Two things this day needs, both already in place from earlier weeks:

Node, for npx (the filesystem MCP server runs over it). Check with node --version.
OPENAI_API_KEY in the repo-root .env. Today's frameworks use OpenAI's gpt-5.4-mini, and that key has been in your .env since Week 1.

Pydantic AI is part of the repo's environment, so a normal uv sync from the repo root installs everything. Open this notebook in Cursor and pick the repo's default Python 3.12.12 kernel, the same one you use every week, then run the cells top to bottom.

Warm the filesystem MCP server once now so the first run is fast, then stop it with Ctrl-C as soon as it prints that it is running:

bash

npx -y @modelcontextprotocol/server-filesystem .

In [ ]python · cell 5

python

import os
from pathlib import Path

from dotenv import load_dotenv
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
from fastmcp.client.transports import StdioTransport

load_dotenv(override=True)

Step 1: Create the agent

In Pydantic AI an agent is an Agent: a model string plus instructions, which is its system prompt. The string "openai-chat:gpt-5.4-mini" picks the provider and the model, with OPENAI_API_KEY coming from the repo-root .env. The -chat names OpenAI's Chat Completions API explicitly. A bare openai: works too, but its default shifts in the coming Pydantic AI v2, and Chat Completions is the API every OpenAI-compatible endpoint speaks, which is what we want for the model swap at the end of the day.

In [ ]python · cell 7

python

MODEL = "openai-chat:gpt-5.4-mini"

agent = Agent(
    MODEL,
    instructions="You are a concise, friendly assistant. Reply in a single short sentence.",
)

Step 2: Run it

Send a message, await the reply, and print the result's .output. With no tools yet there is nothing to loop over, so the agent just answers; this is still only an LLM call. There is a synchronous agent.run_sync(...) too, but inside a notebook we await agent.run(...) so it cooperates with Jupyter's event loop.

In [ ]python · cell 9

python

result = await agent.run("Say hello in Spanish.")
print(result.output)

Our project this week: a SQLite todo board

The worker coordinates through the same tiny SQLite board as Day 1, the same board.py file: one file, one table, no server to run. A worker is handed one goal; to reach it, it writes its own step todos under that goal, ticks each one off as it goes, and closes the goal at the end. Under the hood the board is just a list of dicts (a goal has parent_id of None, a step points at its goal):

In [ ]python · cell 11

python

import board

board.reset_board()
board.add_goal("Read notes.txt, translate its contents into natural Spanish, and write the Spanish to spanish.txt.")
board.list_todos()

show_board() prints that same data nicely, in the rich style from Week 1: each goal with its steps indented beneath, done tasks struck through in green and the one in progress in yellow. There are no steps yet; the worker writes its own when it plans.

In [ ]python · cell 13

python

board.show_board()

Step 3: Add tools

A tool in Pydantic AI is a plain typed Python function. Pydantic reads the type hints and the docstring and builds the JSON schema for you, so there is nothing else to declare. The signature idiom is the @agent.tool decorator; passing the same functions in tools=[...] does exactly the same thing and lets us reuse them across agents, which is what we do here.

We write three small board tools: show_todos to read the board, plan_steps to break a goal into steps, and complete_task to mark a todo done. Here we give a quick agent just two of them and ask what is on the board; watch it decide, on its own, to call show_todos before it answers. That decide, call, read, answer is the agent loop starting to turn. All three tools come together in step 5.

In [ ]python · cell 15

python

def show_todos() -> list[dict]:
    """List every todo on the board. A goal has parent_id None; a step has parent_id set to its goal's id."""
    return board.list_todos()

def plan_steps(goal_id: int, steps: list[str]) -> dict:
    """Break a goal into an ordered checklist of steps on the board. Pass the goal's id and a short list of step descriptions."""
    return {"goal_id": goal_id, "step_ids": [board.add_step(goal_id, step) for step in steps]}

def complete_task(task_id: int, result: str) -> dict:
    """Mark a todo (a step or the goal) with this id as done and record a short result summary."""
    board.complete_todo(task_id, result)
    return {"task_id": task_id, "status": "done"}

In [ ]python · cell 16

python

board_agent = Agent(
    MODEL,
    instructions="You help manage a shared todo board.",
    tools=[show_todos, complete_task],
)

In [ ]python · cell 17

python

result = await board_agent.run("What is on the board right now, and what is its status?")
print(result.output)

Step 4: Add MCP

MCP is just more tools: ones you did not write, connected over a small protocol. We give the agent the filesystem reference server, the same Node server every framework uses this week, scoped to a single workspace folder so it can only touch files in there. In Pydantic AI an MCP server is a toolset: you wrap it in an MCPToolset, pass it via toolsets=[...], and open its connection with async with agent: around the run.

We point log_file at the null device (Path(os.devnull)). That keeps the server's startup logging off the screen and, more importantly, lets the server run from a Jupyter kernel on Windows, where the kernel's own stderr has no real file descriptor for the server to write to. On Mac and Linux it simply keeps the output clean.

In [ ]python · cell 19

python

workspace = Path("workspace").resolve()   # the only folder the agent may touch

filesystem = MCPToolset(
    StdioTransport(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", str(workspace)],
        cwd=str(workspace),  # start the server in the workspace so relative file names resolve there
        log_file=Path(os.devnull),
    ), 
    init_timeout=60
)

In [ ]python · cell 20

python

file_agent = Agent(
    MODEL,
    instructions="You can read and write files in your workspace. Use your tools to do what is asked.",
    toolsets=[filesystem],
)

In [ ]python · cell 21

python

async with file_agent:
    result = await file_agent.run("Read notes.txt and summarize it in one short sentence.")
print(result.output)

Step 5: Put it in a loop with a goal

Now the payoff. Give one agent all three board tools and the filesystem server, hand it the goal, and let it run. It plans its own steps on the board, works them with its file tools, ticks each one off, and closes the goal when the work is done. That is the agent loop running on its own: read, plan, act, check off, repeat. Watch the board fill with steps and then strike them through.

In [ ]python · cell 23

python

INSTRUCTIONS = """
You are a careful worker with a shared todo board and a set of file tools.

Take the pending goal and see it through. Begin by laying out a short plan: the handful of concrete steps the work itself breaks down into, added to the board under the goal. Then carry them out with your file tools, marking each step done as you finish it. Once the steps are all done, close the goal. Your files live in the single folder your tools are allowed to use.
"""

worker = Agent(
    MODEL,
    instructions=INSTRUCTIONS,
    tools=[show_todos, plan_steps, complete_task],
    toolsets=[filesystem],
)

board.reset_board()
goal_id = board.add_goal("Read notes.txt, translate its contents into natural Spanish, and write the Spanish to spanish.txt.")
board.claim_todo(goal_id)

async with worker:
    result = await worker.run("Please work the pending goal on the board.")
print(result.output)
board.show_board()

Run the same worker from the terminal

Everything you just watched in step 5 is also packaged as a small script, pydantic_worker.py, sitting next to this notebook. It seeds the same goal, builds the same agent with the same three board tools and the filesystem MCP server, and runs the same loop, only from the command line instead of the kernel. Open a terminal in this folder and run it:

bash

uv run pydantic_worker.py

You will see the agent plan its steps, work the goal and tick each one off, then the finished board and the Spanish it wrote. This is the shape every worker takes on Day 5, where a Google ADK orchestrator launches one of these per framework as a parallel subprocess against a single shared board.

The one cool thing: typed, validated, observable

Pydantic AI is built around two ideas worth knowing even though our worker does not lean on them:

Typed, validated results. Pass a Pydantic model as output_type= and result.output comes back as a validated instance of that type. Every framework can do typed output; this is the thing Pydantic AI is known for.
Observability in two lines. logfire.configure() and logfire.instrument_pydantic_ai() light up every model request, tool call and validation in the Logfire UI, live, from the same team that builds the framework. It needs the logfire extra, which we leave out to keep the shared environment lean.

And the same one-line spirit as the rest of the week: to run on any OpenAI-compatible endpoint, wrap an OpenAIChatModel around an OpenAIProvider and hand it to the agent. Everything else stays identical:

python

from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider

model = OpenAIChatModel(
    "gpt-5.4-mini",
    provider=OpenAIProvider(base_url="https://openrouter.ai/api/v1", api_key=OPENROUTER_KEY),
)
agent = Agent(model, instructions="...")

Chapter 17

Week 5.2: PydanticAI

Notebook.venv26 cells

Welcome to Week 5 - Agent Frameworks

Day 2: Pydantic AI

Create the agent - give it a model and a system prompt.
Run it - send a message, get a reply.
Add tools - plain typed functions the agent can call.
Add MCP - connect a tool server someone else wrote, wired the same way every time.
Put it in a loop with a goal - hand it an objective and let it work, step after step, until the job is done.

Setup

Two things this day needs, both already in place from earlier weeks:

Node, for npx (the filesystem MCP server runs over it). Check with node --version.
OPENAI_API_KEY in the repo-root .env. Today's frameworks use OpenAI's gpt-5.4-mini, and that key has been in your .env since Week 1.

Warm the filesystem MCP server once now so the first run is fast, then stop it with Ctrl-C as soon as it prints that it is running:

bash

npx -y @modelcontextprotocol/server-filesystem .

In [ ]python · cell 5

python

import os
from pathlib import Path

from dotenv import load_dotenv
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
from fastmcp.client.transports import StdioTransport

load_dotenv(override=True)

Step 1: Create the agent

In [ ]python · cell 7

python

MODEL = "openai-chat:gpt-5.4-mini"

agent = Agent(
    MODEL,
    instructions="You are a concise, friendly assistant. Reply in a single short sentence.",
)

Step 2: Run it

In [ ]python · cell 9

python

result = await agent.run("Say hello in Spanish.")
print(result.output)

Our project this week: a SQLite todo board

In [ ]python · cell 11

python

import board

board.reset_board()
board.add_goal("Read notes.txt, translate its contents into natural Spanish, and write the Spanish to spanish.txt.")
board.list_todos()

In [ ]python · cell 13

python

board.show_board()

Step 3: Add tools

In [ ]python · cell 15

python

def show_todos() -> list[dict]:
    """List every todo on the board. A goal has parent_id None; a step has parent_id set to its goal's id."""
    return board.list_todos()

def plan_steps(goal_id: int, steps: list[str]) -> dict:
    """Break a goal into an ordered checklist of steps on the board. Pass the goal's id and a short list of step descriptions."""
    return {"goal_id": goal_id, "step_ids": [board.add_step(goal_id, step) for step in steps]}

def complete_task(task_id: int, result: str) -> dict:
    """Mark a todo (a step or the goal) with this id as done and record a short result summary."""
    board.complete_todo(task_id, result)
    return {"task_id": task_id, "status": "done"}

In [ ]python · cell 16

python

board_agent = Agent(
    MODEL,
    instructions="You help manage a shared todo board.",
    tools=[show_todos, complete_task],
)

In [ ]python · cell 17

python

result = await board_agent.run("What is on the board right now, and what is its status?")
print(result.output)

Step 4: Add MCP

In [ ]python · cell 19

python

workspace = Path("workspace").resolve()   # the only folder the agent may touch

filesystem = MCPToolset(
    StdioTransport(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", str(workspace)],
        cwd=str(workspace),  # start the server in the workspace so relative file names resolve there
        log_file=Path(os.devnull),
    ), 
    init_timeout=60
)

In [ ]python · cell 20

python

file_agent = Agent(
    MODEL,
    instructions="You can read and write files in your workspace. Use your tools to do what is asked.",
    toolsets=[filesystem],
)

In [ ]python · cell 21

python

async with file_agent:
    result = await file_agent.run("Read notes.txt and summarize it in one short sentence.")
print(result.output)

Step 5: Put it in a loop with a goal

In [ ]python · cell 23

python

INSTRUCTIONS = """
You are a careful worker with a shared todo board and a set of file tools.

Take the pending goal and see it through. Begin by laying out a short plan: the handful of concrete steps the work itself breaks down into, added to the board under the goal. Then carry them out with your file tools, marking each step done as you finish it. Once the steps are all done, close the goal. Your files live in the single folder your tools are allowed to use.
"""

worker = Agent(
    MODEL,
    instructions=INSTRUCTIONS,
    tools=[show_todos, plan_steps, complete_task],
    toolsets=[filesystem],
)

board.reset_board()
goal_id = board.add_goal("Read notes.txt, translate its contents into natural Spanish, and write the Spanish to spanish.txt.")
board.claim_todo(goal_id)

async with worker:
    result = await worker.run("Please work the pending goal on the board.")
print(result.output)
board.show_board()

Run the same worker from the terminal

bash

uv run pydantic_worker.py

The one cool thing: typed, validated, observable

Pydantic AI is built around two ideas worth knowing even though our worker does not lean on them:

Typed, validated results. Pass a Pydantic model as output_type= and result.output comes back as a validated instance of that type. Every framework can do typed output; this is the thing Pydantic AI is known for.
Observability in two lines. logfire.configure() and logfire.instrument_pydantic_ai() light up every model request, tool call and validation in the Logfire UI, live, from the same team that builds the framework. It needs the logfire extra, which we leave out to keep the shared environment lean.

python

from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider

model = OpenAIChatModel(
    "gpt-5.4-mini",
    provider=OpenAIProvider(base_url="https://openrouter.ai/api/v1", api_key=OPENROUTER_KEY),
)
agent = Agent(model, instructions="...")