首页 › tutorials › MCP反向调大模型实战

概述

绝大多数 MCP 服务器是"被动的"——客户端问它要数据，它返回数据。但 MCP 规范里藏着一个常被忽略、却极具威力的原语：Sampling（采样）。它把调用方向翻转过来，让服务器反向请求宿主的大模型帮自己完成推理。本教程带你从零写一个"会自己思考"的 MCP 工具：一个日志智能分诊服务器，它把杂乱的日志丢给宿主的 LLM，让它输出严重等级、根因和处置建议。学完后，你将掌握 Sampling 的完整闭环——服务端发起、宿主审批与转发、结果回流。

适用人群：已会用 Python 写过最基础 MCP 工具（至少跑通过一个 @mcp.tool）、想突破"工具只能搬运数据"瓶颈、希望工具具备判断力的开发者。

前置条件

关键认知：工具调用与采样是相反方向的两条路。工具调用是"宿主 → 服务器"；采样是"服务器 → 宿主"。理解这一点，本教程就懂了一半。

步骤一：理解 Sampling 的反向调用机制

先看清楚方向。普通 MCP 工具调用走的是这条路：

宿主(LLM) --call_tool--> 服务器 --结果--> 宿主

而 Sampling 走的是这条路：

服务器(在工具执行中) --createMessage--> 宿主 --调用自身LLM--> 返回completion --> 服务器

服务器并没有自己持有 API Key，它借用宿主的模型能力。这意味着：

  1. 零密钥：服务器不需要任何 LLM 凭证，模型访问权由宿主统一管理。
  2. 可审计：宿主是"守门人"，可以审查、改写、甚至拒绝服务器发出的采样请求，再决定是否真的转发给 LLM。这把安全控制权牢牢握在用户手里。
  3. 能力升级：原本只能做确定性计算的工具，现在能做"理解、分类、总结、推断"这类需要语言模型的任务，却又不必把敏感数据外泄到第三方。

官方规范定义的方法名是 sampling/createMessage，请求参数包括 messages、systemPrompt、modelPreferences、maxTokens、temperature、stopSequences、includeContext 等；返回结果包含 content（assistant 消息）、model（实际使用的模型）、stopReason（停止原因）和 usage（token 计数）。

步骤二：搭建项目骨架

新建目录并初始化虚拟环境：

mkdir mcp-sampling-triage && cd mcp-sampling-triage
python -m venv .venv
source .venv/bin/activate   # Windows 用 .venv\Scripts\activate
pip install "mcp[cli]"
pip install anthropic        # 可选，用于真实模型调用

建立两个文件：server.py（带采样能力的服务器）和 client.py（支持采样的宿主客户端）。我们一会儿分别填充。这种"服务器 + 自建宿主"的双文件结构是测试 Sampling 的标准姿势——因为并非所有 MCP 宿主都实现了 sampling 客户端能力，自带一个能响应 createMessage 的宿主，能保证端到端可跑。

预期结果：目录里出现 server.py、client.py 两个空文件，依赖安装无报错。

常见问题： - 安装 mcp[cli] 报错：检查 Python 版本 ≥3.10，必要时升级 pip。 - Windows 下 stdio 传输偶尔有缓冲问题：建议在 WSL 或 macOS/Linux 下联调。

步骤三：编写带 Sampling 的 MCP 服务器

server.py 的核心思想：在工具函数内部，不等宿主，主动发起一次采样请求，把日志交给宿主的 LLM 分析，再把分析结果作为工具返回值。

# server.py —— 带 Sampling 的日志分诊服务器
from mcp.server.fastmcp import FastMCP, Context
from mcp.types import SamplingMessage, TextContent

mcp = FastMCP("log-triage")

SYSTEM_PROMPT = (
    "你是一名资深 SRE。读取给定日志，仅输出 JSON，包含三个字段："
    '"severity"(critical/warning/info)、'
    '"root_cause"(一句话根因)、'
    '"action"(建议的下一步操作)。不要输出 JSON 以外的任何内容。'
)

