<!-- title: 让MCP工具反向提问：Elicitation与Sampling实战 -->

## 概述

绝大多数 MCP 教程都停留在"客户端调用工具、服务器被动响应"的单向模型。但 MCP 规范（2025-06-18）真正强大的地方在于两个**服务器主动发起**的原语：**Elicitation**（向用户征求结构化输入）和 **Sampling**（向客户端借用 LLM 推理能力）。它们把工具从"死板的执行器"升级成"会提问、会思考的协作者"。本教程用 FastMCP 从零构建一个同时具备这两种能力的 MCP 服务器，讲解能力协商、双原语组合、客户端回退策略，并在 Claude Code 真实环境与自建测试客户端中验证。适合已写过基础 MCP 工具、想进阶到交互式/智能体化工具链的开发者。

## 前置条件

- Python 3.10+，已安装 `uv`（`curl -LsSf https://astral.sh/uv/install.sh | sh`）
- 理解 MCP 三原语（tools / resources / prompts）的基本概念
- 一台能运行 Claude Code 的机器，用于真实客户端联调
- 可选：一个 OpenAI 或 Anthropic API Key（用于 Sampling 回退场景）

---

## 第一步：认知翻转——什么是"服务器主动发起"

在经典调用链里，控制权始终在客户端（Claude Code）手中：

```
Client → tools/call → Server（执行）→ 返回结果 → Client
```

服务器只能被动应答。但真实业务里，工具经常需要在**执行中途**做两件事：

1. **缺信息**：删除数据库前想确认"确定要删 `production` 库吗？"——这是 **Elicitation**。
2. **要推理**：拿到一段日志后，想直接让 LLM 判断是否异常，而不是把原始日志一股脑塞回给客户端重新编排——这是 **Sampling**。

两者共同点是：**请求由服务器发出，方向是 Server → Client**。客户端可以选择响应、拒绝或回退。这彻底改变了工具的设计哲学——工具不再是"输入→输出"的纯函数，而是一个可与人、与 LLM 多轮交互的小型智能体。

| 维度 | Elicitation（征求） | Sampling（采样） |
|------|---------------------|------------------|
| 请求对象 | 终端用户 | 客户端背后的 LLM |
| JSON-RPC 方法 | `elicitation/create` | `sampling/createMessage` |
| 返回内容 | 结构化用户输入（accept/decline/cancel） | LLM 生成的文本/图像/音频 |
| 典型场景 | 高危确认、缺参补全、多步收集 | 内容总结、风险判定、代码生成 |
| 安全要求 | 用户可拒绝 | 必须有人类在环（human-in-the-loop） |

记住这个区别：**Elicitation 问的是"人"，Sampling 问的是"模型"**。

## 第二步：环境准备

创建项目骨架，安装 FastMCP（它是官方 MCP Python SDK 之上的高层封装，对 Elicitation/Sampling 有最完善支持）：

```bash
uv init mcp-proactive && cd mcp-proactive
uv add "fastmcp>=2.10"           # 2.10+ 才有 ctx.elicit()
uv add "fastmcp[openai]"         # 可选：Sampling 回退到 OpenAI
```

新建 `server.py`，先搭一个能跑的空壳：

```python
from fastmcp import FastMCP

mcp = FastMCP("proactive-server")

if __name__ == "__main__":
    mcp.run(transport="stdio")
```

启动验证：

```bash
uv run server.py
# 预期：进程挂起等待 stdio 输入，无报错即正常
```

**常见问题**：若报 `cannot import name 'elicit'`，说明版本低于 2.10，执行 `uv add "fastmcp>=2.10"` 升级。

## 第三步：Elicitation 实战——让工具开口问人

核心 API 是 `ctx.elicit(message, response_type=...)`。`response_type` 决定了弹给用户的是文本框、数字框、布尔勾选还是下拉枚举。

### 3.1 标量类型：最简问答

```python
from fastmcp import FastMCP, Context

mcp = FastMCP("proactive-server")

@mcp.tool
async def get_user_name(ctx: Context) -> str:
    """询问用户姓名。"""
    result = await ctx.elicit("你叫什么名字？", response_type=str)
    if result.action == "accept":
        return f"你好，{result.data}！"
    return "用户未提供姓名"
```

