View a markdown version of this page

從 AgentCore 閘道接收記錄訊息 - Amazon Bedrock AgentCore

本文為英文版的機器翻譯版本,如內容有任何歧義或不一致之處,概以英文版為準。

從 AgentCore 閘道接收記錄訊息

MCP 伺服器目標可以在工具執行期間,使用 notifications/message方法將日誌訊息傳送至用戶端。這些訊息可讓您即時了解伺服器正在執行的操作,有助於偵錯、稽核和監控工具行為。AgentCore Gateway 會將這些日誌通知從 MCP 伺服器目標轉送到您的用戶端,做為 Server-Sent Events (SSE) 區塊。

先決條件

若要從閘道接收記錄訊息:

  • 啟用回應串流 — 在開放連線期間,日誌訊息會以 SSE 區塊的形式傳遞。在閘道的 true中streamingConfiguration.enableResponseStreaming將 設定為 protocolConfiguration.mcp。

  • MCP 伺服器目標類型 — 日誌訊息來自 MCP 伺服器目標。

  • 用戶端傳送 Accept標頭 — 用戶端必須傳送 ,Accept: application/json, text/event-stream以便閘道可以傳回串流 (SSE) 回應。

日誌層級

MCP 會依嚴重性的增加順序定義下列日誌層級:

Level 說明

debug

故障診斷的詳細診斷資訊。

info

有關正常操作的一般資訊訊息。

notice

正常但重大的事件。

warning

不會阻止操作的潛在有害情況。

error

阻止特定操作的錯誤條件。

critical

需要立即注意的關鍵條件。

alert

必須立即採取動作。

emergency

系統無法使用。

記錄訊息的運作方式

當 MCP 伺服器目標在工具執行notifications/message期間發出 時,閘道會以 SSE 事件的形式將其轉送至用戶端。每個日誌訊息都包含:

  • level — 訊息的嚴重性等級。

  • logger — 識別來源元件的選用名稱。

  • data — 日誌內容 (字串或結構化物件)。

日誌訊息是資訊性的,不需要用戶端的回應。它們與其他 SSE 事件一起交付,例如進度通知和最終工具結果。

程式碼範例

範例
curl (2025-11-25 and earlier)

呼叫工具並在 SSE 串流中接收日誌訊息。將 MCP-Protocol-Version標頭設定為閘道支援的版本。

curl -N -X POST \ https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp \ -H "Content-Type: application/json" \ -H "Accept: application/json, text/event-stream" \ -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \ -H "MCP-Protocol-Version: 2025-11-25" \ -d '{ "jsonrpc": "2.0", "id": "tool-call-1", "method": "tools/call", "params": { "name": "deployService", "arguments": { "serviceName": "my-api", "environment": "staging" } } }'

閘道會傳回具有日誌訊息的 SSE 串流,後面接著最終結果:

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."}]}}
curl (2026-07-28)

在版本 上2026-07-28,用戶端會在 io.modelcontextprotocol/logLevel中設定 以選擇每個請求的日誌訊息 _meta(此版本已淘汰 logging/setLevel操作)。伺服器notifications/message只會在請求的層級或更高層級傳送。同時包含 Mcp-Method和 Mcp-Name request-metadata 標頭以及_meta版本欄位。MCP-Protocol-Version 標頭必須符合 _meta.io.modelcontextprotocol/protocolVersion。閘道的 supportedVersions 必須包含 2026-07-28。

curl -N -X POST \ https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp \ -H "Content-Type: application/json" \ -H "Accept: application/json, text/event-stream" \ -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \ -H "MCP-Protocol-Version: 2026-07-28" \ -H "Mcp-Method: tools/call" \ -H "Mcp-Name: deployService" \ -d '{ "jsonrpc": "2.0", "id": "tool-call-1", "method": "tools/call", "params": { "name": "deployService", "arguments": { "serviceName": "my-api", "environment": "staging" }, "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientInfo": { "name": "my-agent", "version": "1.0.0" }, "io.modelcontextprotocol/clientCapabilities": {}, "io.modelcontextprotocol/logLevel": "info" } } }'

閘道會傳回具有日誌訊息的 SSE 串流,後面接著最終結果:

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 (2025-11-25 and earlier)

將 MCP-Protocol-Version標頭設定為閘道支援的版本。

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": "application/json, text/event-stream", "Authorization": "Bearer YOUR_ACCESS_TOKEN", "MCP-Protocol-Version": "2025-11-25" } # 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
Python requests package (2026-07-28)

在版本 上2026-07-28,透過在 io.modelcontextprotocol/logLevel中設定 選擇加入日誌訊息 _meta(此版本已淘汰 logging/setLevel操作)。同時新增 MCP-Protocol-Version、 Mcp-Method和 Mcp-Name標頭以及_meta版本欄位。閘道的 supportedVersions 必須包含 2026-07-28。

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": "application/json, text/event-stream", "Authorization": "Bearer YOUR_ACCESS_TOKEN", "MCP-Protocol-Version": "2026-07-28", "Mcp-Method": "tools/call", "Mcp-Name": "deployService" } # 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"}, "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientInfo": {"name": "my-agent", "version": "1.0.0"}, "io.modelcontextprotocol/clientCapabilities": {}, "io.modelcontextprotocol/logLevel": "info" } } }, 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
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
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)