4. Stream live results

In step 3 you called research-agent-01 with message/send and got a single reply. Real research takes time, though — and your agent already emits progress as it works. In this step you'll switch to message/stream and watch the task unfold live: status updates as the agent searches, reads, and summarizes, then the result artifact, then a terminal "done" — all over a single Server-Sent Events (SSE) connection.

How streaming works

message/send returns one reply on the same HTTP response. message/stream instead opens a long-lived text/event-stream connection: the gateway uses the agent's Agent Bridge (its per-agent virtual subscriber) as an SSE relay, forwarding each event the agent emits to you as it happens. You read frames until a terminal event closes the stream.

Each frame has an event: name and a JSON data: envelope. The agent's envelope type maps to the SSE event name you see:

The research-agent-01 agent from step 2 emits, in order: three task.status updates (searching → reading → summarizing), one task.artifact, then a terminal task.done. Stop reading on task.done (or task.error). The exact wire format and envelope shapes are in SSE behavior.

curl -N -X POST http://localhost:9090/a2a/research-agent-01 \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "message/stream",
    "params": {
      "message": {
        "parts": [{"text": "Summarize the history of message queues"}]
      }
    }
  }'

import asyncio
import json

import httpx
from httpx_sse import aconnect_sse

KUBEMQ_URL = "http://localhost:9090"
AGENT_ID = "research-agent-01"


async def main() -> None:
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "message/stream",
        "params": {
            "message": {"parts": [{"text": "Summarize the history of message queues"}]},
        },
    }

    async with httpx.AsyncClient(timeout=None) as client:
        print("Connecting to SSE stream...")
        async with aconnect_sse(
            client,
            "POST",
            f"{KUBEMQ_URL}/a2a/{AGENT_ID}",
            json=payload,
            headers={"Accept": "text/event-stream"},
        ) as event_source:
            async for event in event_source.aiter_sse():
                data = json.loads(event.data)
                print(f"[{event.event}] {json.dumps(data)}")
                # task.done and task.error are terminal — stop reading.
                if event.event in ("task.done", "task.error"):
                    break

    print("Stream complete.")


if __name__ == "__main__":
    asyncio.run(main())

Connecting to SSE stream...
[task.status] {"type": "status_update", "payload": {"status": "searching", "progress": 1, "total": 3}}
[task.status] {"type": "status_update", "payload": {"status": "reading", "progress": 2, "total": 3}}
[task.status] {"type": "status_update", "payload": {"status": "summarizing", "progress": 3, "total": 3}}
[task.artifact] {"type": "artifact", "payload": {"name": "summary.json", "data": {"summary": "Research summary for: Summarize the history of message queues", "sources": 3}}}
[task.done] {"type": "done", "payload": {"final_result": "completed", "event_count": 4}}
Stream complete.

Keepalive, idle timeout & disconnect

A few stream behaviors matter once you build real callers — all relayed by the gateway, not the agent:

• Keepalive — during quiet periods the gateway emits an SSE comment line (: keepalive) every 30 seconds so proxies don't drop the idle socket. It's a comment, not an event; SSE client libraries (including httpx-sse) ignore it. If you parse the stream by hand, skip any line starting with :.
• Idle timeout — if the agent emits no events for MaxSSEIdleSeconds (default 300s), the gateway closes the stream with a terminal task.error (code -32001, "stream idle timeout"). The timer measures the gap between events, so a long task stays alive as long as it heartbeats with periodic task.status frames.
• Auto-cancel on disconnect — if you close the connection before a terminal event, the gateway detects it and cancels the work on the agent (via a stream_cancel to the Agent Bridge) rather than letting it run to completion. Closing your SSE reader is a real cancellation signal — it frees the agent's concurrency slot promptly.

Next step

You've watched a task stream live from task.status through task.artifact to a terminal task.done. So far every call has been a hand-written HTTP request. In the final step, an LLM does the discovering and invoking for you — over MCP, through the same fabric.