企业官网建设流程全解析

Pydantic AI 的 MCPSamplingModel：在 MCP 服务器中通过客户端回调驱动 LLM 调用

【免费下载链接】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 的MCPSamplingModel（位于 docs/api/models/mcp-sampling.md 对应的 API 模块）展开，深入讲解 MCP Sampling 这一机制如何在 Pydantic AI 中被建模为一种「模型」：MCP 服务器不直接持有任何 LLM 凭证，而是通过session.create_message回调连接它的 MCP 客户端，让客户端代为发起大模型调用。读完本文，你将掌握MCPSamplingModel的完整参数与请求链路、服务端与客户端的双向实战配置（含可运行示例），并理解其底层消息映射、系统提示词处理、错误约定与当前功能边界。

一、MCP Sampling 是什么：为什么需要它

在 MCP（Model Context Protocol）协议中，Sampling是一种「服务器通过客户端反向调用 LLM」的机制：当 MCP 服务器内的代码需要调用生成式 AI（Gen AI）能力时，它不必自己配置 LLM 的 API Key，而是向与之相连的 MCP 客户端发送一条 "create message" 采样请求，由客户端用自己持有的模型完成调用并把结果回传。

Pydantic AI 的官方文档对它的价值做了精辟概括（见 docs/mcp/client.md）：

• 当 MCP 服务器需要用到 Gen AI，但你不想为每个服务器单独发放 LLM 凭证时，Sampling 让服务器复用客户端的模型能力；
• 当公共 MCP 服务器希望由连接它的客户端来承担 LLM 调用费用时，Sampling 天然实现了「谁连接谁付费」；
• 需要特别澄清的是：这里的 "sampling" 与可观测性领域的采样概念毫无关系，只是协议层面的专有名词。

在 Pydantic AI 中，Sampling 被一等公民化地设计为pydantic_ai.models.mcp_sampling.MCPSamplingModel——它实现了Model接口，因此可以像任何其他模型一样被传入Agent使用。

二、MCPSamplingModel：一个"借道客户端"的模型

MCPSamplingModel定义在 pydantic_ai_slim/pydantic_ai/models/mcp_sampling.py 中，是一个 dataclass，核心字段如下：

从源码结构看，request()方法被调用时，该模型会通过session.create_message(...)将整个请求"外包"给 MCP 客户端，因此它自身不持有任何远程模型句柄。这一点也体现在两个只读属性上：

• model_name永远返回'mcp-sampling'——因为真正的模型名只有等请求发出、CreateMessageResult.model返回后才能知道；
• system永远返回'MCP'——表示模型提供方是 MCP。

provider属性则返回None，说明该模型不绑定任何具体的 provider 实现。

2.1 MCPSamplingModelSettings：采样专用设置

该模块还定义了一个专用的设置类MCPSamplingModelSettings(ModelSettings, total=False)，其唯一新增字段为：

• mcp_model_preferences: ModelPreferences——MCP Sampling 请求使用的模型偏好（对应create_message的model_preferences参数，例如hints、costPriority、speedPriority、intelligencePriority等协议字段）。

源码注释明确强调：该设置类的所有字段必须以mcp_前缀命名，以便在与其他模型共用同一份设置对象时可以安全合并，互不污染。

request()中实际向create_message透传的 settings 键如下（pydantic_ai_slim/pydantic_ai/models/mcp_sampling.py）：

result = await self.session.create_message( sampling_messages, max_tokens=model_settings.get('max_tokens', self.default_max_tokens), system_prompt=system_prompt, temperature=model_settings.get('temperature'), model_preferences=model_settings.get('mcp_model_preferences'), stop_sequences=model_settings.get('stop_sequences'), )

其中max_tokens的取值逻辑正是上文default_max_tokens兜底设计的落地：ModelSettings.max_tokens未设置时，默认取16_384。因此，通过model_settings={'max_tokens': 4096, 'temperature': 0.7}这类标准 Pydantic AI 设置，即可精细化控制每次采样调用的行为。

三、请求链路：Pydantic AI 消息如何变成 MCP 采样消息

MCPSamplingModel的request()完整流程如下：

