View a markdown version of this page

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

Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

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 in tempo reale (SSE) ai clienti durante l'esecuzione degli strumenti. 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 di risposte

Real-time feedback

I clienti ricevono aggiornamenti sullo stato di avanzamento e registrano i messaggi non appena 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. Ricevi messaggi di registrazione dal tuo gateway AgentCore Queste funzionalità richiedono una connessione SSE aperta per fornire eventi avviati dal server durante l'esecuzione dello strumento.

Migliore esperienza utente per strumenti di lunga durata

Per gli strumenti che richiedono pochi 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 questa opzione true nel protocolConfiguration.mcp campo durante la creazione o l'aggiornamento del gateway:

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

L'attivazione dello streaming di risposta introduce una modifica al contratto di input dell'intercettore di risposta. Se utilizzi intercettori di risposta, rivedi la logica degli intercettori per garantire la compatibilità con le risposte in streaming. Per ulteriori informazioni, consulta Intercettori di risposta con streaming abilitato.

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 dal server di destinazione MCP.

Lo stream SSE può includere i seguenti tipi di eventi:

Tipo di evento Description

notifications/progress

Aggiornamenti dei progressi dal target 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 dal target che richiedono l'input dell'utente. Vedi Usa l'elicitazione.

sampling/createMessage

Campionamento delle richieste provenienti dal target che richiedono il completamento di un LLM. Vedi Usa 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 nel buffer la risposta e restituisce una singola risposta JSON al termine della chiamata allo strumento. Gli eventi intermedi (avanzamento, registrazione) non vengono forniti in questo caso.

Requisiti del cliente

Per ricevere risposte in streaming, i clienti devono:

  • Invia l'Accept: text/event-streamintestazione delle loro richieste.

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

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

Esempi di codice

Esempio
curl

  1. Inviate 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 allo stream 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"
))