View a markdown version of this page

Activez le streaming de réponse MCP pour votre passerelle AgentCore - Base rocheuse de l'Amazonie AgentCore
Avantages du streaming de réponsesActivez le streaming des réponsesComment fonctionne le streaming des réponsesLes exigences du clientExemples de code

Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Activez le streaming de réponse MCP pour votre passerelle AgentCore

Le streaming de réponses MCP permet à votre AgentCore passerelle de fournir des Server-Sent événements en temps réel (SSE) aux clients pendant l'exécution de l'outil. Au lieu d'attendre la fin de l'appel à l'outil pour 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 d'élicitation et les demandes d'échantillonnage.

Avantages du streaming de réponses

Real-time feedback

Les clients reçoivent des mises à jour sur la progression et enregistrent les messages au fur et à mesure, au lieu 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 à la journalisation des messages. Ces fonctionnalités nécessitent une connexion SSE ouverte pour fournir des événements initiés par le serveur lors de l'exécution de l'outil.

Une meilleure expérience utilisateur pour les outils de longue durée

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

Activez le streaming des réponses

Pour activer le streaming des réponses, définissez sur streamingConfiguration.enableResponseStreaming 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 de réponses introduit une modification du contrat d'entrée de l'intercepteur de réponse. Si vous utilisez des intercepteurs de réponse, revoyez la logique de votre intercepteur pour vous assurer de la compatibilité avec les réponses en streaming. Consultez la section Intercepteurs de réponse avec diffusion en continu activée pour plus de détails.

Comment fonctionne le streaming des réponses

Lorsque le streaming de 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 diffusés au fur et à mesure de leur réception depuis la cible du serveur MCP.

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

Type d’événement Description

notifications/progress

La progression est mise à jour depuis la cible lors de l'exécution de l'outil. Voir Recevoir des notifications de progression.

notifications/message

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

elicitation/create

Demandes d'élicitation émanant de la cible demandant à l'utilisateur de saisir des informations. Voir Utiliser l'élicitation.

sampling/createMessage

Échantillonnage des 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, envoyé comme 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.

Les 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'à la réception du résultat final.

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