Receba mensagens de registro do seu AgentCore gateway

Os alvos do servidor MCP podem enviar mensagens de log aos clientes durante a execução da ferramenta usando o notifications/message método. Essas mensagens fornecem visibilidade em tempo real do que o servidor está fazendo, o que é útil para depuração, auditoria e monitoramento do comportamento da ferramenta. AgentCore O Gateway encaminha essas notificações de log dos alvos do servidor MCP para seu cliente como partes de Server-Sent Eventos (SSE).

Pré-requisitos

Para receber mensagens de registro do seu gateway:

• Streaming de resposta ativado — As mensagens de log 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 mensagens de log são originadas 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.

Níveis de logs

O MCP define os seguintes níveis de log, em ordem crescente de severidade:

Nível	Description
debug	Informações detalhadas de diagnóstico para solução de problemas.
info	Mensagens informativas gerais sobre a operação normal.
notice	Eventos normais, mas significativos.
warning	Situações potencialmente prejudiciais que não impedem a operação.
error	Condições de erro que impediram uma operação específica.
critical	Condições críticas que requerem atenção imediata.
alert	A ação deve ser tomada imediatamente.
emergency	O sistema está inutilizável.

Como as mensagens de registro funcionam

Quando um alvo do servidor MCP emite um notifications/message durante a execução da ferramenta, o gateway o encaminha para o cliente como um evento SSE. Cada mensagem de registro inclui:

• level— O nível de severidade da mensagem.

• logger— Nome opcional que identifica o componente de origem.

• data— O conteúdo do log (string ou objeto estruturado).

As mensagens de log são informativas e não exigem uma resposta do cliente. Eles são entregues junto com outros eventos da SSE, como notificações de progresso e o resultado final da ferramenta.

Exemplos de código

exemplo

curl

1. Chame uma ferramenta e receba mensagens de registro no fluxo SSE:

   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": "deployService",
         "arguments": {
           "serviceName": "my-api",
           "environment": "staging"
         }
       }
   }'

   O gateway retorna um fluxo SSE com mensagens de log seguidas pelo resultado final:

   event: message
   data: {"jsonrpc":"2.0","method":"notifications/message","params":{"level":"info","logger":"deploy-service","data":"Starting deployment of my-api to staging"}}
   
   event: message
   data: {"jsonrpc":"2.0","method":"notifications/message","params":{"level":"info","logger":"deploy-service","data":"Building container image..."}}
   
   event: message
   data: {"jsonrpc":"2.0","method":"notifications/message","params":{"level":"warning","logger":"deploy-service","data":"Deprecated configuration detected in service manifest"}}
   
   event: message
   data: {"jsonrpc":"2.0","method":"notifications/message","params":{"level":"info","logger":"deploy-service","data":"Deployment complete"}}
   
   event: message
   data: {"jsonrpc":"2.0","id":"tool-call-1","result":{"content":[{"type":"text","text":"Successfully deployed my-api to staging environment."}]}}

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"
   }
   
   # Call tool (streaming response)
   response = requests.post(gateway_url, headers=headers, json={
       "jsonrpc": "2.0",
       "id": "tool-call-1",
       "method": "tools/call",
       "params": {
           "name": "deployService",
           "arguments": {"serviceName": "my-api", "environment": "staging"}
       }
   }, stream=True)
   
   # Process SSE events
   client = sseclient.SSEClient(response)
   for event in client.events():
       data = json.loads(event.data)
       if data.get("method") == "notifications/message":
           params = data["params"]
           print(f"[{params['level'].upper()}] {params.get('logger', '')}: {params['data']}")
       elif "result" in data:
           print(f"Tool result: {data['result']}")
           break

MCP Client

1. from mcp import ClientSession
   from mcp.client.streamable_http import streamablehttp_client
   import asyncio
   
   async def log_handler(level, logger, data):
       """Handle log messages from the server."""
       print(f"[{level.upper()}] {logger or 'server'}: {data}")
   
   async def use_logging(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,
               logging_handler=log_handler
           ) as session:
               await session.initialize()
   
               # Call tool - log messages handled by callback
               result = await session.call_tool(
                   name="deployService",
                   arguments={"serviceName": "my-api", "environment": "staging"}
               )
               print(f"Tool result: {result}")
               return result
   
   asyncio.run(use_logging(
       url="https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp",
       token="YOUR_ACCESS_TOKEN"
   ))

Strands MCP Client

1. 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 log messages automatically
   with mcp_client:
       agent = Agent(tools=mcp_client.list_tools_sync())
       response = agent("Deploy my-api to staging")
       print(response)