先決條件抽樣的運作方式取樣流程 MCP 伺服器目標開發人員指南錯誤處理疑難排解程式碼範例

搭配 AgentCore 閘道使用抽樣

取樣是一種 MCP 功能，可讓 MCP 伺服器在工具呼叫期間向用戶端請求 LLM 完成。這可讓伺服器利用 AI 功能，而無需直接存取語言模型 — 用戶端會處理模型調用並傳回結果。AgentCore Gateway 會將 MCP 伺服器目標的取樣請求轉送到您的用戶端，id以閘道產生的識別符取代請求。

先決條件

若要搭配閘道使用抽樣：

工作階段已啟用 — 取樣需要工作階段支援。請參閱搭配閘道使用 MCP 工作階段。
已啟用回應串流 — 在開放連線期間，取樣請求會以 SSE 區塊的形式傳送。在閘道的 true中streamingConfiguration.enableResponseStreaming將設定為 protocolConfiguration.mcp。
MCP 伺服器目標類型 — 抽樣請求源自 MCP 伺服器目標。
Client 宣告取樣功能 — 用戶端必須宣告支援在initialize請求期間取樣。閘道只會將取樣請求轉送給宣告此功能的用戶端。

抽樣的運作方式

當 MCP 伺服器目標在工具執行期間需要完成 LLM 時，它會傳送sampling/createMessage請求。閘道會將此請求轉送給用戶端做為 SSE 事件，取代請求 id。用戶端會叫用其語言模型，並將結果傳回至閘道，再將其轉送至目標。

抽樣請求包括：

messages — 要傳送至模型的對話訊息。
modelPreferences — 有關所需模型功能的選用提示（智慧、速度、成本）。
systemPrompt — 模型的選用系統提示。
maxTokens — 要產生的字符數目上限。

用戶端會以下列方式回應：

model — 使用的模型。
role — 一律為 assistant。
content — 產生的內容（文字或影像）。

注意

用戶端可完全控制要使用的模型，以及如何處理請求。伺服器modelPreferences是提示，而不是要求。用戶端也可以根據自己的政策修改或拒絕請求。

取樣流程

用戶端使用 Mcp-Session-Id標頭傳送tools/call請求。
Gateway 會將工具呼叫轉送至 MCP 伺服器目標。
目標會開啟 SSE 串流並傳送sampling/createMessage請求。
Gateway 會將取樣請求轉送給用戶端做為 SSE 事件，取代請求 id。
用戶端會使用提供的訊息叫用其語言模型。
用戶端會使用相同的 Mcp-Session-Id和id來自閘道請求的，傳送具有取樣結果的新請求。
Gateway 會將結果轉送至 MCP 伺服器目標。
目標會繼續處理並傳回最終工具結果。
Gateway 會將最終結果轉送至用戶端並關閉串流。

MCP 伺服器目標開發人員指南

重要

傳送取樣請求的 MCP 伺服器目標應將取樣呼叫包裝在 try-catch 區塊中，並處理用戶端不支援取樣的情況。如果閘道的用戶端未宣告取樣功能，閘道不會將其宣告至目標。如果目標仍然傳送取樣請求，閘道會將 -32601（找不到方法）錯誤傳回目標。

當無法使用取樣時，伺服器應實作備用路徑（例如使用內建模型或略過 AI 輔助步驟）。

錯誤處理

案例	錯誤	說明
當沒有等待取樣請求時，用戶端會傳送取樣回應	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


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


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"
))

您的瀏覽器已停用或無法使用 Javascript。

您必須啟用 Javascript，才能使用 AWS 文件。請參閱您的瀏覽器說明頁以取得說明。

文件慣用形式

記錄訊息

搜尋閘道工具