Pydantic AI 接入 MCP 实战：构建类型安全的智能体

为什么要把 Pydantic AI 连接到 MCP

当你用 Pydantic AI 构建了一个基于明确数据类型的智能体后，往往需要让它获得外部能力——比如天气查询、数据库访问或第三方服务调用。传统做法是把这些能力的代码复制进应用、或者改写智能体的运行循环，但这既破坏了边界，也让维护变得困难。

Model Context Protocol（MCP）提供了另一条路径。通过 MCP，Pydantic AI 能够发现外部服务器暴露的工具，让模型选择并调用其中一个，再处理返回的内容。整个过程不需要把工具定义搬进客户端代码。本文基于 Emmanuel Koral 的 Agentic AI 课程 Stage 22 第 10 课，拆解如何用一个课程提供的天气服务器（提供 get_forecast 工具）构建类型安全的客户端应用。

核心思路很清晰：MCP 负责供给能力，Pydantic 负责保证应用层对象的可靠性。

Model Context Protocol（MCP）是由 Anthropic 于 2024 年末提出并开源的标准化协议，旨在解决 AI 智能体与外部工具、数据源之间的集成碎片化问题。在 MCP 出现之前，每个智能体框架都有自己的工具调用约定，导致同一个天气 API 可能需要为 LangChain、AutoGen、Pydantic AI 分别编写适配层。MCP 借鉴了 LSP（Language Server Protocol）的设计思路：定义一套通用的客户端/服务器消息格式，任何遵循该协议的服务器都能被任何兼容客户端发现和调用。协议支持三种传输方式——本地 stdio（子进程通信）、streamable HTTP 和旧版 HTTP+SSE——分别对应本地工具、远程服务和遗留系统的不同部署场景。这种标准化使得一个 MCP 服务器可以同时被 Claude Desktop、Cursor 和 Pydantic AI 等不同宿主复用，工具逻辑只需维护一份。

环境准备与协议契约

在已有的 Python 3.12 项目中执行 uv add pydantic-ai，完整包已内置 MCP 集成；如果使用精简（slim）发行版，则需要额外安装 MCP extra。Pydantic AI 要求 Python 3.10 或更新版本，当前环境满足条件。

连接之前，先检查服务器提供的契约。这台服务器对外公布了 get_forecast，带有清晰的描述，要求传入一个城市字符串（city），并接受一个可选的数字（天数）。这份契约属于未经改动的服务器——我们要做的是构建一个使用它的类型化客户端，而不是去修改服务器。

对于本地 stdio 传输方式，客户端会把 Python 服务器作为子进程启动，并通过标准输入输出交换协议消息。它用 tools/list 发现工具，再用 tools/call 调用选中的工具。这里有一个关键细节：服务器日志必须写入标准错误（stderr），因为任何多余的标准输出都可能污染协议流。

stdio 传输是理解 MCP 工作机制的最直观入口。当客户端以子进程方式启动服务器时，双方通过 stdin/stdout 交换 JSON-RPC 2.0 格式的消息。握手阶段客户端发送 initialize 请求，服务器回应自身的协议版本与能力声明；随后客户端发 tools/list 获取工具清单，每个工具包含名称、描述和符合 JSON Schema 规范的输入定义。这份 schema 既是给模型的语义提示（描述字段说明工具用途），也是运行时的参数校验依据（类型和必填字段由 schema 约束）。正因为整个发现过程发生在运行时而非编译期，客户端代码无需提前知晓服务器的工具列表——这也是为什么"不 import 服务器文件"是 MCP 客户端的正确姿势：一旦直接导入，边界隔离的优势就荡然无存。

定义客户端与输出验证模型

打开 client.py，依次导入所需模块：asyncio 管理生命周期、os 从环境变量配置模型、BaseModel 定义最终结果、Agent 运行智能体，以及 MCPToolset 直接控制 MCP 传输。这个 toolset 把智能体连接到外部服务器所公布的能力上。

接着定义一个 ForecastSummary 模型，包含 city、temperature_c、conditions 和 source 字段。这个模型描述的是智能体运行结束后，应用层必须收到的数据形态。

这个模型定义了应用在智能体激活后必须接收的内容

这里有个容易被忽略的要点：一次成功的 MCP 工具调用，并不能保证最终输出符合预期形状。模型可能在工具返回之后又生成了自由文本。因此，最终的智能体输出需要一个独立的验证边界，而这正是 Pydantic 输出模型所提供的保障。

搭建连接与运行智能体

创建 MCPToolset，以 Python 命令和 course_weather_server.py 作为参数——这定义了一台本地 stdio 服务器。当连接打开时，Pydantic AI 会启动该进程并与之通信。注意，我们的应用并不 import 服务器文件，能力始终被保留在独立的协议边界之后。

然后创建智能体：使用课程配置的模型，给出聚焦的指令，通过 toolset 参数挂载天气工具，并把 output_type 设为 ForecastSummary。这套集成是跨服务商通用的，环境会选择一个支持所需工具交互的模型。

在 main 中使用 async with agent，它会在运行前打开已注册的 MCP 连接并启动本地子进程。调用 agent.run 传入用户请求，再读取 result.output。由于设置了 output_type，这个值是一个经过验证的 ForecastSummary，而非未经结构化的文本猜测。最后用 asyncio.run(main()) 收尾——上下文在整个运行期间（包括发现与工具调用）保持打开，然后干净地关闭。

