用 Elicitation 原语造会问人的 MCP 工具

# 1. 建项目目录
mkdir safe-ops-mcp && cd safe-ops-mcp
uv init .
uv venv
uv add fastmcp           # 会同时装好 mcp 官方 SDK、pydantic、starlette 等

from fastmcp import FastMCP, Context
from fastmcp.server.elicitation import (
    AcceptedElicitation, DeclinedElicitation, CancelledElicitation,
)

mcp = FastMCP("Safe Ops Server")


@mcp.tool
async def greet(ctx: Context) -> str:
    """演示标量征询：问用户名字。"""
    result = await ctx.elicit("你叫什么名字？", response_type=str)

    # 用类型化的结果类做模式匹配，最稳
    match result:
        case AcceptedElicitation(data=name):
            return f"你好，{name}！"
        case DeclinedElicitation():
            return "你选择了不提供信息。"
        case CancelledElicitation():
            return "操作已取消。"

from dataclasses import dataclass

@dataclass
class ConfirmDelete:
    confirm: bool      # 是否确认删除（标量 ✓）
    reason: str        # 删除原因，留作审计（标量 ✓）
    # 注意：不能放嵌套对象/列表，规范不支持

import asyncio
from fastmcp import Client
from fastmcp.client.elicitation import ElicitResult
from fastmcp.client.transports import StdioTransport


async def elicitation_handler(message, response_type, params, ctx):
    print("\n🔔 服务器向你提问：")
    print(message)

    # 受限选项：列出序号让用户选
    if isinstance(response_type, list) and response_type:
        opts = response_type
        for i, o in enumerate(opts, 1):
            print(f"  {i}. {o}")
        idx = int(input("选择序号: ")) - 1
        return ElicitResult(action="accept", content={"value": opts[idx]})

    # 结构化响应（dataclass/BaseModel）：逐字段问
    # FastMCP 会把 schema 拍平，这里简化为按字段名提示
    if hasattr(response_type, "__annotations__"):
        fields = response_type.__annotations__
        data = {}
        for name, typ in fields.items():
            raw = input(f"{name} ({typ.__name__}): ").strip()
            if typ is bool:
                data[name] = raw.lower() in ("y", "yes", "true", "1")
            elif typ is int:
                data[name] = int(raw)
            else:
                data[name] = raw
        return ElicitResult(action="accept", content=data)

    # 标量
    raw = input("你的回答: ").strip()
    if response_type is int:
        return ElicitResult(action="accept", content={"value": int(raw)})
    return ElicitResult(action="accept", content={"value": raw})


async def main():
    transport = StdioTransport(command="uv", args=["run", "safe_ops_server.py"])
    client = Client(transport, elicitation_handler=elicitation_handler)
    async with client:
        res = await client.call_tool("greet", {})
        print("\n=== 工具返回 ===")
        print(res.content[0].text)


asyncio.run(main())

uv run client.py

npx @modelcontextprotocol/inspector uv run safe_ops_server.py

# 方式一：命令行注册
claude mcp add safe-ops -- uv run /绝对路径/safe_ops_server.py

# 方式二：项目级 .mcp.json

{
  "mcpServers": {
    "safe-ops": {
      "command": "uv",
      "args": ["run", "/绝对路径/safe_ops_server.py"]
    }
  }
}

match result:
    case AcceptedElicitation(data):
        ...            # 正常往下走
    case DeclinedElicitation():
        return "⏭️ 用户拒绝提供，按安全策略中止。"   # 用户主动说"不"
    case CancelledElicitation():
        return "🛑 用户取消整个操作。"              # 用户关掉弹窗/Esc

from dataclasses import dataclass
from pathlib import Path
from fastmcp import FastMCP, Context
from fastmcp.server.elicitation import (
    AcceptedElicitation, DeclinedElicitation, CancelledElicitation,
)

mcp = FastMCP("Safe Ops Server")


@dataclass
class ConfirmDelete:
    """结构化二次确认：MCP 规范只支持浅对象+标量属性，这里两个字段都合规。"""
    confirm: bool
    reason: str


