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-ai

Pydantic AI 对 Model Context Protocol（MCP） 提供了双向、多层次的完整支持：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 服务器（凭据、钩子、追踪都由你掌控），并可通过一个native=True标志选择使用模型提供商的原生 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能力 —— 一行代码同时获得本地回退与原生 MCP

MCP是一个提供商自适应能力（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(url='https://mcp.example.com/api'), # 选择原生 MCP —— 若模型不支持则回退到本地 MCP(url='https://mcp.example.com/other', native=True), ], )

关键点：

• 将 URL 作为第一个参数传入，即可同时启用本地回退与（设置native=True时）提供商原生 MCP；
• 本地侧，local=接受任何MCPToolset输入——URL、FastMCP transport、预构建的fastmcp.Client、进程内FastMCP服务器、本地脚本路径等，非工具集输入会自动包装为MCPToolset；
• 原生侧由MCPServerTool支撑。需要完全控制（如自定义id、authorization_token、description）时，可直接传native=MCPServerTool(...)。

从源码看，MCP类继承自NativeOrLocalTool（pydantic_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(local=my_fastmcp_client) # 原生优先；URL 型本地回退 MCP('https://mcp.example.com/api', native=True) # 仅原生（无本地 —— 不需要 `mcp` extra） MCP('https://mcp.example.com/api', native=True, local=False) # 显式原生 + 显式本地 —— 两侧独立配置 MCP( native=MCPServerTool( id='public-mcp', url='https://relay.example.com/mcp', authorization_token='relay-token', ), local=my_fastmcp_client, )

路径二：MCPToolset—— 底层客户端，掌控生命周期

MCPToolset是 Pydantic AI 连接 MCP 服务器的底层工具集，包装了 FastMCP 客户端，同时支持本地（stdio）与远程（Streamable HTTP、SSE）MCP 服务器。完整文档见 docs/mcp/client.md，实现位于 pydantic_ai_slim/pydantic_ai/mcp.py。

安装

需要安装pydantic-ai，或以mcp可选组安装pydantic-ai-slim：

pip/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(transport='streamable-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.

1. 用连接 URL 定义 MCP 工具集；
2. 创建挂载该工具集的 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()

SSE（HTTP + Server-Sent Events 传输同样受支持，但已在 MCP 中被弃用，新部署应优先 Streamable HTTP）。URL 以/sse结尾时自动识别为 SSE；其他路径需显式传入SSETransport：

from 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(command='python', 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( id='deepwiki', url='https://mcp.deepwiki.com/mcp', ) ) ] ) result = agent.run_sync('Tell me about the pydantic/pydantic-ai repo.') print(result.output)

MCPServerTool支持authorization_token（OpenAI/Anthropic/xAI）、allowed_tools（三家）、description（OpenAI/xAI）、headers（OpenAI/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、cwd（stdio 服务器），或url、headers（HTTP 服务器）。加载时会进行校验，类型错误的字段会立即报错而非等到连接时；未知键会被忽略（因此与其他 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', toolsets=toolsets) async def main(): result = await agent.run('What is 7 plus 5?') print(result.output)

客户端侧高级用法

工具调用定制（process_tool_call）

MCPToolset接受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(command='python', args=['mcp_server.py']), process_tool_call=process_tool_call, ) agent = Agent( model=TestModel(call_tools=['echo_deps']), deps_type=int, toolsets=[toolset], ) async def main(): result = await agent.run('Echo with deps set to 42', deps=42) 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让你选择错误应如何表现：

对'retry'与'failed'两种模式，结构化错误内容会以 JSON 序列化进模型可见消息，因此重试提示等机器可读细节对模型仍然可见；协议与传输层错误不会被报告为已完成的失败工具调用。这与本地工具代码中工具重试与失败工具结果的区分是 MCP 层面的对应物。从源码看，tool_error_behavior是MCPToolset的公开字段，默认值为'retry'（pydantic_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_instructions）

MCP 服务器可在初始化期间提供指令，说明如何最好地使用其工具。连接建立后可通过MCPToolset.instructions访问；设置include_instructions=True可自动注入 Agent 的指令集：

from pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset toolset = MCPToolset('http://localhost:8000/mcp', include_instructions=True) 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 Tasks）

