# 4. Stream live results (/aiway/tutorial/stream-live-results)
In [step 3](/aiway/tutorial/discover-and-invoke) 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.
This step assumes KubeMQ is running on port `9090` and `research-agent-01` is registered
and listening (Steps 1 and 2). All calls go to the **Aiway gateway** on `9090`, not to the
agent's own port — the gateway relays the agent's events back to you.
## How streaming works [#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](/aiway/a2a/architecture) (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:
| SSE event | Envelope `type` | Meaning | Terminal |
| --------------- | --------------- | ---------------------------------------------- | -------- |
| `task.status` | `status_update` | Progress update (non-terminal) | No |
| `task.artifact` | `artifact` | A partial or complete result artifact | No |
| `task.done` | `done` | The task completed successfully | Yes |
| `task.error` | `error` | The task failed (carries `code` and `message`) | Yes |
The `research-agent-01` agent from [step 2](/aiway/tutorial/build-and-register-an-agent)
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](/aiway/a2a/guides/sse-behavior).
## Open a stream [#open-a-stream]
Send a `message/stream` request to `POST /a2a/research-agent-01` with
`Accept: text/event-stream`. The body is a JSON-RPC 2.0 envelope identical to `message/send`,
just with the `message/stream` method. (You can also open a stream with
`GET /a2a/research-agent-01/stream` — the POST form is the common case.)
```bash
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"}]
}
}
}'
```
The `-N` flag disables curl's output buffering so frames print the instant they arrive.
This uses `httpx` with the `httpx-sse` helper for clean SSE parsing — install both with
`pip install httpx httpx-sse`.
```python
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())
```
The `httpx-sse` library skips keepalive comment lines for you and yields one object per
event with `.event` (the name) and `.data` (the JSON payload).
## Read the live event sequence [#read-the-live-event-sequence]
As the agent works, the frames arrive in order. You'll see the three status updates, then
the artifact, then the terminal `task.done`:
```text
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.
```
Mapped to the event table above:
* **`task.status`** (`status_update`) — non-terminal progress. The agent emits one per
stage (`searching`, `reading`, `summarizing`) with `progress`/`total` so a UI can show a
bar. There can be any number of these.
* **`task.artifact`** (`artifact`) — the result payload. The envelope carries a `name`
(`summary.json`) and a `data` object with the actual result. An agent can emit several
artifacts; the [artifact envelope is defined in SSE behavior](/aiway/a2a/guides/sse-behavior).
* **`task.done`** (`done`) — **terminal**. After this the gateway closes the connection.
A failed task ends with `task.error` instead, carrying a `code` and `message`. Always
break on either.
## Keepalive, idle timeout & disconnect [#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.
For the full streaming walkthrough, see [Streaming (SSE)](/aiway/a2a/streaming). For
the exact wire format, the artifact envelope, keepalive cadence, idle-timeout behavior, and
disconnect cancellation, see [SSE behavior](/aiway/a2a/guides/sse-behavior).
## Next step [#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.