摘要：MCP Sampling原语让Server反向请求Client端的LLM能力，实现服务端AI推理。本文详解采样请求流程、模型偏好设置、安全审批机制和典型应用场景。

Sampling原语让Server反向请求LLM

前段时间我做一个数据分析工具，工具内部需要让大模型对查询结果做一轮摘要再返回。我一开始的方案是在Server里硬塞一个OpenAI的API Key，直接调GPT。结果上线第二天安全团队找上门，说Server里不该存API Key，而且不同用户想用不同模型也没法满足。后来我改用MCP的Sampling原语，Server不持有任何Key，需要LLM时反向请求Client，由Client用自己的模型能力生成结果。这篇我把这个"逆向通信"机制讲清楚。

---

Sampling的逆向通信机制

前面三篇讲的Tools、Resources、Prompts都是Server向Client暴露能力，方向是Server到Client。Sampling正好反过来，它是Client的能力，Server在执行过程中可以请求Client帮忙调一次LLM。方向变成了Server请求Client。

这个设计解决了一个核心矛盾。Server经常需要在工具执行中借助LLM做推理，比如分析一段数据、生成一段摘要、判断一个分类。但Server不应该自己持有模型API Key，那样既不安全又限制了用户选模型的自由。Sampling让Server说"我需要一次LLM生成"，具体用哪个模型、用什么Key，全由Client决定。

协议上这走的是sampling/createMessage这个JSON-RPC方法，由Server发起请求。请求里带上messages对话内容、modelPreferences模型偏好、systemPrompt系统提示、maxTokens等参数。Client收到后调自己的LLM，把生成结果返回给Server。

整个流程有个重要的安全设计，规范要求必须有人类在环。Client在调LLM前应该让用户审查和编辑请求，生成结果也要给用户过目再返回。我实际用下来，这个审查环节在开发阶段特别有用，能直接看到Server到底给LLM发了什么。

Server如何请求Client的LLM能力

在FastMCP里，Server端通过Context对象的sample方法发起请求。你在工具函数里拿到ctx，调ctx.sample()，框架自动帮你构造sampling/createMessage请求发给Client。

下面是协议层面的请求结构。

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "sampling/createMessage",
  "params": {
    "messages": [
      {"role": "user", "content": {"type": "text", "text": "总结这段数据的趋势"}}
    ],
    "modelPreferences": {
      "hints": [{"name": "claude-3-sonnet"}],
      "intelligencePriority": 0.8,
      "speedPriority": 0.5
    },
    "systemPrompt": "你是一个数据分析助手",
    "maxTokens": 100
  }
}

modelPreferences是Sampling的精华设计。Server不能直接指定模型名，因为Client未必有那个模型。于是Server用hints给提示，用三个优先级表达需求。costPriority越高越想省钱，speedPriority越高越想要快，intelligencePriority越高越想要强。Client综合这些偏好，从自己可用的模型里挑一个合适的。

比如Server说"我想要claude-3-sonnet这个级别的，速度优先"，Client如果只有Gemini，可以映射到一个速度快的Gemini模型。hints是子串匹配，多个hint按优先级排序，Client尽量满足。

安全考量

Sampling的安全模型围绕"控制权在Client"展开。Server全程不接触API Key，不接触模型选择，甚至连最终返回什么内容都是Client审查后决定的。

我总结了几条安全要点。第一，Client必须实现用户审批，Server发来的sampling请求要先给用户看，用户同意才转发给LLM。第二，用户可以编辑请求内容，防止Server通过精心构造的prompt做坏事。第三，生成结果也要审查，避免Server拿到不该拿的信息。第四，Client应该做限流，防止Server疯狂发sampling请求烧token。

我踩过一个真实的坑。有个Server工具在循环里调ctx.sample()，每轮都生成一段分析，循环没设退出条件，token哗哗地烧。后来我在Client端的handler里加了调用计数，超过5次直接拒绝，问题才控制住。Server端的循环一定要有明确的终止条件。

完整代码

下面是完整示例。Server提供一个数据分析工具，内部用ctx.sample()请求Client的LLM做摘要。Client端配置一个自定义sampling_handler，为了不依赖真实API Key，我用了一个mock handler返回模拟结果，真实场景换成OpenAI等内置handler即可。

server.py

