View a markdown version of this page

AgentCore 게이트웨이에서 진행률 알림 수신 - Amazon Bedrock AgentCore
사전 조건진행률 알림 작동 방식코드 샘플

AgentCore 게이트웨이에서 진행률 알림 수신

진행률 알림을 통해 MCP 서버 대상은 장기 실행 도구 호출에 대한 증분 진행 상황을 보고할 수 있습니다. 도구 호출을 완료하는 데 시간이 걸리면 서버는 notifications/progress 이벤트를 전송하여 클라이언트에게 작업 상태를 알릴 수 있습니다. AgentCore Gateway는 MCP 서버 대상에서 클라이언트로 이러한 알림을 SSE(Server-Sent Events) 청크로 전달합니다.

사전 조건

게이트웨이에서 진행률 알림을 수신하려면:

  • 응답 스트리밍 활성화됨 - 진행률 알림은 열린 연결 중에 SSE 청크로 전달됩니다. 게이트웨이의 true에서를 streamingConfiguration.enableResponseStreaming로 설정합니다protocolConfiguration.mcp.

  • MCP 서버 대상 유형 - 진행률 알림은 MCP 서버 대상에서 시작됩니다.

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

진행률 알림 작동 방식

클라이언트가 tools/call 요청 파라미터progressToken에서를 사용하여 요청하면 MCP 서버 대상이 실행 중에 notifications/progress 이벤트를 보낼 수 있습니다. 게이트웨이는 최종 도구 결과 전에 이러한 이벤트를 SSE 청크로 클라이언트에 전달합니다.

각 진행 상황 알림에는 다음이 포함됩니다.

  • progressToken - 원래 요청에 제공된 토큰과 일치합니다.

  • progress - 현재 진행률 값(숫자).

  • total - 완료 대상을 나타내는 선택적 총 값입니다.

  • message - 현재 상태에 대한 사람이 읽을 수 있는 선택적 설명입니다.

코드 샘플

예
curl

  1. 진행 토큰을 사용하여 도구를 호출합니다.

    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": "analyzeDataset",
      "arguments": {
        "datasetId": "ds-12345"
      },
      "_meta": {
        "progressToken": "progress-1"
      }
    }
}'

    게이트웨이는 진행률 알림과 함께 SSE 스트림을 반환한 다음 최종 결과를 반환합니다.

    event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"progress-1","progress":1,"total":4,"message":"Loading dataset..."}}

event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"progress-1","progress":2,"total":4,"message":"Running analysis..."}}

event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"progress-1","progress":3,"total":4,"message":"Generating report..."}}

event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"progress-1","progress":4,"total":4,"message":"Complete"}}

event: message
data: {"jsonrpc":"2.0","id":"tool-call-1","result":{"content":[{"type":"text","text":"Analysis complete. Found 3 anomalies in dataset ds-12345."}]}}
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 with progress token (streaming response)
response = requests.post(gateway_url, headers=headers, json={
    "jsonrpc": "2.0",
    "id": "tool-call-1",
    "method": "tools/call",
    "params": {
        "name": "analyzeDataset",
        "arguments": {"datasetId": "ds-12345"},
        "_meta": {"progressToken": "progress-1"}
    }
}, stream=True)

# Process SSE events
client = sseclient.SSEClient(response)
for event in client.events():
    data = json.loads(event.data)
    if data.get("method") == "notifications/progress":
        params = data["params"]
        print(f"Progress: {params['progress']}/{params.get('total', '?')} - {params.get('message', '')}")
    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 progress_handler(progress_token, progress, total, message=None):
    """Handle progress notifications."""
    print(f"[{progress}/{total}] {message or ''}")

async def use_progress(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,
            progress_handler=progress_handler
        ) as session:
            await session.initialize()

            # Call tool with progress token - notifications handled by callback
            result = await session.call_tool(
                name="analyzeDataset",
                arguments={"datasetId": "ds-12345"}
            )
            print(f"Tool result: {result}")
            return result

asyncio.run(use_progress(
    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 progress notifications automatically
with mcp_client:
    agent = Agent(tools=mcp_client.list_tools_sync())
    response = agent("Analyze dataset ds-12345")
    print(response)