企业官网建设流程全解析

在 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 不仅能作为 MCP（Model Context Protocol）客户端去消费外部工具，其 Agent 也可以被嵌入到 MCP Server 内部，成为"会写诗、会推理"的 MCP 工具实现。本文基于 docs/mcp/server.md 展开，先演示如何用 FastMCP + Pydantic AI Agent 快速构建一个 MCP Server，再深入讲解 MCP Sampling 机制——让服务器不再直连 LLM，而是通过客户端回调完成模型调用，从而把凭据管理与计费负担转移给客户端。读完本文，你将能独立编写可运行的 MCP Server/Client 完整示例，并理解其底层消息转换与实现约束。

一、MCP Server：Pydantic AI 的另一面

在 Pydantic AI 的 MCP 生态中，Agent 有两种截然不同的身份：

1. 作为客户端：Agent 连接 MCP 服务器并调用其暴露的工具（详见 docs/mcp/client.md）；
2. 作为服务端实现：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', 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()

拆解这段代码的要点：

• FastMCP('Pydantic AI Server')：创建名为Pydantic AI Server的 MCP 服务器实例，默认通过 stdio 传输运行（server.run()）；
• Agent('anthropic:claude-haiku-4-5', instructions='always 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( 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) """ 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())

运行流程：

1. StdioServerParameters(command='python', args=['mcp_server.py'], env=os.environ)声明以子进程方式启动服务器，并继承当前环境变量（这样服务器进程能读取到模型 API 凭据，例如ANTHROPIC_API_KEY）；
2. stdio_client(...)建立 stdin/stdout 双向管道；
3. ClientSession内先initialize()完成 MCP 握手，随后call_tool('poet', {'theme': 'socks'})调用服务器上的工具；
4. 工具返回的文本内容位于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(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

与基础示例的差异一目了然：

• Agent 不再指定模型：Agent(instructions='always reply in rhyme')没有传模型名，模型在每次运行时通过model=参数动态注入；
• ctx: Context入参：FastMCP 会把当前会话上下文注入到工具中，ctx.session就是与客户端的 MCP 会话；
• MCPSamplingModel(session=ctx.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_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) """ sampling messages: [ SamplingMessage( role='user', content=TextContent( type='text', text='write a poem about socks', annotations=None, meta=None, ), role='user', ) ] """ # TODO get the response content by calling an 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())

需要理解的关键点：

• sampling_callback的签名：接收RequestContext与CreateMessageRequestParams，后者包含服务器侧传来的systemPrompt（always reply in rhyme）与messages（这里是一条write a poem about socks的用户消息）；
• 回调必须返回CreateMessageResult：包含role、content（TextContent）以及model（表示"由哪个模型生成"，示例中用了虚构名fictional-llm）；
• ClientSession(..., sampling_callback=sampling_callback)：只有注册了回调，客户端才会响应服务器的 sampling 请求；否则服务器侧的create_message调用会失败；
• 回调内是"自由发挥"区域：示例中直接硬编码了'Socks for a fox.'以演示协议流程，生产环境中应在此调用真实 LLM（把params.systemPrompt与params.messages转发给模型服务）。

官方文档注明该示例是完整的，可直接运行（"This example is complete, it can be run as is"），sampling_callback内的TODO注释就是留给读者接入真实 LLM 的位置。

五、源码级原理：MCPSamplingModel 是如何工作的

5.1 核心类与默认值

MCPSamplingModel 继承自Model抽象基类，核心结构如下（源码事实）：

• session: ServerSession：必填字段，即 MCP 服务器会话，采样请求经由它发往客户端；
• default_max_tokens: int = 16_384：MCP Sampling 协议中max_tokens是必填参数，而 Pydantic AI 的ModelSettings.max_tokens是可选参数，因此当未显式设置时使用该默认值兜底；
• model_name属性恒返回'mcp-sampling'——因为模型名只有在请求真正发出后才能由客户端告知（CreateMessageResult.model）；
• system属性返回'MCP'，标识系统/模型提供方。

5.2 request 的完整调用链

request()方法（非流式文本生成入口）的实现逻辑：

1. 调用_mcp.map_from_pai_messages(messages)把 Pydantic AI 内部消息转换为system prompt + MCPSamplingMessage列表（转换细节见 pydantic_ai_slim/pydantic_ai/_mcp.py）：
   • ModelRequest.instructions与SystemPromptPart内容会被聚合进system_prompt字符串；
   • UserPromptPart文本转换为role='user'的TextContent；其中的图片二进制内容（BinaryContent.is_image）会被编码为ImageContent（音频转换仍是 TODO，见源码注释）；
   • ModelResponse转换为role='assistant'的文本消息，其中ThinkingPart会被跳过；
2. 调用session.create_message(...)，透传以下参数：
   • max_tokens：取model_settings['max_tokens']，缺省用default_max_tokens；
   • system_prompt、temperature、stop_sequences；
   • model_preferences：来自MCPSamplingModelSettings.mcp_model_preferences；
3. 校验返回的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 对上述行为给出了可执行的证据，值得关注几个关键用例：

此外，仓库自带的测试 MCP 服务器 tests/mcp_server.py 中的use_sampling工具演示了服务器侧完整的 sampling 调用参数：

result = await ctx.session.create_message( [ SamplingMessage(role='assistant', content=TextContent(type='text', text='')), SamplingMessage(role='user', content=TextContent(type='text', text=foo)), ], max_tokens=1_024, system_prompt='this is a test of MCP sampling', temperature=0.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的采样模型。

七、总结与选型建议

两条路线可以并存于同一个服务器：工具函数内部按需选择直连模型还是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