FastMCP 会自动把 `str` 包装成 MCP 规范要求的对象 schema（字段名 `value`），客户端按 schema 渲染输入框。返回的 `result.action` 有三种：`accept`（用户填了）、`decline`（用户拒绝）、`cancel`（取消整个操作）。

### 3.2 枚举选择：做选择题

```python
from typing import Literal

@mcp.tool
async def pick_env(ctx: Context) -> str:
    """让用户选择部署环境。"""
    result = await ctx.elicit(
        "部署到哪个环境？",
        response_type=Literal["staging", "production"],
        response_title="目标环境",          # 自定义输入框标签
        response_description="选择要发布的环境",
    )
    if result.action != "accept":
        return "已取消部署"
    return f"开始部署到 {result.data}"
```

### 3.3 结构化模型：一次收集多个字段

```python
from pydantic import BaseModel, Field

class DeployConfig(BaseModel):
    service: str = Field(description="服务名")
    replicas: int = Field(ge=1, le=20, description="副本数")
    dry_run: bool = Field(default=False, description="是否空跑")

@mcp.tool
async def deploy(ctx: Context) -> str:
    """通过交互式表单收集部署参数。"""
    result = await ctx.elicit("请填写部署参数：", response_type=DeployConfig)
    if result.action == "accept":
        cfg = result.data
        return f"部署 {cfg.service} ×{cfg.replicas}（dry_run={cfg.dry_run}）"
    if result.action == "decline":
        return "用户拒绝填写"
    return "操作取消"
```

**预期结果**：客户端把 `DeployConfig` 渲染成一张带校验（`ge=1, le=20`）的表单，用户填完点确认，工具拿到一个强类型的 `DeployConfig` 实例。

### 3.4 多轮收集：渐进式披露

工具可以连续多次 `elicit`，像向导一样分步收集：

```python
@mcp.tool
async def plan_meeting(ctx: Context) -> str:
    """分步收集会议信息。"""
    title = await ctx.elicit("会议主题？", response_type=str)
    if title.action != "accept":
        return "已取消"

    duration = await ctx.elicit("持续多少分钟？", response_type=int)
    if duration.action != "accept":
        return "已取消"

    urgent = await ctx.elicit("紧急吗？", response_type=Literal["是", "否"])
    if urgent.action != "accept":
        return "已取消"

    return (f"会议《{title.data}》{duration.data} 分钟"
            f"（紧急：{urgent.data == '是'}）")
```

**常见问题**：如果客户端不支持 Elicitation，`ctx.elicit()` 会直接抛异常。因此**必须**用 `result.action` 判断，并对不支持的情况做 try/except 兜底（见第六步）。

## 第四步：Sampling 实战——让工具借模型来思考

核心 API 是 `ctx.sample(...)`，返回 `SamplingResult`，文本在 `.text`。默认请求路由到客户端的 LLM（如 Claude Code 背后的 Claude）。

### 4.1 最简采样：一句话总结

```python
@mcp.tool
async def summarize(content: str, ctx: Context) -> str:
    """对给定内容生成摘要。"""
    result = await ctx.sample(f"请用一句话总结：\n\n{content}")
    return result.text or "(无内容)"
```

工具自己不再需要持有任何 API Key——模型选择、计费、权限都在客户端侧。

### 4.2 带系统提示与参数

```python
@mcp.tool
async def generate_code(concept: str, ctx: Context) -> str:
    """为某概念生成 Python 示例。"""
    result = await ctx.sample(
        messages=f"写一个演示「{concept}」的 Python 例子。",
        system_prompt="你是资深 Python 工程师，只输出简洁可运行的代码。",
        temperature=0.7,
        max_tokens=300,
    )
    return f"```python\n{result.text}\n```"
```

### 4.3 模型偏好：给客户端一个提示

`model_preferences` 只是**提示**而非强制，客户端可忽略：

