Pydantic MCP Agents with FastAPI
In the previous example I explored the Model-Context-Protocol server gateways after going through some MCP and server deployment options using FastMCP. Now let's go a little bit deeper and explore the agent interaction with the MCP server using Pydantic-ai.
Start by creating your virtual environment using uv:
uv init
uv sync
uv add "pydantic-ai-slim[openai]" fastmcp openai fastapi uvicorn requests pyjwt python-dotenv python-multipart python-weather aiohttp beautifulsoup4 huggingface_hub
git add .
git commit -m "Initial commit"
And make sure your terminal is using it:
which python
/usr/bin/python
source .venv/bin/activate
which python
~/Projects/mcp-servers/langgraph_intro/.venv/bin/python
Getting Started
Building a Pydantic Agent on top of a FastMCP server is surprisingly easy - coming from LangGraph:
import asyncio
from fastmcp import FastMCP
from pathlib import Path
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
# Model
def ollama_model(name):
return OpenAIChatModel(
name,
provider=OpenAIProvider(base_url="http://localhost:11434/v1", api_key="ollama"),
)
model = ollama_model("qwen3.8:latest")
# Tools
mcp_server = FastMCP("FileServer")
@mcp_server.tool()
def write_file(filename: str, content: str) -> str:
"""Save content into a text file. Use this tool whenever asked to save, write or create a file with a given content. Create a camel-case filename based on the content if the user does not specify one."""
print(f"[INFO] :: [TOOL RUN] :: wrtie_file({filename!r}, {len(content)} characters")
Path(filename).write_text(content)
return f"Saved {filename} with {len(content)} content length."
# MCP Agent
agent = Agent(
model,
system_prompt="You are a helpful assistant that can use your tool to save content to text files.",
toolsets = [MCPToolset(mcp_server)]
)
response = asyncio.run(agent.run("Write a Haiku about the computer game series Mass Effect."))
print(response)
Pydantic Orchestrator & Worker Agents
Also more advanced workflows like the multi-agent Orchestrator-Worker is straight forward:
import urllib.parse, aiohttp, python_weather, asyncio
from bs4 import BeautifulSoup
from fastmcp import FastMCP
from datetime import datetime
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
# Model
def ollama_model(name):
return OpenAIChatModel(
name,
provider=OpenAIProvider(base_url="http://localhost:11434/v1", api_key="ollama"),
)
model = ollama_model("qwen3.8:latest")
# Tools
mcp_researcher = FastMCP("ResearchServer") # READ ONLY
@mcp_researcher.tool()
async def check_weather(location: str) -> str:
"""Get tge current weather for a location."""
print(f"[INFO] :: [TOOL RUN] :: check_weather({location!r})")
async with python_weather.Client(unit=python_weather.METRIC) as client:
temperature = await client.get(location)
return f"{location}: {temperature}"
@mcp_researcher.tool()
async def search_web(query: str) -> str:
"""Search the web. Use this tool to find travel destinations."""
print(f"[INFO] :: [TOOL RUN] :: search_web({query!r})")
url = f"https://www.bing.com/search?q={urllib.parse.quote(query)}"
async with aiohttp.ClientSession() as search:
async with search.get(url, headers={"User-Agent": "Mozilla/5.0"}) as request:
soup = BeautifulSoup(await request.text(), "html.parser")
hits = [a.text.strip() for a in soup.find_all("a", class_="result__snippet")][:3]
return "\n".join(hits) or "No results."
mcp_actor = FastMCP("ActionServer")
@mcp_actor.tool()
async def select_hotel(hotel_name: str, dates: str) -> str:
"""Select a hotel. Only call this AFTER you have found a real hotel name."""
print(f" [TOOL] book_hotel({hotel_name!r})")
return f"CONFIRMED: {hotel_name} for {dates}."
# MCP Agent
researcher = Agent(model,
system_prompt="You are a researcher. Report the weather and REAL hotel names you found.",
toolsets=[MCPToolset(mcp_researcher)]) # cannot book
actor = Agent(model,
system_prompt="You are a booking clerk. Call book_hotel with confirm=True.",
toolsets=[MCPToolset(mcp_actor)]) # cannot search
# The manager - agents wrapped as tools
manager = Agent(model, system_prompt=(
"You are a vacation manager. You have staff - you do nothing yourself. "
"1) ask_researcher for weather + real hotels in the given location. 2) pick ONE. 3) ask_booker to book it."))
@manager.tool_plain
async def ask_researcher(question: str) -> str:
"""Ask the researcher about weather and real hotels.""" # <- the MANAGER reads this
print(f"\n[MANAGER -> RESEARCHER] {question}")
r = await researcher.run(question) # a FULL agent loop, nested inside one tool call
return r.output
@manager.tool_plain
async def ask_booker(instruction: str) -> str:
"""Ask the actor to book a hotel. Include the hotel name and dates."""
print(f"\n[MANAGER -> ACTOR] {instruction}")
r = await actor.run(instruction)
return r.output
today = datetime.now().strftime("%A, %B %d, %Y")
response = asyncio.run(manager.run(f"Today is {today}. Plan a day trip to Kuala Lumpur and seleect a hotel for me next Saturday."))
print(response)
Adding a FastAPI Interface with JWT Auth
import secrets, threading, asyncio, uvicorn, requests, time, os, hashlib, jwt
from dotenv import load_dotenv
from datetime import datetime, timedelta, timezone
from pathlib import Path
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastmcp import FastMCP
from pydantic import BaseModel
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
# Model
def ollama_model(name):
return OpenAIChatModel(
name,
provider=OpenAIProvider(base_url="http://localhost:11434/v1", api_key="ollama"),
)
model = ollama_model("qwen3.8:latest")
# Generate user login
def generate_new_secret():
Path(".env").write_text(f"SECRET_KEY={secrets.token_hex(32)}\nALGORITHM=HS256\n")
print("[INFO] :: New secret token generated!")
if not Path(".env").exists():
generate_new_secret()
load_dotenv()
SECRET_KEY, ALGORITHM = os.getenv("SECRET_KEY"), os.getenv("ALGORITHM")
API_BASE = "http://127.0.0.1:8000"
def hash_password(password):
return hashlib.sha256(password.encode()).hexdigest()
def verify(plain, hashed):
return hash_password(plain) == hashed
DB = {"user1": {"username": "user1", "hashed_password": hash_password("password123")}}
# Tools
mcp_server = FastMCP("Utils")
@mcp_server.tool()
def calculate(expression: str) -> str:
"""Evaluate a simple math expression, e.g. '22 * 2 * ((1+1) / 2)."""
print(f"[INFO] [TOOL RUN] :: calculate({expression!r})")
allowed_char = "0123456789+-*/., ()"
if not set(expression) <= set(allowed_char): #whitelist input
return f"[Error] :: Only {allowed_char} allowed!"
return str(eval(expression))
# MCP Agent
agent = Agent(model,
system_prompt="You are a helpful assistant that can evaluate simple mathematical equations using it's tools.",
toolsets=[MCPToolset(mcp_server)])
# REST API
app = FastAPI()
oauth2 = OAuth2PasswordBearer(tokenUrl="login") # "tokens arrive in the Authorization header"
class ChatSchema(BaseModel):
message: str
# ---------- 1. CREATE LOGIN ----------
def create_access_token(username: str) -> str:
return jwt.encode({"sub": username,
"exp": datetime.now(timezone.utc) + timedelta(minutes=30)},
SECRET_KEY, algorithm=ALGORITHM)
# ---------- 2. LOGIN ----------
@app.post("/login")
def login(form: OAuth2PasswordRequestForm = Depends()):
user = DB.get(form.username)
if not user or not verify(form.password, user["hashed_password"]):
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Incorrect username or password")
return {"access_token": create_access_token(form.username), "token_type": "bearer"}
# ---------- 3. GUARD ----------
def get_current_user(token: str = Depends(oauth2)) -> str:
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) # signature + expiry
except jwt.PyJWTError: # any failure -> 401
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Could not validate credentials")
return payload["sub"]
# ---------- 4. ENDPOINTS ----------
class ChatIn(BaseModel):
message: str
@app.get("/health")
def health():
return {"status": "ok"}
@app.post("/chat")
async def chat(request: ChatSchema, user: str = Depends(get_current_user)) -> dict:
# FastAPI runs the guard first. If it raises 401, this function never executes.
result = await agent.run(request.message)
return {"user": user, "reply": result.output}
# ---------- 5. START SERVER ----------
def start_server(app, port):
"""Start uvicorn on a background thread and DON'T return until it is really listening."""
server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=port, log_level="warning"))
def run():
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(server.serve())
t = threading.Thread(target=run, daemon=True)
t.start()
for _ in range(100):
if server.started:
return f"http://127.0.0.1:{port}"
if not t.is_alive(): # thread crashed -> say so LOUDLY.
raise RuntimeError( # otherwise our requests would silently hit
"uvicorn thread died - see the traceback above")
time.sleep(0.3)
raise RuntimeError(f"server did not start on port {port} within 30s")
if __name__ == "__main__":
BASE = start_server(app, 8001)
assert requests.post(f"{BASE}/login", data={}).status_code != 404, "/login is missing - BASE points at the wrong server. Restart the runtime."
# 1. no token -> 401
print("no token :", requests.post(f"{BASE}/chat", json={"message": "hi"}).status_code)
# 2. wrong password -> 401 (data= not json= : OAuth2 expects a FORM body)
print("bad pwd :", requests.post(f"{BASE}/login",
data={"username": "alice", "password": "WRONG"}).status_code)
# 3. correct login -> a JWT
r = requests.post(f"{BASE}/login", data={"username": "alice", "password": "wonderland"})
TOKEN = r.json()["access_token"]
print("login :", r.status_code)
# 4. use it: one header, on every request
headers = {"Authorization": f"Bearer {TOKEN}"}
r = requests.post(f"{BASE}/chat", headers=headers,
json={"message": "What is 17 * 23 + 100?"}, timeout=120)
print("chat :", r.json())
# 5. forged token -> 401 (they don't have our SECRET_KEY, so the signature won't match)
forged = jwt.encode({"sub": "admin", "exp": datetime.now(timezone.utc) + timedelta(minutes=30)},
"f" * 64, algorithm=ALGORITHM)
print("forged :", requests.post(f"{BASE}/chat", headers={"Authorization": f"Bearer {forged}"},
json={"message": "hi"}).status_code)
r = requests.post(f"{BASE}/login", data={"username": "alice", "password": "wonderland"})
print(r.json())
try:
TOKEN = r.json()["access_token"]
headers = {"Authorization": f"Bearer {TOKEN}"}
r = requests.post(f"{BASE}/chat", headers=headers,
json={"message": "What is 17 * 23 + 100?"}, timeout=120)
print("chat :", r.json())
except:
print("No access token found!!!")