首页 › tutorials › 让MCP工具反向提问：Elicitation与Sampling实战

概述

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

前置条件

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

在经典调用链里，控制权始终在客户端（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 有最完善支持）：

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

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

from fastmcp import FastMCP

mcp = FastMCP("proactive-server")

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

启动验证：

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 标量类型：最简问答

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 枚举选择：做选择题

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 结构化模型：一次收集多个字段

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，像向导一样分步收集：

@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 最简采样：一句话总结

@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 带系统提示与参数

@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 只是提示而非强制，客户端可忽略：

@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 吐结构化数据

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 评估风险，再据评估结果决定是否向用户确认：

@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 都是可选能力，客户端在初始化时声明：

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

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

6.1 Sampling 回退处理器

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

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，并退化为"用工具入参"或"拒绝执行"：

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 客户端写测试桩

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

# 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())

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

6.4 接入 Claude Code 联调

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

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

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

第七步：常见问题与陷阱

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

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

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")

运行与测试：

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

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

工作流程图

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

延伸阅读

训练营适用性评估

参考来源

  1. Sampling — Model Context Protocol Specification (2025-06-18) — 官方 Sampling 原语规范，定义 sampling/createMessage、能力声明与人机在环要求
  2. User Elicitation — FastMCP — FastMCP 的 ctx.elicit() 用法、response_type 与 accept/decline/cancel 结果模型
  3. Sampling — FastMCP — FastMCP 的 ctx.sample()、result_type 结构化输出与回退处理器（OpenAI/Anthropic）
  4. Connect Claude Code to tools via MCP — Claude Code 接入 MCP 服务器的官方配置与传输方式说明
  5. Complete Guide to MCP in 2026 — Architecture & Implementation — 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学通信知识星球海报-扫码加入

📎 本分类最新文章