1. 调用_mcp.map_from_pai_messages(messages)，把 Pydantic AI 的ModelMessage列表拆解为「系统提示词 + MCPSamplingMessage列表」；
2. 调用prepare_request(...)合并模型设置与请求参数；
3. 调用session.create_message(...)发起采样；
4. 校验返回的result.role必须为'assistant'，否则抛出exceptions.UnexpectedModelBehavior（错误消息为：Unexpected result from MCP sampling, expected "assistant" role, got {role}.）；
5. 将result.content映射回ModelResponse，并把result.model作为model_name记录。

3.1 消息映射的底层实现

映射逻辑集中在 pydantic_ai_slim/pydantic_ai/_mcp.py 的三个函数中，这是理解"双向转换"的关键：

• map_from_pai_messages（L74-L118）：遍历ModelMessage列表，将ModelRequest上的instructions与SystemPromptPart内容累积为system_prompt字符串（最终以空字符串拼接返回）；UserPromptPart为纯字符串时转为TextContent的 user 消息，内容为str | BinaryContent列表时，字符串块转TextContent，图片块（BinaryContent.is_image）转ImageContent（base64 数据 + MIME 类型），其他类型（含音频）目前抛出NotImplementedError；ModelResponse则经map_from_model_response转成 assistant 消息。
• map_from_model_response（L121-L131）：将响应中的TextPart拼接为文本，ThinkingPart被直接跳过，其余部件类型抛出UnexpectedModelBehavior。这意味着采样场景下模型回复以纯文本为约定。
• map_from_sampling_content（L134-L142）：把采样返回的TextContent映射回 Pydantic AI 的TextPart；返回图片/音频内容目前同样抛出NotImplementedError（源码注释表明计划用FilePart支持，尚未落地）。

系统提示词的处理细节值得注意：指令（instructions）与常驻系统提示词（SystemPromptPart）会优先进入create_message的system_prompt参数；而在没有instructions的历史回放场景下，SystemPromptPart内容会被包装成<system>...</system>文本作为一条 user 采样消息传入——这一点在tests/models/test_mcp_sampling.py的test_standing_system_prompt_history与test_assistant_text_history_complex两个用例中有精确断言。

四、服务端实战：在 MCP 服务器工具里使用 MCPSamplingModel

在 MCP 服务器一侧，Pydantic AI Agent 可以通过采样机制"反向"借用客户端的大模型能力。核心做法是：在 FastMCP 工具的签名中声明ctx: Context，然后把MCPSamplingModel(session=ctx.session)作为model参数传给agent.run()。

以下完整示例取自 docs/mcp/server.md（mcp_server_sampling.py）：

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() # run the server over stdio

与直连 LLM 的版本相比，差异点非常直观：server_agent创建时不指定模型，模型仅在每次run调用时通过model=参数临时注入，且注入的正是包装了当前会话的MCPSamplingModel。于是 LLM 调用不再由服务器直发，而是沿「服务器 → 客户端 → LLM → 客户端 → 服务器」的路径完成。

4.1 客户端必须支持 Sampling，否则会报错

如果客户端像 docs/mcp/server.md 中的"简单客户端"那样仅用ClientSession(read, write)直连，协议层没有注册sampling_callback，那么服务器发起的采样请求将得不到应答——官方文档明确指出这会直接报错。要让上面的服务器跑通，客户端必须在ClientSession构造时提供sampling_callback：

import asyncio from typing import Any from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from mcp.shared.context import RequestContext from mcp.types import ( CreateMessageRequestParams, CreateMessageResult, ErrorData, TextContent, ) async def sampling_callback( context: RequestContext[ClientSession, Any], params: CreateMessageRequestParams ) -> CreateMessageResult | ErrorData: print('sampling system prompt:', params.systemPrompt) #> sampling system prompt: always reply in rhyme print('sampling messages:', params.messages) # 实际场景中在这里调用你选择的 LLM... response_content = 'Socks for a fox.' return CreateMessageResult( role='assistant', content=TextContent(type='text', text=response_content), model='fictional-llm', ) async def client(): server_params = StdioServerParameters(command='python', args=['mcp_server_sampling.py']) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write, sampling_callback=sampling_callback) as session: await session.initialize() result = await session.call_tool('poet', {'theme': 'socks'}) print(result.content[0].text) #> Socks for a fox. if __name__ == '__main__': asyncio.run(client())