@mcp.tool
async def safe_delete_files(ctx: Context, directory: str, pattern: str = "*") -> str:
    """递归删除 directory 下匹配 pattern 的文件；执行真正删除前必须经用户确认。"""
    root = Path(directory).expanduser()
    if not root.is_dir():
        return f"❌ 目录不存在: {directory}"

    # 运行时才知道的事实：到底匹配到多少文件
    targets = [p for p in root.rglob(pattern) if p.is_file()]
    if not targets:
        return "✅ 没有匹配的文件，无需操作。"

    preview = ", ".join(p.name for p in targets[:3])
    await ctx.info(f"扫描完成：在 {directory} 下匹配到 {len(targets)} 个文件（如 {preview}...）")

    # —— 第一轮：选模式（受限选项）——
    mode = await ctx.elicit(
        message=(
            f"在 {directory} 下找到 {len(target_count)} 个匹配 '{pattern}' 的文件。\n"
            f"样本: {preview}\n请选择操作模式："
        ),
        response_type=["delete", "dry-run", "cancel"],
    )
    if not isinstance(mode, AcceptedElicitation):
        return "🛑 已取消。"
    if mode.data == "cancel":
        return "🛑 用户取消。"

    # —— dry-run 分支：只预览，零副作用 ——
    if mode.data == "dry-run":
        lines = "\n".join(str(p) for p in targets[:20])
        more = "" if len(targets) <= 20 else f"\n...（其余 {len(targets)-20} 个略）"
        return f"🔍 [演练] 将删除 {len(targets)} 个文件：\n{lines}{more}"

    # —— delete 分支：结构化二次确认 ——
    conf = await ctx.elicit(
        message=(
            f"⚠️ 即将真正删除 {len(targets)} 个文件，操作不可逆！\n"
            "请确认，并填写删除原因（写入审计日志）："
        ),
        response_type=ConfirmDelete,
    )
    # 安全铁律：非 accept 一律中止
    if not isinstance(conf, AcceptedElicitation) or not conf.data.confirm:
        return "🛑 未获确认，已放弃删除。"

    deleted, failed = 0, []
    for p in targets:
        try:
            p.unlink()
            deleted += 1
        except OSError as e:
            failed.append(f"{p}: {e}")
            await ctx.warning(f"删除失败 {p}: {e}")

    audit = f"原因[{conf.data.reason}]"
    summary = f"✅ 已删除 {deleted}/{len(targets)} 个文件（{audit}）。"
    if failed:
        summary += f"\n⚠️ {len(failed)} 个失败：\n" + "\n".join(failed[:10])
    return summary


if __name__ == "__main__":
    mcp.run()   # 默认 stdio 传输

res = await client.call_tool("safe_delete_files",
                             {"directory": "/tmp/junk", "pattern": "*.log"})
print(res.content[0].text)

mkdir -p /tmp/junk && cd /tmp/junk && for i in $(seq 1 5); do touch "app_$i.log"; done && touch keep.txt

sequenceDiagram
    participant U as 用户
    participant C as MCP客户端<br/>(Claude Code/Inspector/自带客户端)
    participant S as MCP服务器(FastMCP)
    participant T as safe_delete_files

    U->>C: "清理 /tmp/junk 下日志"
    C->>S: tools/call safe_delete_files
    S->>T: 执行(directory, pattern)
    T->>T: rglob 扫描 → 5 个 .log
    T-->>S: 找到5个，需确认
    Note over S,C: 关键：方向反过来，服务器→客户端
    S->>C: elicitation/create (schema: enum[delete,dry-run,cancel])
    C->>U: 渲染选项/提问
    U-->>C: 选 "dry-run"
    C-->>S: result: accept, data=dry-run
    S-->>T: 返回 ElicitResult
    T-->>S: 预览文本
    S-->>C: 工具结果
    C-->>U: 展示 5 个文件清单

flowchart TD
    A[工具被调用] --> B[扫描目录得到 targets]
    B --> C{第一轮 elicit: 选模式}
    C -->|cancel/非accept| Z[🛑 中止,零副作用]
    C -->|dry-run| D[返回文件预览清单]
    C -->|delete| E{第二轮 elicit:<br/>ConfirmDelete结构化}
    E -->|未确认| Z
    E -->|确认 confirm=true| F[执行 unlink 删除]
    F --> G[返回删除统计+审计原因]
    D --> H[结束]
    G --> H