View a markdown version of this page

在 AgentCore 网关上使用采样 - Amazon Bedrock AgentCore
先决条件采样的工作原理采样流程MCP 服务器目标开发者指南错误处理问题排查代码示例

在 AgentCore 网关上使用采样

采样是一项 MCP 功能，允许 MCP 服务器在工具调用期间向客户端请求完成 LLM。这使服务器无需直接访问语言模型即可利用 AI 功能——客户端处理模型调用并返回结果。 AgentCore Gateway 将来自 MCP 服务器目标的采样请求转发给您的客户端，将请求替换为网关生id成的标识符。

先决条件

要在网关上使用采样，请执行以下操作：

  • 启用会话-采样需要会话支持。请参阅在您的网关上使用 MCP 会话。

  • 已启用响应流-在打开的连接期间，采样请求以 SSE 区块的形式发送。true在streamingConfiguration.enableResponseStreaming您的网关中设置为protocolConfiguration.mcp。

  • MCP 服务器目标类型-采样请求源自 MCP 服务器目标。

  • 客户端声明采样能力 — 客户端必须在initialize请求期间声明支持采样。网关仅将采样请求转发给声明此功能的客户端。

采样的工作原理

当 MCP 服务器目标在工具执行期间需要完成 LLM 时，它会发送请求。sampling/createMessage网关将此请求作为 SSE 事件转发给客户端，取代该请求id。客户端调用其语言模型并将结果发送回网关，网关将其转发给目标。

采样请求包括：

  • messages— 要发送给模特的对话消息。

  • modelPreferences— 有关所需模型功能（智能、速度、成本）的可选提示。

  • systemPrompt— 模型的可选系统提示符。

  • maxTokens— 要生成的最大代币数量。

客户回复为：

  • model— 使用的模型。

  • role— 永远assistant。

  • content— 生成的内容（文本或图像）。

注意

客户端可以完全控制使用哪种模型以及如何处理请求。服务器modelPreferences是提示，而不是要求。客户端也可以根据自己的策略修改或拒绝请求。

采样流程

  1. 客户端发送带有Mcp-Session-Id标头的tools/call请求。

  2. Gateway 将工具调用转发给 MCP 服务器目标。

  3. 目标打开 SSE 流并发送sampling/createMessage请求。

  4. Gateway 将采样请求作为 SSE 事件转发给客户端，取代请求id。

  5. 客户端使用提供的消息调用其语言模型。

  6. 客户端发送一个新的请求，其采样结果与网关请求id中的相同Mcp-Session-Id。

  7. Gateway 将结果转发给 MCP 服务器目标。

  8. 目标继续处理并返回最终的刀具结果。

  9. Gateway 将最终结果转发给客户端并关闭直播。

MCP 服务器目标开发者指南

重要

发送采样请求的 MCP 服务器目标应将采样调用封装在 try-catch 块中，并处理客户端不支持采样的情况。如果网关的客户端未声明采样能力，则网关不会向目标声明采样能力。如果目标仍然发送采样请求，则网关会向目标返回-32601（未找到方法）错误。

当无法进行采样时，服务器应实现回退路径（例如使用内置模型或跳过该 AI-assisted 步骤）。

错误处理

场景 错误 说明

当没有待处理的采样请求时，客户端会发送采样响应

JSON-RPC -32600（请求无效）

未找到与该会话匹配的采样请求。

客户端发送的样本响应与待处理请求不匹配 id

JSON-RPC -32600（请求无效）

id必须与网关在sampling/createMessage请求中发送的相匹配。

MCP 服务器发送采样请求但网关未声明支持

JSON-RPC -32601（未找到方法）

已返回到 MCP 服务器目标。参阅故障排除。

问题排查

错误：“调用工具'sample_tool'时出错：找不到方法：” sampling/createMessage

当 MCP 服务器目标发送采样请求但网关的客户端在此期间initialize未声明采样能力时，就会发生此错误。网关向目标返回-32601（未找到方法）错误，目标可能会将此错误作为工具执行错误返回给客户端。

