尧图网站建设 
Pydantic AI 与 MCP从 MCP 客户端到 MCP 服务器的完整接入指南【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-aiPydantic AI 对 Model Context ProtocolMCP 提供了双向、多层次的完整支持Agent 既可以作为 MCP 客户端连接外部 MCP 服务器并使用其工具也可以把 Agent 本身封装进 MCP 服务器、通过工具调用暴露给任意 MCP 客户端。本文以 docs/mcp/overview.md 为主线结合客户端文档 docs/mcp/client.md、服务器文档 docs/mcp/server.md 以及仓库源码系统讲解在 Pydantic AI 中使用 MCP 的推荐路径、底层实现与高级用法。读完本文你将掌握MCP能力、MCPToolset、MCPServerTool三条接入路径的适用场景并能搭建完整的 MCP 客户端与服务端应用。MCP 是什么为什么 Agent 需要它Model Context Protocol 是一种标准化协议允许 AI 应用——包括 Pydantic AI 这类程序化 Agent、Cursor 等编码 Agent、Claude Desktop 等桌面应用——通过统一接口连接外部工具与服务。与所有协议一样MCP 的愿景是让大量应用无需逐一定制集成即可互相通信。官方维护了一份 MCP 服务器清单涵盖搜索、数据库、GitHub、Slack 等常见服务。落到 Pydantic AI 的实际场景中这意味着Pydantic AI 可以接入一个以 MCP 服务器形式实现的网络搜索服务构建深度研究型 Agent其他 MCP 客户端如 Cursor可以连接 Pydantic 官方提供的 MCP 服务器来检索日志、链路与指标辅助排查 Bug任何 MCP 客户端都可以连接 Pydantic 的 Run Python MCP 服务器在沙箱环境中运行任意 Python 代码。Pydantic AI 的 MCP 支持分为两大方向Agent 连接 MCP 服务器客户端侧以及Agent 被封装进 MCP 服务器服务端侧。下面分别展开。连接 MCP 服务器三种接入路径怎么选Pydantic AI 提供三条连接 MCP 服务器的路径按推荐程度排列1.MCP能力推荐声明式接入默认本地运行 MCP 服务器凭据、钩子、追踪都由你掌控并可通过一个nativeTrue标志选择使用模型提供商的原生 MCP 支持同一 Agent 无需改代码即可跨提供商工作。相关实现见 pydantic_ai_slim/pydantic_ai/capabilities/mcp.py 与完整文档 docs/capabilities/mcp.md。2.MCPToolset工具集底层直接管理工具集生命周期、在多个 Agent 间共享同一个 MCP 服务器或传入MCP能力未暴露的高级传输/客户端配置。通过toolsets[...]注册到 Agent详见 docs/mcp/client.md。3.MCPServerTool原生工具仅原生当只需要模型提供商的原生 MCP 支持、不需要本地回退时可直接将MCPServerTool作为原生工具使用见 docs/native-tools.md#mcp-server-tool。路径一MCP能力 —— 一行代码同时获得本地回退与原生 MCPMCP是一个提供商自适应能力provider-adaptive capability是 Pydantic AI 中 MCP 的主要入口。默认本地运行 MCP 服务器让凭据、钩子与追踪保持在你的控制之下同时支持基于 URL 的服务器以及直接的 client / toolset / transport 输入from pydantic_ai import Agent from pydantic_ai.capabilities import MCP agent Agent( openai:gpt-5.2, capabilities[ # 默认在本地运行 MCP 服务器 MCP(urlhttps://mcp.example.com/api), # 选择原生 MCP —— 若模型不支持则回退到本地 MCP(urlhttps://mcp.example.com/other, nativeTrue), ], )关键点将 URL 作为第一个参数传入即可同时启用本地回退与设置nativeTrue时提供商原生 MCP本地侧local接受任何MCPToolset输入——URL、FastMCP transport、预构建的fastmcp.Client、进程内FastMCP服务器、本地脚本路径等非工具集输入会自动包装为MCPToolset原生侧由MCPServerTool支撑。需要完全控制如自定义id、authorization_token、description时可直接传nativeMCPServerTool(...)。从源码看MCP类继承自NativeOrLocalToolpydantic_ai_slim/pydantic_ai/capabilities/mcp.py#L26-L28其设计意图即原生优先、本地兜底模型支持原生 MCP 时走提供商侧执行上下文更优化、缓存更高效、无往返 Pydantic AI 的延迟不支持时自动回退本地。四种常见组合如下from pydantic_ai.capabilities import MCP from pydantic_ai.native_tools import MCPServerTool # URL 型 MCP 服务器本地运行需要 pydantic-ai-slim[mcp] MCP(https://mcp.example.com/api) # 无 URL 的本地客户端 —— 传任意 MCPToolset 输入 MCP(localmy_fastmcp_client) # 原生优先URL 型本地回退 MCP(https://mcp.example.com/api, nativeTrue) # 仅原生无本地 —— 不需要 mcp extra MCP(https://mcp.example.com/api, nativeTrue, localFalse) # 显式原生 显式本地 —— 两侧独立配置 MCP( nativeMCPServerTool( idpublic-mcp, urlhttps://relay.example.com/mcp, authorization_tokenrelay-token, ), localmy_fastmcp_client, )路径二MCPToolset—— 底层客户端掌控生命周期MCPToolset是 Pydantic AI 连接 MCP 服务器的底层工具集包装了 FastMCP 客户端同时支持本地stdio与远程Streamable HTTP、SSEMCP 服务器。完整文档见 docs/mcp/client.md实现位于 pydantic_ai_slim/pydantic_ai/mcp.py。安装需要安装pydantic-ai或以mcp可选组安装pydantic-ai-slimpip/uv-add pydantic-ai-slim[mcp]注意FastMCP 4 目前是预发布版本需显式安装。MCPToolset支持它但其现代协议模式不支持服务器发起的 sampling 与 elicitation也无法应用log_level此时MCPToolset会给出警告请在log_handler中过滤日志。这些选项在 FastMCP 4 的 legacy 协议模式下仍保留 FastMCP 3 的行为。支持的输入形态MCPToolset第一个位置参数接受以下任意一种URL 字符串Streamable HTTP若路径以/sse结尾则自动识别为 SSE本地 Python 或 Node.js 脚本路径通过 stdio 运行FastMCP transport如StdioTransport、StreamableHttpTransport、SSETransport预构建的fastmcp.Client用于 OAuth、工具转换等高级 FastMCP 配置进程内FastMCP服务器用于测试或单进程部署无网络往返。每个MCPToolset实例是一个工具集可通过toolsets参数注册到Agent。生命周期管理上既可以用async with agent统一开关所有已注册工具集的连接stdio 服务器还会随之启停子进程也可以用async with toolset单独管理某个工具集适合跨多 Agent 共享。若未显式进入任一上下文管理器工具集会按需自动打开与关闭。四种连接方式的完整示例Streamable HTTP连接远程 MCP 服务器的推荐方式。需要先运行一个支持该传输的服务端from mcp.server.fastmcp import FastMCP app FastMCP() app.tool() def add(a: int, b: int) - int: return a b if __name__ __main__: app.run(transportstreamable-http)from pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset toolset MCPToolset(http://localhost:8000/mcp) # (1)! agent Agent(openai:gpt-5.2, toolsets[toolset]) # (2)! async def main(): result await agent.run(What is 7 plus 5?) print(result.output) # The answer is 12.用连接 URL 定义 MCP 工具集创建挂载该工具集的 Agent。运行此示例时需导入asyncio并追加asyncio.run(main())其余无需改动。这一过程完整展示了 MCP 客户端的工作链路模型收到 What is 7 plus 5? 提示 → 模型决定调用add工具 → 模型返回工具调用 → Pydantic AI 通过 Streamable HTTP 把工具调用发给 MCP 服务器 → 服务器执行add返回 12 → 模型携带返回值被再次调用 → 模型给出最终答案。如需可视化整个过程甚至直接看到工具调用可在示例中追加三行 logfire 插桩代码import logfire logfire.configure() logfire.instrument_pydantic_ai()SSEHTTP Server-Sent Events 传输同样受支持但已在 MCP 中被弃用新部署应优先 Streamable HTTP。URL 以/sse结尾时自动识别为 SSE其他路径需显式传入SSETransportfrom pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset toolset MCPToolset(http://localhost:3001/sse) agent Agent(openai:gpt-5.2, toolsets[toolset])Stdio服务器作为子进程运行通过stdin/stdout通信。传脚本路径或用StdioTransport完全控制命令、参数与环境from fastmcp.client.transports import StdioTransport from pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset toolset MCPToolset(StdioTransport(commandpython, args[mcp_server.py])) agent Agent(openai:gpt-5.2, toolsets[toolset])进程内 FastMCP 服务器服务器与 Agent 在同一 Python 进程省去网络往返from fastmcp import FastMCP from pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset fastmcp_server FastMCP(my_server) fastmcp_server.tool() async def add(a: int, b: int) - int: return a b toolset MCPToolset(fastmcp_server) agent Agent(openai:gpt-5.2, toolsets[toolset]) async def main(): result await agent.run(What is 7 plus 5?) print(result.output) # The answer is 12.路径三MCPServerTool—— 仅使用模型提供商的原生 MCP当只想要提供商原生 MCP 支持不需要本地回退时可直接使用MCPServerTool作为原生工具。它要求 MCP 服务器位于提供商可访问的公网 URL不支持 Pydantic AI Agent 侧 MCP 的许多高级特性但能获得更优化的上下文使用与缓存、以及省去回程 Pydantic AI 的更低延迟。当前 OpenAI Responses、Anthropic、xAI 三家提供支持Google 等暂不支持。from pydantic_ai import Agent, MCPServerTool from pydantic_ai.capabilities import NativeTool agent Agent( anthropic:claude-sonnet-4-6, capabilities[ NativeTool( MCPServerTool( iddeepwiki, urlhttps://mcp.deepwiki.com/mcp, ) ) ] ) result agent.run_sync(Tell me about the pydantic/pydantic-ai repo.) print(result.output)MCPServerTool支持authorization_tokenOpenAI/Anthropic/xAI、allowed_tools三家、descriptionOpenAI/xAI、headersOpenAI/xAI等配置项。使用 OpenAI Responses 时还可通过x-openai-connector:connector_id形式的特殊 URL 接入 OpenAI 的 MCP Connectors。从配置文件批量加载 MCP 工具集当需要管理多个 MCP 服务器、或希望在不改代码的情况下从外部配置服务器时可以用load_mcp_toolsets()从 JSON 配置文件批量加载工具集。配置格式配置文件包含一个mcpServers对象每个服务器以唯一键标识{ mcpServers: { python-runner: { command: uv, args: [run, mcp-run-python, stdio] }, weather: { command: python, args: [mcp_server.py] }, weather-api: { url: http://localhost:3001/sse }, calculator: { url: http://localhost:8000/mcp } } }每个条目支持command、args、env、cwdstdio 服务器或url、headersHTTP 服务器。加载时会进行校验类型错误的字段会立即报错而非等到连接时未知键会被忽略因此与其他 MCP 客户端共享的配置文件仍可加载但只忽略、绝不生效——特别是disabled不会跳过服务器type不会选择传输方式传输方式由 URL 推断仅以/sse结尾视为 SSE其他一律视为 Streamable HTTP。环境变量展开配置文件支持${VAR}与${VAR:-default}语法展开环境变量与 Claude Code 的 MCP 配置一致便于把 API Key、主机名等敏感信息留在配置文件之外{ mcpServers: { python-runner: { command: ${PYTHON_CMD:-python3}, args: [run, ${MCP_MODULE}, stdio], env: { API_KEY: ${MY_API_KEY} } }, weather-api: { url: https://${SERVER_HOST:-localhost}:${SERVER_PORT:-8080}/sse } } }${VAR}会被替换为对应环境变量值${VAR:-default}在环境变量未设置时使用默认值警告使用${VAR}语法时若环境变量未定义将抛出ValueError请用${VAR:-default}提供回退安全警告配置文件指定了要作为子进程启动的可执行文件与参数能写配置的人即可执行任意命令${VAR}按完整进程环境展开、无白名单配置文件还能读取任何环境变量。因此只加载你控制的配置文件切勿加载不可信来源的配置。用法from pydantic_ai import Agent from pydantic_ai.mcp import load_mcp_toolsets # 从配置文件加载所有工具集 toolsets load_mcp_toolsets(mcp_config.json) # 创建挂载所有工具集的 Agent agent Agent(openai:gpt-5.2, toolsetstoolsets) async def main(): result await agent.run(What is 7 plus 5?) print(result.output)客户端侧高级用法工具调用定制process_tool_callMCPToolset接受process_tool_call回调用于定制工具调用请求及其响应。常见用途是注入服务端处理器需要读取的元数据——例如把 run context 的 deps 传给服务器from typing import Any from fastmcp.client.transports import StdioTransport from pydantic_ai import Agent, RunContext from pydantic_ai.mcp import CallToolFunc, MCPToolset, ToolResult from pydantic_ai.models.test import TestModel async def process_tool_call( ctx: RunContext[int], call_tool: CallToolFunc, name: str, tool_args: dict[str, Any], ) - ToolResult: A tool call processor that passes along the deps. return await call_tool(name, tool_args, {deps: ctx.deps}) toolset MCPToolset( StdioTransport(commandpython, args[mcp_server.py]), process_tool_callprocess_tool_call, ) agent Agent( modelTestModel(call_tools[echo_deps]), deps_typeint, toolsets[toolset], ) async def main(): result await agent.run(Echo with deps set to 42, deps42) print(result.output) # {echo_deps:{echo:This is an echo message,deps:42}}服务端如何读取注入的元数据取决于 MCP 服务器 SDK。例如 MCP Python SDK 中工具处理函数的ctx: Context参数即可访问from typing import Any from mcp.server.fastmcp import Context, FastMCP from mcp.server.session import ServerSession mcp FastMCP(Pydantic AI MCP Server) mcp.tool() async def echo_deps(ctx: Context[ServerSession, None]) - dict[str, Any]: Echo the run context. await ctx.info(This is an info message) deps: Any getattr(ctx.request_context.meta, deps) return {echo: This is an echo message, deps: deps} if __name__ __main__: mcp.run()工具错误处理tool_error_behavior当 MCP 服务器报告工具错误时MCPToolset让你选择错误应如何表现tool_error_behavior行为retry默认。抛出ModelRetry把服务器错误作为重试提示发回模型。适用于模型可能自我纠正调用的情况。failed抛出ToolFailed记录为outcomefailed的工具结果。适用于工具调用已完成但失败、由模型决定下一步的情况。error传播底层 MCP 工具异常并使 Agent 运行失败。适用于需要应用程序代码在模型循环外处理的错误。对retry与failed两种模式结构化错误内容会以 JSON 序列化进模型可见消息因此重试提示等机器可读细节对模型仍然可见协议与传输层错误不会被报告为已完成的失败工具调用。这与本地工具代码中工具重试与失败工具结果的区分是 MCP 层面的对应物。从源码看tool_error_behavior是MCPToolset的公开字段默认值为retrypydantic_ai_slim/pydantic_ai/mcp.py#L764-L770。工具前缀避免命名冲突连接多个可能提供同名工具的 MCP 服务器时用.prefixed(...)包装每个MCPToolset为其工具名加前缀from pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset weather MCPToolset(http://localhost:3001/sse).prefixed(weather) # weather_* calculator MCPToolset(http://localhost:3002/sse).prefixed(calc) # calc_* # 两个服务器可能都暴露 get_data 工具但会被区分为 # weather_get_data 和 calc_get_data。 agent Agent(openai:gpt-5.2, toolsets[weather, calculator])服务器指令注入include_instructionsMCP 服务器可在初始化期间提供指令说明如何最好地使用其工具。连接建立后可通过MCPToolset.instructions访问设置include_instructionsTrue可自动注入 Agent 的指令集from pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset toolset MCPToolset(http://localhost:8000/mcp, include_instructionsTrue) agent Agent(openai:gpt-5.2, toolsets[toolset])源码中include_instructions默认值为False向后兼容开启后服务器初始化返回的指令会被加入 Agent 的指令集pydantic_ai_slim/pydantic_ai/mcp.py#L813-L818。工具元数据与过滤MCP 工具可携带描述工具特征的元数据在过滤工具时非常有用。meta与annotations字段位于传给过滤函数的ToolDefinition对象的metadata字典上工具的输出 schema如有则作为return_schema字段可用。MCPToolset还额外暴露task: bool标志表示该工具集是否会为工具使用任务增强执行task-augmented execution。后台任务Background TasksMCPToolset支持 MCP 的任务增强执行SEP-1686。使用 SEP-1686 的服务器包括 FastMCP 3可通过execution.taskSupport声明每个工具的任务支持MCPToolset据此路由调用execution.taskSupport行为required始终以taskTrue调用。服务器创建任务客户端通过tasks/result等待最终结果。optional默认以taskTrue调用。设置prefer_tasksFalse可改为普通调用。forbidden或缺失普通调用。FastMCP 4 使用更新的 MCP Tasks 扩展SEP-2663由服务器主导任务创建因此上述task元数据与prefer_tasks偏好适用于 FastMCP 3 而非 FastMCP 4。普通调用即可驱动仅任务型工具完成无需额外安装而显式选择 tasks 扩展use_taskTrue需要单独的fastmcp-tasks包可通过mcp-tasks可选组安装pip install pydantic-ai-slim[mcp-tasks]。对 FastMCP 3 服务器用pip install fastmcp[tasks]3,4安装 tasks extra并通过taskTaskConfig(mode...)按工具声明任务支持from fastmcp import FastMCP from fastmcp.server.tasks import TaskConfig mcp FastMCP(long_running_server) mcp.tool(taskTaskConfig(modeoptional)) async def deep_research(topic: str) - str: import asyncio await asyncio.sleep(0) return fResearched {topic} if __name__ __main__: mcp.run(transportstreamable-http)默认MCPToolset在工具支持时即采用任务增强执行偏好普通调用者可设prefer_tasksFalse不影响任务支持为 required 的工具from pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset toolset MCPToolset(http://localhost:8000/mcp, prefer_tasksFalse) agent Agent(openai:gpt-5.2, toolsets[toolset])资源ResourcesMCP 服务器可提供资源——文件、数据或内容供客户端访问。MCP 中的资源是应用驱动的由宿主应用决定如何手动将上下文纳入而不会自动暴露给 LLM除非工具返回ResourceLink或EmbeddedResource。MCPToolset暴露三个方法list_resources()—— 列出服务器上所有可用资源list_resource_templates()—— 列出带参数占位符的资源模板read_resource(uri)—— 按 URI 读取特定资源内容。文本内容返回为str二进制内容返回为BinaryContent。完整示例见 docs/mcp/client.md#resources先运行暴露resource://user_name.txt资源的 FastMCP 服务器客户端用async with toolset打开连接后依次list_resources()与read_resource(resource://user_name.txt)读取输出Alice。HTTP 认证与多用户认证对 HTTP 传输MCPToolset接受auth参数bearer token 字符串、任意httpx.Auth或字面量字符串oauth启用 FastMCP 的 OAuth 流程静态请求头如 API Key可经headers参数传入。多用户/多租户应用中每个用户通常有自己的 MCP 服务器凭据如租户级 bearer token。注意共享的MCPToolset实例是单一身份——它维护一个 MCP 会话被所有并发 Agent 运行共享连接由最先需要它的运行建立认证随之解析直到最后一个运行结束才拆除。在共享实例上从ContextVar等任务局部状态派生逐请求凭据是无效的重叠运行会静默地使用打开会话的那个运行的凭据发送请求。要让每次并发运行使用对应用户的凭据需要为每次运行创建独立的MCPToolset实例以建立各自认证的会话。推荐方式是用agent.toolset装饰器动态构建工具集被装饰函数会收到 run context可从运行的依赖中读取用户凭据from dataclasses import dataclass from pydantic_ai import Agent, RunContext from pydantic_ai.mcp import MCPToolset dataclass class UserDeps: mcp_token: str agent Agent(openai:gpt-5.2, deps_typeUserDeps) agent.toolset(per_run_stepFalse) # (1)! def user_mcp_server(ctx: RunContext[UserDeps]) - MCPToolset: return MCPToolset(http://localhost:8000/mcp, authctx.deps.mcp_token) async def main(): result await agent.run(What is 7 plus 5?, depsUserDeps(mcp_tokentoken)) print(result.output) # The answer is 12.per_run_stepFalse使工具集每次运行构建一次而非每个运行步骤前构建整个运行共享单个 MCP 会话。由于每次运行的工具集会话在运行内部建立ContextVar中持有的凭据在此模式下也能正确解析——但通过 deps 传递更显式、不依赖任务局部状态。另一种替代方案是每次请求自行构造新的MCPToolset并传给运行方法的toolsets参数。自定义 TLS/SSL 配置某些环境需要调整 HTTPS 连接方式——例如信任内部 CA、为mTLS出示客户端证书或仅限本地开发时完全禁用证书校验。MCPToolset提供http_client参数可传入预先配置好的httpx.AsyncClientimport ssl import httpx from pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset # 信任内部/自签名 CA ssl_ctx ssl.create_default_context(cafile/etc/ssl/private/my_company_ca.pem) # 可选为双向 TLS 加载客户端证书 ssl_ctx.load_cert_chain(certfile/etc/ssl/certs/client.crt, keyfile/etc/ssl/private/client.key) http_client httpx.AsyncClient(verifyssl_ctx, timeouthttpx.Timeout(10.0)) toolset MCPToolset(http://localhost:3001/sse, http_clienthttp_client) # (1)! agent Agent(openai:gpt-5.2, toolsets[toolset])提供http_client后Pydantic AI 会为每个请求复用该客户端httpx 支持的一切verify、cert、自定义代理、超时等都适用于所有 MCP 流量。客户端标识client_info连接 MCP 服务器时可指定一个Implementation对象作为客户端信息在初始化期间发送给服务器。用途包括在服务器日志中标识应用、允许服务器基于客户端提供自定义行为、调试监控 MCP 连接、版本特性协商from mcp import types as mcp_types from pydantic_ai.mcp import MCPToolset toolset MCPToolset( http://localhost:3001/sse, client_infomcp_types.Implementation( nameMyApplication, version2.1.0, ), )MCP Sampling客户端视角什么是 MCP samplingMCP 中sampling 是 MCP 服务器通过 MCP 客户端发起 LLM 调用的一套机制——即服务器借助客户端、经由任意传输代理对 LLM 的请求。它在服务器需要使用 Gen AI 但不想为每个服务器单独配置 LLM 凭据时极为有用或当公共 MCP 服务器希望由连接它的客户端来支付 LLM 调用费用时。注意这与可观测性中的 sampling 概念无关。Pydantic AI 同时支持作为客户端和服务端使用 sampling。作为客户端MCPToolset需要设置sampling_model——既可在工具集上用sampling_model构造参数直接设置也可用agent.set_mcp_sampling_model()让 Agent 的模型或参数指定的模型成为其注册的所有MCPToolset的 sampling 模型。从源码看设置sampling_model后且未显式传sampling_handlerPydantic AI 会构建一个委托给该模型、并应用请求中maxTokens/temperature/stopSequences设置的 sampling handler两者同时传入会报错pydantic_ai_slim/pydantic_ai/mcp.py#L833-L839。典型流程示例一个 MCP 服务器希望使用 sampling 生成 SVG。服务器端工具通过ctx.session.create_message(...)发起 sampling 调用携带max_tokens、system_prompt客户端侧只需给Agent设置 sampling 模型即可自动响应from fastmcp.client.transports import StdioTransport from pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset toolset MCPToolset(StdioTransport(commandpython, args[generate_svg.py])) agent Agent(openai:gpt-5.2, toolsets[toolset]) async def main(): agent.set_mcp_sampling_model() result await agent.run(Create an image of a robot in a punk style.) print(result.output) # Image file written to robot_punk.svg.服务器端完整实现见 docs/mcp/client.md#mcp-sampling 中的generate_svg.py本例可原样运行。Elicitation服务器向客户端请求结构化输入MCP 的 elicitation 允许服务器在会话期间就缺失或额外上下文向客户端请求结构化输入——让模型可以说等等我需要先知道 X 才能继续而不是要求一切 upfront 或盲目猜测。工作原理Elicitation 引入了一种名为ElicitRequest的协议消息类型由服务器在需要补充信息时发送给客户端客户端可用ElicitResult或ErrorData消息响应。一次典型交互用户向 MCP 服务器发起请求如预订那家意大利餐厅的桌子→ 服务器识别出缺少信息哪家意大利餐厅什么日期时间→ 服务器向客户端发送ElicitRequest询问缺失信息 → 客户端接收请求并呈现给用户终端提示、GUI 对话框或 Web 界面→ 用户提供信息、拒绝或取消 → 客户端把ElicitResult发回服务器 → 服务器携带结构化数据继续处理原请求。这让多阶段工作流更具交互性不必 upfront 收集全部信息服务器可按需询问。客户端配置创建MCPToolset时提供elicitation_handler即可启用。完整示例餐厅预订见 docs/mcp/client.md#setting-up-elicitation服务器端book_table工具通过ctx.elicit(message..., schemaBookingDetails)请求结构化预订信息restaurant、party_size、date三个字段客户端handle_elicitation处理器根据params.requestedSchema逐字段提示用户输入、按 JSON schema 类型转换并让用户确认后返回ElicitResult(actionaccept, contentdata)或返回decline/cancel。需要注意的限制与安全点MCP elicitation 仅支持 string、number、boolean 与 enum 类型且仅限扁平对象结构服务器不得请求敏感信息客户端必须实现带清晰说明的用户批准控制。把 Agent 封装进 MCP 服务器服务端视角Pydantic AI 模型同样可以用于 MCP 服务器内部。这是 MCP 支持的第二个方向docs/mcp/server.md把 Agent 作为一个工具暴露出去任何 MCP 客户端都能调用。最小服务器示例用 Python MCP SDK 的FastMCP定义一个服务器在工具内运行 Pydantic AI Agentfrom mcp.server.fastmcp import FastMCP from pydantic_ai import Agent server FastMCP(Pydantic AI Server) server_agent Agent( anthropic:claude-haiku-4-5, instructionsalways reply in rhyme ) server.tool() async def poet(theme: str) - str: Poem generator r await server_agent.run(fwrite a poem about {theme}) return r.output if __name__ __main__: server.run()简单客户端该服务器可被任意 MCP 客户端查询。以下是直接用 Python SDK 的客户端示例import asyncio import os from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def client(): server_params StdioServerParameters( commandpython, args[mcp_server.py], envos.environ ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() result await session.call_tool(poet, {theme: socks}) print(result.content[0].text) if __name__ __main__: asyncio.run(client())服务端 Sampling经客户端回调 LLM当 Agent 被用于 MCP 服务器时可通过MCPSamplingModel使用 sampling——不再直接连接 LLM而是回调 MCP 客户端来发起 LLM 调用。将上面的示例扩展为 sampling 版本from mcp.server.fastmcp import Context, FastMCP from pydantic_ai import Agent from pydantic_ai.models.mcp_sampling import MCPSamplingModel server FastMCP(Pydantic AI Server with sampling) server_agent Agent(instructionsalways reply in rhyme) server.tool() async def poet(ctx: Context, theme: str) - str: Poem generator r await server_agent.run(fwrite a poem about {theme}, modelMCPSamplingModel(sessionctx.session)) return r.output if __name__ __main__: server.run() # 通过 stdio 运行服务器前面那个简单客户端不支持 sampling直接使用会报错。支持 sampling 的最简单方式是用 Pydantic AI Agent 作为客户端见上文 sampling 章节若要用原生 MCP SDK 支持则需为ClientSession提供sampling_callback在回调中构造CreateMessageResult返回响应内容完整示例见 docs/mcp/server.md#mcp-sampling。选择指南与测试佐证三种客户端接入路径的选型总结需求推荐方案默认本地运行、可一键切换原生 MCP跨提供商免改代码MCP能力capabilities[...]管理工具集生命周期、跨 Agent 共享、高级传输/客户端配置、配置文件批量加载MCPToolsettoolsets[...]load_mcp_toolsets()仅需要提供商原生 MCP、追求最优上下文与延迟MCPServerTool原生工具仓库测试对上述能力有充分覆盖tests/test_mcp.py覆盖MCPToolset的构造、传输构建、错误处理与配置文件加载tests/durable_exec/系列如tests/durable_exec/test_prefect.py、tests/durable_exec/temporal/test_toolsets.py验证了工具集在 durable execution 环境下的生命周期与序列化行为tests/mcp_server.py与tests/mcp_task_server.py提供了测试用的 MCP 服务器。需要快速上手时可参考 examples/pydantic_ai_examples 与文档 docs/mcp/overview.md、docs/mcp/client.md、docs/mcp/server.md。总而言之日常开发优先使用MCP能力获得本地默认 原生可选的双模体验需要细粒度控制时下沉到MCPToolset服务器场景则用 FastMCP 包装 Agent并视需要启用 sampling 与 elicitation 增强交互。【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Pydantic AI 与 MCP：从 MCP 客户端到 MCP 服务器的完整接入指南
Pydantic AI 与 MCP从 MCP 客户端到 MCP 服务器的完整接入指南【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-aiPydantic AI 对 Model Context ProtocolMCP 提供了双向、多层次的完整支持Agent 既可以作为 MCP 客户端连接外部 MCP 服务器并使用其工具也可以把 Agent 本身封装进 MCP 服务器、通过工具调用暴露给任意 MCP 客户端。本文以 docs/mcp/overview.md 为主线结合客户端文档 docs/mcp/client.md、服务器文档 docs/mcp/server.md 以及仓库源码系统讲解在 Pydantic AI 中使用 MCP 的推荐路径、底层实现与高级用法。读完本文你将掌握MCP能力、MCPToolset、MCPServerTool三条接入路径的适用场景并能搭建完整的 MCP 客户端与服务端应用。MCP 是什么为什么 Agent 需要它Model Context Protocol 是一种标准化协议允许 AI 应用——包括 Pydantic AI 这类程序化 Agent、Cursor 等编码 Agent、Claude Desktop 等桌面应用——通过统一接口连接外部工具与服务。与所有协议一样MCP 的愿景是让大量应用无需逐一定制集成即可互相通信。官方维护了一份 MCP 服务器清单涵盖搜索、数据库、GitHub、Slack 等常见服务。落到 Pydantic AI 的实际场景中这意味着Pydantic AI 可以接入一个以 MCP 服务器形式实现的网络搜索服务构建深度研究型 Agent其他 MCP 客户端如 Cursor可以连接 Pydantic 官方提供的 MCP 服务器来检索日志、链路与指标辅助排查 Bug任何 MCP 客户端都可以连接 Pydantic 的 Run Python MCP 服务器在沙箱环境中运行任意 Python 代码。Pydantic AI 的 MCP 支持分为两大方向Agent 连接 MCP 服务器客户端侧以及Agent 被封装进 MCP 服务器服务端侧。下面分别展开。连接 MCP 服务器三种接入路径怎么选Pydantic AI 提供三条连接 MCP 服务器的路径按推荐程度排列1.MCP能力推荐声明式接入默认本地运行 MCP 服务器凭据、钩子、追踪都由你掌控并可通过一个nativeTrue标志选择使用模型提供商的原生 MCP 支持同一 Agent 无需改代码即可跨提供商工作。相关实现见 pydantic_ai_slim/pydantic_ai/capabilities/mcp.py 与完整文档 docs/capabilities/mcp.md。2.MCPToolset工具集底层直接管理工具集生命周期、在多个 Agent 间共享同一个 MCP 服务器或传入MCP能力未暴露的高级传输/客户端配置。通过toolsets[...]注册到 Agent详见 docs/mcp/client.md。3.MCPServerTool原生工具仅原生当只需要模型提供商的原生 MCP 支持、不需要本地回退时可直接将MCPServerTool作为原生工具使用见 docs/native-tools.md#mcp-server-tool。路径一MCP能力 —— 一行代码同时获得本地回退与原生 MCPMCP是一个提供商自适应能力provider-adaptive capability是 Pydantic AI 中 MCP 的主要入口。默认本地运行 MCP 服务器让凭据、钩子与追踪保持在你的控制之下同时支持基于 URL 的服务器以及直接的 client / toolset / transport 输入from pydantic_ai import Agent from pydantic_ai.capabilities import MCP agent Agent( openai:gpt-5.2, capabilities[ # 默认在本地运行 MCP 服务器 MCP(urlhttps://mcp.example.com/api), # 选择原生 MCP —— 若模型不支持则回退到本地 MCP(urlhttps://mcp.example.com/other, nativeTrue), ], )关键点将 URL 作为第一个参数传入即可同时启用本地回退与设置nativeTrue时提供商原生 MCP本地侧local接受任何MCPToolset输入——URL、FastMCP transport、预构建的fastmcp.Client、进程内FastMCP服务器、本地脚本路径等非工具集输入会自动包装为MCPToolset原生侧由MCPServerTool支撑。需要完全控制如自定义id、authorization_token、description时可直接传nativeMCPServerTool(...)。从源码看MCP类继承自NativeOrLocalToolpydantic_ai_slim/pydantic_ai/capabilities/mcp.py#L26-L28其设计意图即原生优先、本地兜底模型支持原生 MCP 时走提供商侧执行上下文更优化、缓存更高效、无往返 Pydantic AI 的延迟不支持时自动回退本地。四种常见组合如下from pydantic_ai.capabilities import MCP from pydantic_ai.native_tools import MCPServerTool # URL 型 MCP 服务器本地运行需要 pydantic-ai-slim[mcp] MCP(https://mcp.example.com/api) # 无 URL 的本地客户端 —— 传任意 MCPToolset 输入 MCP(localmy_fastmcp_client) # 原生优先URL 型本地回退 MCP(https://mcp.example.com/api, nativeTrue) # 仅原生无本地 —— 不需要 mcp extra MCP(https://mcp.example.com/api, nativeTrue, localFalse) # 显式原生 显式本地 —— 两侧独立配置 MCP( nativeMCPServerTool( idpublic-mcp, urlhttps://relay.example.com/mcp, authorization_tokenrelay-token, ), localmy_fastmcp_client, )路径二MCPToolset—— 底层客户端掌控生命周期MCPToolset是 Pydantic AI 连接 MCP 服务器的底层工具集包装了 FastMCP 客户端同时支持本地stdio与远程Streamable HTTP、SSEMCP 服务器。完整文档见 docs/mcp/client.md实现位于 pydantic_ai_slim/pydantic_ai/mcp.py。安装需要安装pydantic-ai或以mcp可选组安装pydantic-ai-slimpip/uv-add pydantic-ai-slim[mcp]注意FastMCP 4 目前是预发布版本需显式安装。MCPToolset支持它但其现代协议模式不支持服务器发起的 sampling 与 elicitation也无法应用log_level此时MCPToolset会给出警告请在log_handler中过滤日志。这些选项在 FastMCP 4 的 legacy 协议模式下仍保留 FastMCP 3 的行为。支持的输入形态MCPToolset第一个位置参数接受以下任意一种URL 字符串Streamable HTTP若路径以/sse结尾则自动识别为 SSE本地 Python 或 Node.js 脚本路径通过 stdio 运行FastMCP transport如StdioTransport、StreamableHttpTransport、SSETransport预构建的fastmcp.Client用于 OAuth、工具转换等高级 FastMCP 配置进程内FastMCP服务器用于测试或单进程部署无网络往返。每个MCPToolset实例是一个工具集可通过toolsets参数注册到Agent。生命周期管理上既可以用async with agent统一开关所有已注册工具集的连接stdio 服务器还会随之启停子进程也可以用async with toolset单独管理某个工具集适合跨多 Agent 共享。若未显式进入任一上下文管理器工具集会按需自动打开与关闭。四种连接方式的完整示例Streamable HTTP连接远程 MCP 服务器的推荐方式。需要先运行一个支持该传输的服务端from mcp.server.fastmcp import FastMCP app FastMCP() app.tool() def add(a: int, b: int) - int: return a b if __name__ __main__: app.run(transportstreamable-http)from pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset toolset MCPToolset(http://localhost:8000/mcp) # (1)! agent Agent(openai:gpt-5.2, toolsets[toolset]) # (2)! async def main(): result await agent.run(What is 7 plus 5?) print(result.output) # The answer is 12.用连接 URL 定义 MCP 工具集创建挂载该工具集的 Agent。运行此示例时需导入asyncio并追加asyncio.run(main())其余无需改动。这一过程完整展示了 MCP 客户端的工作链路模型收到 What is 7 plus 5? 提示 → 模型决定调用add工具 → 模型返回工具调用 → Pydantic AI 通过 Streamable HTTP 把工具调用发给 MCP 服务器 → 服务器执行add返回 12 → 模型携带返回值被再次调用 → 模型给出最终答案。如需可视化整个过程甚至直接看到工具调用可在示例中追加三行 logfire 插桩代码import logfire logfire.configure() logfire.instrument_pydantic_ai()SSEHTTP Server-Sent Events 传输同样受支持但已在 MCP 中被弃用新部署应优先 Streamable HTTP。URL 以/sse结尾时自动识别为 SSE其他路径需显式传入SSETransportfrom pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset toolset MCPToolset(http://localhost:3001/sse) agent Agent(openai:gpt-5.2, toolsets[toolset])Stdio服务器作为子进程运行通过stdin/stdout通信。传脚本路径或用StdioTransport完全控制命令、参数与环境from fastmcp.client.transports import StdioTransport from pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset toolset MCPToolset(StdioTransport(commandpython, args[mcp_server.py])) agent Agent(openai:gpt-5.2, toolsets[toolset])进程内 FastMCP 服务器服务器与 Agent 在同一 Python 进程省去网络往返from fastmcp import FastMCP from pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset fastmcp_server FastMCP(my_server) fastmcp_server.tool() async def add(a: int, b: int) - int: return a b toolset MCPToolset(fastmcp_server) agent Agent(openai:gpt-5.2, toolsets[toolset]) async def main(): result await agent.run(What is 7 plus 5?) print(result.output) # The answer is 12.路径三MCPServerTool—— 仅使用模型提供商的原生 MCP当只想要提供商原生 MCP 支持不需要本地回退时可直接使用MCPServerTool作为原生工具。它要求 MCP 服务器位于提供商可访问的公网 URL不支持 Pydantic AI Agent 侧 MCP 的许多高级特性但能获得更优化的上下文使用与缓存、以及省去回程 Pydantic AI 的更低延迟。当前 OpenAI Responses、Anthropic、xAI 三家提供支持Google 等暂不支持。from pydantic_ai import Agent, MCPServerTool from pydantic_ai.capabilities import NativeTool agent Agent( anthropic:claude-sonnet-4-6, capabilities[ NativeTool( MCPServerTool( iddeepwiki, urlhttps://mcp.deepwiki.com/mcp, ) ) ] ) result agent.run_sync(Tell me about the pydantic/pydantic-ai repo.) print(result.output)MCPServerTool支持authorization_tokenOpenAI/Anthropic/xAI、allowed_tools三家、descriptionOpenAI/xAI、headersOpenAI/xAI等配置项。使用 OpenAI Responses 时还可通过x-openai-connector:connector_id形式的特殊 URL 接入 OpenAI 的 MCP Connectors。从配置文件批量加载 MCP 工具集当需要管理多个 MCP 服务器、或希望在不改代码的情况下从外部配置服务器时可以用load_mcp_toolsets()从 JSON 配置文件批量加载工具集。配置格式配置文件包含一个mcpServers对象每个服务器以唯一键标识{ mcpServers: { python-runner: { command: uv, args: [run, mcp-run-python, stdio] }, weather: { command: python, args: [mcp_server.py] }, weather-api: { url: http://localhost:3001/sse }, calculator: { url: http://localhost:8000/mcp } } }每个条目支持command、args、env、cwdstdio 服务器或url、headersHTTP 服务器。加载时会进行校验类型错误的字段会立即报错而非等到连接时未知键会被忽略因此与其他 MCP 客户端共享的配置文件仍可加载但只忽略、绝不生效——特别是disabled不会跳过服务器type不会选择传输方式传输方式由 URL 推断仅以/sse结尾视为 SSE其他一律视为 Streamable HTTP。环境变量展开配置文件支持${VAR}与${VAR:-default}语法展开环境变量与 Claude Code 的 MCP 配置一致便于把 API Key、主机名等敏感信息留在配置文件之外{ mcpServers: { python-runner: { command: ${PYTHON_CMD:-python3}, args: [run, ${MCP_MODULE}, stdio], env: { API_KEY: ${MY_API_KEY} } }, weather-api: { url: https://${SERVER_HOST:-localhost}:${SERVER_PORT:-8080}/sse } } }${VAR}会被替换为对应环境变量值${VAR:-default}在环境变量未设置时使用默认值警告使用${VAR}语法时若环境变量未定义将抛出ValueError请用${VAR:-default}提供回退安全警告配置文件指定了要作为子进程启动的可执行文件与参数能写配置的人即可执行任意命令${VAR}按完整进程环境展开、无白名单配置文件还能读取任何环境变量。因此只加载你控制的配置文件切勿加载不可信来源的配置。用法from pydantic_ai import Agent from pydantic_ai.mcp import load_mcp_toolsets # 从配置文件加载所有工具集 toolsets load_mcp_toolsets(mcp_config.json) # 创建挂载所有工具集的 Agent agent Agent(openai:gpt-5.2, toolsetstoolsets) async def main(): result await agent.run(What is 7 plus 5?) print(result.output)客户端侧高级用法工具调用定制process_tool_callMCPToolset接受process_tool_call回调用于定制工具调用请求及其响应。常见用途是注入服务端处理器需要读取的元数据——例如把 run context 的 deps 传给服务器from typing import Any from fastmcp.client.transports import StdioTransport from pydantic_ai import Agent, RunContext from pydantic_ai.mcp import CallToolFunc, MCPToolset, ToolResult from pydantic_ai.models.test import TestModel async def process_tool_call( ctx: RunContext[int], call_tool: CallToolFunc, name: str, tool_args: dict[str, Any], ) - ToolResult: A tool call processor that passes along the deps. return await call_tool(name, tool_args, {deps: ctx.deps}) toolset MCPToolset( StdioTransport(commandpython, args[mcp_server.py]), process_tool_callprocess_tool_call, ) agent Agent( modelTestModel(call_tools[echo_deps]), deps_typeint, toolsets[toolset], ) async def main(): result await agent.run(Echo with deps set to 42, deps42) print(result.output) # {echo_deps:{echo:This is an echo message,deps:42}}服务端如何读取注入的元数据取决于 MCP 服务器 SDK。例如 MCP Python SDK 中工具处理函数的ctx: Context参数即可访问from typing import Any from mcp.server.fastmcp import Context, FastMCP from mcp.server.session import ServerSession mcp FastMCP(Pydantic AI MCP Server) mcp.tool() async def echo_deps(ctx: Context[ServerSession, None]) - dict[str, Any]: Echo the run context. await ctx.info(This is an info message) deps: Any getattr(ctx.request_context.meta, deps) return {echo: This is an echo message, deps: deps} if __name__ __main__: mcp.run()工具错误处理tool_error_behavior当 MCP 服务器报告工具错误时MCPToolset让你选择错误应如何表现tool_error_behavior行为retry默认。抛出ModelRetry把服务器错误作为重试提示发回模型。适用于模型可能自我纠正调用的情况。failed抛出ToolFailed记录为outcomefailed的工具结果。适用于工具调用已完成但失败、由模型决定下一步的情况。error传播底层 MCP 工具异常并使 Agent 运行失败。适用于需要应用程序代码在模型循环外处理的错误。对retry与failed两种模式结构化错误内容会以 JSON 序列化进模型可见消息因此重试提示等机器可读细节对模型仍然可见协议与传输层错误不会被报告为已完成的失败工具调用。这与本地工具代码中工具重试与失败工具结果的区分是 MCP 层面的对应物。从源码看tool_error_behavior是MCPToolset的公开字段默认值为retrypydantic_ai_slim/pydantic_ai/mcp.py#L764-L770。工具前缀避免命名冲突连接多个可能提供同名工具的 MCP 服务器时用.prefixed(...)包装每个MCPToolset为其工具名加前缀from pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset weather MCPToolset(http://localhost:3001/sse).prefixed(weather) # weather_* calculator MCPToolset(http://localhost:3002/sse).prefixed(calc) # calc_* # 两个服务器可能都暴露 get_data 工具但会被区分为 # weather_get_data 和 calc_get_data。 agent Agent(openai:gpt-5.2, toolsets[weather, calculator])服务器指令注入include_instructionsMCP 服务器可在初始化期间提供指令说明如何最好地使用其工具。连接建立后可通过MCPToolset.instructions访问设置include_instructionsTrue可自动注入 Agent 的指令集from pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset toolset MCPToolset(http://localhost:8000/mcp, include_instructionsTrue) agent Agent(openai:gpt-5.2, toolsets[toolset])源码中include_instructions默认值为False向后兼容开启后服务器初始化返回的指令会被加入 Agent 的指令集pydantic_ai_slim/pydantic_ai/mcp.py#L813-L818。工具元数据与过滤MCP 工具可携带描述工具特征的元数据在过滤工具时非常有用。meta与annotations字段位于传给过滤函数的ToolDefinition对象的metadata字典上工具的输出 schema如有则作为return_schema字段可用。MCPToolset还额外暴露task: bool标志表示该工具集是否会为工具使用任务增强执行task-augmented execution。后台任务Background TasksMCPToolset支持 MCP 的任务增强执行SEP-1686。使用 SEP-1686 的服务器包括 FastMCP 3可通过execution.taskSupport声明每个工具的任务支持MCPToolset据此路由调用execution.taskSupport行为required始终以taskTrue调用。服务器创建任务客户端通过tasks/result等待最终结果。optional默认以taskTrue调用。设置prefer_tasksFalse可改为普通调用。forbidden或缺失普通调用。FastMCP 4 使用更新的 MCP Tasks 扩展SEP-2663由服务器主导任务创建因此上述task元数据与prefer_tasks偏好适用于 FastMCP 3 而非 FastMCP 4。普通调用即可驱动仅任务型工具完成无需额外安装而显式选择 tasks 扩展use_taskTrue需要单独的fastmcp-tasks包可通过mcp-tasks可选组安装pip install pydantic-ai-slim[mcp-tasks]。对 FastMCP 3 服务器用pip install fastmcp[tasks]3,4安装 tasks extra并通过taskTaskConfig(mode...)按工具声明任务支持from fastmcp import FastMCP from fastmcp.server.tasks import TaskConfig mcp FastMCP(long_running_server) mcp.tool(taskTaskConfig(modeoptional)) async def deep_research(topic: str) - str: import asyncio await asyncio.sleep(0) return fResearched {topic} if __name__ __main__: mcp.run(transportstreamable-http)默认MCPToolset在工具支持时即采用任务增强执行偏好普通调用者可设prefer_tasksFalse不影响任务支持为 required 的工具from pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset toolset MCPToolset(http://localhost:8000/mcp, prefer_tasksFalse) agent Agent(openai:gpt-5.2, toolsets[toolset])资源ResourcesMCP 服务器可提供资源——文件、数据或内容供客户端访问。MCP 中的资源是应用驱动的由宿主应用决定如何手动将上下文纳入而不会自动暴露给 LLM除非工具返回ResourceLink或EmbeddedResource。MCPToolset暴露三个方法list_resources()—— 列出服务器上所有可用资源list_resource_templates()—— 列出带参数占位符的资源模板read_resource(uri)—— 按 URI 读取特定资源内容。文本内容返回为str二进制内容返回为BinaryContent。完整示例见 docs/mcp/client.md#resources先运行暴露resource://user_name.txt资源的 FastMCP 服务器客户端用async with toolset打开连接后依次list_resources()与read_resource(resource://user_name.txt)读取输出Alice。HTTP 认证与多用户认证对 HTTP 传输MCPToolset接受auth参数bearer token 字符串、任意httpx.Auth或字面量字符串oauth启用 FastMCP 的 OAuth 流程静态请求头如 API Key可经headers参数传入。多用户/多租户应用中每个用户通常有自己的 MCP 服务器凭据如租户级 bearer token。注意共享的MCPToolset实例是单一身份——它维护一个 MCP 会话被所有并发 Agent 运行共享连接由最先需要它的运行建立认证随之解析直到最后一个运行结束才拆除。在共享实例上从ContextVar等任务局部状态派生逐请求凭据是无效的重叠运行会静默地使用打开会话的那个运行的凭据发送请求。要让每次并发运行使用对应用户的凭据需要为每次运行创建独立的MCPToolset实例以建立各自认证的会话。推荐方式是用agent.toolset装饰器动态构建工具集被装饰函数会收到 run context可从运行的依赖中读取用户凭据from dataclasses import dataclass from pydantic_ai import Agent, RunContext from pydantic_ai.mcp import MCPToolset dataclass class UserDeps: mcp_token: str agent Agent(openai:gpt-5.2, deps_typeUserDeps) agent.toolset(per_run_stepFalse) # (1)! def user_mcp_server(ctx: RunContext[UserDeps]) - MCPToolset: return MCPToolset(http://localhost:8000/mcp, authctx.deps.mcp_token) async def main(): result await agent.run(What is 7 plus 5?, depsUserDeps(mcp_tokentoken)) print(result.output) # The answer is 12.per_run_stepFalse使工具集每次运行构建一次而非每个运行步骤前构建整个运行共享单个 MCP 会话。由于每次运行的工具集会话在运行内部建立ContextVar中持有的凭据在此模式下也能正确解析——但通过 deps 传递更显式、不依赖任务局部状态。另一种替代方案是每次请求自行构造新的MCPToolset并传给运行方法的toolsets参数。自定义 TLS/SSL 配置某些环境需要调整 HTTPS 连接方式——例如信任内部 CA、为mTLS出示客户端证书或仅限本地开发时完全禁用证书校验。MCPToolset提供http_client参数可传入预先配置好的httpx.AsyncClientimport ssl import httpx from pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset # 信任内部/自签名 CA ssl_ctx ssl.create_default_context(cafile/etc/ssl/private/my_company_ca.pem) # 可选为双向 TLS 加载客户端证书 ssl_ctx.load_cert_chain(certfile/etc/ssl/certs/client.crt, keyfile/etc/ssl/private/client.key) http_client httpx.AsyncClient(verifyssl_ctx, timeouthttpx.Timeout(10.0)) toolset MCPToolset(http://localhost:3001/sse, http_clienthttp_client) # (1)! agent Agent(openai:gpt-5.2, toolsets[toolset])提供http_client后Pydantic AI 会为每个请求复用该客户端httpx 支持的一切verify、cert、自定义代理、超时等都适用于所有 MCP 流量。客户端标识client_info连接 MCP 服务器时可指定一个Implementation对象作为客户端信息在初始化期间发送给服务器。用途包括在服务器日志中标识应用、允许服务器基于客户端提供自定义行为、调试监控 MCP 连接、版本特性协商from mcp import types as mcp_types from pydantic_ai.mcp import MCPToolset toolset MCPToolset( http://localhost:3001/sse, client_infomcp_types.Implementation( nameMyApplication, version2.1.0, ), )MCP Sampling客户端视角什么是 MCP samplingMCP 中sampling 是 MCP 服务器通过 MCP 客户端发起 LLM 调用的一套机制——即服务器借助客户端、经由任意传输代理对 LLM 的请求。它在服务器需要使用 Gen AI 但不想为每个服务器单独配置 LLM 凭据时极为有用或当公共 MCP 服务器希望由连接它的客户端来支付 LLM 调用费用时。注意这与可观测性中的 sampling 概念无关。Pydantic AI 同时支持作为客户端和服务端使用 sampling。作为客户端MCPToolset需要设置sampling_model——既可在工具集上用sampling_model构造参数直接设置也可用agent.set_mcp_sampling_model()让 Agent 的模型或参数指定的模型成为其注册的所有MCPToolset的 sampling 模型。从源码看设置sampling_model后且未显式传sampling_handlerPydantic AI 会构建一个委托给该模型、并应用请求中maxTokens/temperature/stopSequences设置的 sampling handler两者同时传入会报错pydantic_ai_slim/pydantic_ai/mcp.py#L833-L839。典型流程示例一个 MCP 服务器希望使用 sampling 生成 SVG。服务器端工具通过ctx.session.create_message(...)发起 sampling 调用携带max_tokens、system_prompt客户端侧只需给Agent设置 sampling 模型即可自动响应from fastmcp.client.transports import StdioTransport from pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset toolset MCPToolset(StdioTransport(commandpython, args[generate_svg.py])) agent Agent(openai:gpt-5.2, toolsets[toolset]) async def main(): agent.set_mcp_sampling_model() result await agent.run(Create an image of a robot in a punk style.) print(result.output) # Image file written to robot_punk.svg.服务器端完整实现见 docs/mcp/client.md#mcp-sampling 中的generate_svg.py本例可原样运行。Elicitation服务器向客户端请求结构化输入MCP 的 elicitation 允许服务器在会话期间就缺失或额外上下文向客户端请求结构化输入——让模型可以说等等我需要先知道 X 才能继续而不是要求一切 upfront 或盲目猜测。工作原理Elicitation 引入了一种名为ElicitRequest的协议消息类型由服务器在需要补充信息时发送给客户端客户端可用ElicitResult或ErrorData消息响应。一次典型交互用户向 MCP 服务器发起请求如预订那家意大利餐厅的桌子→ 服务器识别出缺少信息哪家意大利餐厅什么日期时间→ 服务器向客户端发送ElicitRequest询问缺失信息 → 客户端接收请求并呈现给用户终端提示、GUI 对话框或 Web 界面→ 用户提供信息、拒绝或取消 → 客户端把ElicitResult发回服务器 → 服务器携带结构化数据继续处理原请求。这让多阶段工作流更具交互性不必 upfront 收集全部信息服务器可按需询问。客户端配置创建MCPToolset时提供elicitation_handler即可启用。完整示例餐厅预订见 docs/mcp/client.md#setting-up-elicitation服务器端book_table工具通过ctx.elicit(message..., schemaBookingDetails)请求结构化预订信息restaurant、party_size、date三个字段客户端handle_elicitation处理器根据params.requestedSchema逐字段提示用户输入、按 JSON schema 类型转换并让用户确认后返回ElicitResult(actionaccept, contentdata)或返回decline/cancel。需要注意的限制与安全点MCP elicitation 仅支持 string、number、boolean 与 enum 类型且仅限扁平对象结构服务器不得请求敏感信息客户端必须实现带清晰说明的用户批准控制。把 Agent 封装进 MCP 服务器服务端视角Pydantic AI 模型同样可以用于 MCP 服务器内部。这是 MCP 支持的第二个方向docs/mcp/server.md把 Agent 作为一个工具暴露出去任何 MCP 客户端都能调用。最小服务器示例用 Python MCP SDK 的FastMCP定义一个服务器在工具内运行 Pydantic AI Agentfrom mcp.server.fastmcp import FastMCP from pydantic_ai import Agent server FastMCP(Pydantic AI Server) server_agent Agent( anthropic:claude-haiku-4-5, instructionsalways reply in rhyme ) server.tool() async def poet(theme: str) - str: Poem generator r await server_agent.run(fwrite a poem about {theme}) return r.output if __name__ __main__: server.run()简单客户端该服务器可被任意 MCP 客户端查询。以下是直接用 Python SDK 的客户端示例import asyncio import os from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def client(): server_params StdioServerParameters( commandpython, args[mcp_server.py], envos.environ ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() result await session.call_tool(poet, {theme: socks}) print(result.content[0].text) if __name__ __main__: asyncio.run(client())服务端 Sampling经客户端回调 LLM当 Agent 被用于 MCP 服务器时可通过MCPSamplingModel使用 sampling——不再直接连接 LLM而是回调 MCP 客户端来发起 LLM 调用。将上面的示例扩展为 sampling 版本from mcp.server.fastmcp import Context, FastMCP from pydantic_ai import Agent from pydantic_ai.models.mcp_sampling import MCPSamplingModel server FastMCP(Pydantic AI Server with sampling) server_agent Agent(instructionsalways reply in rhyme) server.tool() async def poet(ctx: Context, theme: str) - str: Poem generator r await server_agent.run(fwrite a poem about {theme}, modelMCPSamplingModel(sessionctx.session)) return r.output if __name__ __main__: server.run() # 通过 stdio 运行服务器前面那个简单客户端不支持 sampling直接使用会报错。支持 sampling 的最简单方式是用 Pydantic AI Agent 作为客户端见上文 sampling 章节若要用原生 MCP SDK 支持则需为ClientSession提供sampling_callback在回调中构造CreateMessageResult返回响应内容完整示例见 docs/mcp/server.md#mcp-sampling。选择指南与测试佐证三种客户端接入路径的选型总结需求推荐方案默认本地运行、可一键切换原生 MCP跨提供商免改代码MCP能力capabilities[...]管理工具集生命周期、跨 Agent 共享、高级传输/客户端配置、配置文件批量加载MCPToolsettoolsets[...]load_mcp_toolsets()仅需要提供商原生 MCP、追求最优上下文与延迟MCPServerTool原生工具仓库测试对上述能力有充分覆盖tests/test_mcp.py覆盖MCPToolset的构造、传输构建、错误处理与配置文件加载tests/durable_exec/系列如tests/durable_exec/test_prefect.py、tests/durable_exec/temporal/test_toolsets.py验证了工具集在 durable execution 环境下的生命周期与序列化行为tests/mcp_server.py与tests/mcp_task_server.py提供了测试用的 MCP 服务器。需要快速上手时可参考 examples/pydantic_ai_examples 与文档 docs/mcp/overview.md、docs/mcp/client.md、docs/mcp/server.md。总而言之日常开发优先使用MCP能力获得本地默认 原生可选的双模体验需要细粒度控制时下沉到MCPToolset服务器场景则用 FastMCP 包装 Agent并视需要启用 sampling 与 elicitation 增强交互。【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考