Menerima pesan logging dari AgentCore gateway Anda

Target server MCP dapat mengirim pesan log ke klien selama eksekusi alat menggunakan notifications/message metode ini. Pesan-pesan ini memberikan visibilitas real-time ke dalam apa yang dilakukan server, berguna untuk debugging, audit, dan perilaku alat pemantauan. AgentCore Gateway meneruskan pemberitahuan log ini dari target server MCP ke klien Anda sebagai potongan Server-Sent Acara (SSE).

Prasyarat

Untuk menerima pesan logging dari gateway Anda:

• Streaming respons diaktifkan - Pesan log dikirimkan sebagai potongan SSE selama koneksi terbuka. Setel streamingConfiguration.enableResponseStreaming ke true dalam gateway AndaprotocolConfiguration.mcp.

• Jenis target server MCP - Pesan log berasal dari target server MCP.

• Klien mengirim Accept: text/event-stream header - Klien harus meminta respons SSE untuk menerima acara streaming.

Tingkat log

MCP mendefinisikan tingkat log berikut, dalam rangka meningkatkan keparahan:

Tingkat	Deskripsi
debug	Informasi diagnostik terperinci untuk pemecahan masalah.
info	Pesan informasi umum tentang operasi normal.
notice	Peristiwa normal tapi signifikan.
warning	Situasi yang berpotensi berbahaya yang tidak mencegah operasi.
error	Kondisi kesalahan yang mencegah operasi tertentu.
critical	Kondisi kritis yang membutuhkan perhatian segera.
alert	Tindakan harus segera diambil.
emergency	Sistem tidak dapat digunakan.

Cara kerja pesan logging

Ketika target server MCP memancarkan notifications/message selama eksekusi alat, gateway meneruskannya ke klien sebagai peristiwa SSE. Setiap pesan log meliputi:

• levelTingkat keparahan pesan.

• logger— Nama opsional yang mengidentifikasi komponen sumber.

• data— Konten log (string atau objek terstruktur).

Pesan log bersifat informasi dan tidak memerlukan respons dari klien. Mereka dikirimkan bersama acara SSE lainnya seperti pemberitahuan kemajuan dan hasil alat akhir.

Sampel Kode

contoh

curl

1. Panggil alat dan terima pesan log di aliran 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"
         }
       }
   }'

   Gateway mengembalikan aliran SSE dengan pesan log diikuti dengan hasil akhir:

   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)