PydanticAI 作为 MCP 客户端示例

什么是 MCP？

MCP (Model Context Protocol) 是一个标准化协议，允许 AI 模型与外部工具和服务进行交互。PydanticAI 可以作为 MCP 客户端，连接到 MCP 服务器来扩展其功能。

环境准备

1. 安装依赖

见：PydanticAI快速入门示例
另外安装logfire，方便观察。logfire的key申请可参考logfire文档

uv pip install logfire

代码实现

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerHTTP
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.providers.openai import OpenAIProvider
import asyncio
import logfire# 配置 Logfire 用于监控和调试
logfire.configure(token='your_logfire_token_here')
logfire.instrument_pydantic_ai()# 创建 MCP 服务器连接
server = MCPServerHTTP(url='http://localhost:3001/sse')# 配置模型（使用 DashScope 的 Qwen 模型）
model = OpenAIModel('qwen-turbo',provider=OpenAIProvider(base_url='https://dashscope.aliyuncs.com/compatible-mode/v1',api_key='your_api_key_here'),
)# 创建 Agent 并连接 MCP 服务器
agent = Agent(model, mcp_servers=[server])async def main():# 启动 MCP 服务器连接async with agent.run_mcp_servers():print('MCP 客户端已启动')# 执行需要计算能力的查询result = await agent.run('How many days between 2000-01-01 and 2025-03-18?')print(f"结果: {result.output}")if __name__ == "__main__":asyncio.run(main())

核心组件说明

1. MCPServerHTTP

• 作用: 创建与 MCP 服务器的 HTTP 连接
• 参数: url - MCP 服务器的地址（这里使用 SSE 传输）

2. Agent 配置

• model: 指定使用的 AI 模型
• mcp_servers: 连接的 MCP 服务器列表

3. 异步上下文管理

async with agent.run_mcp_servers():# 在这个上下文中，MCP 服务器处于活跃状态# 可以执行需要外部工具的任务

运行步骤

1. 启动 MCP 服务器

deno run -N -R=node_modules -W=node_modules --node-modules-dir=auto jsr:@pydantic/mcp-run-python sse

输出应该显示：

2. 运行客户端代码

uv run mcp_test.py

预期输出

MCP 的优势

1. 扩展能力: 通过 MCP 服务器，AI 可以执行 Python 代码、进行复杂计算
2. 标准化: 使用标准协议，易于集成不同的工具和服务
3. 安全性: 在隔离环境中执行代码，保证安全性
4. 实时性: 支持实时数据处理和计算

常见用例

• 数学计算: 复杂的数学运算和统计分析
• 数据处理: 处理和分析大量数据
• 代码执行: 动态执行 Python 代码
• API 调用: 与外部服务进行交互

注意事项

1. 确保 MCP 服务器在客户端启动前已经运行
2. 检查网络连接和端口是否可用（默认 3001）
3. 替换示例中的 API 密钥为您的实际密钥
4. Logfire token 是可选的，用于监控和调试

参考链接：https://ai.pydantic.dev/mcp/client/#sse-client