# MCP 反向交互三原语开发实战：Sampling、Elicitation、Roots ## 概述 绝大多数 MCP 教程只讲 Tools——服务器被动响应客户端调用。但真正让 MCP 服务器"活起来"的，是三个反向交互原语：**Sampling**（服务器反向调用 LLM）、**Elicitation**（服务器向用户征询结构化输入）、**Roots**（客户端向服务器声明可操作边界）。本教程面向已能写出基础 MCP Tool 的开发者，带你从协议层到 SDK 层吃透这三个"客户端特性"（Client Features），并用 FastMCP 构建一个三者联动的智能代码重构助手。 ## 前置条件 - 已掌握 MCP Tools 的基本开发（`@mcp.tool` 装饰器、JSON Schema 参数） - Python 3.11+，熟悉 `async/await` - 已安装 `pip install fastmcp`（本教程基于 FastMCP 2.x） - 一个支持这三个特性的客户端用于联调：Claude Code（命令行运行 `claude mcp add`）或官方 MCP Inspector（`npx @modelcontextprotocol/inspector`） - 理解 JSON-RPC 2.0 基本报文结构 ## 一、为什么需要"反向交互"：理解 MCP 的方向性 MCP 的报文分两个方向。**客户端 → 服务器**的方向有三个 Server Features（Primitives）：Prompts、Resources、Tools，这是大家最熟悉的——Host/客户端主动向服务器要资源、调工具。**服务器 → 客户端**的反向方向，对应三个 Client Features：Sampling、Elicitation、Roots。这三个特性的共同点是：**由服务器发起请求，客户端决定是否执行**。 这种"反向 + 客户端把关"的设计是 MCP 安全模型的核心。服务器代码可能运行在第三方环境，如果它直接持有 API Key 调用模型、直接读写任意磁盘，就是一个巨大的风险面。MCP 把"模型访问权""用户交互权""文件边界"统统收归客户端，服务器只能"请求"，客户端可以"审批、改写、拒绝"。 理解这一点，就理解了为什么这三个原语都带着 `capabilities` 声明、都强调 `user approval`、都要求 `Security Considerations`。它们不是可有可无的锦上添花，而是构建可信智能体工作流的地基。 ```mermaid flowchart LR subgraph C[客户端 / Host / LLM] LL[大模型] UI[用户交互层] FS[文件系统边界] end subgraph S[MCP 服务器] T[Tools] SP[Sampling 引擎] EL[Elicitation 引擎] RT[Roots 读取] end C -->|tools/call 调用| T T -->|sampling/createMessage 反向请求模型| LL T -->|elicitation/create 反向征询用户| UI RT -->|roots/list 拉取边界| FS LL -.审批.-> C UI -.审批.-> C ``` ## 二、Roots：让服务器知道"自己的地盘在哪" ### 协议层 Roots 是三者中最简单的，也是最该先实现的。客户端通过 `roots/list` 向服务器暴露一组 `file://` URI，告知服务器"你只能在这些目录里干活"。它解决的是一个朴素问题：服务器如何知道当前工作区在哪、有哪些项目？ 初始化握手时，客户端声明该能力： ```json { "capabilities": { "roots": { "listChanged": true } } } ``` 服务器通过 `roots/list` 拉取，客户端返回： ```json { "jsonrpc": "2.0", "id": 1, "result": { "roots": [ { "uri": "file:///home/user/projects/myproject", "name": "My Project" }, { "uri": "file:///home/user/repos/backend", "name": "Backend" } ] } } ``` 当用户在 IDE 里切换了工作区，客户端通过 `notifications/roots/list_changed` 通知服务器重新拉取。注意：当前规范要求 root 必须是 `file://` URI，未来可能扩展。 ### SDK 层（FastMCP） ```python from fastmcp import FastMCP, Context mcp = FastMCP("root-aware-server") @mcp.tool async def where_can_i_work(ctx: Context) -> str: """返回服务器当前可操作的所有根目录。""" roots = await ctx.list_roots() if not roots: return "客户端未暴露任何 roots，服务器不知道边界在哪。" lines = [f"- {r.name}: {r.uri}" for r in roots] return "可操作边界:\n" + "\n".join(lines) ``` ### 预期结果 在 Claude Code 中，把当前项目目录设为 root 后调用该工具，返回形如 `- myproject: file:///home/user/projects/myproject` 的列表。 ### 常见问题 **关键设计原则**：服务器永远不要硬编码路径，而应"先问 roots，再干活"。这是后续两个原语能安全运行的前提。 ## 三、Elicitation：服务器向用户"要数据" ### 协议层 Elicitation 是 **2025-06-18 规范版本才正式引入**的新原语（并在 2025-11-25 持续演进），设计目的是让服务器能在工具执行中途，动态向用户征询结构化输入——比如"请填写重构参数""请选择数据库环境"。 服务器发送 `elicitation/create` 请求，携带一段消息和一个 JSON Schema： ```json { "jsonrpc": "2.0", "id": 2, "method": "elicitation/create", "params": { "message": "请提供数据库连接信息", "requestedSchema": { "type": "object", "properties": { "host": { "type": "string", "title": "主机" }, "port": { "type": "integer", "minimum": 1, "maximum": 65535 }, "env": { "type": "string", "enum": ["dev","staging","prod"], "enumNames": ["开发","预发","生产"] } }, "required": ["host", "port"] } } } ``` 响应采用**三态动作模型**，这是 Elicitation 最值得称道的设计： - `accept`：用户明确提交，`content` 字段携带符合 schema 的数据 - `decline`：用户明确拒绝（点了"拒绝/否"），`content` 通常省略 - `cancel`：用户未做明确选择就关闭（点 ESC、点空白处），`content` 省略 ```json { "jsonrpc": "2.0", "id": 2, "result": { "action": "accept", "content": { "host": "db.local", "port": 5432, "env": "dev" } } } ``` ### Schema 的"有限子集"约束 这是新手最容易踩坑的地方。为了降低客户端实现成本，Elicitation 的 `requestedSchema` **被刻意限制为扁平对象 + 原始类型**： - 仅支持 `string` / `number`(`integer`) / `boolean` / `enum` - 字符串支持 `format`：`email`、`uri`、`date`、`date-time` - 数字支持 `minimum` / `maximum` - 枚举支持 `enum` + `enumNames` - **不支持**嵌套对象、对象数组、`oneOf`/`anyOf` 等高级 JSON Schema 特性 如果你需要复杂结构，必须把它"拍平"成多个原始字段。 ### 安全红线 规范明确写道：**Servers MUST NOT use elicitation to request sensitive information.**——服务器绝不能用 Elicitation 索要密码、私钥、Token 等敏感信息。这类数据应通过 OAuth、环境变量等受控通道获取。客户端应清晰展示"是哪个服务器在问"，并允许用户随时 decline/cancel。 ### SDK 层（FastMCP） FastMCP 把 schema 约束直接编码进 Pydantic 模型，自动转换为合法的扁平 schema： ```python from enum import Enum from pydantic import BaseModel, Field from fastmcp import FastMCP, Context mcp = FastMCP("elicitation-demo") class Env(str, Enum): dev = "dev" staging = "staging" prod = "prod" class DBConfig(BaseModel): host: str = Field(title="数据库主机", description="例如 db.internal") port: int = Field(ge=1, le=65535, default=5432) env: Env = Field(default=Env.dev, description="目标环境") @mcp.tool async def connect_db(ctx: Context) -> str: """连接数据库，但参数由用户现场填写。""" result = await ctx.elicit( message="为了连接数据库，请确认以下信息（不会存储）", response_type=DBConfig, ) if result.action != "accept": return f"用户已 {result.action}，连接已取消。" cfg: DBConfig = result.data # 注意：这里不要处理密码！敏感信息走 OAuth/环境变量 return f"将连接 {cfg.host}:{cfg.port}（{cfg.env} 环境）" ``` ### 预期结果 调用 `connect_db` 时，客户端弹出表单，用户填完点提交，工具返回连接信息字符串；用户点拒绝，工具返回"用户已 decline"。 ### 常见问题 | 现象 | 原因 | 解决 | |------|------|------| | Pydantic 模型含嵌套对象，客户端报 schema 错误 | 违反扁平约束 | 拆成多个扁平字段，或分多轮 elicit | | `result.data` 为 `None` | 用户非 accept | 永远先判断 `result.action == "accept"` 再取 `data` | | 想要密码但被规范禁止 | 误用原语 | 密码改用 MCP Authorization（OAuth）或环境变量注入 | | 客户端不弹表单 | 客户端未声明 `elicitation` capability | 升级到支持 Elicitation 的客户端；服务器侧 `try/except` 降级为参数默认值 | ## 四、Sampling：服务器反向"借用"LLM ### 协议层 Sampling 让服务器不持有任何模型 API Key，就能请求一次模型生成。客户端收到 `sampling/createMessage` 后，自己挑选模型、自己审批、自己付费，把结果回传给服务器。这是构建"递归智能体""服务器内部需要推理"场景的关键。 ```json { "jsonrpc": "2.0", "id": 1, "method": "sampling/createMessage", "params": { "messages": [ { "role": "user", "content": { "type": "text", "text": "这段代码有什么坏味道?" } } ], "modelPreferences": { "hints": [ { "name": "claude-sonnet" }, { "name": "claude" } ], "intelligencePriority": 0.8, "speedPriority": 0.3, "costPriority": 0.2 }, "systemPrompt": "你是资深代码审查员，回答精炼。", "maxTokens": 300 } } ``` 响应： ```json { "jsonrpc": "2.0", "id": 1, "result": { "role": "assistant", "content": { "type": "text", "text": "..." }, "model": "claude-sonnet-4-6", "stopReason": "endTurn" } } ``` ### 模型偏好：hints + 三维优先级 这是 Sampling 最精妙的设计。**服务器永远不知道客户端有哪些模型可用**，所以不能写死模型名。MCP 的解法是： - **hints**：模型名子串提示（如 `claude-sonnet`），按优先级排列，客户端可跨厂商映射（如把 sonnet 映射到 gemini-pro） - **三维归一化优先级**（0-1）： - `intelligencePriority`：越重视能力，值越大，倾向选更强模型 - `speedPriority`：越重视低延迟，值越大 - `costPriority`：越重视省钱，值越大 消息内容支持 `text`、`image`（base64）、`audio`（base64），因此可以做多模态采样。 ### 安全考量 客户端应对每次采样实现**用户审批**（或在配置里授权某服务器免审批）、实现**速率限制**、并对双方内容做校验。错误情形返回标准 JSON-RPC 错误，如 `User rejected sampling request`。 ### SDK 层（FastMCP） ```python from fastmcp import FastMCP, Context mcp = FastMCP("sampling-demo") @mcp.tool async def review_code(snippet: str, ctx: Context) -> str: """让宿主端的 LLM 审查一段代码。""" result = await ctx.sample( f"请用 3 条要点审查这段代码的潜在 bug:\n\n{snippet}", system_prompt="你是严谨的代码审查员，只说真问题。", ) # 当采样为纯文本时，FastMCP 直接返回 str；含结构时返回 Message 对象 return result if isinstance(result, str) else result.content.text ``` ### 预期结果 调用 `review_code` 时，工具把代码片段交给宿主端的 Claude（或其他被批准的模型），返回审查要点。注意：宿主可能弹出"服务器请求使用模型，是否允许？"的审批框。 ### 常见问题 | 现象 | 原因 | 解决 | |------|------|------| | 返回 `-1 User rejected` | 用户拒绝审批 | 属正常行为，服务器应给出友好提示而非崩溃 | | 模型选得太贵/太慢 | 优先级配置不当 | 优先级按场景调：批处理重 `costPriority`，实时交互重 `speedPriority` | | 客户端无 `sampling` capability | 客户端不支持 | 降级为本地规则逻辑，或提示用户换客户端 | | 死循环调用 | 服务器在采样回调里又触发工具调用 | 服务器侧设最大采样轮次上限，避免无限递归 | ## 五、三者联动：一个完整的智能重构助手 下面把三个原语串成一个真实可运行的项目。场景：用户让服务器"重构当前项目的某个目录"。服务器先用 **Roots** 确认地盘，用 **Elicitation** 向用户征询重构风格与安全参数，再用 **Sampling** 借用宿主 LLM 生成每文件的重构方案。 ```python # refactor_assistant/server.py from enum import Enum from pathlib import Path from urllib.parse import urlparse from pydantic import BaseModel, Field from fastmcp import FastMCP, Context mcp = FastMCP("refactor-assistant") class Style(str, Enum): concise = "concise" verbose = "verbose" safe = "safe" class RefactorParams(BaseModel): style: Style = Field(description="重构风格: concise 精简 / verbose 详尽 / safe 保守") dry_run: bool = Field(default=True, description="True=只出方案不写盘") max_files: int = Field(ge=1, le=50, default=10, description="单次最大处理文件数") def _root_paths(roots) -> set[Path]: out = set() for r in roots: p = urlparse(str(r.uri)).path out.add(Path(p).resolve()) return out @mcp.tool async def refactor_directory(rel_path: str, ctx: Context) -> str: """对项目内某目录进行安全重构。""" # 1. Roots：确认边界 try: roots = await ctx.list_roots() except Exception: return "错误: 客户端未提供 roots，无法确认可操作边界，已拒绝执行。" allowed = _root_paths(roots) if not allowed: return "错误: roots 为空，拒绝执行。" target = (max(allowed) / rel_path).resolve() # 简化:取第一个根拼接 # 防路径穿越:必须落在某个 root 内 if not any(target == a or a in target.parents for a in allowed): return f"错误: {rel_path} 不在任何 root 范围内，拒绝执行（防止路径穿越）。" py_files = sorted(target.rglob("*.py"))[:0] # 占位,下面按用户上限裁剪 # 2. Elicitation:征询参数 res = await ctx.elicit( message=f"即将分析 {rel_path}，请确认重构参数。敏感信息请勿填写。", response_type=RefactorParams, ) if res.action != "accept": return f"用户 {res.action}，重构已取消。" params: RefactorParams = res.data py_files = sorted(target.rglob("*.py"))[: params.max_files] if not py_files: return "目标目录下没有 .py 文件。" # 3. Sampling:借用宿主 LLM 逐文件出方案 plans = [] for f in py_files: code = f.read_text(encoding="utf-8", errors="ignore")[:6000] prompt = ( f"请按 {params.style.value} 风格，给出这段 Python 代码的重构要点(不超过4条):\n\n{code}" ) try: idea = await ctx.sample( prompt, system_prompt="你是 Python 重构专家，只给可操作建议。", ) except Exception as e: idea = f"[采样失败: {e}]" plans.append(f"### {f.name}\n{idea}") header = f"模式: {'dry-run(仅方案)' if params.dry_run else '写盘'} | 风格: {params.style.value} | 文件数: {len(py_files)}\n" return header + "\n\n".join(plans) if __name__ == "__main__": mcp.run() # stdio 传输 ``` 运行与联调： ```bash # 1. 启动服务器(调试期可单独跑) python refactor_assistant/server.py # 2. 在 Claude Code 中注册 claude mcp add refactor -- python /abs/path/to/refactor_assistant/server.py # 3. 在对话里触发 # 用户: 帮我重构 src/utils 目录 # Claude 会调用 refactor_directory -> 你填表单 -> 服务器采样出方案 ``` 注意三处安全设计的配合：Roots 提供"能碰哪里"的硬边界（防路径穿越），Elicitation 提供"用户授权 + 参数注入"的软控制（且不收敏感数据），Sampling 提供"借脑"能力但由客户端审批付费。三者缺一，要么不安全，要么不智能。 ```mermaid sequenceDiagram participant U as 用户 participant H as Host/LLM(Claude Code) participant S as 重构服务器 participant FS as 文件系统 U->>H: 重构 src/utils H->>S: tools/call refactor_directory("src/utils") S->>H: roots/list H->>S: [file://.../project] S->>FS: 校验目标在 root 内(防穿越) S->>H: elicitation/create (RefactorParams) H->>U: 弹出参数表单 U->>H: accept {style:safe, dry_run:true} H->>S: action=accept, content=... loop 每个文件 S->>H: sampling/createMessage (审查 prompt) H->>U: 审批采样?(可选) H->>S: assistant 回答 end S->>H: 汇总重构方案 H->>U: 展示结果 ``` ## 六、工程化建议 **降级策略必须显式**。三个原语都依赖客户端 capability，但并非所有客户端都支持（尤其 Elicitation 是新特性）。每个 `ctx.list_roots()`、`ctx.elicit()`、`ctx.sample()` 调用都应包在 `try/except` 里，并提供合理的本地降级：没有 roots 就拒接危险操作，没有 elicitation 就用工具参数默认值，没有 sampling 就回退到本地规则引擎。**静默失败**是 MCP 服务器最常见的 bug 来源。 **避免采样递归**。服务器在 Sampling 的结果处理逻辑里，不要再去触发会引发新采样的工具调用，否则容易形成"服务器↔客户端"的无限循环。给每条业务路径设一个最大采样次数。 **Capability 探测要提前**。可以在服务器启动后、首次工具调用前，用一个轻量 `inspect_capabilities` 工具读 `ctx.session` 暴露的客户端能力，把"哪些原语可用"返回给用户，避免运行到一半才发现不支持。 **版本兼容**。Elicitation 在 2025-06-18 引入、2025-11-25 继续演进；Roots 与 Sampling 相对稳定。面向多客户端发布时，把 Elicitation 视为"增强体验"而非"必需"，核心功能要能脱离它独立工作。 ## 七、训练营适用性评估 | 维度 | 评估 | |------|------| | 适合训练营 | 中高级 MCP / AI Agent 工程化训练营；不适合零基础入门营 | | 前置门槛 | 已掌握 MCP Tools 开发与 Python 异步 | | 教学时长 | 建议 1 个全天（6 学时）：协议层 1.5h + 三原语各 1h + 联动项目 1.5h | | 实操难度 | ★★★☆☆（Roots/Sampling 易，Elicitation 的 schema 约束与三态处理是难点）| | 评估方式 | 让学员各写一个"反向交互"工具，用 Inspector 跑通 accept/decline/cancel 三态 | ## 延伸阅读 - MCP 官方文档 Client Features 总览：`modelcontextprotocol.io/specification/2025-06-18/client/roots` - Microsoft `mcp-for-beginners` 高级主题仓库（含 Sampling 完整示例代码） - FastMCP 官方文档 Context 章节（`fastmcp.com`），了解 `ctx.sample/elicit/list_roots` 的最新签名 - MCP 2025-11-25 草案新增的 Tasks 特性，可视为下一个反向交互原语的研究方向 ## 参考来源 1. [Elicitation — Model Context Protocol 规范](https://modelcontextprotocol.io/specification/2025-06-18/client/elicitation) — 三态动作模型与扁平 schema 约束的权威定义 2. [Sampling — Model Context Protocol 规范](https://modelcontextprotocol.io/specification/2025-06-18/client/sampling) — modelPreferences 三维优先级与 hints 设计的原文 3. [Roots — Model Context Protocol 规范](https://modelcontextprotocol.io/specification/2025-06-18/client/roots) — file:// 边界声明与 list_changed 通知机制 4. [Microsoft mcp-for-beginners: Sampling](https://github.com/microsoft/mcp-for-beginners/blob/main/05-AdvancedTopics/mcp-sampling/README.md) — 微软官方教程中 sampling/createMessage 流程与示例 5. [Roots, Sampling, and Elicitation — Durgesh Rai](https://www.durgesh.dev/blog/advanced-features) — 三个高级特性的工程化视角对比与实践建议 本文由「51学通信」原创出品，站长爱卫生（微信：gprshome201101）。聚焦通信与 AI 工程化前沿，硬核技术，做深做透。 > [!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 等场景全覆盖。