# server.py  MCP Sampling原语完整示例
# 运行方式  python server.py
# 依赖安装  pip install fastmcp

from fastmcp import FastMCP, Context

# 创建服务器实例
mcp = FastMCP(name="SamplingDemoServer")


@mcp.tool
async def analyze_and_summarize(data: str, ctx: Context) -> str:
    """分析数据并用客户端的LLM生成摘要.

    这个工具先做本地处理, 再请求Client的LLM做摘要,
    整个过程Server不持有任何模型API Key.
    """
    # 第一步, 本地预处理, 统计数据基本特征
    lines = data.strip().split("\n")
    local_stats = f"共 {len(lines)} 行数据"

    # 第二步, 通过ctx.sample请求Client的LLM生成摘要
    # 这里Server只表达需求, 具体用哪个模型由Client决定
    result = await ctx.sample(
        messages=f"请用一句话总结以下数据的核心趋势.\n\n{data}",
        system_prompt="你是一个简洁的数据分析助手, 只输出结论.",
        max_tokens=80,
        # hints告诉Client倾向用什么级别的模型
        model_preferences=["claude-3-sonnet", "gpt-4o"],
    )

    # result.text是LLM生成的文本
    summary = result.text or "摘要生成失败"

    # 组合本地统计和LLM摘要一起返回
    return f"本地统计  {local_stats}\nLLM摘要  {summary}"


@mcp.tool
async def classify_text(text: str, ctx: Context) -> str:
    """用客户端LLM对文本做情感分类.

    演示Sampling在分类场景的应用,
    Server定义分类规则, LLM负责判断.
    """
    result = await ctx.sample(
        messages=(
            f"判断以下文本的情感倾向, 只回复 正面/负面/中性 三个词之一.\n\n"
            f"文本  {text}"
        ),
        system_prompt="你是一个情感分析器, 严格只输出一个分类标签.",
        max_tokens=10,
        # 分类任务速度优先, 不需要最强模型
        model_preferences=["gpt-4o-mini", "claude-3-haiku"],
    )
    return result.text or "分类失败"


if __name__ == "__main__":
    mcp.run()

client_test.py

# client_test.py  带sampling_handler的客户端测试
# 运行方式  python client_test.py
# 这个脚本连接server.py, 并提供一个mock的sampling_handler

import asyncio
from fastmcp import Client
from fastmcp.client.sampling import SamplingMessage, SamplingParams, RequestContext


# 自定义sampling_handler, 处理Server发来的LLM生成请求
# 真实场景替换成OpenAI或Anthropic的内置handler
async def mock_sampling_handler(
    messages: list[SamplingMessage],
    params: SamplingParams,
    context: RequestContext,
) -> str:
    """模拟LLM生成的handler, 不依赖真实API Key.

    真实项目中用内置handler替代, 例如
    from fastmcp.client.sampling.handlers.openai import OpenAISamplingHandler
    sampling_handler=OpenAISamplingHandler(default_model="gpt-4o")

    这里为了演示可独立运行, 返回模拟文本.
    """
    # 调用计数, 防止Server死循环烧请求
    # 生产环境也应该做限流
    count = getattr(mock_sampling_handler, "_count", 0) + 1
    mock_sampling_handler._count = count
    if count > 10:
        raise RuntimeError("sampling调用次数超限, 疑似死循环")

    # 提取最后一条用户消息的文本
    last_msg = messages[-1]
    # SamplingMessage的content可能是TextContent, 有text属性
    user_text = last_msg.content.text if hasattr(last_msg.content, "text") else str(last_msg.content)

    # 根据system_prompt决定返回什么, 模拟不同LLM行为
    system = params.systemPrompt or ""
    if "情感" in system or "分类" in user_text:
        return "正面"
    # 默认返回一个模拟摘要
    return f"[模拟摘要] 该数据呈现稳定上升趋势, 共涉及{len(user_text)}个字符."


