# SSE Behavior (/aiway/a2a/guides/sse-behavior) A2A streaming rides on **Server-Sent Events (SSE)**: the gateway proxies a long-lived `text/event-stream` response from the agent back to the caller. This guide is the ground-truth reference for *how that stream behaves on the wire* — the exact event names, the JSON envelope each event carries, the keepalive cadence, when the gateway closes a stream, and what happens when the caller hangs up. For the task-level walkthrough of *using* streaming, see [Streaming (SSE)](/aiway/a2a/streaming). This page focuses on protocol details you need when writing or debugging an SSE client. ## Overview [#overview] A stream is opened either by `POST /a2a/:agent_id` with `method: "message/stream"`, or by `GET /a2a/:agent_id/stream`. In both cases the gateway responds with `Content-Type: text/event-stream` and relays events the agent emits, framed in the standard SSE format, until a terminal event arrives or the stream is torn down. The gateway is a transparent relay: the agent produces the events, KubeMQ's per-agent **virtual subscriber** bridges them from the agent's HTTP SSE connection onto a temporary Events channel (`_AGENTS_.stream/`), and the A2A connector replays them to the caller. The caller only ever sees a normal SSE stream. ## How it works [#how-it-works] The diagram below traces a single streaming request through the gateway and virtual subscriber and back to the caller's SSE reader. *The gateway and virtual subscriber relay the agent's SSE events to the caller; a `task.done` or `task.error` envelope ends the stream.* ## Wire format [#wire-format] Each SSE message is an `event:` line and a `data:` line, terminated by a blank line. The `data:` payload is a single-line JSON envelope: ```text event: task.status data: {"stream_id":"...","type":"status_update","payload":{"status":"working","progress":3,"total":10}} ``` Multi-line JSON is not used — every envelope is serialized to one line so it fits a single `data:` field. Responses carry `Content-Type: text/event-stream`. ## Event types [#event-types] The agent's envelope `type` maps to a named SSE `event`. `task.done` and `task.error` are **terminal** — after either, stop reading; the gateway closes the connection and the `kubemq_a2a_sse_streams_active` gauge decrements. | SSE event | Envelope `type` | Meaning | Terminal | | --------------- | --------------- | ----------------------------------------------- | -------- | | `task.status` | `status_update` | Progress update (`status`, `progress`, `total`) | No | | `task.artifact` | `artifact` | Intermediate artifact delivery | No | | `task.done` | `done` | Successful completion | Yes | | `task.error` | `error` | Failure (carries `code` and `message`) | Yes | | `message` | (default) | Any envelope without a recognized type | No | Example payloads: ```json {"type": "status_update", "payload": {"status": "working", "progress": 3, "total": 10}} {"type": "artifact", "payload": {"name": "result.json", "data": {"key": "value"}}} {"type": "done", "payload": {"final_result": "completed", "event_count": 10}} {"type": "error", "payload": {"code": -32001, "message": "agent timeout"}} ``` ## Keepalive comments [#keepalive-comments] To keep proxies and load balancers from dropping an idle connection, the gateway emits an SSE **comment** line every 30 seconds (`sseKeepaliveInterval`): ```text : keepalive ``` Comment lines start with `:`, carry no `event:` or `data:` line, and are not stream events. Standard SSE client libraries ignore them automatically. If you parse the stream by hand, skip any line beginning with `:`. ## Idle timeout [#idle-timeout] If no events flow for `MaxSSEIdleSeconds` (default **300s**, set on `A2aConfig`), the gateway closes the stream. Before closing it sends a terminal `task.error` with code `-32001` and message `"stream idle timeout"`, then issues a best-effort cancel to the agent. Agents handling long-running work should emit periodic `task.status` events to keep the stream alive. The idle timer measures time *between events*, not total stream duration. A stream can run indefinitely as long as the agent keeps emitting events (including `task.status` heartbeats) more often than `MaxSSEIdleSeconds`. ## Client disconnect and cancellation [#client-disconnect-and-cancellation] When the caller disconnects from an open stream, the gateway detects it and propagates the cancellation to the agent rather than leaking the upstream connection: 1. The gateway detects the closed caller connection. 2. It sends a Query to the agent's virtual subscriber on `_AGENTS_.agents/` with `a2a_method: "stream_cancel"` and the `stream_id` (10s timeout). 3. The virtual subscriber cancels its SSE relay goroutine, closing the HTTP SSE connection to the agent. 4. `kubemq_a2a_sse_streams_active` decrements. This means closing your SSE reader is a real cancellation signal — the agent is told to stop, freeing its work and the agent's concurrency slot. ## Reconnection [#reconnection] A2A streams are **not resumable**. There is no event ID and no `Last-Event-ID` support — a dropped stream cannot be resumed from where it stopped, and any undelivered events are lost. To recover, start a fresh stream. Do not rely on automatic SSE reconnection to continue a task. Because there is no replay, a reconnect starts a brand-new request. For long-running work, make the agent idempotent and correlate retries with `context_id`. For resilience across reconnects: * Reuse `context_id` so the agent can correlate the new stream with the original request. * Make agent-side processing idempotent. * Where supported, check task status before re-streaming so you do not duplicate work. ## Concurrent streams [#concurrent-streams] Multiple SSE streams can be open at once: * Multiple streams to the **same agent** are allowed; each one counts against that agent's `AgentMaxConcurrency` limit (default 100). * Streams to **different agents** are fully independent. * `kubemq_a2a_sse_streams_active` tracks all active streams. ## Consuming the stream [#consuming-the-stream] Read the stream line by line, track the most recent `event:`, parse each `data:` line as JSON, and stop when you see `task.done` or `task.error`. The snippets below read a `message/stream` response end to end. ```bash curl -N -X POST http://localhost:9090/a2a/stream-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": "Stream me some updates" }] } } }' ``` ```csharp using System.Text; using System.Text.Json; const string KubeMqUrl = "http://localhost:9090"; const string AgentId = "stream-agent-01"; var payload = new { jsonrpc = "2.0", id = 1, method = "message/stream", @params = new { message = new { parts = new[] { new { text = "Stream me some updates" } } } } }; using var client = new HttpClient { Timeout = TimeSpan.FromSeconds(60) }; var request = new HttpRequestMessage(HttpMethod.Post, $"{KubeMqUrl}/a2a/{AgentId}") { Content = new StringContent(JsonSerializer.Serialize(payload), Encoding.UTF8, "application/json") }; request.Headers.Add("Accept", "text/event-stream"); Console.WriteLine("Connecting to SSE stream..."); using var resp = await client.SendAsync(request, HttpCompletionOption.ResponseHeadersRead); using var stream = await resp.Content.ReadAsStreamAsync(); using var reader = new StreamReader(stream); string? eventType = null; int eventCount = 0; while (!reader.EndOfStream) { var line = await reader.ReadLineAsync(); if (line == null) break; if (line.StartsWith("event: ")) eventType = line[7..]; else if (line.StartsWith("data: ") && eventType != null) { eventCount++; var data = line[6..]; Console.WriteLine($"[{eventType}] {data}"); if (eventType is "task.done" or "task.error") break; } else if (line.Length == 0) eventType = null; } Console.WriteLine($"\nReceived {eventCount} events"); ``` ```go package main import ( "bufio" "bytes" "encoding/json" "fmt" "net/http" "os" "strings" ) const ( kubemqURL = "http://localhost:9090" agentID = "stream-agent-01" ) func main() { payload := map[string]interface{}{ "jsonrpc": "2.0", "id": 1, "method": "message/stream", "params": map[string]interface{}{ "message": map[string]interface{}{ "parts": []map[string]interface{}{{"text": "Stream me some updates"}}, }, }, } data, err := json.Marshal(payload) if err != nil { fmt.Fprintf(os.Stderr, "Marshal failed: %v\n", err) os.Exit(1) } req, err := http.NewRequest(http.MethodPost, kubemqURL+"/a2a/"+agentID, bytes.NewReader(data)) if err != nil { fmt.Fprintf(os.Stderr, "Request build failed: %v\n", err) os.Exit(1) } req.Header.Set("Content-Type", "application/json") req.Header.Set("Accept", "text/event-stream") resp, err := http.DefaultClient.Do(req) if err != nil { fmt.Fprintf(os.Stderr, "Request failed: %v\n", err) os.Exit(1) } defer resp.Body.Close() fmt.Println("Connecting to SSE stream...") eventCount := 0 eventType := "" scanner := bufio.NewScanner(resp.Body) for scanner.Scan() { line := scanner.Text() if strings.HasPrefix(line, "event: ") { eventType = strings.TrimPrefix(line, "event: ") } else if strings.HasPrefix(line, "data: ") { eventCount++ dataStr := strings.TrimPrefix(line, "data: ") fmt.Printf("[%s] %s\n", eventType, dataStr) if eventType == "task.done" || eventType == "task.error" { break } } } fmt.Printf("\nReceived %d events\n", eventCount) } ``` ```java import com.fasterxml.jackson.databind.ObjectMapper; import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; import java.time.Duration; import java.util.List; import java.util.Map; public class Client { static final String KUBEMQ_URL = "http://localhost:9090"; static final String AGENT_ID = "stream-agent-01"; static final ObjectMapper MAPPER = new ObjectMapper(); public static void main(String[] args) throws Exception { var payload = Map.of( "jsonrpc", "2.0", "id", 1, "method", "message/stream", "params", Map.of( "message", Map.of("parts", List.of(Map.of("text", "Stream me some updates"))) ) ); var client = HttpClient.newBuilder() .connectTimeout(Duration.ofSeconds(60)) .build(); var req = HttpRequest.newBuilder() .uri(URI.create(KUBEMQ_URL + "/a2a/" + AGENT_ID)) .header("Content-Type", "application/json") .header("Accept", "text/event-stream") .timeout(Duration.ofSeconds(60)) .POST(HttpRequest.BodyPublishers.ofString(MAPPER.writeValueAsString(payload))) .build(); System.out.println("Connecting to SSE stream..."); var resp = client.send(req, HttpResponse.BodyHandlers.ofLines()); int eventCount = 0; String currentEvent = null; for (var it = resp.body().iterator(); it.hasNext(); ) { String line = it.next(); if (line.startsWith("event: ")) { currentEvent = line.substring(7).trim(); } else if (line.startsWith("data: ")) { eventCount++; String data = line.substring(6); System.out.println("[" + currentEvent + "] " + data); if ("task.done".equals(currentEvent) || "task.error".equals(currentEvent)) { break; } } } System.out.println("\nReceived " + eventCount + " events"); } } ``` ```python import asyncio import json import httpx from httpx_sse import aconnect_sse KUBEMQ_URL = "http://localhost:9090" AGENT_ID = "stream-agent-01" async def main() -> None: payload = { "jsonrpc": "2.0", "id": 1, "method": "message/stream", "params": { "message": {"parts": [{"text": "Stream me some updates"}]}, }, } async with httpx.AsyncClient(timeout=60) 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: event_count = 0 async for event in event_source.aiter_sse(): event_count += 1 data = json.loads(event.data) print(f"[{event.event}] {json.dumps(data)}") if event.event in ("task.done", "task.error"): break print(f"\nReceived {event_count} events") if __name__ == "__main__": asyncio.run(main()) ``` ```typescript const KUBEMQ_URL = "http://localhost:9090"; const AGENT_ID = "stream-agent-01"; async function main() { const request = { jsonrpc: "2.0", id: 1, method: "message/stream", params: { message: { parts: [{ text: "Stream me some updates" }] } }, }; const resp = await fetch(`${KUBEMQ_URL}/a2a/${AGENT_ID}`, { method: "POST", headers: { "Content-Type": "application/json", Accept: "text/event-stream", }, body: JSON.stringify(request), }); const reader = resp.body!.getReader(); const decoder = new TextDecoder(); let eventCount = 0; let buffer = ""; while (true) { const { done, value } = await reader.read(); if (done) break; buffer += decoder.decode(value, { stream: true }); const frames = buffer.split("\n\n"); buffer = frames.pop() ?? ""; for (const frame of frames) { if (!frame.trim()) continue; let eventType = ""; let eventData = ""; for (const line of frame.split("\n")) { if (line.startsWith("event: ")) eventType = line.slice(7).trim(); else if (line.startsWith("data: ")) eventData = line.slice(6).trim(); } if (!eventType) continue; eventCount++; const payload = JSON.parse(eventData); console.log(`[${eventType}] ${JSON.stringify(payload)}`); if (eventType === "task.done" || eventType === "task.error") { console.log(`\nStream complete. Total events: ${eventCount}`); reader.cancel(); return; } } } } main().catch(console.error); ``` Always break on `task.done` or `task.error`. Closing the reader after a terminal event is how the gateway and agent learn the stream is finished — and, mid-stream, how a caller cancels the agent's work (see [Client disconnect and cancellation](#client-disconnect-and-cancellation)). ## Related [#related]