MCP 反向交互三原语开发实战：Sampling、Elicitation、Roots

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

{
  "capabilities": {
    "roots": { "listChanged": true }
  }
}

{
  "jsonrpc": "2.0", "id": 1,
  "result": {
    "roots": [
      { "uri": "file:///home/user/projects/myproject", "name": "My Project" },
      { "uri": "file:///home/user/repos/backend", "name": "Backend" }
    ]
  }
}

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)

{
  "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"]
    }
  }
}

{ "jsonrpc": "2.0", "id": 2,
  "result": { "action": "accept", "content": { "host": "db.local", "port": 5432, "env": "dev" } } }

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} 环境）"

{
  "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
  }
}

{
  "jsonrpc": "2.0", "id": 1,
  "result": {
    "role": "assistant",
    "content": { "type": "text", "text": "..." },
    "model": "claude-sonnet-4-6",
    "stopReason": "endTurn"
  }
}

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

# 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 传输

# 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 -> 你填表单 -> 服务器采样出方案

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: 展示结果