```python
@mcp.tool
async def risk_analysis(data: str, ctx: Context) -> str:
    """用推理型模型做风险分析。"""
    result = await ctx.sample(
        messages=f"分析以下数据的风险点：\n\n{data}",
        model_preferences=["claude-opus-4-5", "claude-sonnet-4-6"],
        temperature=0.2,
    )
    return result.text or "(无分析)"
```

### 4.4 结构化输出：让 LLM 吐结构化数据

```python
from pydantic import BaseModel

class Verdict(BaseModel):
    is_anomaly: bool
    confidence: float = Field(ge=0, le=1)
    reason: str

@mcp.tool
async def classify_log(log_line: str, ctx: Context) -> str:
    """判断日志是否异常，返回结构化判定。"""
    result: Verdict = await ctx.sample(
        messages=f"判断这条日志是否异常：\n{log_line}",
        result_type=Verdict,
    )
    return (f"异常={result.is_anomaly} "
            f"(置信度 {result.confidence:.0%})：{result.reason}")
```

FastMCP 会自动注入一个 `final_response` 工具让 LLM 调用以产出结构化结果，校验失败会自动重试。

## 第五步：组合拳——一个会问又会想的工具

真正的杀手锏是**两原语组合**。下面这个工具先借 LLM 评估风险，再据评估结果决定是否向用户确认：

```python
@mcp.tool
async def safe_delete(path: str, ctx: Context) -> str:
    """删除高危路径：先用模型评估，再向用户确认。"""
    # 第 1 步：借用客户端 LLM 做风险评估
    risk = await ctx.sample(
        messages=(
            f"判断删除路径「{path}」的风险等级，"
            f"只回答 low / medium / high。"
        ),
        temperature=0.0,
        max_tokens=10,
    )
    level = (risk.text or "high").strip().lower()

    # 第 2 步：低风险直接删；中高风险必须人工确认
    if level == "low":
        _do_delete(path)
        return f"风险低，已删除 {path}"

    result = await ctx.elicit(
        f"⚠️ 风险等级 {level}，确认删除「{path}」吗？",
        response_type=bool,
        response_title="确认删除",
        response_description="此操作不可恢复",
    )

    if result.action == "accept" and result.data:
        _do_delete(path)
        return f"用户确认，已删除 {path}"
    return "用户拒绝，未删除"


def _do_delete(path: str) -> None:
    # 实际删除逻辑，此处仅占位
    print(f"[delete] {path}")
```

这是一个微型智能体：**Sampling 负责"思考"，Elicitation 负责"决策"，工具本体负责"执行"**。三者各司其职，比把所有逻辑塞进提示词干净得多。

## 第六步：能力协商与客户端测试

Elicitation 和 Sampling 都是**可选能力**，客户端在初始化时声明：

```json
{ "capabilities": { "sampling": {}, "elicitation": {} } }
```

**关键现实**：并非所有客户端都支持。截至 2026 年中，Claude Code 对服务器主动发起原语的支持仍在演进（社区已有 `claude-code#1785` 等能力请求）。因此工业级做法是**双重保险**：

### 6.1 Sampling 回退处理器

当客户端不支持 Sampling 时，让服务器自己调用 LLM，而非崩溃：

```python
from fastmcp.client.sampling.handlers.openai import OpenAISamplingHandler
import os

mcp = FastMCP(
    "proactive-server",
    sampling_handler=OpenAISamplingHandler(
        default_model="gpt-4o-mini",
        api_key=os.environ["OPENAI_API_KEY"],
    ),
    sampling_handler_behavior="fallback",   # 仅在客户端不支持时回退
)
```

`sampling_handler_behavior` 两种取值：`"fallback"`（客户端支持就用客户端，否则用自带）和 `"always"`（永远用自带，绕开客户端，适合成本/合规管控）。

### 6.2 Elicitation 的兜底

Elicitation 没有内置回退处理器，需手动 try/except，并退化为"用工具入参"或"拒绝执行"：

