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: text/event-stream標頭 — 用戶端必須請求 SSE 回應才能接收串流事件。

日誌層級

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

Level 說明

debug

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

info

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

notice

正常但重大的事件。

warning

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

error

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

critical

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

alert

必須立即採取動作。

emergency

系統無法使用。

記錄訊息的運作方式

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

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

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

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

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

程式碼範例

範例
curl

  1. 呼叫工具並在 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"
      }
    }
}'

    閘道會傳回具有日誌訊息的 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

  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)