在 MCP Server 中集成 Pydantic AI Agent从 FastMCP 工具调用到 Sampling 采样的完整实践指南【免费下载链接】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 不仅能作为 MCPModel Context Protocol客户端去消费外部工具其 Agent 也可以被嵌入到 MCP Server 内部成为会写诗、会推理的 MCP 工具实现。本文基于 docs/mcp/server.md 展开先演示如何用 FastMCP Pydantic AI Agent 快速构建一个 MCP Server再深入讲解 MCP Sampling 机制——让服务器不再直连 LLM而是通过客户端回调完成模型调用从而把凭据管理与计费负担转移给客户端。读完本文你将能独立编写可运行的 MCP Server/Client 完整示例并理解其底层消息转换与实现约束。一、MCP ServerPydantic AI 的另一面在 Pydantic AI 的 MCP 生态中Agent 有两种截然不同的身份作为客户端Agent 连接 MCP 服务器并调用其暴露的工具详见 docs/mcp/client.md作为服务端实现Agent 被用在 MCP Server 内部作为某个 tool 的执行逻辑即本文 docs/mcp/server.md 的主题。后一种场景意味着任何支持 MCP 的客户端Claude Desktop、Cursor、其他编程框架等都可以通过标准协议调用到 Pydantic AI 驱动的能力而无需为这些客户端编写定制集成。这是 MCP 一次实现、处处复用思想的直接体现——服务器端负责业务逻辑工具定义、编排Pydantic AI Agent 负责其中需要大模型智能的部分文本生成、推理、工具调用。官方概览docs/mcp/overview.md将这条路线概括为Agents can be used within MCP servers与客户端方向Agents can connect to MCP servers and use their tools并列。二、基础示例在 FastMCP 工具内运行 Agent以下是一个完整的 Python MCP Server基于官方 python-sdk 的FastMCP封装它在poet工具内部调用 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, 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()拆解这段代码的要点FastMCP(Pydantic AI Server)创建名为Pydantic AI Server的 MCP 服务器实例默认通过 stdio 传输运行server.run()Agent(anthropic:claude-haiku-4-5, instructionsalways reply in rhyme)在服务器进程内实例化一个 Pydantic AI Agent模型选用 Anthropic 的 claude-haiku-4-5指令要求始终用押韵回复——注意 instructions 会作为系统提示词参与每次运行server.tool()装饰的poet把poet注册为 MCP 工具入参theme会根据类型注解自动生成 JSON Schema 供客户端发现await server_agent.run(...)在工具内部同步等待 Agent 完成一轮推理r.output是模型的文本输出直接作为工具返回值返回给 MCP 客户端。这里的关键思想是MCP 工具不需要自己实现如何调用 LLM只需把智能推理委托给内嵌的 Pydantic AI Agent。工具签名入参/出参是 MCP 协议的一部分而 Agent 的输出被自然地桥接为工具结果。三、简单客户端通过 stdio 连接并调用工具上面的服务器可以被任何 MCP 客户端查询。下面是直接用 Python SDK 写的 stdio 客户端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) Oh, socks, those garments soft and sweet, That nestle softly round our feet, From cotton, wool, or blended thread, They keep our toes from feeling dread. if __name__ __main__: asyncio.run(client())运行流程StdioServerParameters(commandpython, args[mcp_server.py], envos.environ)声明以子进程方式启动服务器并继承当前环境变量这样服务器进程能读取到模型 API 凭据例如ANTHROPIC_API_KEYstdio_client(...)建立 stdin/stdout 双向管道ClientSession内先initialize()完成 MCP 握手随后call_tool(poet, {theme: socks})调用服务器上的工具工具返回的文本内容位于result.content[0].text打印即为诗歌。这个客户端本身不持有任何 LLM 凭据——模型调用完全发生在服务器进程内直连claude-haiku-4-5。这正是服务器自持模型的典型形态。接下来要讨论的 Sampling 则彻底反转了这一格局。四、MCP Sampling让服务器借客户端的模型能力4.1 什么是 MCP Sampling关于 MCP Sampling 的完整定义与客户端侧支持方式参见 docs/mcp/client.md#mcp-sampling。MCP 协议中的Sampling采样是一套机制MCP Server 不再直连 LLM而是通过 MCP Client 发起创建消息sampling/createMessage请求由客户端代为调用大模型并将结果回传给服务器。从数据流看LLM 调用被代理到了客户端一侧经过传输层stdio / HTTP往返一次。这带来两个显著收益凭据集中管理服务器无需为每个部署环境配置自己的 LLM API Key客户端负责持有凭据服务器按需借用成本归属清晰公共 MCP 服务器可以让连接它的客户端为 LLM 调用付费而不是服务器运营方承担。需要澄清的是这里的 sampling 与可观测性领域的采样sampling以及任何其他领域的同名概念毫无关系它是 MCP 协议中专有的命名。典型的调用时序客户端侧文档中的 mermaid 示意图为可以看到一轮完整的工具调用过程中LLM 可能被调用两次一次在客户端侧用于决定调用哪个工具一次由服务器通过 sampling 回调发起用于生成工具执行所需的模型输出。4.2 服务器端用 MCPSamplingModel 替代直连模型在 Pydantic AI 中服务器侧启用 Sampling 的方式是使用MCPSamplingModel——一个专门封装通过 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() # run the server over stdio与基础示例的差异一目了然Agent 不再指定模型Agent(instructionsalways reply in rhyme)没有传模型名模型在每次运行时通过model参数动态注入ctx: Context入参FastMCP 会把当前会话上下文注入到工具中ctx.session就是与客户端的 MCP 会话MCPSamplingModel(sessionctx.session)构造一个通过该会话回调客户端做模型调用的模型对象。Agent 每次run()时请求都会被打包为sampling/createMessage发给客户端由客户端去调用真正的 LLM。因此服务器进程内部不再需要任何模型 API 凭据——推理能力完全来自客户端侧。4.3 客户端侧实现 sampling_callback上文#simple-client中的简单客户端不支持 Sampling——它的ClientSession没有注册采样回调。如果直接用它与 sampling 版服务器通信会得到协议错误。最简单的支持方式是用 Pydantic AI Agent 作为 MCP 客户端通过MCPToolset(sampling_model...)或agent.set_mcp_sampling_model()自动处理详见 docs/mcp/client.md#mcp-sampling。但如果想用原生的 MCP Python SDK 手写支持可以像下面这样实现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) sampling messages: [ SamplingMessage( roleuser, contentTextContent( typetext, textwrite a poem about socks, annotationsNone, metaNone, ), roleuser, ) ] # TODO get the response content by calling an 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())需要理解的关键点sampling_callback的签名接收RequestContext与CreateMessageRequestParams后者包含服务器侧传来的systemPromptalways reply in rhyme与messages这里是一条write a poem about socks的用户消息回调必须返回CreateMessageResult包含role、contentTextContent以及model表示由哪个模型生成示例中用了虚构名fictional-llmClientSession(..., sampling_callbacksampling_callback)只有注册了回调客户端才会响应服务器的 sampling 请求否则服务器侧的create_message调用会失败回调内是自由发挥区域示例中直接硬编码了Socks for a fox.以演示协议流程生产环境中应在此调用真实 LLM把params.systemPrompt与params.messages转发给模型服务。官方文档注明该示例是完整的可直接运行This example is complete, it can be run as issampling_callback内的TODO注释就是留给读者接入真实 LLM 的位置。五、源码级原理MCPSamplingModel 是如何工作的5.1 核心类与默认值MCPSamplingModel 继承自Model抽象基类核心结构如下源码事实session: ServerSession必填字段即 MCP 服务器会话采样请求经由它发往客户端default_max_tokens: int 16_384MCP Sampling 协议中max_tokens是必填参数而 Pydantic AI 的ModelSettings.max_tokens是可选参数因此当未显式设置时使用该默认值兜底model_name属性恒返回mcp-sampling——因为模型名只有在请求真正发出后才能由客户端告知CreateMessageResult.modelsystem属性返回MCP标识系统/模型提供方。5.2 request 的完整调用链request()方法非流式文本生成入口的实现逻辑调用_mcp.map_from_pai_messages(messages)把 Pydantic AI 内部消息转换为system prompt MCPSamplingMessage列表转换细节见 pydantic_ai_slim/pydantic_ai/_mcp.pyModelRequest.instructions与SystemPromptPart内容会被聚合进system_prompt字符串UserPromptPart文本转换为roleuser的TextContent其中的图片二进制内容BinaryContent.is_image会被编码为ImageContent音频转换仍是 TODO见源码注释ModelResponse转换为roleassistant的文本消息其中ThinkingPart会被跳过调用session.create_message(...)透传以下参数max_tokens取model_settings[max_tokens]缺省用default_max_tokenssystem_prompt、temperature、stop_sequencesmodel_preferences来自MCPSamplingModelSettings.mcp_model_preferences校验返回的result.role若为assistant将result.content通过map_from_sampling_content转为 Pydantic AI 的TextPart并以result.model作为model_name构造ModelResponse返回否则抛出UnexpectedModelBehavior错误信息形如Unexpected result from MCP sampling, expected assistant role, got {result.role}.。5.3 采样设置与限制MCPSamplingModelSettings继承ModelSettings新增mcp_model_preferences: ModelPreferences字段。源码特别强调所有字段必须以mcp_前缀命名以便与其他模型的设置安全合并不支持流式request_stream()直接raise NotImplementedError(MCP Sampling does not support streaming)——这是 Sampling 协议的固有限制使用时必须用非流式run()。5.4 客户端侧消息转换_mcp.py同时提供反向转换map_from_mcp_params(params: CreateMessageRequestParams)把 MCP 的采样请求参数映射回 Pydantic AI 消息system_prompt转为SystemPromptPart用户消息的TextContent/ImageContent/AudioContent转为UserPromptPart其中的图片会以 base64 解码为BinaryContent助手消息转为ModelResponse。这两组映射函数共同构成了 Pydantic AI 与 MCP Sampling 协议之间的双向翻译层。六、测试验证协议行为的关键断言仓库的测试 tests/models/test_mcp_sampling.py 对上述行为给出了可执行的证据值得关注几个关键用例测试断言要点test_mcp_sampling_model构造MCPSamplingModel(fake_session(AsyncMock()))后model_name mcp-sampling、system MCPtest_assistant_text用MCPSamplingModel作为 Agent 模型执行run_sync(Hello)输出即CreateMessageResult中的文本内容且ModelResponse.model_name来自result.modeltest_user_text当采样结果roleuser时run_sync抛出UnexpectedModelBehavior错误信息精确匹配expected assistant role, got usertest_standing_system_prompt_history历史中的 standing system prompt 会被放进create_message的system_prompt参数而不会混入sampling_messages文本test_assistant_text_history_complex包含SystemPromptPart、BinaryContent图片的复杂历史可正确转换SystemPromptPart会以system.../system文本形式出现在采样消息中此外仓库自带的测试 MCP 服务器 tests/mcp_server.py 中的use_sampling工具演示了服务器侧完整的 sampling 调用参数result await ctx.session.create_message( [ SamplingMessage(roleassistant, contentTextContent(typetext, text)), SamplingMessage(roleuser, contentTextContent(typetext, textfoo)), ], max_tokens1_024, system_promptthis is a test of MCP sampling, temperature0.5, stop_sequences[potato], )可见max_tokens、system_prompt、temperature、stop_sequences都是实际可用的采样参数与MCPSamplingModel.request的透传字段一一对应。若要在客户端侧用 Pydantic AI Agent 自动化处理 sampling调用 agent.set_mcp_sampling_model()不传参数时使用 Agent 自身的模型即可把 Agent 的模型注册为所有MCPToolset的采样模型。七、总结与选型建议形态模型凭据归属适用场景服务器直连 LLMAgent(anthropic:claude-haiku-4-5, ...)服务器进程服务器自己持有 API Key、对客户端无信任假设的私有部署服务器使用 SamplingMCPSamplingModel客户端进程公共/共享服务器希望客户端为 LLM 调用付费或自带凭据或希望客户端控制模型选择两条路线可以并存于同一个服务器工具函数内部按需选择直连模型还是MCPSamplingModel。需要注意的是Sampling 模式不支持流式输出且要求连接的客户端必须注册sampling_callback或使用 Pydantic AI 客户端自动处理否则协议调用会失败。从架构视角看Pydantic AI Agent 嵌入 MCP Server打通了协议壁垒任何 MCP 客户端都能获得类型化、可复用的智能工具而 Sampling 机制则进一步解耦了能力提供者与算力/凭据提供者是构建安全、可审计的多方 AI 系统的重要基石。【免费下载链接】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），仅供参考 在 MCP Server 中集成 Pydantic AI Agent：从 FastMCP 工具调用到 Sampling 采样的完整实践指南
在 MCP Server 中集成 Pydantic AI Agent从 FastMCP 工具调用到 Sampling 采样的完整实践指南【免费下载链接】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 不仅能作为 MCPModel Context Protocol客户端去消费外部工具其 Agent 也可以被嵌入到 MCP Server 内部成为会写诗、会推理的 MCP 工具实现。本文基于 docs/mcp/server.md 展开先演示如何用 FastMCP Pydantic AI Agent 快速构建一个 MCP Server再深入讲解 MCP Sampling 机制——让服务器不再直连 LLM而是通过客户端回调完成模型调用从而把凭据管理与计费负担转移给客户端。读完本文你将能独立编写可运行的 MCP Server/Client 完整示例并理解其底层消息转换与实现约束。一、MCP ServerPydantic AI 的另一面在 Pydantic AI 的 MCP 生态中Agent 有两种截然不同的身份作为客户端Agent 连接 MCP 服务器并调用其暴露的工具详见 docs/mcp/client.md作为服务端实现Agent 被用在 MCP Server 内部作为某个 tool 的执行逻辑即本文 docs/mcp/server.md 的主题。后一种场景意味着任何支持 MCP 的客户端Claude Desktop、Cursor、其他编程框架等都可以通过标准协议调用到 Pydantic AI 驱动的能力而无需为这些客户端编写定制集成。这是 MCP 一次实现、处处复用思想的直接体现——服务器端负责业务逻辑工具定义、编排Pydantic AI Agent 负责其中需要大模型智能的部分文本生成、推理、工具调用。官方概览docs/mcp/overview.md将这条路线概括为Agents can be used within MCP servers与客户端方向Agents can connect to MCP servers and use their tools并列。二、基础示例在 FastMCP 工具内运行 Agent以下是一个完整的 Python MCP Server基于官方 python-sdk 的FastMCP封装它在poet工具内部调用 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, 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()拆解这段代码的要点FastMCP(Pydantic AI Server)创建名为Pydantic AI Server的 MCP 服务器实例默认通过 stdio 传输运行server.run()Agent(anthropic:claude-haiku-4-5, instructionsalways reply in rhyme)在服务器进程内实例化一个 Pydantic AI Agent模型选用 Anthropic 的 claude-haiku-4-5指令要求始终用押韵回复——注意 instructions 会作为系统提示词参与每次运行server.tool()装饰的poet把poet注册为 MCP 工具入参theme会根据类型注解自动生成 JSON Schema 供客户端发现await server_agent.run(...)在工具内部同步等待 Agent 完成一轮推理r.output是模型的文本输出直接作为工具返回值返回给 MCP 客户端。这里的关键思想是MCP 工具不需要自己实现如何调用 LLM只需把智能推理委托给内嵌的 Pydantic AI Agent。工具签名入参/出参是 MCP 协议的一部分而 Agent 的输出被自然地桥接为工具结果。三、简单客户端通过 stdio 连接并调用工具上面的服务器可以被任何 MCP 客户端查询。下面是直接用 Python SDK 写的 stdio 客户端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) Oh, socks, those garments soft and sweet, That nestle softly round our feet, From cotton, wool, or blended thread, They keep our toes from feeling dread. if __name__ __main__: asyncio.run(client())运行流程StdioServerParameters(commandpython, args[mcp_server.py], envos.environ)声明以子进程方式启动服务器并继承当前环境变量这样服务器进程能读取到模型 API 凭据例如ANTHROPIC_API_KEYstdio_client(...)建立 stdin/stdout 双向管道ClientSession内先initialize()完成 MCP 握手随后call_tool(poet, {theme: socks})调用服务器上的工具工具返回的文本内容位于result.content[0].text打印即为诗歌。这个客户端本身不持有任何 LLM 凭据——模型调用完全发生在服务器进程内直连claude-haiku-4-5。这正是服务器自持模型的典型形态。接下来要讨论的 Sampling 则彻底反转了这一格局。四、MCP Sampling让服务器借客户端的模型能力4.1 什么是 MCP Sampling关于 MCP Sampling 的完整定义与客户端侧支持方式参见 docs/mcp/client.md#mcp-sampling。MCP 协议中的Sampling采样是一套机制MCP Server 不再直连 LLM而是通过 MCP Client 发起创建消息sampling/createMessage请求由客户端代为调用大模型并将结果回传给服务器。从数据流看LLM 调用被代理到了客户端一侧经过传输层stdio / HTTP往返一次。这带来两个显著收益凭据集中管理服务器无需为每个部署环境配置自己的 LLM API Key客户端负责持有凭据服务器按需借用成本归属清晰公共 MCP 服务器可以让连接它的客户端为 LLM 调用付费而不是服务器运营方承担。需要澄清的是这里的 sampling 与可观测性领域的采样sampling以及任何其他领域的同名概念毫无关系它是 MCP 协议中专有的命名。典型的调用时序客户端侧文档中的 mermaid 示意图为可以看到一轮完整的工具调用过程中LLM 可能被调用两次一次在客户端侧用于决定调用哪个工具一次由服务器通过 sampling 回调发起用于生成工具执行所需的模型输出。4.2 服务器端用 MCPSamplingModel 替代直连模型在 Pydantic AI 中服务器侧启用 Sampling 的方式是使用MCPSamplingModel——一个专门封装通过 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() # run the server over stdio与基础示例的差异一目了然Agent 不再指定模型Agent(instructionsalways reply in rhyme)没有传模型名模型在每次运行时通过model参数动态注入ctx: Context入参FastMCP 会把当前会话上下文注入到工具中ctx.session就是与客户端的 MCP 会话MCPSamplingModel(sessionctx.session)构造一个通过该会话回调客户端做模型调用的模型对象。Agent 每次run()时请求都会被打包为sampling/createMessage发给客户端由客户端去调用真正的 LLM。因此服务器进程内部不再需要任何模型 API 凭据——推理能力完全来自客户端侧。4.3 客户端侧实现 sampling_callback上文#simple-client中的简单客户端不支持 Sampling——它的ClientSession没有注册采样回调。如果直接用它与 sampling 版服务器通信会得到协议错误。最简单的支持方式是用 Pydantic AI Agent 作为 MCP 客户端通过MCPToolset(sampling_model...)或agent.set_mcp_sampling_model()自动处理详见 docs/mcp/client.md#mcp-sampling。但如果想用原生的 MCP Python SDK 手写支持可以像下面这样实现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) sampling messages: [ SamplingMessage( roleuser, contentTextContent( typetext, textwrite a poem about socks, annotationsNone, metaNone, ), roleuser, ) ] # TODO get the response content by calling an 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())需要理解的关键点sampling_callback的签名接收RequestContext与CreateMessageRequestParams后者包含服务器侧传来的systemPromptalways reply in rhyme与messages这里是一条write a poem about socks的用户消息回调必须返回CreateMessageResult包含role、contentTextContent以及model表示由哪个模型生成示例中用了虚构名fictional-llmClientSession(..., sampling_callbacksampling_callback)只有注册了回调客户端才会响应服务器的 sampling 请求否则服务器侧的create_message调用会失败回调内是自由发挥区域示例中直接硬编码了Socks for a fox.以演示协议流程生产环境中应在此调用真实 LLM把params.systemPrompt与params.messages转发给模型服务。官方文档注明该示例是完整的可直接运行This example is complete, it can be run as issampling_callback内的TODO注释就是留给读者接入真实 LLM 的位置。五、源码级原理MCPSamplingModel 是如何工作的5.1 核心类与默认值MCPSamplingModel 继承自Model抽象基类核心结构如下源码事实session: ServerSession必填字段即 MCP 服务器会话采样请求经由它发往客户端default_max_tokens: int 16_384MCP Sampling 协议中max_tokens是必填参数而 Pydantic AI 的ModelSettings.max_tokens是可选参数因此当未显式设置时使用该默认值兜底model_name属性恒返回mcp-sampling——因为模型名只有在请求真正发出后才能由客户端告知CreateMessageResult.modelsystem属性返回MCP标识系统/模型提供方。5.2 request 的完整调用链request()方法非流式文本生成入口的实现逻辑调用_mcp.map_from_pai_messages(messages)把 Pydantic AI 内部消息转换为system prompt MCPSamplingMessage列表转换细节见 pydantic_ai_slim/pydantic_ai/_mcp.pyModelRequest.instructions与SystemPromptPart内容会被聚合进system_prompt字符串UserPromptPart文本转换为roleuser的TextContent其中的图片二进制内容BinaryContent.is_image会被编码为ImageContent音频转换仍是 TODO见源码注释ModelResponse转换为roleassistant的文本消息其中ThinkingPart会被跳过调用session.create_message(...)透传以下参数max_tokens取model_settings[max_tokens]缺省用default_max_tokenssystem_prompt、temperature、stop_sequencesmodel_preferences来自MCPSamplingModelSettings.mcp_model_preferences校验返回的result.role若为assistant将result.content通过map_from_sampling_content转为 Pydantic AI 的TextPart并以result.model作为model_name构造ModelResponse返回否则抛出UnexpectedModelBehavior错误信息形如Unexpected result from MCP sampling, expected assistant role, got {result.role}.。5.3 采样设置与限制MCPSamplingModelSettings继承ModelSettings新增mcp_model_preferences: ModelPreferences字段。源码特别强调所有字段必须以mcp_前缀命名以便与其他模型的设置安全合并不支持流式request_stream()直接raise NotImplementedError(MCP Sampling does not support streaming)——这是 Sampling 协议的固有限制使用时必须用非流式run()。5.4 客户端侧消息转换_mcp.py同时提供反向转换map_from_mcp_params(params: CreateMessageRequestParams)把 MCP 的采样请求参数映射回 Pydantic AI 消息system_prompt转为SystemPromptPart用户消息的TextContent/ImageContent/AudioContent转为UserPromptPart其中的图片会以 base64 解码为BinaryContent助手消息转为ModelResponse。这两组映射函数共同构成了 Pydantic AI 与 MCP Sampling 协议之间的双向翻译层。六、测试验证协议行为的关键断言仓库的测试 tests/models/test_mcp_sampling.py 对上述行为给出了可执行的证据值得关注几个关键用例测试断言要点test_mcp_sampling_model构造MCPSamplingModel(fake_session(AsyncMock()))后model_name mcp-sampling、system MCPtest_assistant_text用MCPSamplingModel作为 Agent 模型执行run_sync(Hello)输出即CreateMessageResult中的文本内容且ModelResponse.model_name来自result.modeltest_user_text当采样结果roleuser时run_sync抛出UnexpectedModelBehavior错误信息精确匹配expected assistant role, got usertest_standing_system_prompt_history历史中的 standing system prompt 会被放进create_message的system_prompt参数而不会混入sampling_messages文本test_assistant_text_history_complex包含SystemPromptPart、BinaryContent图片的复杂历史可正确转换SystemPromptPart会以system.../system文本形式出现在采样消息中此外仓库自带的测试 MCP 服务器 tests/mcp_server.py 中的use_sampling工具演示了服务器侧完整的 sampling 调用参数result await ctx.session.create_message( [ SamplingMessage(roleassistant, contentTextContent(typetext, text)), SamplingMessage(roleuser, contentTextContent(typetext, textfoo)), ], max_tokens1_024, system_promptthis is a test of MCP sampling, temperature0.5, stop_sequences[potato], )可见max_tokens、system_prompt、temperature、stop_sequences都是实际可用的采样参数与MCPSamplingModel.request的透传字段一一对应。若要在客户端侧用 Pydantic AI Agent 自动化处理 sampling调用 agent.set_mcp_sampling_model()不传参数时使用 Agent 自身的模型即可把 Agent 的模型注册为所有MCPToolset的采样模型。七、总结与选型建议形态模型凭据归属适用场景服务器直连 LLMAgent(anthropic:claude-haiku-4-5, ...)服务器进程服务器自己持有 API Key、对客户端无信任假设的私有部署服务器使用 SamplingMCPSamplingModel客户端进程公共/共享服务器希望客户端为 LLM 调用付费或自带凭据或希望客户端控制模型选择两条路线可以并存于同一个服务器工具函数内部按需选择直连模型还是MCPSamplingModel。需要注意的是Sampling 模式不支持流式输出且要求连接的客户端必须注册sampling_callback或使用 Pydantic AI 客户端自动处理否则协议调用会失败。从架构视角看Pydantic AI Agent 嵌入 MCP Server打通了协议壁垒任何 MCP 客户端都能获得类型化、可复用的智能工具而 Sampling 机制则进一步解耦了能力提供者与算力/凭据提供者是构建安全、可审计的多方 AI 系统的重要基石。【免费下载链接】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），仅供参考