MCP Sampling 反向调用：构建零密钥的智能 MCP 工具

flowchart LR
    subgraph 正向["普通工具调用（Client → Server）"]
        A1[Client LLM] -->|tools/call| A2[Server 工具]
        A2 -->|结果| A1
    end
    subgraph 反向["Sampling（Server → Client）"]
        B1[Server 工具] -->|sampling/createMessage| B2[Client LLM]
        B2 -->|补全| B1
    end

{
  "capabilities": { "sampling": {} }
}

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "sampling/createMessage",
  "params": {
    "messages": [
      { "role": "user", "content": { "type": "text", "text": "法国的首都是哪里？" } }
    ],
    "modelPreferences": {
      "hints": [{ "name": "claude-3-sonnet" }],
      "intelligencePriority": 0.8,
      "speedPriority": 0.5
    },
    "systemPrompt": "你是一个简洁的助手。",
    "maxTokens": 100
  }
}

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "role": "assistant",
    "content": { "type": "text", "text": "法国的首都是巴黎。" },
    "model": "claude-3-sonnet-20240307",
    "stopReason": "endTurn"
  }
}

{ "hints": [{ "name": "claude-3-sonnet" }, { "name": "claude" }],
  "costPriority": 0.3, "speedPriority": 0.8, "intelligencePriority": 0.5 }

sequenceDiagram
    participant S as Server
    participant C as Client
    participant U as User
    participant L as LLM
    Note over S,C: 1. 服务器发起采样
    S->>C: sampling/createMessage
    Note over C,U: 2. 人审请求
    C->>U: 展示请求供审核
    U-->>C: 审阅/修改/批准
    Note over C,L: 3. 模型交互
    C->>L: 转发已批准请求
    L-->>C: 返回补全
    Note over C,U: 4. 人审响应
    C->>U: 展示响应供审核
    U-->>C: 审阅/修改/批准
    C-->>S: 返回已批准响应

# server_sentiment.py
from fastmcp import FastMCP, Context
from pydantic import BaseModel

mcp = FastMCP("SentimentBySampling")

class SentimentResult(BaseModel):
    text: str
    sentiment: str   # positive / negative / neutral

@mcp.tool
async def analyze_sentiment(text: str, ctx: Context) -> SentimentResult:
    """对文本做情感分类（推理由宿主 LLM 完成，本服务器不持有任何密钥）。"""
    prompt = (
        "把下面的文本分类为 'positive'、'negative' 或 'neutral'，"
        "只返回一个单词，不要标点。\n\n"
        f"Text: {text}"
    )
    # 关键：把推理外包给宿主
    resp = await ctx.sample(
        messages=prompt,
        temperature=0.0,   # 分类任务要稳定
        max_tokens=8,
    )
    label = (resp.text or "").strip().lower()
    # 兜底：宿主可能返回越界值
    if label not in {"positive", "negative", "neutral"}:
        label = "neutral"
    return SentimentResult(text=text, sentiment=label)

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

// server_loganalyzer.ts
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { CallToolRequestSchema } from "@modelcontextprotocol/sdk/types.js";

const server = new Server(
  { name: "log-analyzer", version: "1.0.0" },
  { capabilities: { tools: {}, sampling: {} } } // 声明将使用 sampling
);

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name !== "analyze-log") {
    throw new Error(`Unknown tool: ${request.params.name}`);
  }
  const rawLog = request.params.arguments?.log as string;
  if (!rawLog) throw new Error("Missing required argument: log");

  let samplingResult;
  try {
    samplingResult = await server.createMessage({
      messages: [{
        role: "user",
        content: { type: "text", text:
          `分析日志并返回 JSON：severity(critical|warning|info)、root_cause(一句话)、action_required(bool)。\n` +
          `只返回合法 JSON，不要 markdown。\n日志：${rawLog}` }
      }],
      // 分类任务：要快、要便宜、不需要深度推理
      modelPreferences: { speedPriority: 0.9, costPriority: 0.8, intelligencePriority: 0.3 },
      systemPrompt: "你是日志分析 API，只返回合法 JSON。",
      maxTokens: 300,
    });
  } catch (err: any) {
    // 降级：采样失败时不要让整个工具崩掉
    return { content: [{ type: "text", text: `Analysis unavailable: ${err.message}` }], isError: true };
  }

  const content: any = samplingResult.content;
  if (content.type !== "text") throw new Error("Expected text response");
  let parsed: any;
  try { parsed = JSON.parse(content.text); }
  catch { throw new Error(`Sampling returned invalid JSON: ${content.text}`); }

  return { content: [{ type: "text", text: JSON.stringify({
    analysis: parsed,
    model_used: samplingResult.model,   // 记录实际使用的模型，便于审计
    raw_log_length: rawLog.length,
  }, null, 2) }] };
});

async function main() {
  await server.connect(new StdioServerTransport());
  console.error("log-analyzer MCP server running");
}
main().catch(console.error);

// host_client.ts
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { CreateMessageRequestSchema } from "@modelcontextprotocol/sdk/types.js";
import Anthropic from "@anthropic-ai/sdk";

const anthropic = new Anthropic();
let samplingCallCount = 0;
const MAX_SAMPLING_CALLS = 5; // 防止服务器陷入采样死循环

// 必须在 client.connect() 之前注册
client.setRequestHandler(CreateMessageRequestSchema, async (request) => {
  const { messages, systemPrompt, maxTokens, modelPreferences } = request.params;

  // 采样深度护栏：防止恶意/有 bug 的服务器无限采样
  samplingCallCount++;
  if (samplingCallCount > MAX_SAMPLING_CALLS) {
    throw new Error("Sampling depth limit exceeded");
  }

  // 模型路由：把 hint 当建议，自己拍板
  const hint = modelPreferences?.hints?.[0]?.name;
  const model = resolveModel(hint);

  const response = await anthropic.messages.create({
    model,
    max_tokens: maxTokens ?? 1024,
    system: systemPrompt,
    messages: messages.map((m: any) => ({
      role: m.role,
      content: typeof m.content === "string" ? m.content : m.content.text,
    })),
  });
  const textBlock: any = response.content.find((b: any) => b.type === "text");
  return {
    role: "assistant",
    content: { type: "text", text: textBlock?.text ?? "" },
    model: response.model,
    stopReason: response.stop_reason ?? "endTurn",
  };
});

function resolveModel(hint?: string): string {
  const map: Record<string, string> = {
    "claude-sonnet": "claude-sonnet-4-6",
    "claude-haiku": "claude-haiku-4-5-20251001",
    "claude-opus":  "claude-opus-4-8",
  };
  return hint && map[hint] ? map[hint] : "claude-haiku-4-5-20251001";
}

flowchart TD
    U([用户触发工具]) --> S1[服务器：确定性逻辑]
    S1 -->|需要语言理解?| D{决策}
    D -->|否| S2[纯代码完成]
    D -->|是| SM[sampling/createMessage]
    SM --> H[宿主：人审 + 模型路由]
    H --> L[LLM 推理]
    L --> H2[宿主：人审响应]
    H2 -->|深度未超限| S3[服务器：解析 + 继续]
    H2 -->|超过 MAX 深度| ERR[降级/报错]
    S3 --> R([返回最终工具结果])
    S2 --> R