View a markdown version of this page

Aktifkan streaming respons MCP untuk gateway Anda AgentCore - Batuan Dasar Amazon AgentCore
Manfaat streaming responsAktifkan streaming responsCara kerja streaming responsPersyaratan klienSampel Kode

Aktifkan streaming respons MCP untuk gateway Anda AgentCore

Streaming respons MCP memungkinkan AgentCore gateway Anda untuk mengirimkan Server-Sent Acara real-time (SSE) ke klien selama eksekusi alat. Alih-alih menunggu seluruh panggilan alat selesai sebelum mengembalikan respons, gateway mengalirkan peristiwa saat terjadi — termasuk pemberitahuan kemajuan, pesan log, permintaan elisitasi, dan permintaan pengambilan sampel.

Manfaat streaming respons

Real-time umpan balik

Klien menerima pembaruan kemajuan dan mencatat pesan saat terjadi, daripada menunggu respons alat lengkap.

Mengaktifkan fitur MCP interaktif

Streaming respons adalah prasyarat untuk elisitasi, pengambilan sampel, pemberitahuankemajuan, dan pesan pencatatan. Fitur-fitur ini memerlukan koneksi SSE terbuka untuk mengirimkan peristiwa yang dimulai server selama eksekusi alat.

Pengalaman pengguna yang lebih baik untuk alat yang berjalan lama

Untuk alat yang membutuhkan waktu beberapa detik atau menit untuk diselesaikan, streaming membuat klien tetap terinformasi dan responsif.

Aktifkan streaming respons

Untuk mengaktifkan streaming respons, atur streamingConfiguration.enableResponseStreaming ke true di protocolConfiguration.mcp bidang saat membuat atau memperbarui gateway Anda:

{
  "protocolConfiguration": {
    "mcp": {
      "streamingConfiguration": {
        "enableResponseStreaming": true
      }
    }
  }
}
catatan

Mengaktifkan streaming respons memperkenalkan perubahan pada kontrak input pencegat respons. Jika Anda menggunakan pencegat respons, tinjau logika pencegat Anda untuk memastikan kompatibilitas dengan respons streaming. Lihat Pencegat respons dengan streaming diaktifkan untuk detailnya.

Cara kerja streaming respons

Saat streaming respons diaktifkan dan klien mengirimkan permintaanAccept: text/event-stream, gateway mengembalikan aliran SSE, bukan respons JSON tunggal. Acara dikirimkan saat diterima dari target server MCP.

Aliran SSE dapat mencakup jenis acara berikut:

Tipe peristiwa Deskripsi

notifications/progress

Kemajuan pembaruan dari target selama eksekusi alat. Lihat Menerima pemberitahuan kemajuan.

notifications/message

Log pesan dari target. Lihat Menerima pesan logging.

elicitation/create

Permintaan elisitasi dari target yang meminta masukan pengguna. Lihat Gunakan elisitasi.

sampling/createMessage

Permintaan pengambilan sampel dari target yang meminta penyelesaian LLM. Lihat Gunakan sampling.

Hasil akhir

Hasil panggilan alat, dikirim sebagai acara terakhir sebelum aliran ditutup.

Jika klien tidak mengirimAccept: text/event-stream, gateway menyangga respons dan mengembalikan respons JSON tunggal setelah panggilan alat selesai. Acara menengah (kemajuan, pencatatan) tidak dikirimkan dalam kasus ini.

Persyaratan klien

Untuk menerima tanggapan streaming, klien harus:

  • Kirim Accept: text/event-stream header dalam permintaan mereka.

  • Menangani peristiwa SSE saat mereka tiba, mengurai setiap data: baris sebagai JSON-RPC pesan.

  • Tetap buka koneksi sampai acara hasil akhir diterima.

Sampel Kode

contoh
curl

  1. Kirim permintaan panggilan alat dengan header terima 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"
      }
    }
}'

    Contoh respons aliran 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"
))