View a markdown version of this page

Aktivieren Sie das MCP-Antwort-Streaming für Ihr Gateway AgentCore - Amazon Grundgestein AgentCore
Vorteile von Response-StreamingAktivieren Sie das Antwort-StreamingSo funktioniert Response-StreamingAnforderungen des KundenCodebeispiele

Aktivieren Sie das MCP-Antwort-Streaming für Ihr Gateway AgentCore

MCP-Antwort-Streaming ermöglicht es Ihrem AgentCore Gateway, während der Ausführung des Tools Server-Sent Ereignisse (SSE) in Echtzeit an Clients zu senden. Anstatt zu warten, bis der gesamte Tool-Aufruf abgeschlossen ist, bevor eine Antwort zurückgegeben wird, streamt das Gateway Ereignisse, sobald sie auftreten — einschließlich Fortschrittsbenachrichtigungen, Protokollmeldungen, Auslöseanforderungen und Sampling-Anfragen.

Vorteile von Response-Streaming

Real-time Feedback

Kunden erhalten Statusaktualisierungen und Protokollmeldungen, sobald sie geschehen, anstatt auf die vollständige Antwort des Tools zu warten.

Aktiviert interaktive MCP-Funktionen

Das Streaming von Antworten ist eine Grundvoraussetzung für die Erfassung, Probenahme, Fortschrittsmeldungen und Protokollierung von Meldungen. Für diese Funktionen ist eine offene SSE-Verbindung erforderlich, um serverinitiierte Ereignisse während der Ausführung des Tools zu übertragen.

Bessere Benutzererfahrung für Tools mit langer Laufzeit

Bei Tools, deren Fertigstellung Sekunden oder Minuten dauert, sorgt das Streaming dafür, dass der Kunde stets auf dem Laufenden bleibt und schnell reagieren kann.

Aktivieren Sie das Antwort-Streaming

Um das Antwort-Streaming streamingConfiguration.enableResponseStreaming zu aktivieren, stellen Sie true bei der Erstellung oder Aktualisierung Ihres Gateways in dem protocolConfiguration.mcp Feld Folgendes ein:

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

Durch die Aktivierung des Antwort-Streamings wird der Eingabevertrag für den Response Interceptor geändert. Wenn Sie Antwort-Interzeptoren verwenden, überprüfen Sie Ihre Interceptor-Logik, um die Kompatibilität mit Streaming-Antworten sicherzustellen. Einzelheiten finden Sie unter Response Interceptors mit aktiviertem Streaming.

So funktioniert Response-Streaming

Wenn das Antwort-Streaming aktiviert ist und der Client eine Anfrage mit sendetAccept: text/event-stream, gibt das Gateway einen SSE-Stream statt einer einzelnen JSON-Antwort zurück. Ereignisse werden so zugestellt, wie sie vom MCP-Serverziel empfangen wurden.

Der SSE-Stream kann die folgenden Ereignistypen enthalten:

Ereignistyp Description

notifications/progress

Fortschrittsaktualisierungen vom Ziel während der Ausführung des Tools. Siehe Fortschrittsbenachrichtigungen erhalten.

notifications/message

Nachrichten vom Ziel protokollieren. Siehe Empfangen von Protokollnachrichten.

elicitation/create

Fordere Anfragen von der Zielperson auf, die nach Benutzereingaben fragen. Siehe Evizitation verwenden.

sampling/createMessage

Stichprobenanfragen von der Zielperson, in der ein LLM-Abschluss beantragt wird. Siehe Stichprobenverfahren verwenden.

Endgültiges Ergebnis

Das Ergebnis des Werkzeugaufrufs, das als letztes Ereignis vor dem Schließen des Streams geliefert wird.

Wenn der Client nicht sendetAccept: text/event-stream, puffert das Gateway die Antwort und gibt nach Abschluss des Werkzeugaufrufs eine einzige JSON-Antwort zurück. Zwischenereignisse (Fortschritt, Protokollierung) werden in diesem Fall nicht übermittelt.

Anforderungen des Kunden

Um Streaming-Antworten zu erhalten, müssen Kunden:

  • Senden Sie den Accept: text/event-stream Header in ihren Anfragen.

  • Behandelt SSE-Ereignisse, sobald sie eintreffen, und analysiert jede data: Zeile als JSON-RPC Nachricht.

  • Halten Sie die Verbindung offen, bis das Endergebnisereignis empfangen wird.

Codebeispiele

Beispiel
curl

  1. Senden Sie eine Tool-Call-Anfrage mit SSE-Accept-Header:

    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"
      }
    }
}'

    Beispiel für eine SSE-Stream-Antwort:

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