```python
async def confirm_or_default(ctx, msg) -> bool:
    try:
        r = await ctx.elicit(msg, response_type=bool)
        return r.action == "accept" and bool(r.data)
    except Exception:
        # 客户端不支持 elicitation，回退为"需用户显式传参确认"
        return False
```

### 6.3 用 FastMCP 客户端写测试桩

不要依赖真实客户端做单元测试。写一个自己控制能力的客户端：

```python
# test_client.py
import asyncio
from fastmcp import Client
from fastmcp.client.sampling import RequestContext, SamplingMessage, SamplingParams

async def fake_sampling(messages, params: SamplingParams, ctx: RequestContext):
    return "high"   # 模拟 LLM 永远判定高风险

async def fake_elicitation(message, schema, ctx):
    return {"action": "accept", "data": True}  # 模拟用户点确认

async def main():
    async with Client(
        "server.py",
        sampling_handler=fake_sampling,
    ) as client:
        out = await client.call_tool("safe_delete", {"path": "/etc"})
        print(out)

asyncio.run(main())
```

```bash
uv run test_client.py
# 预期：高风险 → 触发 elicitation（被桩函数接受）→ 打印"用户确认，已删除"
```

### 6.4 接入 Claude Code 联调

把服务器登记到 Claude Code 后，先确认它被加载：

```bash
claude mcp add proactive -- uv run --directory $(pwd) server.py
claude mcp list            # 应能看到 proactive
```

在交互会话里让 Claude 调用工具，观察 Claude Code 是否触发征求式输入框或借用自身模型。若你的版本暂不支持，`"always"` 回退 + 手动 elicitation 兜底能保证功能不中断。

## 第七步：常见问题与陷阱

- **工具卡死/超时**：`ctx.elicit()` 和 `ctx.sample()` 都是阻塞等待，务必设超时与 `action` 短路返回，避免长时间挂起。
- **schema 太复杂客户端拒绝渲染**：Elicitation 的 `requestedSchema` 仅支持对象 + 基本类型（string/number/integer/boolean/enum），不支持 `$ref` 和深层嵌套递归。复杂结构请拆成多轮 `elicit`。
- **Sampling 静默不返回**：多半是客户端未声明 `sampling` 能力。用 `"always"` 回退处理器可立即定位。
- **安全红线**：Sampling 规范要求**始终有人类在环**可拒绝；任何破坏性操作必须经 Elicitation 二次确认，不要仅凭 LLM 的 risk 判定就执行。
- **隐私**：`ctx.sample` 会把内容发往客户端模型供应商，敏感数据慎用，或用本地模型回退。

---

## 完整实战项目：风险操作审批器

把上述能力组装成一个"破坏性操作审批流水线"服务器 `server.py`：

```python
import os
from typing import Literal
from pydantic import BaseModel, Field
from fastmcp import FastMCP, Context
from fastmcp.client.sampling.handlers.openai import OpenAISamplingHandler

mcp = FastMCP(
    "risk-approver",
    sampling_handler=OpenAISamplingHandler(
        default_model="gpt-4o-mini",
        api_key=os.environ["OPENAI_API_KEY"],
    ),
    sampling_handler_behavior="fallback",
)

class RiskReport(BaseModel):
    level: Literal["low", "medium", "high"]
    reason: str

@mcp.tool
async def approve_and_delete(path: str, ctx: Context) -> str:
    """评估风险 → 必要时人工确认 → 执行删除。"""
    # 1) LLM 风险评估（结构化）
    report: RiskReport = await ctx.sample(
        messages=f"评估删除「{path}」的风险。系统级路径/含通配/递归视为 high。",
        result_type=RiskReport,
        temperature=0.0,
    )

    # 2) 低风险直删
    if report.level == "low":
        return f"[已删除] {path}（{report.reason}）"

    # 3) 中高风险征求确认
    try:
        r = await ctx.elicit(
            f"风险={report.level}，原因：{report.reason}\n确认删除「{path}」？",
            response_type=bool,
            response_title="破坏性操作确认",
        )
        confirmed = r.action == "accept" and bool(r.data)
    except Exception:
        confirmed = False  # 不支持 elicitation → 默认拒绝

    if confirmed:
        return f"[已删除] {path}（用户已确认高风险）"
    return f"[已拒绝] {path}（风险={report.level}，未获确认）"

if __name__ == "__main__":
    mcp.run(transport="stdio")
```