要解决这个问题，请执行以下操作：

  • 如果你是 MCP 服务器开发人员：在采样调用周围添加错误处理。在不支持采样时实现回退路径：

    重要

    您必须在create_message通话related_request_id=ctx.request_context.request_id中包含。网关需要这样才能将采样请求与原始工具调用正确关联起来。没有它，采样就不起作用。

    try:
    result = await ctx.session.create_message(
        messages=[{"role": "user", "content": {"type": "text", "text": "Summarize this document"}}],
        max_tokens=500,
        related_request_id=ctx.request_context.request_id,
    )
except Exception as e:
    # Fallback when client doesn't support sampling
    logger.warning(f"Sampling not supported: {e}")
    result = fallback_summarization(document)

  • 如果您是网关客户端开发人员：请确保您的客户端在以下期间声明采样功能initialize：

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

代码示例

注意

LangGraph MCP 客户端 (langchain-mcp-adapters) 和 Strands MCP 客户端目前不支持采样。使用下面显示的 MCP 客户端方法来处理来自您的网关的采样请求。

例
Python requests package

  1. import requests
import json
import sseclient

gateway_url = "https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp"
headers = {
    "Content-Type": "application/json",
    "Accept": "text/event-stream",
    "Authorization": "Bearer YOUR_ACCESS_TOKEN"
}

# Step 1: Initialize with sampling capability
init_response = requests.post(gateway_url, headers=headers, json={
    "jsonrpc": "2.0",
    "id": "init-request",
    "method": "initialize",
    "params": {
        "protocolVersion": "2025-06-18",
        "capabilities": {"sampling": {}},
        "clientInfo": {"name": "my-agent", "version": "1.0.0"}
    }
})
session_id = init_response.headers["Mcp-Session-Id"]
headers["Mcp-Session-Id"] = session_id

# Step 2: Call tool (streaming response)
response = requests.post(gateway_url, headers=headers, json={
    "jsonrpc": "2.0",
    "id": "tool-call-1",
    "method": "tools/call",
    "params": {
        "name": "summarizeDocument",
        "arguments": {"documentId": "doc-789"}
    }
}, stream=True)

# Step 3: Process SSE events
client = sseclient.SSEClient(response)
for event in client.events():
    data = json.loads(event.data)
    if data.get("method") == "sampling/createMessage":
        sampling_id = data["id"]
        print(f"Sampling request: {data['params']['messages']}")

        # Step 4: Invoke your LLM and send result
        llm_result = invoke_your_model(data["params"])  # Your LLM invocation
        requests.post(gateway_url, headers=headers, json={
            "jsonrpc": "2.0",
            "id": sampling_id,
            "result": {
                "model": "claude-sonnet-4-20250514",
                "role": "assistant",
                "content": {"type": "text", "text": llm_result}
            }
        })
    elif "result" in data:
        print(f"Tool result: {data['result']}")
        break
MCP Client

  1. from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
import asyncio

async def sampling_handler(request):
    """Handle sampling requests from the server by invoking an LLM."""
    messages = request.params.messages
    llm_response = await invoke_your_model(messages, max_tokens=request.params.maxTokens)
    return {
        "model": "claude-sonnet-4-20250514",
        "role": "assistant",
        "content": {"type": "text", "text": llm_response}
    }

async def use_sampling(url, token):
    headers = {"Authorization": f"Bearer {token}"}

    async with streamablehttp_client(url=url, headers=headers) as (
        read_stream, write_stream, _
    ):
        async with ClientSession(
            read_stream, write_stream,
            sampling_handler=sampling_handler
        ) as session:
            await session.initialize()
            result = await session.call_tool(
                name="summarizeDocument",
                arguments={"documentId": "doc-789"}
            )
            print(f"Tool result: {result}")
            return result

asyncio.run(use_sampling(
    url="https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp",
    token="YOUR_ACCESS_TOKEN"
))