
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 是什么为什么需要它在 MCPModel Context Protocol协议中Sampling是一种「服务器通过客户端反向调用 LLM」的机制当 MCP 服务器内的代码需要调用生成式 AIGen 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核心字段如下字段类型说明sessionmcp.ServerSession用于采样请求的 MCP 服务器会话必填default_max_tokensint 16_384默认最大 token 数。MCP Sampling 要求max_tokens为必填参数而ModelSettings.max_tokens是可选的因此当用户在设置中未显式给出时使用此默认值兜底从源码结构看request()方法被调用时该模型会通过session.create_message(...)将整个请求外包给 MCP 客户端因此它自身不持有任何远程模型句柄。这一点也体现在两个只读属性上model_name永远返回mcp-sampling——因为真正的模型名只有等请求发出、CreateMessageResult.model返回后才能知道system永远返回MCP——表示模型提供方是 MCP。provider属性则返回None说明该模型不绑定任何具体的 provider 实现。2.1 MCPSamplingModelSettings采样专用设置该模块还定义了一个专用的设置类MCPSamplingModelSettings(ModelSettings, totalFalse)其唯一新增字段为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.pyresult await self.session.create_message( sampling_messages, max_tokensmodel_settings.get(max_tokens, self.default_max_tokens), system_promptsystem_prompt, temperaturemodel_settings.get(temperature), model_preferencesmodel_settings.get(mcp_model_preferences), stop_sequencesmodel_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()完整流程如下调用_mcp.map_from_pai_messages(messages)把 Pydantic AI 的ModelMessage列表拆解为「系统提示词 MCPSamplingMessage列表」调用prepare_request(...)合并模型设置与请求参数调用session.create_message(...)发起采样校验返回的result.role必须为assistant否则抛出exceptions.UnexpectedModelBehavior错误消息为Unexpected result from MCP sampling, expected assistant role, got {role}.将result.content映射回ModelResponse并把result.model作为model_name记录。3.1 消息映射的底层实现映射逻辑集中在 pydantic_ai_slim/pydantic_ai/_mcp.py 的三个函数中这是理解双向转换的关键map_from_pai_messagesL74-L118遍历ModelMessage列表将ModelRequest上的instructions与SystemPromptPart内容累积为system_prompt字符串最终以空字符串拼接返回UserPromptPart为纯字符串时转为TextContent的 user 消息内容为str | BinaryContent列表时字符串块转TextContent图片块BinaryContent.is_image转ImageContentbase64 数据 MIME 类型其他类型含音频目前抛出NotImplementedErrorModelResponse则经map_from_model_response转成 assistant 消息。map_from_model_responseL121-L131将响应中的TextPart拼接为文本ThinkingPart被直接跳过其余部件类型抛出UnexpectedModelBehavior。这意味着采样场景下模型回复以纯文本为约定。map_from_sampling_contentL134-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(sessionctx.session)作为model参数传给agent.run()。以下完整示例取自 docs/mcp/server.mdmcp_server_sampling.pyfrom 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() # 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_callbackimport 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( roleassistant, contentTextContent(typetext, textresponse_content), modelfictional-llm, ) async def client(): server_params StdioServerParameters(commandpython, args[mcp_server_sampling.py]) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write, sampling_callbacksampling_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())注意回调返回值必须满足两个协议约束roleassistant若返回其他角色MCPSamplingModel会按上文所述抛出UnexpectedModelBehavior以及必须携带model字段标明实际使用的模型名。五、客户端实战用 Pydantic AI Agent 自动充当采样方更省事的客户端方案是直接让 Pydantic AI Agent 同时扮演 MCP 客户端与采样方。此时不需要手写sampling_callback只需让与服务器关联的MCPToolset带上一个sampling_model。有两种设置方式见 docs/mcp/client.md创建MCPToolset时通过sampling_model关键字参数显式指定采样模型调用 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(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 中generate_svg.py服务端示例工具内部通过ctx.session.create_message([...], max_tokens1_024, system_promptGenerate 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 MCPtest_mcp_sampling_model。正常文本响应返回roleassistant、TextContent与modeltest-model的CreateMessageResult时Agent 输出即为文本内容且ModelResponse.model_name记录为test-modeltest_assistant_text。角色校验返回roleuser时agent.run_sync(Hello)抛出UnexpectedModelBehavior错误信息精确匹配expected assistant role, got user.test_user_text。多轮历史带message_history的连续两轮运行都能正确透传采样消息test_assistant_text_history。系统提示词抽取instructionstesting或常驻系统提示词会被放入create_message的system_prompt参数而不会以 user 消息形式重复出现在采样消息列表中无instructions时历史中的SystemPromptPart则以system.../system文本形式出现在采样消息中test_standing_system_prompt_history、test_assistant_text_history_complex。这些测试同时揭示了三条使用边界不支持流式request_stream()直接抛出NotImplementedError(MCP Sampling does not support streaming)因此MCPSamplingModel只能用于非流式run响应仅支持文本采样返回图片/音频内容会抛出NotImplementedError角色必须为 assistant任何非assistant角色都会触发UnexpectedModelBehavior。七、小结适用场景与设计取舍MCPSamplingModel的适用场景非常聚焦当你希望 MCP 服务器内的 Agent 不持有任何 LLM 凭证、把模型调用与计费完全交给客户端时。它把协议回调抽象成 Pydantic AI 标准Model接口从而无缝复用了 Agent 的全部能力指令、工具、历史、设置这是其设计上的最大亮点。同时选择它之前应确认以下前提与限制连接的客户端必须支持 Sampling注册sampling_callback或本身就是 Pydantic AI Agent仅支持单轮非流式文本响应真正的模型名只有在请求完成后才可知model_name属性恒为占位符mcp-samplingMCPSamplingModelSettings的自定义字段需遵循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创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考