跳转到主要内容

采样

处理来自服务器的 LLM 补全请求
MCP 服务器可以在执行工具时向 Client 请求 LLM 补全。这样服务器就可以将 AI 推理委托给由 Client 控制的一侧,由 Client 决定使用哪种模型。

处理器

为了支持采样功能,请在建立连接时注册一个采样处理器。该处理器会接收服务器的请求(CreateMessageRequestParams),并应返回:
  • CreateMessageResult(成功),或
  • ErrorData(失败)
下面是一个使用 Anthropic 的完整示例:
from anthropic import AsyncAnthropic
from dedalus_mcp import types

anthropic = AsyncAnthropic()

async def sampling_handler(
    _ctx: object,
    params: types.CreateMessageRequestParams,
) -> types.CreateMessageResult | types.ErrorData:
    try:
        messages = [{"role": m.role, "content": m.content.text} for m in params.messages]

        resp = await anthropic.messages.create(
            model="claude-sonnet-4-20250514",
            messages=messages,
            max_tokens=params.maxTokens,
        )

        return types.CreateMessageResult(
            model=resp.model,
            role="assistant",
            content=types.TextContent(type="text", text=resp.content[0].text),
            stopReason="end_turn",
        )
    except Exception as e:
        return types.ErrorData(code=types.INTERNAL_ERROR, message=str(e))

用法

在连接时,通过在 ClientCapabilitiesConfig 中传入该处理程序来启用采样:
capabilities = ClientCapabilitiesConfig(sampling=sampling_handler)

async with open_connection(
    url="http://127.0.0.1:8000/mcp",
    capabilities=capabilities,
) as client:
    # 如果服务器在此 tool 运行期间调用 sampling/createMessage,
    # 将会调用您的 sampling_handler。
    result = await client.call_tool("analyze", {"data": "..."})

错误处理

当你的 handler 内部出现问题时,请返回一个 ErrorData(不要抛出异常)。服务器会将其视为对其采样请求的 MCP 错误响应。
from dedalus_mcp import types

async def sampling_handler(context: object, params: types.CreateMessageRequestParams):
    try:
        # ... 调用大语言模型 ...
        return types.CreateMessageResult(
            model="claude-3-5-sonnet-20241022",
            role="assistant",
            content=types.TextContent(type="text", text="ok"),
            stopReason="end_turn",
        )
    except Exception as e:
        return types.ErrorData(code=types.INTERNAL_ERROR, message=str(e))