MCPToolset支持 MCP 的任务增强执行（SEP-1686）。使用 SEP-1686 的服务器（包括 FastMCP 3）可通过execution.taskSupport声明每个工具的任务支持，MCPToolset据此路由调用：

FastMCP 4 使用更新的 MCP Tasks 扩展（SEP-2663），由服务器主导任务创建，因此上述task元数据与prefer_tasks偏好适用于 FastMCP 3 而非 FastMCP 4。普通调用即可驱动仅任务型工具完成，无需额外安装；而显式选择 tasks 扩展（use_task=True）需要单独的fastmcp-tasks包，可通过mcp-tasks可选组安装：pip install "pydantic-ai-slim[mcp-tasks]"。

对 FastMCP 3 服务器，用pip install "fastmcp[tasks]>=3,<4"安装 tasks extra，并通过task=TaskConfig(mode=...)按工具声明任务支持：

from fastmcp import FastMCP from fastmcp.server.tasks import TaskConfig mcp = FastMCP('long_running_server') @mcp.tool(task=TaskConfig(mode='optional')) async def deep_research(topic: str) -> str: import asyncio await asyncio.sleep(0) return f'Researched {topic}' if __name__ == '__main__': mcp.run(transport='streamable-http')

默认MCPToolset在工具支持时即采用任务增强执行；偏好普通调用者可设prefer_tasks=False（不影响任务支持为 required 的工具）：

from pydantic_ai import Agent from pydantic_ai.mcp import MCPToolset toolset = MCPToolset('http://localhost:8000/mcp', prefer_tasks=False) agent = Agent('openai:gpt-5.2', toolsets=[toolset])

资源（Resources）

MCP 服务器可提供资源——文件、数据或内容，供客户端访问。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_type=UserDeps) @agent.toolset(per_run_step=False) # (1)! def user_mcp_server(ctx: RunContext[UserDeps]) -> MCPToolset: return MCPToolset('http://localhost:8000/mcp', auth=ctx.deps.mcp_token) async def main(): result = await agent.run('What is 7 plus 5?', deps=UserDeps(mcp_token='<token>')) print(result.output) #> The answer is 12.

1. per_run_step=False使工具集每次运行构建一次（而非每个运行步骤前构建），整个运行共享单个 MCP 会话。

由于每次运行的工具集会话在运行内部建立，ContextVar中持有的凭据在此模式下也能正确解析——但通过 deps 传递更显式、不依赖任务局部状态。另一种替代方案是每次请求自行构造新的MCPToolset并传给运行方法的toolsets参数。

自定义 TLS/SSL 配置

某些环境需要调整 HTTPS 连接方式——例如信任内部 CA、为mTLS出示客户端证书，或（仅限本地开发时）完全禁用证书校验。MCPToolset提供http_client参数，可传入预先配置好的httpx.AsyncClient：

import 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(verify=ssl_ctx, timeout=httpx.Timeout(10.0)) toolset = MCPToolset('http://localhost:3001/sse', http_client=http_client) # (1)! agent = Agent('openai:gpt-5.2', toolsets=[toolset])

1. 提供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_info=mcp_types.Implementation( name='MyApplication', version='2.1.0', ), )

MCP Sampling：客户端视角

什么是 MCP sampling？MCP 中，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_handler），Pydantic 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(command='python', 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=..., schema=BookingDetails)请求结构化预订信息（restaurant、party_size、date三个字段）；客户端handle_elicitation处理器根据params.requestedSchema逐字段提示用户输入、按 JSON schema 类型转换，并让用户确认后返回ElicitResult(action='accept', content=data)，或返回'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 Agent：

from mcp.server.fastmcp import FastMCP from pydantic_ai import Agent server = FastMCP('Pydantic AI Server') server_agent = Agent( 'anthropic:claude-haiku-4-5', instructions='always reply in rhyme' ) @server.tool() async def poet(theme: str) -> str: """Poem generator""" r = await server_agent.run(f'write 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( command='python', args=['mcp_server.py'], env=os.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(instructions='always reply in rhyme') @server.tool() async def poet(ctx: Context, theme: str) -> str: """Poem generator""" r = await server_agent.run(f'write a poem about {theme}', model=MCPSamplingModel(session=ctx.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）。

选择指南与测试佐证

三种客户端接入路径的选型总结：

仓库测试对上述能力有充分覆盖：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