@mcp.tool()
async def triage_log(log_text: str, ctx: Context) -> str:
    """对一个日志片段做智能分诊，返回严重等级、根因和建议操作。"""
    result = await ctx.session.create_message(
        messages=[
            SamplingMessage(
                role="user",
                content=TextContent(
                    type="text",
                    text=f"请分诊以下日志：\n\n{log_text}",
                ),
            )
        ],
        system_prompt=SYSTEM_PROMPT,
        max_tokens=400,
        temperature=0.2,
    )
    # 采样结果是一条 assistant 消息，取其文本
    if result.content.type == "text":
        return result.content.text
    return str(result.content)

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

逐行拆解关键点：

预期结果：用官方 SDK 的 snippet 做对照，这段代码与服务端采样范例的结构完全一致，能被任何实现了 sampling 的宿主调用。

常见问题： - 报 AttributeError: 'Context' object has no attribute 'session'：极旧版本 SDK 用 ctx.get_context() 或不同接口，升级到最新 mcp 包即可。 - create_message 触发但无响应：说明连接的宿主没有实现 sampling 客户端能力（见下一步如何自建宿主）。

步骤四：编写支持 Sampling 的宿主客户端

Sampling 的另一半在宿主侧：服务器发来 createMessage 请求，宿主要决定怎么响应——是直接转发给真实模型，还是先用规则改写、甚至拒绝。下面这个 client.py 就是这样一个"会审批"的宿主：

# client.py —— 支持 Sampling 的宿主：把请求转发给真正的 LLM
import asyncio
import os
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.types import (
    CreateMessageResult, TextContent,
    SamplingMessage, CreateMessageParams,
)

API_KEY = os.environ.get("ANTHROPIC_API_KEY")

async def sampling_callback(
    messages: list[SamplingMessage],
    params: CreateMessageParams,
) -> CreateMessageResult:
    """宿主对 sampling/createMessage 的响应：这里调用真正的模型。"""
    user_msg = messages[-1]
    user_text = (user_msg.content.text
                 if hasattr(user_msg.content, "text")
                 else str(user_msg.content))

    if API_KEY:
        # 真实路径：转发给大模型
        from anthropic import AsyncAnthropic
        client = AsyncAnthropic(api_key=API_KEY)
        resp = await client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=params.maxTokens or 512,
            system=params.systemPrompt or "",
            messages=[{"role": "user", "content": user_text}],
        )
        text = resp.content[0].text
        used = "claude-haiku-4-5"
    else:
        # 离线回退：无 Key 时也能跑通整个链路
        text = ('{"severity":"info",'
                '"root_cause":"离线回退：未配置 ANTHROPIC_API_KEY",'
                '"action":"设置环境变量后重试以获得真实分析"}')
        used = "stub"

    return CreateMessageResult(
        role="assistant",
        content=TextContent(type="text", text=text),
        model=used,
        stopReason="endTurn",
    )

async def main():
    params = StdioServerParameters(command="python", args=["server.py"])
    async with stdio_client(params) as (read, write):
        async with ClientSession(
            read, write, sampling_callback=sampling_callback
        ) as session:
            await session.initialize()
            tools = await session.list_tools()
            print("已注册工具：", [t.name for t in tools.tools])
            res = await session.call_tool(
                "triage_log",
                {"log_text":
                 "ERROR ConnectionRefused db:5432 timeout after 30s"},
            )
            print("分诊结果：\n", res.content[0].text)

if __name__ == "__main__":
    asyncio.run(main())

几个容易踩的细节：

预期结果：连接成功后打印出已注册工具 ['triage_log']，以及一段 JSON 形态的分诊结论。

步骤五：端到端联调

在项目目录下运行：

# 不配 Key，先验证链路通不通
python client.py

预期输出（离线回退）：

已注册工具： ['triage_log']
分诊结果：
 {"severity":"info","root_cause":"离线回退：未配置 ANTHROPIC_API_KEY","action":"设置环境变量后重试以获得真实分析"}

看到这行，说明反向调用链路已完全打通——服务器成功向宿主发了 createMessage，宿主成功返回了结果。接着配置真实模型再跑一次：

export ANTHROPIC_API_KEY=sk-ant-...
python client.py

把日志换成更接近真实的故障：

"分诊结果：
 {"severity":"critical","root_cause":"数据库连接被拒，端口5432无服务监听或网络不通","action":"检查 db 实例是否存活、安全组是否放行5432、连接串是否指向正确主机"}'

