FastMCP高级特性之Sampling（待重新排版）

在 LLM（大语言模型） 的语境中，Sampling（采样） 指的是模型从预测的候选词概率分布中，选择下一个词来生成文本的过程。
简单来说，大语言模型生成文本时，不会直接 “确定” 下一个词，而是先计算出当前语境下所有可能出现的词的概率值，再通过特定的采样策略，从这个概率分布里挑选一个词，逐步拼接成完整的文本。

一、给MCP服务绑定LLM

importasyncioimportosfromopenaiimportOpenAIfromfastmcpimportFastMCPfromfastmcp.experimental.sampling.handlers.openaiimportOpenAISamplingHandler# 1. 初始化 FastMCP 服务器，配置大模型处理器server=FastMCP(name="Server-Controlled LLM",# 配置 OpenAI 大模型处理器（支持官方 API 或兼容 API）sampling_handler=OpenAISamplingHandler(default_model="gpt-4o-mini",# 默认使用的大模型client=OpenAI(api_key=os.getenv("OPENAI_API_KEY"),# 大模型 API 密钥base_url=os.getenv("OPENAI_BASE_URL"),# 可选：自定义 API 地址（如代理、私有部署）),),# 触发策略：always（强制用服务端大模型）/ fallback（客户端不支持时兜底）sampling_handler_behavior="always",)

二、通过 ctx.sample() 与大模型问答

方法一

*ctx.sample# async method请求大语言模型生成文本，并自动运行直至完成。

场景 1：简单文本生成（基础提问）

@server.toolasyncdefgenerate_summary(content:str,ctx:Context)->str:"""向大模型请求生成文本摘要"""# 构造提问提示prompt=f"请简洁总结以下内容：\n\n{content}"# 调用大模型response=awaitctx.sample(messages=prompt,# 简单字符串提问temperature=0.3,# 低随机性，保证摘要稳定max_tokens=300,# 限制摘要长度)# 返回大模型生成的结果（response.text 为文本内容）returnresponse.text

场景 2：带系统提示的专业问答（定义大模型角色）

