Receba notificações de progresso do seu AgentCore gateway
As notificações de progresso permitem que os alvos do servidor MCP relatem o progresso incremental em chamadas de ferramentas de longa duração. Quando uma chamada de ferramenta demora para ser concluída, o servidor pode enviar notifications/progress eventos para manter o cliente informado sobre o status da operação. AgentCore O Gateway encaminha essas notificações dos alvos do servidor MCP para seu cliente como blocos de Server-Sent Eventos (SSE).
Pré-requisitos
Para receber notificações de progresso do seu gateway:
-
Streaming de resposta ativado — As notificações de progresso são entregues como partes SSE durante uma conexão aberta. streamingConfiguration.enableResponseStreamingDefina como true no seu gatewayprotocolConfiguration.mcp.
-
Tipo de alvo do servidor MCP — As notificações de progresso se originam dos alvos do servidor MCP.
-
O cliente envia o Accept: text/event-stream cabeçalho — O cliente deve solicitar uma resposta SSE para receber eventos de streaming.
Como as notificações de progresso funcionam
Quando um cliente faz uma tools/call solicitação com um progressToken nos parâmetros da solicitação, o destino do servidor MCP pode enviar notifications/progress eventos durante a execução. O gateway encaminha esses eventos para o cliente como partes SSE antes do resultado final da ferramenta.
Cada notificação de progresso inclui:
-
progressToken— Corresponde ao token fornecido na solicitação original.
-
progress— O valor do progresso atual (numérico).
-
total— Valor total opcional indicando a meta de conclusão.
-
message— Descrição opcional legível por humanos do status atual.
Exemplos de código
exemplo
- curl
-
-
Chame uma ferramenta com um token de progresso:
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"
}
}
}'
O gateway retorna um fluxo SSE com notificações de progresso seguidas pelo resultado 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)
