Reciba mensajes de registro desde su AgentCore puerta de enlace

Los servidores MCP de destino pueden enviar mensajes de registro a los clientes durante la ejecución de la herramienta mediante notifications/message este método. Estos mensajes proporcionan visibilidad en tiempo real de lo que está haciendo el servidor, lo que resulta útil para depurar, auditar y supervisar el comportamiento de las herramientas. AgentCore Gateway reenvía estas notificaciones de registro desde los destinos del servidor MCP a su cliente en forma de fragmentos de Server-Sent eventos (SSE).

Requisitos previos

Para recibir mensajes de registro de su puerta de enlace:

• Transmisión de respuestas habilitada: los mensajes de registro se envían como fragmentos de SSE durante una conexión abierta. streamingConfiguration.enableResponseStreamingConfigúrelo true en la puerta de enlace. protocolConfiguration.mcp

• Tipo de destino del servidor MCP: los mensajes de registro se originan en los destinos del servidor MCP.

• El cliente envía el Accept: text/event-stream encabezado: el cliente debe solicitar una respuesta de SSE para recibir eventos de transmisión.

Niveles de registro

El MCP define los siguientes niveles de registro, en orden de gravedad creciente:

Nivel	Description (Descripción)
debug	Información de diagnóstico detallada para la solución de problemas.
info	Mensajes de información general sobre el funcionamiento normal.
notice	Acontecimientos normales pero significativos.
warning	Situaciones potencialmente dañinas que no impiden su funcionamiento.
error	Condiciones de error que impidieron una operación específica.
critical	Condiciones críticas que requieren atención inmediata.
alert	Se deben tomar medidas de inmediato.
emergency	El sistema no se puede utilizar.

Cómo funcionan los mensajes de registro

Cuando un servidor MCP objetivo emite un mensaje notifications/message durante la ejecución de la herramienta, la puerta de enlace lo reenvía al cliente como un evento SSE. Cada mensaje de registro incluye:

• level— El nivel de gravedad del mensaje.

• logger— Nombre opcional que identifica el componente de origen.

• data— El contenido del registro (cadena u objeto estructurado).

Los mensajes de registro son informativos y no requieren una respuesta del cliente. Se envían junto con otros eventos de la ESS, como las notificaciones de progreso y el resultado final de la herramienta.

Ejemplos de código

ejemplo

curl

1. Llame a una herramienta y reciba los mensajes de registro en la transmisión de la 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"
         }
       }
   }'

   La puerta de enlace devuelve un flujo de SSE con los mensajes de registro seguidos del 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)