View a markdown version of this page

Habilite la transmisión de respuesta MCP para su puerta de enlace AgentCore - Amazon Bedrock AgentCore
Ventajas de la transmisión de respuestasHabilite la transmisión de respuestasCómo funciona la transmisión de respuestasRequisitos del cliente:Ejemplos de código

Habilite la transmisión de respuesta MCP para su puerta de enlace AgentCore

La transmisión de respuestas de MCP permite a su AgentCore pasarela enviar Server-Sent eventos (SSE) en tiempo real a los clientes durante la ejecución de la herramienta. En lugar de esperar a que se complete toda la llamada a la herramienta antes de devolver una respuesta, la pasarela transmite los eventos a medida que se producen, incluidas las notificaciones de progreso, los mensajes de registro, las solicitudes de obtención y las solicitudes de muestreo.

Ventajas de la transmisión de respuestas

Real-time comentarios

Los clientes reciben actualizaciones de progreso y mensajes de registro a medida que se producen, en lugar de esperar a que la herramienta les dé una respuesta completa.

Habilita funciones MCP interactivas

La transmisión de respuestas es un requisito previo para la obtención, el muestreo, las notificaciones de progreso y el registro de los mensajes. Estas funciones requieren una conexión SSE abierta para enviar los eventos iniciados por el servidor durante la ejecución de la herramienta.

Mejor experiencia de usuario para herramientas de larga duración

En el caso de las herramientas que tardan unos segundos o minutos en completarse, la transmisión mantiene al cliente informado y receptivo.

Habilite la transmisión de respuestas

Para habilitar la transmisión de respuestas, streamingConfiguration.enableResponseStreaming configúrelo true en el protocolConfiguration.mcp campo al crear o actualizar su puerta de enlace:

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

La activación de la transmisión de respuestas introduce un cambio en el contrato de entrada del interceptor de respuesta. Si utilizas interceptores de respuesta, revisa la lógica del interceptor para garantizar la compatibilidad con las respuestas de transmisión. Consulta los interceptores de respuesta con la transmisión habilitada para obtener más información.

Cómo funciona la transmisión de respuestas

Cuando la transmisión de respuestas está habilitada y el cliente envía una solicitud con ellaAccept: text/event-stream, la puerta de enlace devuelve una transmisión SSE en lugar de una única respuesta JSON. Los eventos se entregan tal como se reciben del servidor MCP de destino.

La transmisión SSE puede incluir los siguientes tipos de eventos:

Tipo de evento Description (Descripción)

notifications/progress

El progreso se actualiza desde el objetivo durante la ejecución de la herramienta. Consulte Recibir notificaciones de progreso.

notifications/message

Registra los mensajes del objetivo. Consulte Recibir mensajes de registro.

elicitation/create

Solicitudes de obtención del objetivo solicitando la participación del usuario. Consulte Elicitación de usuarios.

sampling/createMessage

Muestreo de solicitudes del objetivo que solicitan completar un LLM. Consulte Utilizar el muestreo.

Resultado final

El resultado de la llamada a la herramienta, que se entrega como último evento antes del cierre de la transmisión.

Si el cliente no envíaAccept: text/event-stream, la puerta de enlace almacena la respuesta en búfer y devuelve una única respuesta JSON una vez finalizada la llamada a la herramienta. En este caso, los eventos intermedios (progreso, registro) no se entregan.

Requisitos del cliente:

Para recibir respuestas en streaming, los clientes deben:

  • Enviar el Accept: text/event-stream encabezado en sus solicitudes.

  • Gestione los eventos de SSE a medida que llegan, analizando cada data: línea como un JSON-RPC mensaje.

  • Mantenga la conexión abierta hasta que se reciba el evento con el resultado final.

Ejemplos de código

ejemplo
curl

  1. Envíe una solicitud de llamada de herramienta con el encabezado 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"
      }
    }
}'

    Ejemplo de respuesta de transmisión 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"
))