Recevez des notifications de progression depuis votre AgentCore passerelle
Les notifications de progression permettent aux cibles du serveur MCP de signaler la progression incrémentielle des appels d'outils de longue durée. Lorsqu'un appel d'outil prend du temps, le serveur peut envoyer notifications/progress des événements pour informer le client de l'état de l'opération. AgentCore Gateway transmet ces notifications des cibles du serveur MCP à votre client sous forme de fragments d' Server-Sent événements (SSE).
Conditions préalables
Pour recevoir des notifications de progression depuis votre passerelle, procédez comme suit :
-
Streaming de réponses activé — Les notifications de progression sont envoyées sous forme de fragments SSE lors d'une connexion ouverte. streamingConfiguration.enableResponseStreamingRéglez sur true celui de votre passerelleprotocolConfiguration.mcp.
-
Type de cible du serveur MCP : les notifications de progression proviennent des cibles du serveur MCP.
-
Le client envoie un Accept: text/event-stream en-tête : le client doit demander une réponse SSE pour recevoir des événements en streaming.
Comment fonctionnent les notifications de progression
Lorsqu'un client fait une tools/call demande avec un progressToken dans les paramètres de la demande, la cible du serveur MCP peut envoyer notifications/progress des événements pendant l'exécution. La passerelle transmet ces événements au client sous forme de fragments SSE avant le résultat final de l'outil.
Chaque notification de progression inclut :
-
progressToken— Correspond au jeton fourni dans la demande initiale.
-
progress— La valeur de progression actuelle (numérique).
-
total— Valeur totale facultative indiquant l'objectif d'achèvement.
-
message— Description lisible par l'homme facultative de l'état actuel.
Exemples de code
Exemple
- curl
-
-
Appelez un outil avec un jeton de progression :
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"
}
}
}'
La passerelle renvoie un flux SSE avec des notifications de progression suivies du résultat final :
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)
