View a markdown version of this page

AgentCore 게이트웨이에 대한 MCP 응답 스트리밍 활성화 - Amazon Bedrock AgentCore
응답 스트리밍의 이점응답 스트리밍 활성화응답 스트리밍 작동 방식클라이언트 요구 사항코드 샘플

AgentCore 게이트웨이에 대한 MCP 응답 스트리밍 활성화

MCP 응답 스트리밍을 사용하면 AgentCore 게이트웨이가 도구 실행 중에 클라이언트에 실시간 서버 전송 이벤트(SSE)를 제공할 수 있습니다. 게이트웨이는 응답을 반환하기 전에 전체 도구 호출이 완료될 때까지 기다리는 대신 진행률 알림, 로그 메시지, 유도 요청, 샘플링 요청 등의 이벤트를 스트리밍합니다.

응답 스트리밍의 이점

실시간 피드백

클라이언트는 전체 도구 응답을 기다리지 않고 진행 상황 업데이트 및 로그 메시지를 수신합니다.

대화형 MCP 기능 활성화

응답 스트리밍은 메시지를 유도, 샘플링, 진행률 알림 및 로깅하기 위한 사전 조건입니다. 이러한 기능을 사용하려면 도구 실행 중에 서버 시작 이벤트를 전달하기 위해 열린 SSE 연결이 필요합니다.

장기 실행 도구를 위한 사용자 경험 개선

몇 초 또는 몇 분 정도 걸리는 도구의 경우 스트리밍은 클라이언트에 정보를 제공하고 응답합니다.

응답 스트리밍 활성화

응답 스트리밍을 활성화하려면 게이트웨이true를 생성하거나 업데이트할 때 protocolConfiguration.mcp 필드에서를 streamingConfiguration.enableResponseStreaming로 설정합니다.

{
  "protocolConfiguration": {
    "mcp": {
      "streamingConfiguration": {
        "enableResponseStreaming": true
      }
    }
  }
}
참고

응답 스트리밍을 활성화하면 응답 인터셉터 입력 계약이 변경됩니다. 응답 인터셉터를 사용하는 경우 인터셉터 로직을 검토하여 스트리밍 응답과의 호환성을 확인합니다. 자세한 내용은 스트리밍이 활성화된 응답 인터셉터를 참조하세요.

응답 스트리밍 작동 방식

응답 스트리밍이 활성화되고 클라이언트가 로 요청을 보내면 Accept: text/event-stream게이트웨이는 단일 JSON 응답 대신 SSE 스트림을 반환합니다. 이벤트는 MCP 서버 대상에서 수신될 때 전달됩니다.

SSE 스트림에는 다음 이벤트 유형이 포함될 수 있습니다.

이벤트 유형 설명

notifications/progress

도구 실행 중 대상의 진행 상황 업데이트. 진행률 알림 수신을 참조하세요.

notifications/message

대상의 메시지를 로깅합니다. 로깅 메시지 수신을 참조하세요.

elicitation/create

사용자 입력을 요청하는 대상의 유도 요청입니다. 유도 사용을 참조하세요.

sampling/createMessage

LLM 완료를 요청하는 대상의 샘플링 요청입니다. 샘플링 사용을 참조하세요.

최종 결과

스트림이 닫히기 전에 마지막 이벤트로 전달되는 도구 호출 결과입니다.

클라이언트가를 전송하지 않으면 게이트웨이Accept: text/event-stream는 응답을 버퍼링하고 도구 호출이 완료된 후 단일 JSON 응답을 반환합니다. 이 경우 중간 이벤트(진행, 로깅)는 전송되지 않습니다.

클라이언트 요구 사항

스트리밍 응답을 수신하려면 클라이언트가 다음을 수행해야 합니다.

  • 요청에서 Accept: text/event-stream 헤더를 전송합니다.

  • 각 data: 줄을 JSON-RPC 메시지로 구문 분석하여 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": "analyzeDataset",
      "arguments": {
        "datasetId": "ds-12345"
      }
    }
}'

    SSE 스트림 응답의 예:

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

event: message
data: {"jsonrpc":"2.0","method":"notifications/message","params":{"level":"info","logger":"analyzer","data":"Processing 10,000 records"}}

event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"auto-1","progress":2,"total":3,"message":"Analyzing..."}}

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

event: message
data: {"jsonrpc":"2.0","id":"tool-call-1","result":{"content":[{"type":"text","text":"Analysis complete. Found 3 anomalies."}]}}
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"
}

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"}
    }
}, stream=True)

client = sseclient.SSEClient(response)
for event in client.events():
    data = json.loads(event.data)
    method = data.get("method")

    if method == "notifications/progress":
        params = data["params"]
        print(f"Progress: {params['progress']}/{params.get('total', '?')} - {params.get('message', '')}")
    elif method == "notifications/message":
        params = data["params"]
        print(f"[{params['level'].upper()}] {params['data']}")
    elif "result" in data:
        print(f"Final result: {data['result']}")
        break
MCP Client

  1. from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
import asyncio

async def use_streaming(url, token):
    headers = {"Authorization": f"Bearer {token}"}

    # The MCP client uses streamable HTTP transport which handles SSE automatically
    async with streamablehttp_client(url=url, headers=headers) as (
        read_stream, write_stream, _
    ):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()

            # Tool calls automatically receive streaming events
            result = await session.call_tool(
                name="analyzeDataset",
                arguments={"datasetId": "ds-12345"}
            )
            print(f"Tool result: {result}")
            return result

asyncio.run(use_streaming(
    url="https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp",
    token="YOUR_ACCESS_TOKEN"
))