# MCP Sampling 反向调用：构建零密钥的智能 MCP 工具 ## 概述 本教程聚焦 MCP（Model Context Protocol）中一个被严重低估的能力——**Sampling（采样）**。绝大多数 MCP 教程只讲"宿主调用工具"的正向流，而 Sampling 恰好相反：**让工具服务器反过来借用宿主的 LLM 完成推理**。读完本文，你将掌握 Sampling 的协议契约、消息流、决策矩阵，并用 Python（FastMCP 的 `ctx.sample`）与 TypeScript（`createMessage`）分别实现一台**不需要任何 API Key 即可拥有智能**的服务器，同时补齐大多数教程缺失的另一半——如何实现 Host Client 侧的 sampling handler。 **适用人群**：已了解 MCP 三原语（Tool/Resource/Prompt）、希望让自建工具获得 LLM 推理能力但不想在服务器里硬编码模型供应商与密钥的中高级开发者。 ## 前置条件 - Node.js 18+ 与 Python 3.10+ 双环境（实战案例两种语言各一套） - 已掌握 MCP 基础概念，能独立用 `@modelcontextprotocol/sdk` 或 `fastmcp` 跑通一个 stdio 服务器 - 一个声明了 `sampling` 能力的 Host（见第九节关于 Claude Code 支持现状的说明） - 安装依赖：`npm i @modelcontextprotocol/sdk` 与 `pip install fastmcp` --- ## 一、为什么需要 Sampling：反向调用的本质 理解 Sampling，先看普通 MCP 流向：**Client（宿主 LLM，例如 Claude）驱动一切**——它调用工具、读取资源、渲染提示词，服务器始终被动等待。这套模型有一个硬伤：当工具执行到一半，发现自己需要一个"判断"（这段日志是不是严重？这封邮件该不该归档？这段代码有没有坏味道？），服务器就只能把半成品结果丢回给宿主，让宿主再判断一轮，往返冗长且服务器逻辑被割裂。 Sampling 引入了一条**反向通道**：服务器在工具执行过程中，主动向宿主发起一次 `sampling/createMessage` 请求，请宿主的 LLM 生成一段补全，拿到结果后继续自己的逻辑。**服务器全程不触碰任何 LLM API，不持有任何密钥，不绑定任何供应商**——模型选择、计费、限流、安全审核全部留在宿主侧。 ```mermaid flowchart LR subgraph 正向["普通工具调用（Client → Server）"] A1[Client LLM] -->|tools/call| A2[Server 工具] A2 -->|结果| A1 end subgraph 反向["Sampling（Server → Client）"] B1[Server 工具] -->|sampling/createMessage| B2[Client LLM] B2 -->|补全| B1 end ``` 这张图是整篇文章的认知锚点：**Sampling 把"服务器需要智能"这件事从服务器里剥离出来，外包给宿主**。gyliu 在其架构指南中总结得最到位——它消除了服务器侧的模型运行时依赖、GPU 需求与凭证管理，让服务器"保持专注在确定性业务逻辑上"。 ## 二、协议契约：createMessage 与能力声明 ### 2.1 能力声明（Capability） 不是所有 Host 都支持 Sampling。按照 MCP 2025-06-18 规范，**支持采样的 Client 必须在初始化握手时声明 `sampling` 能力**： ```json { "capabilities": { "sampling": {} } } ``` 服务器侧同样需要声明自己将使用 sampling（不同 SDK 写法不同，FastMCP 通过 `Context` 类型注入，TypeScript SDK 在 server capabilities 里写 `sampling: {}`）。这份双向声明让不支持采样的 Host 能在连接早期就拒绝握手，而不是在工具执行到一半时才崩溃。 ### 2.2 请求结构 服务器请求一次补全，发送 `sampling/createMessage`： ```json { "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 } } ``` 四个关键字段： - **`messages`**：类聊天的角色/内容列表，内容支持三种类型——`text`、`image`（base64 + mimeType）、`audio`（base64 + mimeType）。这意味着采样天然支持多模态。 - **`systemPrompt`**：服务器可注入一个系统提示，约束本次补全的行为。 - **`maxTokens`**：建议上限，宿主可酌情调整。 - **`modelPreferences`**：模型偏好，这是 Sampling 设计中最精巧的部分。 ### 2.3 响应结构 ```json { "jsonrpc": "2.0", "id": 1, "result": { "role": "assistant", "content": { "type": "text", "text": "法国的首都是巴黎。" }, "model": "claude-3-sonnet-20240307", "stopReason": "endTurn" } } ``` 注意响应里带回了 `model` 与 `stopReason`——服务器据此知道**实际**用哪个模型生成的、为什么停止（`endTurn` / `maxTokens` / 其他），便于审计与降级判断。 ### 2.4 modelPreferences：偏好而非命令 服务器不能指定"必须用某个模型"，因为宿主可能根本没有那个模型。MCP 用一套**抽象偏好系统**化解跨供应商差异： - 三个归一化优先级（0–1）：`costPriority`（越省钱越好）、`speedPriority`（越快越好）、`intelligencePriority`（越聪明越好）。 - `hints`：建议的模型名，作为**子串**灵活匹配，按优先级排列。宿主**可以**把 hint 映射到其他供应商的等价模型——例如服务器 hint 写 `claude-3-sonnet`，宿主若只配了 Gemini，可映射成 `gemini-1.5-pro`。 ``` { "hints": [{ "name": "claude-3-sonnet" }, { "name": "claude" }], "costPriority": 0.3, "speedPriority": 0.8, "intelligencePriority": 0.5 } ``` 这是 **advisory（建议性）**——宿主拥有最终决定权。理解这一点很重要：你的采样请求只是"许愿"，不是"下单"。 ## 三、消息流：人在环路的四段式 规范明确要求 Sampling **应当始终有人类在环（human-in-the-loop）**，可拒绝任何采样请求。完整时序如下： ```mermaid sequenceDiagram participant S as Server participant C as Client participant U as User participant L as LLM Note over S,C: 1. 服务器发起采样 S->>C: sampling/createMessage Note over C,U: 2. 人审请求 C->>U: 展示请求供审核 U-->>C: 审阅/修改/批准 Note over C,L: 3. 模型交互 C->>L: 转发已批准请求 L-->>C: 返回补全 Note over C,U: 4. 人审响应 C->>U: 展示响应供审核 U-->>C: 审阅/修改/批准 C-->>S: 返回已批准响应 ``` 这套四段式（发起 → 人审请求 → 模型 → 人审响应）是 Sampling 安全性的根基。对于不受信任的服务器，宿主会默认弹出审核；对于处理敏感数据的服务器，文档应当写清"它会发送哪些采样提示词"，让用户能审计自己的宿主 LLM 到底看到了什么。 ## 四、何时用、何时不用：决策矩阵 Sampling 不是万能药。markaicode 给出了一张非常实用的取舍表，浓缩如下： | 场景 | 用 Sampling | 用确定性逻辑/工具链 | |---|:---:|:---:| | 执行中途需要分类/解读数据 | ✅ | ❌ | | 从非结构化文本抽取结构化 JSON | ✅ | ❌ | | 逻辑完全确定（能用正则/计算解决） | ❌ | ✅ | | 服务器要分发给不同供应商环境的用户 | ✅（模型无关） | ❌ | | 需要可复现、可测试的输出 | ❌ | ✅ | **一句话原则**：当你的工具逻辑里真有那么一步"需要语言理解"，且你不想在服务器里硬编码模型供应商时，才用 Sampling；纯计算、正则、确定性变换，老老实实写代码，别给一次 LLM 往返徒增延迟和不确定性。 ## 五、实战案例一：Python FastMCP 构建情感分析服务器 FastMCP 把 Sampling 封装成了优雅的 `ctx.sample(...)`，它会让出工具协程直到宿主回复。下面是一台**零密钥情感分析服务器**——它自己没有模型，却能给文本打情感标签： ```python # server_sentiment.py from fastmcp import FastMCP, Context from pydantic import BaseModel mcp = FastMCP("SentimentBySampling") class SentimentResult(BaseModel): text: str sentiment: str # positive / negative / neutral @mcp.tool async def analyze_sentiment(text: str, ctx: Context) -> SentimentResult: """对文本做情感分类（推理由宿主 LLM 完成，本服务器不持有任何密钥）。""" prompt = ( "把下面的文本分类为 'positive'、'negative' 或 'neutral'，" "只返回一个单词，不要标点。\n\n" f"Text: {text}" ) # 关键：把推理外包给宿主 resp = await ctx.sample( messages=prompt, temperature=0.0, # 分类任务要稳定 max_tokens=8, ) label = (resp.text or "").strip().lower() # 兜底：宿主可能返回越界值 if label not in {"positive", "negative", "neutral"}: label = "neutral" return SentimentResult(text=text, sentiment=label) if __name__ == "__main__": mcp.run() # 默认 stdio 传输 ``` **预期结果**：当支持 sampling 的宿主调用 `analyze_sentiment`，服务器会反向发起一次采样，宿主 LLM 返回一个词，服务器校验后返回结构化结果。整个过程中服务器进程**没有任何 API Key 环境变量**。 **常见问题**：若报 `ctx.sample() throws because client doesn't implement sampling interface`，说明宿主未实现采样处理器——这正是下一节要解决的事。 ## 六、实战案例二：TypeScript 构建日志根因分析服务器 对于需要返回结构化 JSON 的场景，TypeScript SDK 的 `server.createMessage(...)` 更显式。下面是一台**日志根因分析服务器**，它用采样把一段混乱的错误日志压成结构化结论： ```typescript // server_loganalyzer.ts import { Server } from "@modelcontextprotocol/sdk/server/index.js"; import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"; import { CallToolRequestSchema } from "@modelcontextprotocol/sdk/types.js"; const server = new Server( { name: "log-analyzer", version: "1.0.0" }, { capabilities: { tools: {}, sampling: {} } } // 声明将使用 sampling ); server.setRequestHandler(CallToolRequestSchema, async (request) => { if (request.params.name !== "analyze-log") { throw new Error(`Unknown tool: ${request.params.name}`); } const rawLog = request.params.arguments?.log as string; if (!rawLog) throw new Error("Missing required argument: log"); let samplingResult; try { samplingResult = await server.createMessage({ messages: [{ role: "user", content: { type: "text", text: `分析日志并返回 JSON：severity(critical|warning|info)、root_cause(一句话)、action_required(bool)。\n` + `只返回合法 JSON，不要 markdown。\n日志：${rawLog}` } }], // 分类任务：要快、要便宜、不需要深度推理 modelPreferences: { speedPriority: 0.9, costPriority: 0.8, intelligencePriority: 0.3 }, systemPrompt: "你是日志分析 API，只返回合法 JSON。", maxTokens: 300, }); } catch (err: any) { // 降级：采样失败时不要让整个工具崩掉 return { content: [{ type: "text", text: `Analysis unavailable: ${err.message}` }], isError: true }; } const content: any = samplingResult.content; if (content.type !== "text") throw new Error("Expected text response"); let parsed: any; try { parsed = JSON.parse(content.text); } catch { throw new Error(`Sampling returned invalid JSON: ${content.text}`); } return { content: [{ type: "text", text: JSON.stringify({ analysis: parsed, model_used: samplingResult.model, // 记录实际使用的模型，便于审计 raw_log_length: rawLog.length, }, null, 2) }] }; }); async function main() { await server.connect(new StdioServerTransport()); console.error("log-analyzer MCP server running"); } main().catch(console.error); ``` **关键设计点**： 1. `speedPriority: 0.9` 把分类这种"不需要深度推理"的任务推向更快的模型，省钱省时。 2. `model_used` 回写到结果里——采样是黑盒，记录"到底是哪个模型分析的"对事后追溯至关重要。 3. `try/catch` 包裹 `createMessage`——宿主可能超时、拒绝、返回非法 JSON，必须降级而非崩溃。 ## 七、被忽视的另一半：实现 Host Client 的 sampling handler 这是绝大多数教程回避的部分。**Sampling 是客户端能力，如果你自建 Host（而不是用现成的桌面客户端），就必须自己实现 `CreateMessageRequestSchema` 处理器，否则任何采样调用都会抛错。** 下面是一个最小可用 Host Client： ```typescript // host_client.ts import { Client } from "@modelcontextprotocol/sdk/client/index.js"; import { CreateMessageRequestSchema } from "@modelcontextprotocol/sdk/types.js"; import Anthropic from "@anthropic-ai/sdk"; const anthropic = new Anthropic(); let samplingCallCount = 0; const MAX_SAMPLING_CALLS = 5; // 防止服务器陷入采样死循环 // 必须在 client.connect() 之前注册 client.setRequestHandler(CreateMessageRequestSchema, async (request) => { const { messages, systemPrompt, maxTokens, modelPreferences } = request.params; // 采样深度护栏：防止恶意/有 bug 的服务器无限采样 samplingCallCount++; if (samplingCallCount > MAX_SAMPLING_CALLS) { throw new Error("Sampling depth limit exceeded"); } // 模型路由：把 hint 当建议，自己拍板 const hint = modelPreferences?.hints?.[0]?.name; const model = resolveModel(hint); const response = await anthropic.messages.create({ model, max_tokens: maxTokens ?? 1024, system: systemPrompt, messages: messages.map((m: any) => ({ role: m.role, content: typeof m.content === "string" ? m.content : m.content.text, })), }); const textBlock: any = response.content.find((b: any) => b.type === "text"); return { role: "assistant", content: { type: "text", text: textBlock?.text ?? "" }, model: response.model, stopReason: response.stop_reason ?? "endTurn", }; }); function resolveModel(hint?: string): string { const map: Record = { "claude-sonnet": "claude-sonnet-4-6", "claude-haiku": "claude-haiku-4-5-20251001", "claude-opus": "claude-opus-4-8", }; return hint && map[hint] ? map[hint] : "claude-haiku-4-5-20251001"; } ``` **三个要点**： 1. **采样深度护栏**（`MAX_SAMPLING_CALLS`）：一台服务器理论上可以"采样→解读→再采样"无限循环。markaicode 建议每个工具调用上限设为 5 次，这是宿主必须自建的第一道防线。 2. **`resolveModel` 是你的主权**：hint 只是建议，真正用哪个模型由宿主决定，可按成本/策略路由。 3. **必须 `connect()` 前注册**：处理器缺失会导致连接后采样直接失败。 ## 八、生产红线：延迟、深度、传输与可观测性 把 Sampling 推向生产，有四条红线必须守住： **① 延迟会叠加。** 每次采样都是一次完整的 LLM 往返。服务器若在一次工具调用里串行采样三次，就是 3 倍单次推理延迟。设计原则：**尽量单发**——用一次结构化输出提示词把需要的字段一次性全要回来，而不是来回三轮对话。 **② 采样循环是真实风险。** 见第七节的深度护栏。同时建议服务器侧也设自检：超过 N 次采样就主动放弃并降级。 **③ 传输层有坑。** gyliu 指出，Sampling 在 **stdio 与 SSE** 上工作良好，但**某些无状态 HTTP 部署会出现 hanging（挂起）**——因为采样需要双向通信，而无状态 HTTP 不维护会话。若目标环境是 HTTP，必须用 Streamable HTTP 的有状态会话模式，并验证不挂起。 **④ 必须可观测。** 采样是分布式调用链里隐藏的一跳，强烈建议接入 OpenTelemetry 做分布式追踪，结构化记录请求/响应耗时、token 消耗、实际模型与成本，否则线上排障时你根本不知道某次工具调用慢在哪一跳。 ```mermaid flowchart TD U([用户触发工具]) --> S1[服务器：确定性逻辑] S1 -->|需要语言理解?| D{决策} D -->|否| S2[纯代码完成] D -->|是| SM[sampling/createMessage] SM --> H[宿主：人审 + 模型路由] H --> L[LLM 推理] L --> H2[宿主：人审响应] H2 -->|深度未超限| S3[服务器：解析 + 继续] H2 -->|超过 MAX 深度| ERR[降级/报错] S3 --> R([返回最终工具结果]) S2 --> R ``` 这张总览图把"是否采样—人审—深度护栏—降级"整条链路串了起来，可作为生产实现的检查清单。 ## 九、在 Claude Code 中接入与验证 Sampling 的成败完全取决于 Host 是否声明 `sampling` capability。**这一点必须以你当前使用的 Claude Code 版本为准核实**——MCP 客户端对 `sampling/createMessage` 的接线一直在演进，不同版本支持度不同。验证路径有两条： 1. **直接在 Claude Code 中接入**：把服务器加入 `.mcp.json`（或 `claude mcp add`），声明 `sampling` 能力后用一个明确需要推理的提示触发工具。若工具返回 `Analysis unavailable: ...` 或报"宿主不支持采样"，说明当前版本的 Claude Code 尚未完整接线该能力。 2. **自建 Host Client 验证（最稳）**：按第七节实现自己的 `CreateMessageRequestSchema` 处理器，用 Anthropic SDK 作为模型后端。这是官方规范保证可行的路径，也是验证你 sampling 服务器逻辑正确的金标准。 > 实践建议：开发期一律用自建 Host Client 验证（行为可控、可打点），集成期再切到 Claude Code 等产品级 Host 做兼容性确认。 ## 延伸阅读 - **MCP 官方规范 · Sampling**（2025-06-18）—— 协议契约的权威来源，含 `createMessage` 请求/响应、modelPreferences、多模态内容与安全条款 - **MCP 官方规范 · Lifecycle** —— `capabilities` 握手与 `sampling` 能力声明的初始化时序 - **FastMCP 文档 · Context.sample** —— Python 侧 `ctx.sample(...)` 的协程式封装与参数约定 - **@modelcontextprotocol/typescript-sdk** —— TypeScript SDK，含 `server.createMessage` 与 `CreateMessageRequestSchema` ## 训练营适用性评估 - **适合类型**：面向"已会建 MCP 工具、想让工具变智能"的进阶营，或"MCP 深度定制 / 自建 Host"的架构师营。不适合零基础入门营（需要先掌握三原语）。 - **预计教学时长**：6–8 学时。其中协议与决策矩阵约 1.5 学时，Python/TypeScript 双服务器实战约 3 学时，Host Client 采样处理器 + 深度护栏约 2 学时，生产红线与排障（HTTP hanging、可观测性）约 1–1.5 学时。 - **实训交付物**：一台零密钥情感分析服务器 + 一台日志根因分析服务器 + 一个带深度护栏的自建 Host Client，三者跑通端到端采样链路即合格。 ## 参考来源 1. [Sampling — Model Context Protocol 官方规范（2025-06-18）](https://modelcontextprotocol.io/specification/2025-06-18/client/sampling) — `createMessage` 协议契约、capabilities 声明、modelPreferences 偏好系统、多模态内容与人在环路安全条款的权威来源 2. [MCP Sampling: How LLMs Call Other LLMs via Servers — Markaicode](https://markaicode.com/mcp-sampling-llms-calling-llms/) — TypeScript 服务器与 Host Client 的可运行实现、采样深度护栏、何时用/何时不用的决策矩阵、错误降级模式 3. [MCP Sampling: Architecture, Workflow, and Practical Guide — gyliu513](https://gyliu513.github.io/jekyll/update/2025/08/21/mcp-sampling.html) — Python FastMCP `ctx.sample` 用法、架构与工作流图、传输层（stdio/SSE/HTTP hanging）生产注意事项与 OpenTelemetry 可观测性建议 4. [modelcontextprotocol/typescript-sdk — GitHub](https://github.com/modelcontextprotocol/typescript-sdk) — 官方 TypeScript SDK，`server.createMessage` 与 `CreateMessageRequestSchema` 处理器注册的实现依据 5. [Implement Sampling in MCP — RapidDev Tutorial](https://www.rapidevelopers.com/mcp-tutorial/how-to-implement-sampling-in-mcp) — `createMessage` 流程、模型偏好与 Host 侧处理的补充教程视角 > [!tip] 51学通信知识星球 > **星主：爱卫生**（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。 > **星球定位**：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。 > **当前规模**：5200 篇主题 / 990 个回答，198 元/年。 > **加入福利**：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。 > **AI 加持**：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。