View a markdown version of this page

AgentCore 게이트웨이에서 로깅 메시지 수신 - Amazon Bedrock AgentCore
사전 조건로그 수준로깅 메시지 작동 방식코드 샘플

AgentCore 게이트웨이에서 로깅 메시지 수신

MCP 서버 대상은 notifications/message 메서드를 사용하여 도구 실행 중에 클라이언트에 로그 메시지를 보낼 수 있습니다. 이러한 메시지는 서버가 수행하는 작업에 대한 실시간 가시성을 제공하여 도구 동작을 디버깅, 감사 및 모니터링하는 데 유용합니다. AgentCore Gateway는 MCP 서버 대상의 이러한 로그 알림을 클라이언트에 SSE(Server-Sent Events) 청크로 전달합니다.

사전 조건

게이트웨이에서 로깅 메시지를 수신하려면:

  • 응답 스트리밍 활성화됨 - 열린 연결 중에 로그 메시지가 SSE 청크로 전달됩니다. 게이트웨이의 true에서를 streamingConfiguration.enableResponseStreaming로 설정합니다protocolConfiguration.mcp.

  • MCP 서버 대상 유형 - 로그 메시지는 MCP 서버 대상에서 비롯됩니다.

  • 클라이언트가 Accept: text/event-stream 헤더 전송 - 클라이언트가 스트리밍 이벤트를 수신하려면 SSE 응답을 요청해야 합니다.

로그 수준

MCP는 심각도가 증가하는 순서대로 다음과 같은 로그 수준을 정의합니다.

수준 설명

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)