注意回调返回值必须满足两个协议约束：role='assistant'（若返回其他角色，MCPSamplingModel会按上文所述抛出UnexpectedModelBehavior），以及必须携带model字段标明实际使用的模型名。

五、客户端实战：用 Pydantic AI Agent 自动充当采样方

更省事的客户端方案是：直接让 Pydantic AI Agent 同时扮演 MCP 客户端与采样方。此时不需要手写sampling_callback，只需让与服务器关联的MCPToolset带上一个sampling_model。

有两种设置方式（见 docs/mcp/client.md）：

1. 创建MCPToolset时通过sampling_model=关键字参数显式指定采样模型；
2. 调用 Agent 的agent.set_mcp_sampling_model()，该方法（pydantic_ai_slim/pydantic_ai/agent/init.py）会为 Agent 上注册的所有MCPToolset统一设置采样模型；不传参数时默认复用 Agent 自身的模型，也可传入Model实例或模型名来覆盖。

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 中generate_svg.py服务端示例（工具内部通过ctx.session.create_message([...], max_tokens=1_024, system_prompt='Generate an SVG image as per the user input')发起采样），可以看到完整闭环：Agent 的模型先执行工具调用 → 服务器收到工具调用 → 服务器发起采样 → Agent 侧sampling_model代为完成 LLM 调用 → 服务器拿到 SVG 文本落盘并返回结果。

六、测试验证：行为约定与边界条件

tests/models/test_mcp_sampling.py用AsyncMock伪造会话，把MCPSamplingModel的关键行为固化为测试：

• 标识约定：model.model_name == 'mcp-sampling'、model.system == 'MCP'（test_mcp_sampling_model）。
• 正常文本响应：返回role='assistant'、TextContent与model='test-model'的CreateMessageResult时，Agent 输出即为文本内容，且ModelResponse.model_name记录为'test-model'（test_assistant_text）。
• 角色校验：返回role='user'时，agent.run_sync('Hello')抛出UnexpectedModelBehavior，错误信息精确匹配expected "assistant" role, got user.（test_user_text）。
• 多轮历史：带message_history的连续两轮运行都能正确透传采样消息（test_assistant_text_history）。
• 系统提示词抽取：instructions='testing'或常驻系统提示词会被放入create_message的system_prompt参数，而不会以 user 消息形式重复出现在采样消息列表中；无instructions时，历史中的SystemPromptPart则以<system>...</system>文本形式出现在采样消息中（test_standing_system_prompt_history、test_assistant_text_history_complex）。

这些测试同时揭示了三条使用边界：

1. 不支持流式：request_stream()直接抛出NotImplementedError('MCP Sampling does not support streaming')，因此MCPSamplingModel只能用于非流式run；
2. 响应仅支持文本：采样返回图片/音频内容会抛出NotImplementedError；
3. 角色必须为 assistant：任何非assistant角色都会触发UnexpectedModelBehavior。

七、小结：适用场景与设计取舍

MCPSamplingModel的适用场景非常聚焦：当你希望 MCP 服务器内的 Agent 不持有任何 LLM 凭证、把模型调用与计费完全交给客户端时。它把"协议回调"抽象成 Pydantic AI 标准Model接口，从而无缝复用了 Agent 的全部能力（指令、工具、历史、设置），这是其设计上的最大亮点。

同时，选择它之前应确认以下前提与限制：

• 连接的客户端必须支持 Sampling（注册sampling_callback，或本身就是 Pydantic AI Agent）；
• 仅支持单轮非流式文本响应；
• 真正的模型名只有在请求完成后才可知，model_name属性恒为占位符'mcp-sampling'；
• MCPSamplingModelSettings的自定义字段需遵循mcp_前缀约定，以便与普通ModelSettings安全合并。

如需进一步查阅，可继续阅读 pydantic_ai_slim/pydantic_ai/models/mcp_sampling.py、pydantic_ai_slim/pydantic_ai/_mcp.py、tests/models/test_mcp_sampling.py，以及 MCP 专题文档 docs/mcp/server.md 与 docs/mcp/client.md。

【免费下载链接】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