@server.toolasyncdefgenerate_python_code(concept:str,ctx:Context)->str:"""让大模型生成 Python 代码示例（指定角色为专家）"""response=awaitctx.sample(messages=f"写一个简单的 Python 代码示例，演示：{concept}",system_prompt="你是资深 Python 工程师，只提供可运行的简洁代码，不附加解释",# 系统提示定义角色temperature=0.7,# 适度随机性，允许代码变体max_tokens=500,model_preferences="gpt-4",# 本次请求优先使用 gpt-4（覆盖默认的 gpt-4o-mini）)# 格式化代码输出returnf"```python\n{response.text}\n```"

场景 3：多轮对话（结构化消息）
通过 SamplingMessage 构造多轮对话上下文，支持角色区分（user/assistant）：

fromfastmcp.client.samplingimportSamplingMessage@server.toolasyncdefmulti_turn_analysis(user_query:str,context_data:str,ctx:Context)->str:"""多轮对话式分析（带历史上下文）"""# 构造多轮消息（角色+内容）messages=[SamplingMessage(role="user",content=f"我的数据：{context_data}"),# 第一轮：用户提供数据SamplingMessage(role="assistant",content="已收到数据，你想分析什么？"),# 第二轮：助手回应SamplingMessage(role="user",content=user_query)# 第三轮：用户具体提问]response=awaitctx.sample(messages=messages,# 传入结构化消息列表system_prompt="你是数据分析师，基于对话上下文提供详细分析",temperature=0.2,# 低随机性，保证分析严谨)returnresponse.text

场景 4：情感分析（简单分类任务）

@server.toolasyncdefanalyze_sentiment(text:str,ctx:Context)->dict:"""让大模型分析文本情感（正面/负面/中性）"""prompt=f""" 分析以下文本的情感，仅输出一个词：positive（正面）、negative（负面）或 neutral（中性）。 文本：{text}"""response=awaitctx.sample(prompt,temperature=0.0)# 0 随机性，保证分类一致sentiment=response.text.strip().lower()# 标准化结果if"positive"insentiment:result="positive"elif"negative"insentiment:result="negative"else:result="neutral"return{"text":text,"sentiment":result}

方法二

ctx.sample_step()

三、结构化输出

当你需要经过验证的、有类型的数据而非自由格式的文本时，请使用 result_type 参数。FastMCP 确保大语言模型返回与你的类型相匹配的数据，并自动处理验证和重试操作。result_type 参数可接受 Pydantic 模型、数据类以及基本类型，如 int、list [str] 或 dict [str, int]。

frompydanticimportBaseModelfromfastmcpimportFastMCP,Context mcp=FastMCP()classSentimentResult(BaseModel):sentiment:strconfidence:floatreasoning:str@mcp.toolasyncdefanalyze_sentiment(text:str,ctx:Context)->SentimentResult:"""Analyze text sentiment with structured output."""result=awaitctx.sample(messages=f"Analyze the sentiment of:{text}",result_type=SentimentResult,)returnresult.result# A validated SentimentResult object

当你调用这个工具时，大语言模型（LLM）会返回一个结构化响应，FastMCP 会根据你的 Pydantic 模型对该响应进行验证。你可以通过result.result访问经过验证的对象，而result.text则包含 JSON 格式的表示。

当你传入result_type时，sample ()会自动创建一个 final_response 工具，大语言模型会调用该工具来提供其响应。如果验证失败，错误会被发送回大语言模型以进行重试。这种自动处理仅适用于sample ()—— 对于sample_step ()，你必须自行管理结构化输出。

四、使用工具进行采样

借助工具进行采样能够实现智能体工作流，让大语言模型（LLM）在做出回应前可以调用函数来收集信息。这一功能实现了 SEP-1577，允许大语言模型自主编排多步骤操作。

将 Python 函数传递给 tools 参数，FastMCP 会自动处理执行循环 —— 调用工具、将结果返回给大语言模型，并持续这一过程，直到大语言模型给出最终回应。

4.1 定义工具

使用类型提示和文档字符串定义常规的 Python 函数。FastMCP 会提取函数的名称、文档字符串和参数类型，以创建大语言模型能够理解的工具模式。

fromfastmcpimportFastMCP,Contextdefsearch(query:str)->str:"""Search the web for information."""returnf"Results for:{query}"defget_time()->str:"""Get the current time."""fromdatetimeimportdatetimereturndatetime.now().strftime("%H:%M:%S")mcp=FastMCP()@mcp.toolasyncdefresearch(question:str,ctx:Context)->str:"""Answer questions using available tools."""result=awaitctx.sample(messages=question,tools=[search,get_time],)returnresult.textor""

大语言模型（LLM）会查看每个函数的签名和文档字符串，并利用这些信息来决定何时以及如何调用它们。工具错误会被捕获并反馈给大语言模型，使其能够平稳地恢复。一个内部安全限制可防止无限循环。

4.2 工具错误处理

默认情况下，当采样工具抛出异常时，错误消息（包括详细信息）会发送回大语言模型，以便其尝试恢复。为防止敏感信息泄露给大语言模型，请使用 mask_error_details 参数：

result=awaitctx.sample(messages=question,tools=[search],mask_error_details=True,# Generic error messages only)

当 mask_error_details=True 时，工具错误会变成诸如 “执行工具‘search’时出错” 之类的通用消息，而不会暴露堆栈跟踪或内部细节。

若要不管是否启用屏蔽，都特意向 LLM 提供特定的错误消息，请抛出 ToolError：

fromfastmcp.exceptionsimportToolErrordefsearch(query:str)->str:"""Search for information."""ifnotquery.strip():raiseToolError("Search query cannot be empty")returnf"Results for:{query}"

工具错误消息总会传递给大语言模型（LLM），使其成为你希望大语言模型看到并处理错误的应急出口。

对于自定义名称或描述，请使用 SamplingTool.from_function ()：

fromfastmcp.server.samplingimportSamplingTool tool=SamplingTool.from_function(my_func,name="custom_name",description="Custom description")result=awaitctx.sample(messages="...",tools=[tool])

4.3 组合结构化输出

将工具与结果类型相结合，用于生成能返回经过验证的结构化数据的智能代理工作流。大语言模型（LLM）会使用你的工具来收集信息，然后返回与你的类型相匹配的响应。

result=awaitctx.sample(messages="Research Python async patterns",tools=[search,fetch_url],result_type=ResearchResult,)

五、循环控制

虽然 sample () 会自动处理工具执行循环，但在某些场景下，需要对每个步骤进行精细控制。sample_step () 方法会进行一次 LLM 调用，并返回一个包含响应和更新后历史记录的 SampleStep。

与 sample () 不同，sample_step () 是无状态的 —— 它不会记住之前的调用。你需要通过每次传递完整的消息历史来控制对话。

返回的 step.history 包含截至当前响应的所有消息，便于继续循环。

在以下情况时使用 sample_step ()：

• 在工具调用执行前对其进行检查
• 实现自定义终止条件
• 在步骤之间添加日志记录、指标监控或检查点设置
• 利用特定领域逻辑构建自定义智能体循环

5.1 使用 sample_step ()

默认情况下，sample_step () 会执行所有工具调用，并将结果包含在历史记录中。在循环中调用它，每次传递更新后的历史记录，直到满足停止条件。

frommcp.typesimportSamplingMessage@mcp.toolasyncdefcontrolled_agent(question:str,ctx:Context)->str:"""Agent with manual loop control."""messages:list[str|SamplingMessage]=[question]# strings auto-convertwhileTrue:step=awaitctx.sample_step(messages=messages,tools=[search,get_time],)ifstep.is_tool_use:# Tools already executed (execute_tools=True by default)# Log what was called before continuingforcallinstep.tool_calls:print(f"Called tool:{call.name}")ifnotstep.is_tool_use:returnstep.textor""# Continue with updated historymessages=step.history

5.2 SampleStep属性

每个 SampleStep 都会提供有关大语言模型（LLM）返回内容的信息：

• step.is_tool_use— 如果大语言模型请求调用工具，则为 True
• step.tool_calls— 请求的工具调用列表（如有）
• step.text— 文本内容（如有）
• step.history— 到目前为止交换的所有消息
• step.history的内容取决于 execute_tools：
• execute_tools=True（默认值）：包含工具结果，为下一次迭代做好准备
• execute_tools=False：包含助手的工具请求，但需要您自行添加结果

5.3 手动执行工具

将 execute_tools 设置为 False，以便自行处理工具执行。禁用该功能时，step.history 会包含用户消息以及带有工具调用的助手回复，但不包含工具结果。您需要执行这些工具，并将结果作为用户消息追加进去。

frommcp.typesimportSamplingMessage,ToolResultContent,TextContentfromfastmcpimportFastMCP,Context mcp=FastMCP()@mcp.toolasyncdefresearch(question:str,ctx:Context)->str:"""Research with manual tool handling."""defsearch(query:str)->str:returnf"Results for:{query}"defget_time()->str:return"12:00 PM"# Map tool names to functionstools={"search":search,"get_time":get_time}messages:list[SamplingMessage]=[question]# strings are converted automaticallywhileTrue:step=awaitctx.sample_step(messages=messages,tools=list(tools.values()),execute_tools=False,)ifnotstep.is_tool_use:returnstep.textor""# Execute tools and collect resultstool_results=[]forcallinstep.tool_calls:fn=tools[call.name]result=fn(**call.input)tool_results.append(ToolResultContent(type="tool_result",toolUseId=call.id,content=[TextContent(type="text",text=result)],))messages=list(step.history)messages.append(SamplingMessage(role="user",content=tool_results))

错误处理
要报告错误，请将 isError 设置为 True。大语言模型（LLM）会看到该错误，并能决定如何继续处理：

tool_result=ToolResultContent(type="tool_result",toolUseId=call.id,content=[TextContent(type="text",text="Permission denied")],isError=True,)

六、备用处理

MCP 默认优先调用客户端已配置的大模型，服务端无需额外配置大模型信息，直接通过 ctx.sample() 触发即可。

客户端对采样的支持是可选的 —— 有些客户端可能不会实现这一功能。为确保你的工具无论客户端具备何种能力都能正常运行，请配置一个直接向大语言模型（LLM）提供商发送请求的采样处理器（sampling_handler）。

FastMCP 为 OpenAI 和 Anthropic 的应用程序接口（API）提供了内置处理器。这些处理器支持完整的采样应用程序接口，包括工具相关功能，能自动将你的 Python 函数转换为各个提供商所需的格式。

使用 pip install fastmcp [openai] 或 pip install fastmcp [anthropic] 安装处理器。

fromfastmcpimportFastMCPfromfastmcp.client.sampling.handlers.openaiimportOpenAISamplingHandler server=FastMCP(name="My Server",sampling_handler=OpenAISamplingHandler(default_model="gpt-4o-mini"),sampling_handler_behavior="fallback",)

或者

fromfastmcpimportFastMCPfromfastmcp.client.sampling.handlers.anthropicimportAnthropicSamplingHandler server=FastMCP(name="My Server",sampling_handler=AnthropicSamplingHandler(default_model="claude-sonnet-4-5"),sampling_handler_behavior="fallback",)

6.1 行为模式

sampling_handler_behavior参数控制handler的使用时机：

• “fallback”（默认值）：仅当客户端不支持采样时使用处理器。这使得有能力的客户端可以使用自己的大语言模型（LLM），同时确保你的工具在缺乏采样支持的客户端上仍然能正常工作。
• “always”：始终使用handler，完全绕过客户端。当你需要确保对处理请求的大语言模型（LLM）拥有控制权时使用此选项，例如出于成本控制、合规要求，或者当特定的模型特性至关重要时。

使用工具进行采样要求客户端声明具备sampling.tools能力。FastMCP 客户端会自动进行声明。对于不支持工具启用采样的外部客户端，请将 fallback 处理器配置为 sampling_handler_behavior=“always”。