运行与测试：

```bash
export OPENAI_API_KEY=sk-...
claude mcp add approver -- uv run --directory $(pwd) server.py
# 在 Claude Code 中：
#   "帮我删除 ./tmp/cache 目录"  → 低风险直删
#   "帮我删除 /var/lib/mysql"    → 高风险，弹出确认框
```

这个服务器体现了"思考—决策—执行"三段式智能体结构，且对不支持双原语的客户端也安全降级。

## 工作流程图

```mermaid
flowchart TD
    A[Client 调用 approve_and_delete] --> B{服务器: Sampling 风险评估}
    B -->|ctx.sample result_type=RiskReport| C{客户端支持 sampling?}
    C -->|是| D[借用客户端 LLM]
    C -->|否| E[回退 OpenAI 处理器]
    D --> F[得到 RiskReport]
    E --> F
    F --> G{level == low?}
    G -->|是| H[直接删除并返回]
    G -->|否| I{客户端支持 elicitation?}
    I -->|是| J[ctx.elicit 弹确认框]
    J -->|accept & true| K[删除]
    J -->|decline/cancel| L[拒绝]
    I -->|否| M[兜底: 默认拒绝]
    K --> N[返回结果]
    L --> N
    M --> N
    H --> N
```

## 延伸阅读

- MCP 规范——Sampling（2025-06-18）：`https://modelcontextprotocol.io/specification/2025-06-18/client/sampling`
- MCP 规范——Elicitation（2025-06-18）：`https://modelcontextprotocol.io/specification/2025-06-18/client/elicitation`
- FastMCP 文档——User Elicitation：`https://gofastmcp.com/servers/elicitation`
- FastMCP 文档——Sampling：`https://gofastmcp.com/servers/sampling`
- Claude Code MCP 集成文档：`https://code.claude.com/docs/en/mcp`

## 训练营适用性评估

- **适用类型**：中高级 LLM 应用开发训练营、智能体（Agent）工程训练营、MCP 工具链专项训练营。不适合零基础，需先完成 MCP 三原语入门。
- **教学时长**：建议 4–5 课时。第 1 课时讲双原语概念与规范；第 2–3 课时分别实操 Elicitation 与 Sampling；第 4 课时做组合案例与能力协商/回退；第 5 课时联调 Claude Code 与测试桩、做安全讨论。
- **配套练习**：要求学员独立实现"危险 SQL 执行审批器"（Sampling 判定 SQL 是否含 DROP/TRUNCATE，Elicitation 确认），并提交包含 `sampling_handler` 回退与 elicitation 兜底的完整代码。

## 参考来源

1. [Sampling — Model Context Protocol Specification (2025-06-18)](https://modelcontextprotocol.io/specification/2025-06-18/client/sampling) — 官方 Sampling 原语规范，定义 `sampling/createMessage`、能力声明与人机在环要求
2. [User Elicitation — FastMCP](https://gofastmcp.com/servers/elicitation) — FastMCP 的 `ctx.elicit()` 用法、`response_type` 与 accept/decline/cancel 结果模型
3. [Sampling — FastMCP](https://gofastmcp.com/servers/sampling) — FastMCP 的 `ctx.sample()`、`result_type` 结构化输出与回退处理器（OpenAI/Anthropic）
4. [Connect Claude Code to tools via MCP](https://code.claude.com/docs/en/mcp) — Claude Code 接入 MCP 服务器的官方配置与传输方式说明
5. [Complete Guide to MCP in 2026 — Architecture & Implementation](https://dev.to/x4nent/complete-guide-to-mcp-model-context-protocol-in-2026-architecture-implementation-and-4a11) — 2026 年 MCP 架构、传输与能力协商的综合综述

> [!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 等场景全覆盖。

![51学通信知识星球海报-扫码加入](https://wiki.500nian.com/images/zhixingqiu_qrcode.png)