Pydantic AI MCP

MCP与Pydantic AI-完整学习指南

使用Pydantic AI学习模型上下文协议（MCP）的全面实践指南。本指南包括每个概念的详细解释和工作代码示例。

🎯 你将学到什么

• 什么是MCP 以及为什么它对人工智能应用具有革命性
• MCP架构 -客户端、服务器、传输和功能
• 构建MCP客户端 -通过STDIO、HTTP/SSE连接到服务器
• 配置管理 -JSON配置、环境变量、多环境设置
• FastMCP集成 -零延迟直接集成的增强功能
• 构建MCP服务器 -创建工具、资源和提示
• 生产模式 -错误处理、重试、安全、最佳实践

📚 导向结构

mcp-pydantic-ai-guide/
├── 01-building-blocks/          # Core concepts and fundamentals
│   ├── 01-what-is-mcp.md       # Introduction to MCP
│   ├── 02-mcp-architecture.md   # Understanding the architecture
│   └── basic_concepts.py        # Runnable Python examples
│
├── 02-mcp-clients/              # Working with MCP clients
│   ├── 01_stdio_transport.py   # STDIO (subprocess) connections
│   ├── 02_http_sse_transport.py # HTTP/SSE remote connections
│   └── 03_json_config.py        # JSON configuration patterns
│
├── 03-fastmcp-client/           # FastMCP enhanced features
│   └── 01_fastmcp_basics.py     # Direct integration and usage
│
├── 04-mcp-servers/              # Building your own servers
│   └── 01_basic_server.py       # Creating tools and servers
│
├── examples/                     # Complete working examples
│   └── complete_example.py      # Full research assistant app
│
└── README.md                    # This file

🚀 快速开始

先决条件

1. Python 3.10+
2. 包管理器 （建议使用pip或uv）
3. API密钥 （可选-用于运行示例）：

- OpenAI API密钥（用于GPT-4） - 或任何其他受支持的LLM提供商

安装

选项1：使用pip

# Install Pydantic AI with MCP support
pip install "pydantic-ai-slim[openai,fastmcp]"

选项2：使用紫外线（推荐）

# Install uv if you don't have it
curl -LsSf https://astral.sh/uv/install.sh | sh

# Install dependencies
uv add "pydantic-ai-slim[openai,fastmcp]"

设置环境变量

# Create a .env file
echo "OPENAI_API_KEY=your-api-key-here" > .env

# Or export directly
export OPENAI_API_KEY="your-api-key-here"

📖 学习路径

路径1：适合初学者

如果您是MCP的新手，请从这里开始：

1. 阅读: 01-building-blocks/01-what-is-mcp.md

- 了解MCP是什么以及它存在的原因

1. 阅读: 01-building-blocks/02-mcp-architecture.md

- 了解客户端、服务器和传输

1. 跑: 01-building-blocks/basic_concepts.py

   python 01-building-blocks/basic_concepts.py

- 查看实际应用中的概念

1. 跑: 02-mcp-clients/01_stdio_transport.py

   python 02-mcp-clients/01_stdio_transport.py

- 学习连接到本地MCP服务器

1. 跑: 02-mcp-clients/02_http_sse_transport.py

   python 02-mcp-clients/02_http_sse_transport.py

- 学习连接到远程服务器

1. 跑: examples/complete_example.py

   python examples/complete_example.py

- 查看完整的工作应用程序

路径2：开发人员构建MCP服务器

如果您想构建MCP服务器，请从这里开始：

1. 阅读：核心概念（构建块）

- 获得基础

1. 跑: 04-mcp-servers/01_basic_server.py

   python 04-mcp-servers/01_basic_server.py

- 学习创建自己的MCP服务器

1. 跑: 03-fastmcp-client/01_fastmcp_basics.py

   python 03-fastmcp-client/01_fastmcp_basics.py

- 了解FastMCP的直接集成

1. 学习: examples/complete_example.py

- 查看生产就绪模式

路径3：用于生产应用

从这里开始供企业/生产使用：

1. 学习：配置管理

   python 02-mcp-clients/03_json_config.py

- 了解特定于环境的配置

1. 学习：HTTP/SSE传输

   python 02-mcp-clients/02_http_sse_transport.py

- 生产部署模式 - SSL/TLS配置 - 错误处理和重试

1. 学习：完整示例

   python examples/complete_example.py

- 生产就绪代码结构

💡 关键概念

MCP在60秒内

主控程序 =模型上下文协议=AI应用程序使用外部工具的标准方式

# Without MCP: Custom integration for each tool
def my_agent():
    # Custom code for Tool A
    # Custom code for Tool B
    # Custom code for Tool C
    # = Lots of work!

# With MCP: One protocol, many tools
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio

server = MCPServerStdio('uv', args=['run', 'mcp-server'])
agent = Agent('openai:gpt-4', toolsets=[server])
# All tools from server are automatically available!

将MCP与Pydantic AI结合使用的三种方法

1.MCPServer（标准协议）

from pydantic_ai.mcp import MCPServerStdio

# Connect to subprocess server
server = MCPServerStdio('uv', args=['run', 'mcp-server'])
agent = Agent('openai:gpt-4', toolsets=[server])

使用时间：使用标准MCP服务器，需要完全的MCP规范支持

2.FastMCP工具集（增强功能）

from fastmcp import FastMCP
from pydantic_ai.toolsets.fastmcp import FastMCPToolset

# Embed server directly (zero latency!)
server = FastMCP('my_server')

@server.tool()
async def my_tool(x: int) -> int:
    return x * 2

