View a markdown version of this page

AgentCore ゲートウェイの MCP レスポンスストリーミングを有効にする - Amazon Bedrock AgentCore
レスポンスストリーミングの利点レスポンスストリーミングを有効にするレスポンスストリーミングの仕組みクライアント要件コードサンプル

AgentCore ゲートウェイの MCP レスポンスストリーミングを有効にする

MCP レスポンスストリーミングにより、AgentCore ゲートウェイはツールの実行中にリアルタイムサーバー送信イベント (SSE) をクライアントに配信できます。ツール呼び出し全体が完了するのを待ってからレスポンスを返す代わりに、ゲートウェイは進行状況通知、ログメッセージ、誘発リクエスト、サンプリングリクエストなど、発生したイベントをストリーミングします。

レスポンスストリーミングの利点

リアルタイムのフィードバック

クライアントは、ツールの完全な応答を待つのではなく、進行状況の更新とログメッセージを発生時に受け取ります。

インタラクティブ MCP 機能を有効にする

レスポンスストリーミングは、メッセージの誘発、サンプリング、進行状況通知、ログ記録の前提条件です。これらの機能を使用するには、ツールの実行中にサーバーが開始したイベントを配信するためのオープン SSE 接続が必要です。

長時間実行されるツールのユーザーエクスペリエンスの向上

完了までに数秒または数分かかるツールの場合、ストリーミングはクライアントに情報を提供し、応答性を維持します。

レスポンスストリーミングを有効にする

レスポンスストリーミングを有効にするには、ゲートウェイを作成または更新するときに protocolConfiguration.mcpフィールドtrueで streamingConfiguration.enableResponseStreaming を に設定します。

{
  "protocolConfiguration": {
    "mcp": {
      "streamingConfiguration": {
        "enableResponseStreaming": true
      }
    }
  }
}
注記

レスポンスストリーミングを有効にすると、レスポンスインターセプター入力契約が変更されます。レスポンスインターセプターを使用する場合は、インターセプターロジックを確認して、ストリーミングレスポンスとの互換性を確認します。詳細については、「ストリーミングが有効になっているレスポンスインターセプター」を参照してください。

レスポンスストリーミングの仕組み

レスポンスストリーミングが有効で、クライアントが でリクエストを送信するとAccept: text/event-stream、ゲートウェイは単一の JSON レスポンスではなく SSE ストリームを返します。イベントは、MCP サーバーターゲットから受信したときに配信されます。

SSE ストリームには、次のイベントタイプを含めることができます。

イベントタイプ 説明

notifications/progress

ツール実行中のターゲットからの進捗状況の更新。「進行状況通知の受信」を参照してください。

notifications/message

ターゲットからのメッセージをログに記録します。「ログメッセージの受信」を参照してください。

elicitation/create

ユーザー入力を求めるターゲットからの誘発リクエスト。「誘発を使用する」を参照してください。

sampling/createMessage

LLM の完了を求めるターゲットからのサンプリングリクエスト。「サンプリングを使用する」を参照してください。

最終結果

ストリームが終了する前に最後のイベントとして配信されるツール呼び出し結果。

クライアントが を送信しない場合Accept: text/event-stream、ゲートウェイはレスポンスをバッファし、ツール呼び出しの完了後に 1 つの JSON レスポンスを返します。この場合、中間イベント (進行状況、ログ記録) は配信されません。

クライアント要件

ストリーミングレスポンスを受信するには、クライアントは以下を行う必要があります。

  • リクエストで Accept: text/event-streamヘッダーを送信します。

  • 到着時に SSE イベントを処理し、各data:行を JSON-RPC メッセージとして解析します。

  • 最終結果イベントを受信するまで接続を開いたままにします。

コードサンプル

例
curl

  1. SSE accept ヘッダーを使用してツール呼び出しリクエストを送信します。

    curl -N -X POST \
  https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -d '{
    "jsonrpc": "2.0",
    "id": "tool-call-1",
    "method": "tools/call",
    "params": {
      "name": "analyzeDataset",
      "arguments": {
        "datasetId": "ds-12345"
      }
    }
}'

    SSE ストリームレスポンスの例:

    event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"auto-1","progress":1,"total":3,"message":"Loading data..."}}

event: message
data: {"jsonrpc":"2.0","method":"notifications/message","params":{"level":"info","logger":"analyzer","data":"Processing 10,000 records"}}

event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"auto-1","progress":2,"total":3,"message":"Analyzing..."}}

event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"auto-1","progress":3,"total":3,"message":"Complete"}}

event: message
data: {"jsonrpc":"2.0","id":"tool-call-1","result":{"content":[{"type":"text","text":"Analysis complete. Found 3 anomalies."}]}}
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"
}

response = requests.post(gateway_url, headers=headers, json={
    "jsonrpc": "2.0",
    "id": "tool-call-1",
    "method": "tools/call",
    "params": {
        "name": "analyzeDataset",
        "arguments": {"datasetId": "ds-12345"}
    }
}, stream=True)

client = sseclient.SSEClient(response)
for event in client.events():
    data = json.loads(event.data)
    method = data.get("method")

    if method == "notifications/progress":
        params = data["params"]
        print(f"Progress: {params['progress']}/{params.get('total', '?')} - {params.get('message', '')}")
    elif method == "notifications/message":
        params = data["params"]
        print(f"[{params['level'].upper()}] {params['data']}")
    elif "result" in data:
        print(f"Final result: {data['result']}")
        break
MCP Client

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

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

    # The MCP client uses streamable HTTP transport which handles SSE automatically
    async with streamablehttp_client(url=url, headers=headers) as (
        read_stream, write_stream, _
    ):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()

            # Tool calls automatically receive streaming events
            result = await session.call_tool(
                name="analyzeDataset",
                arguments={"datasetId": "ds-12345"}
            )
            print(f"Tool result: {result}")
            return result

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