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，閘道會傳回 SSE 串流，而不是單一 JSON 回應。事件會在從 MCP 伺服器目標接收時交付。

SSE 串流可包含下列事件類型：

事件類型 說明

notifications/progress

工具執行期間目標的進度更新。請參閱接收進度通知。

notifications/message

從目標記錄訊息。請參閱接收記錄訊息。

elicitation/create

來自目標的請求要求使用者輸入。請參閱使用引出。

sampling/createMessage

取樣來自目標的請求，要求 LLM 完成。請參閱使用取樣。

最終結果

工具呼叫結果，在串流關閉之前做為最後一個事件傳送。

如果用戶端未傳送 Accept: text/event-stream，閘道會緩衝回應，並在工具呼叫完成後傳回單一 JSON 回應。在這種情況下，不會交付中繼事件 （進行中、記錄）。

用戶端需求

若要接收串流回應，用戶端必須：

  • 在其請求中傳送 Accept: text/event-stream標頭。

  • 處理到達時的 SSE 事件，將每data:行剖析為 JSON-RPC 訊息。

  • 保持連線開啟，直到收到最終結果事件為止。

程式碼範例

範例
curl

  1. 使用 SSE 接受標頭傳送工具呼叫請求：

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