<!-- title: MCP反向调大模型实战 -->

## 概述

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

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

## 前置条件

- Python 3.10 及以上
- 安装官方 Python SDK：`pip install "mcp[cli]"`（FastMCP 已并入官方包）
- 联网时可选装 `pip install anthropic`，用于让宿主调用真实模型
- 基本理解 MCP 三原语（Tools / Resources / Prompts），知道客户端与服务器靠 JSON-RPC 通信
- 一个能跑 `python server.py` 的终端环境

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

## 步骤一：理解 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 计数）。

## 步骤二：搭建项目骨架

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

```bash
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 分析，再把分析结果作为工具返回值。

```python
# 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")
```

逐行拆解关键点：

- `ctx: Context` 是 FastMCP 注入的上下文，`ctx.session` 就是当前会话，`.create_message()` 即发起采样的入口。
- `SamplingMessage` 的 `content` 必须是 `TextContent`（也支持 `ImageContent`），与普通对话消息结构一致。
- `system_prompt`、`max_tokens`、`temperature` 控制宿主侧模型的行为，和我们直接调 API 的参数含义相同。
- 工具函数最终返回的是**采样结果里的文本**——对调用方（无论是人还是 LLM）而言，这就像一个普通的、能"看懂"日志的工具。

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

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

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

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

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

几个容易踩的细节：

- `ClientSession(..., sampling_callback=...)` 是把"宿主如何响应采样"注入会话的官方钩子。回调签名收到服务器发来的 `messages` 和 `params`，必须返回一个 `CreateMessageResult`。
- 没有设置 `ANTHROPIC_API_KEY` 时走 `stub` 分支——这是教学场景的保险阀，确保整条链路在没有密钥、不联网时也能跑通，方便你验证"反向调用"是否接通。
- **安全要点**：生产环境里，这个回调就是你的审批关口。你可以在转发前对 `user_text` 做脱敏（去掉密钥、身份证号）、设白名单（只允许特定 `systemPrompt`）、加额度上限，甚至弹窗让用户确认。这把"服务器借宿主之力"可能带来的风险，锁在了你能看见、能控制的地方。

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

## 步骤五：端到端联调

在项目目录下运行：

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

**预期输出**（离线回退）：

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

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

```bash
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`：

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

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

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

- 想验证"反向调用"本身，**用步骤四的自建宿主最稳妥**，行为完全可控。
- 想在具体某个产品里用，先查它是否声明支持 sampling；不支持时，可把工具改为"采样降级方案"——即工具内部直接调自己的 API（需要服务器持有密钥），代价是失去宿主统一管控与零密钥的优势。

一种兼顾两者的工程做法：在 `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 的价值定位：它不是"再发明一次工具调用"，而是**让工具具备"按需借脑"的能力，同时把借脑的审批权留给用户**。

## 工作流程图

```mermaid
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 的全部精髓。

## 常见问题速查

- **Q：采样和让服务器自己调 OpenAI/Anthropic API 有什么区别？**
  A：自己调 API 需要服务器持有密钥、自行付费、自行管控安全；采样则把这些责任上交给宿主，服务器零密钥、可审计、能力随宿主走。代价是依赖宿主实现 sampling 能力。

- **Q：服务器发起采样会泄漏数据吗？**
  A：会经过宿主。数据走向由宿主决定——这正是宿主做脱敏、白名单、审批的着力点。相比"服务器直连第三方"，采样反而让数据流更可控。

- **Q：`stopReason` 有哪些取值？**
  A：`endTurn`（自然结束）、`stopSequence`（命中停止序列）、`maxTokens`（达上限）。可用于在服务器侧判断是否截断、是否需要续采。

- **Q：能否在采样请求里带图片？**
  A：可以。`SamplingMessage.content` 支持 `ImageContent`，宿主若接入多模态模型即可处理，常用于"截图→诊断"类工具。

## 延伸阅读

- **MCP 官方规范 · Sampling**：`https://modelcontextprotocol.io/specification/2025-06-18/client/sampling`——`createMessage` 的权威定义与安全模型。
- **官方 Python SDK 采样示例**：`https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/sampling.py`——本文服务端代码的直接原型。
- **MCP 官方客户端构建教程**：`https://modelcontextprotocol.io/docs/develop/build-client`——`ClientSession` 与 `sampling_callback` 的接入方式。
- **FastMCP 文档**：`https://gofastmcp.com/`——`Context` 对象与 `create_message` 的高层封装。

## 训练营适用性评估

- **适合训练营类型**：中高级 AI 工程师训练营、Agent 开发实战营、企业级 MCP 集成专题营。要求学员已具备基础 MCP 经验，不适合作为入门第一课。
- **预计教学时长**：3–4 学时。其中概念与方向辨析约 0.5 学时，编码与联调约 2 学时，扩展练习与安全讨论约 1–1.5 学时。
- **考核建议**：以"能否独立解释工具调用与采样的方向差异"为理论考点，以"自建宿主跑通反向链路 + 至少实现一个安全审批规则"为实操交付物。
- **延伸价值**：本主题是把 MCP 从"数据搬运层"提升到"能力协作层"的关键一跃，学完后学员可自然过渡到 Roots、订阅通知、Elicitation 等其余进阶原语。

## 参考来源

1. [Sampling — Model Context Protocol 官方规范](https://modelcontextprotocol.io/specification/2025-06-18/client/sampling) — `sampling/createMessage` 方向、参数、安全模型的权威定义
2. [python-sdk/examples/snippets/servers/sampling.py — GitHub](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/sampling.py) — 官方 Python SDK 的服务端采样代码原型，本文 `server.py` 直接参照
3. [Build an MCP client — Model Context Protocol](https://modelcontextprotocol.io/docs/develop/build-client) — `ClientSession` 与宿主侧 `sampling_callback` 的官方接入教程
4. [MCP Sampling Explained — mcpevals.io](https://www.mcpevals.io/blog/mcp-sampling-explained) — 对采样架构、工作流与"宿主守门人"价值的技术解读
5. [MCP Sampling: Architecture, Workflow, and Practical Guide](https://gyliu513.github.io/jekyll/update/2025/08/21/mcp-sampling.html) — 采样全链路架构图与实践指引，补充了多步骤 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学通信知识星球海报-扫码加入](https://wiki.500nian.com/images/zhixingqiu_qrcode.png)