PydanticAI SDK Track Part 10: MCP Integration

1. MCP Overview in PydanticAI

The Model Context Protocol (MCP) is an open standard that enables AI agents to connect to external tool providers through a unified interface. PydanticAI supports MCP on both sides — as a client consuming tools from MCP servers, and as a platform for building MCP servers that expose agent capabilities.

1.1 Architecture: Agent ↔ MCP Client ↔ MCP Server

The MCP architecture separates tool providers (servers) from tool consumers (clients). PydanticAI agents use an MCP client to discover and invoke tools hosted on remote or local MCP servers:

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerHTTP

# Basic MCP connection — agent discovers tools from server
agent = Agent(
    "openai:gpt-4o",
    system_prompt="You are a helpful assistant with access to external tools.",
    mcp_servers=[
        MCPServerHTTP(url="http://localhost:8080/mcp")
    ]
)

# The agent automatically discovers all tools exposed by the MCP server
# and can invoke them during conversation
import asyncio

async def main():
    async with agent.run_mcp_servers():
        result = await agent.run("Search for recent Python releases on GitHub")
        print(result.data)

asyncio.run(main())

2. MCP Client

PydanticAI provides two transport mechanisms for connecting to MCP servers: HTTP (for remote servers) and Stdio (for local subprocess servers).

2.1 HTTP & Stdio Transports

Choose the transport based on where your MCP server runs:

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerHTTP, MCPServerStdio

# Transport 1: HTTP — for remote or containerized MCP servers
http_server = MCPServerHTTP(
    url="https://mcp-tools.example.com/sse",
    headers={"Authorization": "Bearer your-token"}
)

# Transport 2: Stdio — for local subprocess servers
stdio_server = MCPServerStdio(
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", "/path/to/allowed/dir"]
)

# Agent with both transports — tools from all servers are merged
agent = Agent(
    "openai:gpt-4o",
    system_prompt="You have access to remote APIs and local filesystem tools.",
    mcp_servers=[http_server, stdio_server]
)

import asyncio

async def main():
    async with agent.run_mcp_servers():
        # Agent can use tools from either server
        result = await agent.run("List files in the project directory")
        print(result.data)

asyncio.run(main())

2.2 Auto-Discovery of Server-Provided Tools

When an MCP server is connected, PydanticAI automatically discovers all available tools and makes them callable by the agent. No manual tool registration needed:

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerHTTP
import asyncio

# Connect to a GitHub MCP server
github_server = MCPServerHTTP(url="http://localhost:3000/mcp")

agent = Agent(
    "openai:gpt-4o",
    system_prompt="You can interact with GitHub repositories.",
    mcp_servers=[github_server]
)

async def demonstrate_discovery():
    async with agent.run_mcp_servers():
        # The agent now knows about all tools the GitHub server exposes:
        # - create_issue, list_issues, search_repositories, etc.
        # These are injected into the agent's tool list automatically
        
        result = await agent.run(
            "Create a new issue titled 'Bug: login timeout' in the main repo"
        )
        print(result.data)

asyncio.run(demonstrate_discovery())

3. FastMCP Client

FastMCP provides a simplified, high-level client for common MCP interaction patterns. It reduces boilerplate when you need direct access to MCP resources and prompt templates beyond just tool calling.

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerHTTP
import asyncio

# FastMCP-style simplified connection
# The server exposes tools, resources, and prompt templates
knowledge_server = MCPServerHTTP(
    url="http://localhost:8080/mcp"
)

agent = Agent(
    "openai:gpt-4o",
    system_prompt="You are a knowledge assistant with access to documentation.",
    mcp_servers=[knowledge_server]
)

async def use_fastmcp_features():
    async with agent.run_mcp_servers():
        # Tools are auto-discovered and available
        # The agent can call search_docs, get_page, list_resources, etc.
        result = await agent.run(
            "Find documentation about PydanticAI dependency injection"
        )
        print(result.data)

asyncio.run(use_fastmcp_features())

3.1 Resource & Prompt Template Access

MCP servers can expose resources (data files, database records) and prompt templates alongside tools. Access them through the agent’s MCP integration:

from pydantic_ai import Agent, RunContext
from pydantic_ai.mcp import MCPServerHTTP
from dataclasses import dataclass
import asyncio

@dataclass
class AppDeps:
    user_context: str

# Server that exposes resources and prompts
docs_server = MCPServerHTTP(url="http://localhost:8080/mcp")

agent = Agent(
    "openai:gpt-4o",
    deps_type=AppDeps,
    system_prompt="You help users navigate project documentation.",
    mcp_servers=[docs_server]
)

@agent.tool
def get_user_preferences(ctx: RunContext[AppDeps]) -> str:
    """Get user context for personalized responses."""
    return ctx.deps.user_context

async def main():
    deps = AppDeps(user_context="Senior Python developer, prefers concise answers")
    
    async with agent.run_mcp_servers():
        result = await agent.run(
            "How do I set up structured outputs in PydanticAI?",
            deps=deps
        )
        print(result.data)

asyncio.run(main())

4. Building MCP Servers

Build your own MCP servers using the FastMCP class from the mcp package. This exposes your custom tools, resources, and prompts to any MCP-compatible client.

from mcp.server.fastmcp import FastMCP
from datetime import datetime

