View a markdown version of this page

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

Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Aktifkan streaming respons MCP untuk gateway Anda AgentCore

Streaming respons MCP memungkinkan AgentCore gateway Anda mengirimkan Per Server-Sent istiwa 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 elicitasi, dan permintaan pengambilan sampel.

Manfaat streaming respons

Real-time Umpan balik

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

Mengaktifkan fitur MCP interaktif

Streaming respons adalah prasyarat untuk el icitasi, pengambilan sampel, pemberitahuan kemajuan, dan pen catatan pesan. Fitur-fitur ini memerlukan koneksi SSE terbuka untuk mengirimkan peristiwa yang diprakarsai server selama eksekusi alat.

Pengalaman pengguna yang lebih baik untuk alat yang berjalan lama

Untuk alat yang membutuhkan waktu beberapa detik atau menit untuk menyelesaikannya, streaming membuat klien tetap mendapat informasi dan responsif.

Aktifkan streaming respons

Untuk mengaktifkan streaming respons, set streamingConfiguration.enableResponseStreaming el ke true dalam 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 Respons pencegat dengan streaming diaktifkan untuk detailnya.

Cara kerja streaming respons

Ketika streaming respons diaktifkan dan klien mengirim permintaan denganAccept: text/event-stream, gateway mengembalikan aliran SSE alih-alih respons JSON tunggal. Peristiwa dikirimkan saat diterima dari target server MCP.

Stream SSE dapat menyertakan jenis acara berikut:

Tipe peristiwa Deskripsi

notifications/progress

Pembaruan kemajuan dari target selama eksekusi alat. Lihat M enerima pemberitahuan kemajuan.

notifications/message

Pesan log dari target. Lihat M enerima pesan pencatatan.

elicitation/create

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

sampling/createMessage

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

Hasil akhir

Hasil panggilan alat, dikirimkan sebagai peristiwa terakhir sebelum streaming ditutup.

Jika klien tidak mengirimAccept: text/event-stream, gateway menyangga respons dan mengembalikan respons JSON tunggal setelah panggilan alat selesai. Peristiwa 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.

  • Tangani peristiwa SSE saat tiba, parsing setiap data: baris sebagai JSON-RPC pesan.

  • Biarkan koneksi tetap terbuka sampai acara hasil akhir diterima.

Sampel Kode

contoh
curl

  1. Kirim permintaan panggilan alat dengan header accept 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"
))