View a markdown version of this page

为您的 AgentCore 网关启用 MCP 响应流 - Amazon Bedrock AgentCore
响应流式传输的好处启用响应流响应流的工作原理客户端要求代码示例

为您的 AgentCore 网关启用 MCP 响应流

MCP 响应流使您的 AgentCore 网关能够在工具执行期间向客户端传送实时 Server-Sent 事件 (SSE)。网关不会等待整个工具调用完成后再返回响应，而是在事件发生时对事件进行流式传输，包括进度通知、日志消息、引发请求和采样请求。

响应流式传输的好处

Real-time 反馈

客户端会立即收到进度更新和日志消息，而不必等待完整的工具响应。

启用交互式 MCP 功能

响应流是引发、采样、进度通知和记录消息的先决条件。这些功能需要开放的 SSE 连接才能在工具执行期间传送服务器启动的事件。

为长时间运行的工具提供更好的用户体验

对于需要几秒钟或几分钟才能完成的工具，流媒体可以让客户了解情况并做出响应。

启用响应流

要启用响应流，请在创建或更新网关时true在protocolConfiguration.mcp字段中设置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"
))