Erhalten Sie Fortschrittsbenachrichtigungen von Ihrem AgentCore Gateway
Mithilfe von Fortschrittsbenachrichtigungen können MCP-Serverziele schrittweise Fortschritte bei lang andauernden Toolaufrufen melden. Wenn der Abschluss eines Tool-Aufrufs einige Zeit in Anspruch nimmt, kann der Server notifications/progress Ereignisse senden, um den Client über den Status des Vorgangs auf dem Laufenden zu halten. AgentCore Gateway leitet diese Benachrichtigungen von MCP-Serverzielen als Server-Sent Event-Chunks (SSE) an Ihren Client weiter.
Voraussetzungen
So erhalten Sie Fortschrittsbenachrichtigungen von Ihrem Gateway:
-
Antwort-Streaming aktiviert — Fortschrittsbenachrichtigungen werden während einer offenen Verbindung als SSE-Chunks zugestellt. trueIn streamingConfiguration.enableResponseStreaming Ihren Gateways auf eingestellt. protocolConfiguration.mcp
-
MCP-Serverzieltyp — Fortschrittsbenachrichtigungen stammen von MCP-Serverzielen.
-
Client sendet Accept: text/event-stream Header — Der Client muss eine SSE-Antwort anfordern, um Streaming-Ereignisse zu empfangen.
Wie funktionieren Fortschrittsbenachrichtigungen
Wenn ein Client eine tools/call Anfrage mit einem progressToken in den Anforderungsparametern stellt, kann das MCP-Serverziel während der Ausführung notifications/progress Ereignisse senden. Das Gateway leitet diese Ereignisse vor dem endgültigen Tool-Ergebnis als SSE-Blöcke an den Client weiter.
Jede Fortschrittsmeldung beinhaltet:
-
progressToken— Entspricht dem in der ursprünglichen Anfrage angegebenen Token.
-
progress— Der aktuelle Fortschrittswert (numerisch).
-
total— Optionaler Gesamtwert, der das Abschlussziel angibt.
-
message— Optionale menschenlesbare Beschreibung des aktuellen Status.
Codebeispiele
Beispiel
- curl
-
-
Rufen Sie ein Tool mit einem Fortschrittstoken auf:
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"
},
"_meta": {
"progressToken": "progress-1"
}
}
}'
Das Gateway gibt einen SSE-Stream mit Fortschrittsbenachrichtigungen zurück, gefolgt vom Endergebnis:
event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"progress-1","progress":1,"total":4,"message":"Loading dataset..."}}
event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"progress-1","progress":2,"total":4,"message":"Running analysis..."}}
event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"progress-1","progress":3,"total":4,"message":"Generating report..."}}
event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"progress-1","progress":4,"total":4,"message":"Complete"}}
event: message
data: {"jsonrpc":"2.0","id":"tool-call-1","result":{"content":[{"type":"text","text":"Analysis complete. Found 3 anomalies in dataset ds-12345."}]}}
- Python requests package
-
-
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"
}
# Call tool with progress token (streaming response)
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"},
"_meta": {"progressToken": "progress-1"}
}
}, stream=True)
# Process SSE events
client = sseclient.SSEClient(response)
for event in client.events():
data = json.loads(event.data)
if data.get("method") == "notifications/progress":
params = data["params"]
print(f"Progress: {params['progress']}/{params.get('total', '?')} - {params.get('message', '')}")
elif "result" in data:
print(f"Tool result: {data['result']}")
break
- MCP Client
-
-
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
import asyncio
async def progress_handler(progress_token, progress, total, message=None):
"""Handle progress notifications."""
print(f"[{progress}/{total}] {message or ''}")
async def use_progress(url, token):
headers = {"Authorization": f"Bearer {token}"}
async with streamablehttp_client(url=url, headers=headers) as (
read_stream, write_stream, _
):
async with ClientSession(
read_stream, write_stream,
progress_handler=progress_handler
) as session:
await session.initialize()
# Call tool with progress token - notifications handled by callback
result = await session.call_tool(
name="analyzeDataset",
arguments={"datasetId": "ds-12345"}
)
print(f"Tool result: {result}")
return result
asyncio.run(use_progress(
url="https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp",
token="YOUR_ACCESS_TOKEN"
))
- Strands MCP Client
-
-
from mcp.client.streamable_http import streamablehttp_client
from strands import Agent
from strands.tools.mcp import MCPClient
mcp_url = "https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp"
access_token = "YOUR_ACCESS_TOKEN"
mcp_client = MCPClient(
lambda: streamablehttp_client(
mcp_url, headers={"Authorization": f"Bearer {access_token}"}
)
)
# Strands handles streaming and progress notifications automatically
with mcp_client:
agent = Agent(tools=mcp_client.list_tools_sync())
response = agent("Analyze dataset ds-12345")
print(response)