toolset = FastMCPToolset(server)
agent = Agent('openai:gpt-4', toolsets=[toolset])

使用时间：构建单个应用程序，需要性能，想要简单的Python工具

3.JSON配置

from pydantic_ai.mcp import load_mcp_servers

# Load multiple servers from config
servers = load_mcp_servers('mcp_config.json')
agent = Agent('openai:gpt-4', toolsets=list(servers.values()))

使用时间：管理多个服务器，跨工具共享配置

运输类型

🔧 常见模式

模式1：本地开发服务器

from fastmcp import FastMCP
from pydantic_ai import Agent
from pydantic_ai.toolsets.fastmcp import FastMCPToolset

# Create server with tools
server = FastMCP('dev-server')

@server.tool()
async def hello(name: str) -> str:
    return f"Hello, {name}!"

# Use directly
toolset = FastMCPToolset(server)
agent = Agent('openai:gpt-4', toolsets=[toolset])

# Run
result = await agent.run('Say hello to Alice')

模式2：生产多服务器

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerSSE

# Multiple remote services
search = MCPServerSSE('https://search-api.com/mcp')
database = MCPServerSSE('https://db-api.com/mcp')
analytics = MCPServerSSE('https://analytics-api.com/mcp')

# Agent with all servers
agent = Agent(
    'openai:gpt-4',
    toolsets=[search, database, analytics]
)

模式3：基于环境的配置

import os
from pathlib import Path
from pydantic_ai.mcp import load_mcp_servers
from pydantic_ai import Agent

# Load environment-specific config
env = os.getenv('APP_ENV', 'development')
config_path = Path(f'mcp_config_{env}.json')

# Create agent
servers = load_mcp_servers(config_path)
agent = Agent('openai:gpt-4', toolsets=list(servers.values()))

📝 示例用例

研究助理（见 examples/complete_example.py)

• 搜索网页
• 保存和组织笔记
• 分析文本
• 生成摘要
• 导出报告

代码分析工具

• 读取代码文件
• 运行静态分析
• 执行测试
• 生成文档

数据管道代理

• 查询数据库
• 转换数据
• 生成报告
• 发送通知

DevOps助理

• 监控系统
• 检查日志
• 部署服务
• 管理基础设施

🐛 故障排除

常见问题

1. ModuleNotFoundError: No module named 'fastmcp'

解决方案:

pip install "pydantic-ai-slim[fastmcp]"

2. Server timeout 或连接错误

解决方案:

# Increase timeout
server = MCPServerStdio(
    'uv',
    args=['run', 'server'],
    timeout=30  # Increase from default
)

3.API关键错误

解决方案:

# Ensure environment variable is set
export OPENAI_API_KEY="your-key"

# Or use .env file
echo "OPENAI_API_KEY=your-key" > .env

4.SSL证书错误

解决方案:

import httpx
import ssl

# Custom SSL context
ssl_context = ssl.create_default_context()
client = httpx.AsyncClient(verify=ssl_context)

server = MCPServerSSE('https://...', http_client=client)

🔐 安全最佳实践

1. 永远不要提交API密钥 到版本控制
2. 使用环境变量 为了秘密
3. 启用SSL/TLS 生产中
4. 验证输入 在MCP工具中
5. 优雅地处理错误
6. 使用连接超时
7. 实施速率限制
8. 记录安全事件

🚢 部署

Docker示例

FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install -r requirements.txt

# Copy code
COPY . .

# Set environment variables
ENV OPENAI_API_KEY=""
ENV APP_ENV="production"

# Run server
CMD ["python", "-m", "your_mcp_server"]

环境变量

# Development
APP_ENV=development
OPENAI_API_KEY=your-dev-key
DEBUG=true

# Production
APP_ENV=production
OPENAI_API_KEY=your-prod-key
DEBUG=false
SSL_VERIFY=true

📚 额外资源

• 官方文件: https://ai.pydantic.dev/mcp/overview/
• MCP规范: https://spec.modelcontextprotocol.io/
• Pydantic AI: https://ai.pydantic.dev/
• FastMCP: https://github.com/jlowin/fastmcp
• MCP服务器: https://github.com/modelcontextprotocol/servers

🤝 贡献

发现问题或想改进指南？欢迎投稿！

📄 许可证

本指南作为教育材料提供。请随意使用和适应您的学习。

🎓 学习技巧

1. 运行每个示例 -不要只是阅读，要执行代码
2. 修改示例 -更改参数，添加功能
3. 建造一些东西 -将所学应用于实际项目
4. 读取错误消息 -他们很有教育意义！
5. 实验 -打破事物并修复它们
6. 分享 -把你学到的东西教给别人

✅ 清单

使用此清单跟踪您的进度：

• \[\]了解MCP是什么以及它存在的原因
• \[\]了解客户端和服务器之间的区别
• \[\]可以连接到STDIO服务器
• \[\]可以连接到HTTP/SSE服务器
• \[\]了解JSON配置
• \[\]了解何时使用FastMCP与MCPServer
• \[\]可以构建基本的MCP服务器
• \[\]可以定义具有适当类型的工具
• \[\]了解错误处理模式
• \[\]可以部署到生产环境

🎉 后续步骤

完成本指南后：

1. 构建自己的MCP服务器 对于一个真实的用例
2. 集成多个MCP服务器 在应用程序中
3. 部署到生产环境 有适当的安全保障
4. 共享您的服务器 与MCP社区
5. 贡献 MCP生态系统

______________________________________________________________________

快乐学习！ 🚀

如果你觉得这本指南很有帮助，可以考虑与其他学习MCP的人分享！