本地 stdio 下客户端会启动并停止其子进程

虽然 Pydantic AI 也能在运行期间自动管理 toolset，但显式的上下文有助于学习者看清进程启动、连接存续和关闭的全过程。

发现、调用与类型化结果

用 uv run python client.py 运行客户端，应用会启动服务器进程并打开 MCP 会话。第一个重要事件是发现：客户端请求可用工具，服务器返回 get_forecast 及其描述和输入 schema，Pydantic AI 把这份契约转化为模型可以选择的能力。我们没有把工具定义复制进 client.py，而是直接复用了 MCP 提供的内容。

模型看到被发现的能力后选择 get_forecast，参数指定了 Nairobi 和两天。应用把调用发送给服务器（而不是本地 Python 函数），在这个外部边界上，服务器的 schema 决定了可接受的参数。服务器返回工具内容——在课程 fixture 中包含请求的城市、预报数值、天气状况和来源标签。Pydantic AI 把结果送回模型，为完成应用响应提供依据。

应用可以显示城市或比较数值温度

最终模型产生智能体结果，Pydantic 据 ForecastSummary 完成验证。下游代码可以直接把 result.output.city 或 result.output.temperature_c 当作类型化的值来用，无需解析散文式文本，也不必依赖标签和格式的一致性。

这里要保持两类 schema 相互独立：MCP 输入 schema 保护进入外部能力的调用；（若提供）MCP 输出 schema 描述该能力的返回内容；而我们的 Pydantic 输出类型保护进入应用的最终智能体结果。三条边界各司其职，谁也不能替代谁。

处理失败：重试、暴露与传播

要测试连接边界，可以制造一次受控的失败。临时请求 0 天预报——服务器契约只允许 1 到 7 天，这就构成了一个非法的工具参数。它比随机破坏实现更安全、信息更明确，因为我们清楚知道哪条规则应该拒绝这次调用。

服务器拒绝 days=0 并返回工具错误。默认情况下，模型会把这个失败当作指引接收，并可以重试，而不会收到编造的天气数据。追踪日志会显示被拒绝的参数和验证消息，让问题暴露而非被掩盖。随后模型把参数改为 1 天再次调用，服务器成功，ForecastSummary 完成最终校验。

当更改参数可能成功时，可以让模型重试

不过要限制重试次数，重复的非法调用会浪费时间和模型用量。而“服务器不可用”是另一回事——命令可能缺失、脚本路径可能错误、远程端点可能无法连通，此时模型根本没有工具结果可供修复。这应被当作应用集成失败处理：记录哪台服务器不可用，并给出清晰的恢复提示，而不是凭空编造答案。

MCPToolset 提供三种错误处理方式：

重试模型——当更改参数可能让调用成功时；
暴露失败的工具结果——当模型应当解释错误或改选其他允许的能力时；
传播服务器异常——当应用代码必须自行管理日志、降级或用户提示时。

具体选哪种，取决于操作后果的严重程度。

Pydantic AI 的重试机制与底层模型的工具调用循环密切相关。当工具返回错误时，框架会将错误内容作为一条新的"工具结果消息"追加到对话上下文，模型在下一轮推理中可以读到这条失败信息并决定是否修正参数重试。这与人类使用命令行工具的调试方式非常相似——看到报错、理解原因、调整参数、再次执行。max_retries 参数（默认值通常为 1）控制这个循环的上限，防止模型陷入无效重试的死循环。值得注意的是，重试仅对"参数可修正"的场景有意义：如果错误来自服务器不可用或网络超时，模型没有可供修正的参数，此时应当跳过重试直接走传播路径，否则只是在浪费 token 配额和等待时间。

远程连接与安全边界

如果要接入远程服务，给它一个 streamable HTTP 端点而非本地命令，并把端点与凭证存放在基于环境变量的配置里，再通过同样的 toolset 参数挂载。对新集成，应使用 streamable HTTP，而不是已弃用的 legacy HTTP + SSE 传输。

远程连接会扩大信任边界，意味着更多系统需要保护。应在合适的位置做认证、在服务器端校验请求来源、把仅限本地的服务绑定到 localhost。除非工具契约明确要求，绝不要把凭证放进 prompt 或工具参数。

连接服务器等于给智能体赋予能力，网络访问与工具权限同样重要。原则是只暴露任务所需的工具：Pydantic AI 的 toolset 支持动态过滤，前缀（prefix）还能区分来自不同服务器的同名工具。过滤移除无关选项并限制权限，前缀防止命名冲突并标明每个能力的来源。给模型一组精简、清晰、权限适当的工具，是安全设计的基本功。

完整路径回顾

至此，整条路径已经清晰：MCPToolset 连接到供给服务器 → toolset 把发现到的契约暴露给智能体 → async 上下文管理连接 → 服务器运行被选中的能力 → output_type 在应用代码接收前验证结果。

你现在已经能够从一个类型化的 Pydantic AI 应用中使用 MCP 服务器能力，并把工具契约与最终输出验证彻底分离。课程的下一课（Stage 22 Lesson 11）将对比 Strands 与 Pydantic AI 的架构差异。