本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

# 为您的 AgentCore 网关启用 MCP 响应流
<a name="gateway-mcp-streaming"></a>

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

## 响应流的好处
<a name="gateway-mcp-streaming-benefits"></a>

 **Real-time 反馈 **   
客户端会收到进度更新和日志消息，而不是等待完整的工具响应。

 **启用交互式 MCP 功能 **   
响应流是[激发](gateway-mcp-elicitation.md)、[采样](gateway-mcp-sampling.md)、[进度通知](gateway-mcp-progress.md)和[记录消息的先决条件。](gateway-mcp-logging.md)这些功能需要开放的 SSE 连接才能在工具执行期间传送服务器启动的事件。

 **为长时间运行的工具提供更好的用户体验 **   
对于需要几秒或几分钟才能完成的工具，直播可以让客户随时了解情况并做出响应。

## 启用响应直播
<a name="gateway-mcp-streaming-enable"></a>

要启用响应流，请在创建或更新网关时`true`在`protocolConfiguration.mcp`字段中设置`streamingConfiguration.enableResponseStreaming`为：

```
{
  "protocolConfiguration": {
    "mcp": {
      "streamingConfiguration": {
        "enableResponseStreaming": true
      }
    }
  }
}
```

**注意**  
启用响应流会对响应拦截器输入合约进行更改。如果您使用响应拦截器，请检查您的拦截器逻辑以确保与流式响应兼容。有关详细信息，请参阅启用了流式传输的[响应拦截器](gateway-interceptors-types.md#gateway-interceptors-types-streaming)。

## 响应流媒体的工作原理
<a name="gateway-mcp-streaming-how"></a>

启用响应流且客户端使用发送请求时`Accept: text/event-stream`，网关会返回 SSE 流，而不是单个 JSON 响应。从 MCP 服务器目标接收到事件时即进行传送。

SSE 直播可以包括以下事件类型：


| 事件类型 | 说明 | 
| --- | --- | 
|  `notifications/progress`  | 在工具执行期间从目标系统进行进度更新。请参阅[接收进度通知](gateway-mcp-progress.md)。 | 
|  `notifications/message`  | 记录来自目标的消息。请参阅[接收日志消息](gateway-mcp-logging.md)。 | 
|  `elicitation/create`  | 来自目标的引用请求要求用户输入。请参见[使用激励](gateway-mcp-elicitation.md)。 | 
|  `sampling/createMessage`  | 对来自目标的请求进行采样，要求完成 LLM。请参见[使用采样](gateway-mcp-sampling.md)。 | 
| 最终结果 | 工具调用结果，作为直播关闭前的最后一个事件传送。 | 

如果客户端不发送`Accept: text/event-stream`，则网关会缓冲响应，并在工具调用完成后返回单个 JSON 响应。在这种情况下，不会传送中间事件（进度、日志）。

## 客户端要求
<a name="gateway-mcp-streaming-client"></a>

要接收直播响应，客户必须：
+ 在他们的请求中发送`Accept: text/event-stream`标头。
+ 在 SSE 事件到达时对其进行处理，将每`data:`行解析为 JSON-RPC 消息。
+ 保持连接处于打开状态，直到收到最终结果事件。

## 代码示例
<a name="gateway-mcp-streaming-example"></a>

**Example**  

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."}]}}
   ```

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
   ```

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