async def main():
    # 创建带sampling_handler的Client
    # Client会自动声明sampling能力, Server就能发createMessage请求了
    async with Client(
        "server.py",
        sampling_handler=mock_sampling_handler,
    ) as client:

        # 测试数据分析工具, 它内部会触发sampling
        print("=== 调用 analyze_and_summarize ===")
        result = await client.call_tool(
            "analyze_and_summarize",
            {"data": "1月销量100\n2月销量120\n3月销量150\n4月销量180"},
        )
        print(f"  结果  {result.structured_content}")
        print()

        # 测试分类工具, 同样内部触发sampling
        print("=== 调用 classify_text ===")
        result = await client.call_tool(
            "classify_text",
            {"text": "今天天气真好, 心情特别愉快!"},
        )
        print(f"  分类  {result.structured_content}")


if __name__ == "__main__":
    asyncio.run(main())

效果验证

装好fastmcp后跑client_test.py。因为用了mock handler，不需要任何API Key就能看到完整流程。输出大致如下。

=== 调用 analyze_and_summarize ===
  结果  {'本地统计': '共 4 行数据', 'LLM摘要': '[模拟摘要] 该数据呈现稳定上升趋势, 共涉及38个字符.'}

=== 调用 classify_text ===
  分类  正面

可以看到Server的工具在执行中调用了Client的LLM能力，Client的handler处理后把结果返回给Server，Server再组合成最终结果。整个过程中Server没接触任何模型API Key。

真实场景下，把mock_sampling_handler换成内置的OpenAISamplingHandler或AnthropicSamplingHandler就行。装好对应扩展后，一行代码切换。

Sampling与Tools的对比

Sampling和Tools经常被放一起讨论，因为它们都涉及"调用"，但方向完全相反。我做了个对比。

维度	Sampling	Tools
通信方向	Server请求Client	Client请求Server
谁发起	Server在工具执行中发起	Client/模型发起
能力归属	Client的LLM能力	Server的函数能力
API Key	Server不需要, Client持有	Server自己执行逻辑
典型场景	Server需要AI推理时借力	模型需要执行外部操作
控制方	Client控制模型选择和审批	Server控制工具逻辑

简单记，Tools是"模型要干活，问Server要工具"，Sampling是"Server要思考，问Client借大脑"。两者经常配合使用，Server工具内部用Sampling做推理，推理结果再返回给模型。

我做过一个最典型的组合场景。一个数据分析Server，工具先查数据库拿到原始数据，再用Sampling让Client的LLM分析趋势，最后把分析结果返回给用户。Server只负责数据获取，AI推理交给Client，职责分得清清楚楚。

常见问题与避坑

坑1，循环里调sample导致token爆炸。 Server工具在while循环里反复ctx.sample()，没设退出条件，每轮都烧token。我亲历过一晚上烧了几十刀。Server端循环必须有明确终止条件，Client端handler也要加调用计数限流。

坑2，客户端没声明sampling能力直接报错。 Sampling是可选的Client能力，Client没配sampling_handler时Server调ctx.sample()会失败。用sampling_handler_behavior="fallback"配一个兜底handler，Client不支持时自动走自己的LLM。

坑3，modelPreferences的hints写得太具体匹配不到。 hints是子串匹配，写一个完整版本号可能Client没有完全一致的模型。写模型族名比如"claude-3-sonnet"比写"claude-3-sonnet-20240229"更容易匹配到，Client会映射到同级别的模型。

坑4，忽略了人类在环的审批延迟。 规范要求Client在调LLM前让用户审批，这会引入延迟。如果你的工具对延迟敏感，在prompt里把审批必要的信息写清楚，让用户快速判断要不要批准。别在时序要求极高的场景盲目用Sampling。

坑5，把敏感数据塞进sampling请求。 Server把数据库里的用户隐私数据原样发给Client的LLM，可能违反数据合规要求。发送前做脱敏处理，或者只发聚合统计结果让LLM分析趋势，别发原始明细。

小结

Sampling原语实现了MCP里唯一的逆向通信，Server在执行中请求Client的LLM能力。核心要点有三个，通信方向是Server请求Client，modelPreferences用hints和优先级表达模型偏好不强制指定，安全模型把控制权完全交给Client包括模型选择和人类审批。它和Tools形成对偶关系，Tools是Client借Server的能力，Sampling是Server借Client的大脑。下一篇我们看最后一个原语Elicitation，它让Server能在执行中向用户要输入。

---

相关推荐

• Prompts原语：标准化提示词模板
• • Elicitation原语：Server向Client请求用户输入
• • MCP协议全景：Host、Client、Server架构详解