# Create an MCP server
mcp = FastMCP("My Custom Tools Server")

@mcp.tool()
def get_current_time(timezone: str = "UTC") -> str:
    """Get the current time in the specified timezone."""
    from zoneinfo import ZoneInfo
    now = datetime.now(ZoneInfo(timezone))
    return now.strftime("%Y-%m-%d %H:%M:%S %Z")

@mcp.tool()
def calculate_compound_interest(
    principal: float, 
    rate: float, 
    years: int, 
    compounds_per_year: int = 12
) -> str:
    """Calculate compound interest and return a summary."""
    amount = principal * (1 + rate / compounds_per_year) ** (compounds_per_year * years)
    interest = amount - principal
    return (
        f"Principal: ${principal:,.2f}\n"
        f"Rate: {rate*100:.1f}%\n"
        f"Period: {years} years\n"
        f"Final Amount: ${amount:,.2f}\n"
        f"Interest Earned: ${interest:,.2f}"
    )

@mcp.resource("config://app-settings")
def get_app_settings() -> str:
    """Expose application settings as an MCP resource."""
    return '{"version": "2.1.0", "debug": false, "max_retries": 3}'

# Run the server (typically in a separate process)
if __name__ == "__main__":
    mcp.run(transport="stdio")  # or transport="sse" for HTTP

4.1 Exposing PydanticAI Agents as MCP Server Tools

Wrap a PydanticAI agent inside an MCP tool to make it accessible from any MCP client:

from mcp.server.fastmcp import FastMCP
from pydantic_ai import Agent
import asyncio

# Create the MCP server
mcp = FastMCP("AI Agent Server")

# Define a PydanticAI agent
summarizer = Agent(
    "openai:gpt-4o",
    system_prompt="You summarize text concisely. Return bullet points of key ideas."
)

analyzer = Agent(
    "openai:gpt-4o",
    system_prompt="You analyze sentiment and tone of text. Return: sentiment, tone, confidence."
)

@mcp.tool()
async def summarize_text(text: str, max_bullets: int = 5) -> str:
    """Summarize text into key bullet points using AI."""
    result = await summarizer.run(
        f"Summarize this in {max_bullets} bullet points:\n\n{text}"
    )
    return result.data

@mcp.tool()
async def analyze_sentiment(text: str) -> str:
    """Analyze the sentiment and tone of the provided text."""
    result = await analyzer.run(f"Analyze sentiment:\n\n{text}")
    return result.data

@mcp.tool()
def word_count(text: str) -> str:
    """Count words, sentences, and paragraphs in text."""
    words = len(text.split())
    sentences = text.count('.') + text.count('!') + text.count('?')
    paragraphs = len([p for p in text.split('\n\n') if p.strip()])
    return f"Words: {words}, Sentences: {sentences}, Paragraphs: {paragraphs}"

if __name__ == "__main__":
    mcp.run(transport="stdio")

5. Tool Interoperability

PydanticAI agents can use MCP-provided tools alongside locally defined function tools. This hybrid approach lets you combine community MCP servers with custom business logic.

5.1 Agent with Mixed MCP and Local Tools

from pydantic_ai import Agent, RunContext
from pydantic_ai.mcp import MCPServerHTTP, MCPServerStdio
from dataclasses import dataclass
import asyncio

@dataclass
class ProjectDeps:
    project_id: str
    api_token: str

# MCP servers for external capabilities
github_server = MCPServerHTTP(url="http://localhost:3000/mcp")
filesystem_server = MCPServerStdio(
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", "./project"]
)

# Agent combines MCP tools with local tools
agent = Agent(
    "openai:gpt-4o",
    deps_type=ProjectDeps,
    system_prompt="""You are a project management assistant. You can:
    - Access GitHub (via MCP) for issues, PRs, and repos
    - Read/write files (via MCP filesystem)
    - Use local tools for project-specific logic""",
    mcp_servers=[github_server, filesystem_server]
)

# Local tool — not available via MCP, only in this agent
@agent.tool
def calculate_sprint_velocity(
    ctx: RunContext[ProjectDeps], 
    completed_points: list[int]
) -> str:
    """Calculate sprint velocity from completed story points."""
    if not completed_points:
        return "No data available"
    avg = sum(completed_points) / len(completed_points)
    return (
        f"Project: {ctx.deps.project_id}\n"
        f"Sprints analyzed: {len(completed_points)}\n"
        f"Average velocity: {avg:.1f} points/sprint\n"
        f"Range: {min(completed_points)}-{max(completed_points)} points"
    )

@agent.tool
def format_release_notes(ctx: RunContext[ProjectDeps], items: list[str]) -> str:
    """Format a list of changes into release notes."""
    notes = f"# Release Notes — Project {ctx.deps.project_id}\n\n"
    for i, item in enumerate(items, 1):
        notes += f"{i}. {item}\n"
    return notes

async def main():
    deps = ProjectDeps(project_id="alpha-v2", api_token="ghp_xxx")
    
    async with agent.run_mcp_servers():
        # This query may use GitHub MCP tools + local velocity tool
        result = await agent.run(
            "Check our GitHub issues, calculate our velocity from the last 5 sprints "
            "(points: 21, 34, 28, 31, 25), and draft release notes for the closed issues.",
            deps=deps
        )
        print(result.data)

asyncio.run(main())