View a markdown version of this page

Abilita lo streaming di risposta MCP per il tuo gateway AgentCore - Amazon Bedrock AgentCore
Vantaggi dello streaming delle risposteAbilita lo streaming delle risposteCome funziona lo streaming di rispostaRequisiti del clienteEsempi di codice

Abilita lo streaming di risposta MCP per il tuo gateway AgentCore

Lo streaming di risposta MCP consente al AgentCore gateway di fornire Server-Sent eventi (SSE) in tempo reale ai client durante l'esecuzione dello strumento. Invece di attendere il completamento dell'intera chiamata allo strumento prima di restituire una risposta, il gateway trasmette gli eventi man mano che si verificano, tra cui notifiche di avanzamento, messaggi di registro, richieste di elicitazione e richieste di campionamento.

Vantaggi dello streaming delle risposte

Real-time feedback

I clienti ricevono aggiornamenti sullo stato di avanzamento e messaggi di registro man mano che si verificano, anziché attendere la risposta completa dello strumento.

Abilita le funzionalità MCP interattive

Lo streaming delle risposte è un prerequisito per l'elicitazione, il campionamento, le notifiche di avanzamento e la registrazione dei messaggi. Queste funzionalità richiedono una connessione SSE aperta per fornire eventi avviati dal server durante l'esecuzione dello strumento.

Esperienza utente migliore per strumenti a lunga durata

Per strumenti che richiedono secondi o minuti per essere completati, lo streaming mantiene il cliente informato e reattivo.

Abilita lo streaming delle risposte

Per abilitare lo streaming delle risposte, imposta streamingConfiguration.enableResponseStreaming l'opzione true nel protocolConfiguration.mcp campo durante la creazione o l'aggiornamento del gateway:

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

L'abilitazione dello streaming di risposta introduce una modifica al contratto di input dell'intercettore di risposta. Se utilizzi intercettori di risposta, rivedi la logica dell'intercettore per garantire la compatibilità con le risposte in streaming. Vedi Intercettori di risposta con streaming abilitato per i dettagli.

Come funziona lo streaming di risposta

Quando lo streaming di risposta è abilitato e il client invia una richiesta conAccept: text/event-stream, il gateway restituisce un flusso SSE anziché una singola risposta JSON. Gli eventi vengono consegnati non appena vengono ricevuti dalla destinazione del server MCP.

Il flusso SSE può includere i seguenti tipi di eventi:

Tipo di evento Description

notifications/progress

Aggiornamenti dell'avanzamento dalla destinazione durante l'esecuzione dello strumento. Vedi Ricevere notifiche sullo stato di avanzamento.

notifications/message

Registra i messaggi dalla destinazione. Vedi Ricevere messaggi di registrazione.

elicitation/create

Richieste di sollecitazione da parte del target che richiedono l'input dell'utente. Vedi Use elicitation.

sampling/createMessage

Richieste di campionamento provenienti dal target che richiedono il completamento di un LLM. Vedi Utilizzare il campionamento.

Risultato finale

Il risultato della chiamata allo strumento, fornito come ultimo evento prima della chiusura dello stream.

Se il client non inviaAccept: text/event-stream, il gateway memorizza la risposta nel buffer e restituisce una singola risposta JSON al termine della chiamata allo strumento. In questo caso gli eventi intermedi (avanzamento, registrazione) non vengono forniti.

Requisiti del cliente

Per ricevere risposte in streaming, i clienti devono:

  • Inviare l'Accept: text/event-streamintestazione nelle loro richieste.

  • Gestisci gli eventi SSE non appena arrivano, analizzando ogni data: riga come messaggio. JSON-RPC

  • Mantieni aperta la connessione fino alla ricezione del risultato finale.

Esempi di codice

Esempio
curl

  1. Invia una richiesta di chiamata allo strumento con l'intestazione 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"
      }
    }
}'

    Esempio di risposta al flusso 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"
))