MCP with Pydantic AI

In my prior article on building a streaming approach with Pydantic AI, I built a pattern around streaming with API calls to Anthropic. In this article, we’ll look to expand to use Pydantic AI MCP Clients.

MCP Overview

Before implementing a connection to a MCP server via Pydantic AI, let’s review what MCP is at a very high level as well as implement a simple server with Fast API.

At a high level, an MCP server allows for a standardized way to define how an LLM interacts with tools. Instead of defining tools on a one-off basis in our LLM application, we can utilize prebuilt or custom servers that expose tools. This allows for both reusability for servers we may build ourselves or plugging into various vendor or open source MCP servers - preventing us from reinventing the wheel when we want to use a new tool.

For more information, I’d recommend reading through Anthropic’s release post, the model context protocol site, and browsing through the python sdk github repo.

MCP Server

Define Server and Tools

For our MCP server, we’ll define one very basic tool - getting a user’s name. This allows us to hardcode a name and verify the LLM is picking up the information as expected. Using the FastMCP class, we setup a simple app and use the tool decorator to add this tool.

from mcp.server.fastmcp import FastMCP

# Create an MCP server
mcp = FastMCP("Demo")


@mcp.tool()
async def get_user_name() -> str:
    """
    This function is used to retrieve the user name.
    """
    return "Joe"  # Note: this would be dynamically retrieved in an actual app

Integrate with FastAPI

To connect into a FastAPI server, we can mount this MCP as a sub application. This allows us to setup an endpoint, “/messages” in this case, which then points to a function of server to handle these post messages. We also need to setup a route for the initial connection.

from src.build_mcp.create_mcp import *
from src.startup.app import app
from mcp.server.sse import SseServerTransport
from starlette.routing import Mount
from fastapi import Request


# Mount to fastapi app
sse = SseServerTransport("/messages/")
app.router.routes.append(Mount("/messages", app=sse.handle_post_message))

# Add routes for sse
@app.get('/sse')
async def handle_sse(request: Request) -> None:
    # See https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/server/fastmcp/server.py for more details
    async with sse.connect_sse(
        request.scope,
        request.receive,
        request._send,  # type: ignore[reportPrivateUsage]
    ) as streams:
        await mcp._mcp_server.run(
            streams[0],
            streams[1],
            mcp._mcp_server.create_initialization_options(),
        )

Testing the connection

At this point we can startup the FastAPI server and test that the endpoints are working correctly. The MCP site provides client examples to use to connect to servers. With a few simple changes, we can add a client.py file to test out connecting and using our new tool:

from mcp import ClientSession, types
from mcp.client.sse import sse_client
import asyncio

# This is a simple example client just utilized to test. The fast api server should already be running and this executed in a different session.

async def interact_with_mcp():
    print("Starting MCP client...")
    try:
        # Connect to the remote MCP server
        server_url = "http://127.0.0.1:8000/sse"  # SSE endpoint
        print(f"Connecting to MCP server at {server_url}...")
        
        # First establish the SSE connection
        async with sse_client(server_url) as (read, write):
            print("Connected to SSE endpoint")
            async with ClientSession(read, write) as session:
                print("Created client session")
                
                # Initialize the connection
                print("Initializing connection...")
                await session.initialize()
                print("Connection initialized")

                # 1. List available tools
                print("Listing tools...")
                tools = await session.list_tools()
                print("Available tools:", tools)

                # 2. Call the greet tool
                print("Calling get name tool...")
                result = await session.call_tool("get_user_name")
                print("Get name", result)

    except Exception as e:
        print(f"Error occurred: {str(e)}")
        raise

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

When running this code, we get the following output, which demonstrates successful tool execution and name retrieval:

Starting MCP client...
Connecting to MCP server at http://127.0.0.1:8000/sse...
Connected to SSE endpoint
Created client session
Initializing connection...
Connection initialized
Listing tools...
Available tools: meta=None nextCursor=None tools=[Tool(name='get_user_name', description='\n    This function is used to retrieve the user name.\n    ', inputSchema={'properties': {}, 'title': 'get_user_nameArguments', 'type': 'object'})]
Calling get name tool...
Get name meta=None content=[TextContent(type='text', text='Joe', annotations=None)] isError=False

Pydantic AI Integration

Now with the MCP server setup and connected successfully - we need to integrate into our previously built Pydantic AI pattern. There are only a few changes required.

First, we need to add an outer context manager to our main_stream function. I won’t go through the function in full (see previous article), but the key piece is as follows:

async with agent.run_mcp_servers():
    async with agent.iter(user_prompt) as run:
        # Continue with streaming logic from previous article

The above assumes the MCP servers are listed in the agent. To do this, we populate one additional variable on that initialization.

from pydantic_ai.mcp import MCPServerHTTP

# Setup mcp server objects for pydantic
pydantic_mcp_server = MCPServerHTTP(url='http://127.0.0.1:8000/sse')
pydantic_mcp_servers = [pydantic_mcp_server]

agent = Agent(  
    'anthropic:claude-3-5-sonnet-20241022',
    system_prompt=(
        "You are a helpful assistant that can answer questions and help with tasks. Greet the user by name first before answering any questions."
    ),
    mcp_servers=pydantic_mcp_servers
)

async for text_response in main_stream(agent, 'What is 2+2?'):
    print(text_response, end='', flush=True)

When running the above, we get the following - you’ll see the tool call as expected in debug output (abbreviated).

Hi Joe! The answer to 2+2 is 4. This is a basic mathematical addition that I can help you with directly. Is there anything else you'd like to know?


Debug:
<class 'pydantic_ai.messages.FunctionToolCallEvent'>: FunctionToolCallEvent(part=ToolCallPart(tool_name='get_user_name', args={}, tool_call_id='toolu_01YFmhgYeWnbbxeXvVj89xVY', part_kind='tool-call'), call_id='toolu_01YFmhgYeWnbbxeXvVj89xVY', event_kind='function_tool_call')
<class 'pydantic_ai.messages.FunctionToolResultEvent'>: FunctionToolResultEvent(result=ToolReturnPart(tool_name='get_user_name', content=CallToolResult(meta=None, content=[TextContent(type='text', text='Joe', annotations=None)], isError=False), tool_call_id='toolu_01YFmhgYeWnbbxeXvVj89xVY', timestamp=datetime.datetime(2025, 4, 21, 15, 11, 58, 593954, tzinfo=datetime.timezone.utc), part_kind='tool-return'), tool_call_id='toolu_01YFmhgYeWnbbxeXvVj89xVY', event_kind='function_tool_result')

Summary

MCP servers can provide a standardized way to implement new or integrate existing tools. Building together with frameworks such as FastAPI and PydanticAI allows us to integrate various patterns that work well together.

All examples and files available on Github.