View a markdown version of this page

Activez le streaming des réponses MCP pour votre passerelle AgentCore - Amazon Bedrock AgentCore
Avantages du streaming de réponsesActiver le streaming des réponsesComment fonctionne le streaming de réponsesExigences du clientExemples de code

Activez le streaming des réponses MCP pour votre passerelle AgentCore

Le streaming des réponses MCP permet à votre AgentCore passerelle de transmettre des Server-Sent événements en temps réel (SSE) aux clients lors de l'exécution de l'outil. Au lieu d'attendre la fin de l'appel à l'outil avant de renvoyer une réponse, la passerelle diffuse les événements au fur et à mesure qu'ils se produisent, notamment les notifications de progression, les messages de journal, les demandes de sollicitation et les demandes d'échantillonnage.

Avantages du streaming de réponses

Real-time commentaires

Les clients reçoivent des mises à jour sur la progression et consignent les messages au fur et à mesure, plutôt que d'attendre la réponse complète de l'outil.

Active les fonctionnalités MCP interactives

Le streaming des réponses est une condition préalable à l'élicitation, à l'échantillonnage, aux notifications de progression et à l'enregistrement des messages. Ces fonctionnalités nécessitent une connexion SSE ouverte pour transmettre des événements initiés par le serveur lors de l'exécution de l'outil.

Meilleure expérience utilisateur pour les outils de longue durée

Pour les outils dont l'exécution prend quelques secondes ou minutes, le streaming permet au client de rester informé et réactif.

Activer le streaming des réponses

Pour activer le streaming des réponses, définissez ce paramètre streamingConfiguration.enableResponseStreaming sur true dans le protocolConfiguration.mcp champ lors de la création ou de la mise à jour de votre passerelle :

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

L'activation du streaming des réponses introduit une modification du contrat d'entrée de l'intercepteur de réponse. Si vous utilisez des intercepteurs de réponse, passez en revue votre logique d'interception pour vous assurer de la compatibilité avec les réponses en streaming. Voir Intercepteurs de réponse avec diffusion en continu activée pour plus de détails.

Comment fonctionne le streaming de réponses

Lorsque le streaming des réponses est activé et que le client envoie une demande avecAccept: text/event-stream, la passerelle renvoie un flux SSE au lieu d'une seule réponse JSON. Les événements sont transmis au fur et à mesure qu'ils sont reçus de la cible du serveur MCP.

Le flux SSE peut inclure les types d'événements suivants :

Type d’événement Description

notifications/progress

Mises à jour de progression depuis la cible pendant l'exécution de l'outil. Consultez la section Recevoir des notifications de progression.

notifications/message

Enregistrez les messages provenant de la cible. Consultez la section Recevoir des messages de journalisation.

elicitation/create

Demandes d'élicitation de la part de la cible demandant l'avis de l'utilisateur. Voir Utiliser l'élicitation.

sampling/createMessage

Échantillonnage de demandes émanant de la cible demandant l'achèvement d'un LLM. Voir Utiliser l'échantillonnage.

Résultat final

Le résultat de l'appel à l'outil, fourni en tant que dernier événement avant la fermeture du flux.

Si le client n'envoie pasAccept: text/event-stream, la passerelle met la réponse en mémoire tampon et renvoie une seule réponse JSON une fois l'appel à l'outil terminé. Les événements intermédiaires (progression, journalisation) ne sont pas fournis dans ce cas.

Exigences du client

Pour recevoir des réponses en streaming, les clients doivent :

  • Envoyez l'Accept: text/event-streamen-tête dans leurs demandes.

  • Gérez les événements SSE au fur et à mesure qu'ils arrivent, en analysant chaque data: ligne comme un JSON-RPC message.

  • Maintenez la connexion ouverte jusqu'à ce que le résultat final soit reçu.

Exemples de code

Exemple
curl

  1. Envoyez une demande d'appel d'outil avec l'en-tête 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"
      }
    }
}'

    Exemple de réponse au flux 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"
))