常见问题： - 卡在 initialize() 无响应：八成是 server.py 有语法错误或 mcp 包版本不匹配，单独跑 python server.py 看报错。 - 真实模型返回非 JSON：把 SYSTEM_PROMPT 收紧，或在服务器侧加一层"提取首个 {...}"的后处理。 - token 超限：调大 max_tokens，或缩短喂进去的日志。

步骤六：接入 Claude Code 与宿主兼容性

如果你想把这个分诊工具直接挂到 Claude Code 里用，在项目根目录放一个 .mcp.json：

{
  "mcpServers": {
    "log-triage": {
      "command": "python",
      "args": ["server.py"]
    }
  }
}

随后在该目录运行 claude，Claude Code 会自动加载它。

重要提醒：triage_log 这个工具能否在某个宿主里成功执行，取决于该宿主是否实现了 sampling 客户端能力。并非所有 MCP 客户端都会响应 createMessage：若宿主未实现，工具运行时会因采样请求无人应答而失败。因此：

一种兼顾两者的工程做法：在 triage_log 里先尝试采样，捕获异常后回退到"服务器自带轻量模型"，既享受宿主能力，又保证健壮性。

实战案例：日志智能分诊服务器

把前面的零件组装成一个可复用的小项目。目录结构：

mcp-sampling-triage/
├── server.py        # 带 Sampling 的分诊工具
├── client.py        # 支持 Sampling 的宿主
├── .mcp.json        # 供 Claude Code 加载
└── samples/
    └── crash.log    # 待分诊的真实日志样本

进阶扩展方向（留作练习）：

  1. 批量化：新增 triage_batch(paths: list[str])，对一批日志并发采样，用 asyncio.gather 提速。
  2. 结构化校验：在服务器侧用 pydantic 校验采样返回的 JSON，失败时自动让宿主重试一次。
  3. 审批 UI：在 sampling_callback 里加入"高危请求拦截"——比如 systemPrompt 包含越权指令时直接拒绝返回，演示宿主的守门人价值。
  4. 多模型路由：根据 params.modelPreferences，在宿主侧按成本/速度动态选择 Haiku 还是 Sonnet。

这些扩展共同印证了 Sampling 的价值定位：它不是"再发明一次工具调用"，而是让工具具备"按需借脑"的能力，同时把借脑的审批权留给用户。

工作流程图

sequenceDiagram
    autonumber
    participant U as 用户/LLM
    participant H as 宿主(Claude Code/自建Client)
    participant S as MCP服务器(triage_log)
    participant M as 宿主内的真实模型

    U->>H: "帮我看看这条日志"
    H->>S: tools/call triage_log(log_text)
    Note over S: 工具执行中，需要"理解"日志
    S->>H: sampling/createMessage(日志+系统提示)
    Note over H: 守门人：可审查/改写/拒绝
    H->>M: 调用模型(Haiku/Sonnet)
    M-->>H: completion(JSON分诊结论)
    H-->>S: CreateMessageResult
    S-->>H: tools/call 结果(分诊JSON)
    H-->>U: 呈现结论给用户

注意第 4–7 步构成了一个"回环"：服务器在工具执行过程中，临时把球踢回给宿主，宿主处理完再把球还回来，服务器才继续往下走。这个回环就是 Sampling 的全部精髓。

常见问题速查

延伸阅读

训练营适用性评估

参考来源

  1. Sampling — Model Context Protocol 官方规范 — sampling/createMessage 方向、参数、安全模型的权威定义
  2. python-sdk/examples/snippets/servers/sampling.py — GitHub — 官方 Python SDK 的服务端采样代码原型，本文 server.py 直接参照
  3. Build an MCP client — Model Context Protocol — ClientSession 与宿主侧 sampling_callback 的官方接入教程
  4. MCP Sampling Explained — mcpevals.io — 对采样架构、工作流与"宿主守门人"价值的技术解读
  5. MCP Sampling: Architecture, Workflow, and Practical Guide — 采样全链路架构图与实践指引，补充了多步骤 Agent 场景下的应用

[!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学通信知识星球海报-扫码加入

📎 本分类最新文章