View a markdown version of this page

Ricevi messaggi di registrazione dal tuo gateway AgentCore - Amazon Bedrock AgentCore
PrerequisitiLivelli di logCome funziona la registrazione dei messaggiEsempi di codice

Ricevi messaggi di registrazione dal tuo gateway AgentCore

Le destinazioni del server MCP possono inviare messaggi di registro ai client durante l'esecuzione dello strumento utilizzando questo metodo. notifications/message Questi messaggi forniscono visibilità in tempo reale su ciò che sta facendo il server, utile per il debug, il controllo e il monitoraggio del comportamento degli strumenti. AgentCore Gateway inoltra queste notifiche di registro dalle destinazioni del server MCP al client come Server-Sent blocchi di eventi (SSE).

Prerequisiti

Per ricevere messaggi di registrazione dal gateway:

  • Streaming di risposta abilitato: i messaggi di registro vengono consegnati come blocchi SSE durante una connessione aperta. Imposta streamingConfiguration.enableResponseStreaming su true nel tuo gateway. protocolConfiguration.mcp

  • Tipo di destinazione del server MCP: i messaggi di registro provengono dalle destinazioni del server MCP.

  • Il client invia l'Accept: text/event-streamintestazione: il client deve richiedere una risposta SSE per ricevere eventi in streaming.

Livelli di log

MCP definisce i seguenti livelli di registro, in ordine crescente di gravità:

Livello Description

debug

Informazioni diagnostiche dettagliate per la risoluzione dei problemi.

info

Messaggi informativi generali sul normale funzionamento.

notice

Eventi normali ma significativi.

warning

Situazioni potenzialmente dannose che non impediscono il funzionamento.

error

Condizioni di errore che hanno impedito un'operazione specifica.

critical

Condizioni critiche che richiedono attenzione immediata.

alert

L'azione deve essere intrapresa immediatamente.

emergency

Il sistema è inutilizzabile.

Come funziona la registrazione dei messaggi

Quando una destinazione del server MCP emette un messaggio notifications/message durante l'esecuzione dello strumento, il gateway lo inoltra al client come evento SSE. Ogni messaggio di registro include:

  • level— Il livello di gravità del messaggio.

  • logger— Nome opzionale che identifica il componente di origine.

  • data— Il contenuto del registro (stringa o oggetto strutturato).

I messaggi di registro sono informativi e non richiedono una risposta da parte del client. Vengono forniti insieme ad altri eventi SSE come le notifiche di avanzamento e il risultato finale dello strumento.

Esempi di codice

Esempio
curl

  1. Chiama uno strumento e ricevi messaggi di registro nel flusso 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"
      }
    }
}'

    Il gateway restituisce un flusso SSE con messaggi di registro seguiti dal risultato finale:

    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)