View a markdown version of this page

AgentCore ゲートウェイでサンプリングを使用する - Amazon Bedrock AgentCore
前提条件サンプリングの仕組みサンプリングフローMCP サーバーターゲット開発者向けガイダンスエラー処理トラブルシューティングコードサンプル

AgentCore ゲートウェイでサンプリングを使用する

サンプリングは、 MCP サーバーがツール呼び出し中にクライアントから LLM の完了をリクエストできるようにする MCP 機能です。これにより、サーバーは言語モデルに直接アクセスすることなく 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 アシストステップのスキップなど）。

エラー処理

シナリオ エラー 説明

保留中のサンプリングリクエストがない場合にクライアントがサンプリングレスポンスを送信する

JSON-RPC -32600 (無効なリクエスト)

このセッションに一致するサンプリングリクエストが見つかりませんでした。

クライアントidが保留中のリクエストと一致しない でサンプリングレスポンスを送信する

JSON-RPC -32600 (無効なリクエスト)

は、sampling/createMessageリクエストでゲートウェイによって送信されたものと一致するid必要があります。

MCP サーバーがサンプリングリクエストを送信したが、ゲートウェイがサポートを宣言しなかった

JSON-RPC -32601 (メソッドが見つかりません)

MCP サーバーターゲットに戻ります。トラブルシューティングを参照してください。

トラブルシューティング

エラー: "Error calling tool 'sample_tool': Method not found: 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"
))