View a markdown version of this page

为您的 AgentCore 网关启用 MCP 响应流 - 亚马逊基岩 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"
))