用 MCP Sampling 让 Server 反向调用大模型（sampling/createMessage 实操）

用官方 mcp 包（含 FastMCP）。注意：还有一个独立的 fastmcp v2 包，API 不同，本教程统一用 mcp.server.fastmcp。

python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install "mcp[cli]"

新建 server.py。工具的参数里加一个 ctx: Context，框架会自动注入；在工具内部用 await ctx.session.create_message(...) 反向请求一次模型推理。下面这个工具让服务器"借"客户端的模型把一段文本总结一下。

from mcp.server.fastmcp import FastMCP, Context
from mcp.types import SamplingMessage, TextContent

mcp = FastMCP(name="sampling-demo")

@mcp.tool()
async def summarize(text_to_summarize: str, ctx: Context) -> str:
    """借用宿主模型总结一段文本，服务端本身不持有任何 LLM Key。"""
    result = await ctx.session.create_message(
        messages=[
            SamplingMessage(
                role="user",
                content=TextContent(
                    type="text",
                    text=f"请用一句话总结以下内容：\n{text_to_summarize}",
                ),
            )
        ],
        max_tokens=4000,
        system_prompt="你是一个严谨的研究助手，只输出总结，不发挥。",
    )
    if result.content.type == "text":
        return result.content.text
    raise ValueError("Sampling 返回内容不是文本")

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

客户端要提供一个 sampling 回调，真正去跑模型。下面用 OpenAI 兼容接口举例；回调签名在新版为 (context, params) 形式。

import os
from openai import OpenAI
from mcp import ClientSession, types

client = OpenAI()   # 读环境变量 OPENAI_API_KEY

async def handle_sampling(context, params: types.CreateMessageRequestParams):
    # 客户端用自己的 Key 跑推理，返回结果给服务器
    resp = client.chat.completions.create(
        model="gpt-4.1-mini",
        max_tokens=params.max_tokens or 1000,
        messages=[m.model_dump() for m in params.messages],
    )
    return types.CreateMessageResult(
        role="assistant",
        content=types.TextContent(type="text", text=resp.choices[0].message.content),
        model="gpt-4.1-mini",
        stop_reason="endTurn",
    )

客户端在 initialize 握手时，必须在 capabilities 里声明 sampling: {}，否则服务器发 sampling/createMessage 会报错。用 stdio 拉起服务器并传入回调：

async with sdio_server.server() as (read, write):
    async with ClientSession(read, write, sampling_callback=handle_sampling) as session:
        await session.initialize()           # 握手声明 sampling 能力
        result = await session.call_tool("summarize", {"text_to_summarize": "一段很长的原文..."})
        print(result.content[0].text)

create_message 还支持系统提示、模型偏好和上下文范围。把一次需要的判断尽量放进同一条请求，减少来回推理轮次。

result = await ctx.session.create_message(
    messages=[SamplingMessage(role="user", content=TextContent(type="text", text=prompt))],
    max_tokens=512,
    system_prompt="你是安全审查员，只返回与鉴权、输入校验、数据暴露相关的发现。",
    model_preferences=types.ModelPreference(
        hints=[types.ModelHint(name="claude")],
        intelligence_priority=0.8,
        speed_priority=0.2,
        cost_priority=0.0,
    ),
    include_context="thisServer",
)