# Streaming (SSE) (/aiway/a2a/streaming)



When an agent produces results incrementally — progress updates, partial artifacts, a
final answer — A2A streams them back to the caller as **Server-Sent Events (SSE)**. The
caller sends a JSON-RPC `message/stream` request and reads a sequence of typed event
frames until a terminal `task.done` or `task.error`.

## Overview [#overview]

Synchronous [`message/send`](/aiway/a2a/sync-messaging) returns a single reply.
**`message/stream`** instead opens a long-lived SSE connection and relays each event the
agent emits as it happens, so callers can show progress or consume artifacts before the
task finishes.

There are two ways to start a stream against the same agent:

| Trigger                                                  | Request       | Pre-stream error format                                   |
| -------------------------------------------------------- | ------------- | --------------------------------------------------------- |
| `POST /a2a/{agent_id}` with `"method": "message/stream"` | JSON-RPC body | JSON-RPC 2.0 error at HTTP 200                            |
| `GET /a2a/{agent_id}/stream`                             | query/headers | HTTP status code + `{"is_error": true, "message": "..."}` |

The POST form is the common case and what the examples below use. In both cases the
response is `Content-Type: text/event-stream` and the wire protocol is identical.

## How it works [#how-it-works]

The gateway does not hold a socket open to your agent on the caller's behalf. Instead it
uses the agent's [virtual subscriber](/aiway/a2a/architecture) as an **SSE
relay**: a temporary internal channel carries the agent's events to the gateway, which
forwards them to the caller. This subscribe-first ordering guarantees no event is lost
between the query and the agent's first emission.

<Mermaid
  chart="`
sequenceDiagram
participant Caller
participant Gateway as A2A Gateway
participant VSub as Virtual Subscriber
participant Agent
Caller->>Gateway: POST /a2a/{id} message/stream
Gateway->>Gateway: subscribe to _AGENTS_.stream/{streamID}
Gateway->>VSub: Query (a2a_method=message/stream)
VSub->>Agent: HTTP POST (Accept: text/event-stream)
Agent-->>VSub: 200 text/event-stream
VSub-->>Gateway: ack (Executed: true)
Gateway-->>Caller: 200 text/event-stream
loop until done/error
  Agent-->>VSub: SSE event
  VSub-->>Gateway: Event on stream channel
  Gateway-->>Caller: SSE frame (task.status / task.artifact)
end
Agent-->>VSub: final SSE event
VSub-->>Gateway: Event
Gateway-->>Caller: task.done (stream closes)
`"
/>

*The virtual subscriber relays the agent's SSE events over a temporary internal channel; the gateway forwards them to the caller until a terminal envelope closes the stream.*

## Event types [#event-types]

Each SSE frame has an `event:` name and a JSON `data:` payload. The agent's envelope
`type` maps to the SSE event name the caller sees:

| Envelope `type` | SSE event name  | Meaning                                    |
| --------------- | --------------- | ------------------------------------------ |
| `status_update` | `task.status`   | Progress or status update (non-terminal)   |
| `artifact`      | `task.artifact` | A partial or complete result artifact      |
| `done`          | `task.done`     | Terminal — the task completed successfully |
| `error`         | `task.error`    | Terminal — the task failed                 |
| (other)         | `message`       | Default event name for untyped envelopes   |

Each `data:` line is a stream envelope:

```json
{
  "stream_id": "f3c1...",
  "type": "status_update",
  "payload": { "status": "working", "progress": 3, "total": 5 }
}
```

A caller reads frames until it sees `task.done` or `task.error`, then stops — both are
terminal and the gateway closes the connection after sending them.

## Stream a task [#stream-a-task]

Send a `message/stream` request with `Accept: text/event-stream` and read the event
frames as they arrive. Stop on the terminal `task.done` / `task.error`.

<Tabs groupId="language" items="['curl','C#','Go','Java','Python','TypeScript']">
  <Tab value="curl">
    ```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"}]}
        }
      }'
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    using System.Text;
    using System.Text.Json.Nodes;

    const string KubeMqUrl = "http://localhost:9090";
    const string AgentId = "stream-agent-01";

    var payload = new JsonObject
    {
        ["jsonrpc"] = "2.0",
        ["id"] = 1,
        ["method"] = "message/stream",
        ["params"] = new JsonObject
        {
            ["message"] = new JsonObject
            {
                ["parts"] = new JsonArray(new JsonObject { ["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(payload.ToJsonString(), Encoding.UTF8, "application/json")
    };

    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");
    Console.WriteLine("Stream completed!");
    ```
  </Tab>

  <Tab value="Go">
    ```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)
    	fmt.Println("Stream completed!")
    }
    ```
  </Tab>

  <Tab value="Java">
    ```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");
            System.out.println("Stream completed!");
        }
    }
    ```
  </Tab>

  <Tab value="Python">
    ```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")
        print("Stream completed!")


    if __name__ == "__main__":
        asyncio.run(main())
    ```
  </Tab>

  <Tab value="TypeScript">
    ```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" }] } },
      };

      console.log("=== POST-based streaming ===");
      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;
          }
        }
      }

      console.log(`\nStream ended. Total events: ${eventCount}`);
    }

    main().catch(console.error);
    ```
  </Tab>
</Tabs>

A typical stream is a sequence of `task.status` frames followed by a terminal `task.done`:

```text
Connecting to SSE stream...
[task.status] {"type": "status_update", "payload": {"status": "working", "progress": 1, "total": 5}}
[task.status] {"type": "status_update", "payload": {"status": "working", "progress": 2, "total": 5}}
[task.status] {"type": "status_update", "payload": {"status": "working", "progress": 5, "total": 5}}
[task.done] {"type": "done", "payload": {"final_result": "completed", "event_count": 5}}
```

## Keepalive and idle timeout [#keepalive-and-idle-timeout]

To keep the connection alive through proxies during quiet periods, the gateway emits an
SSE comment line — `: keepalive` — every **30 seconds**. It is a comment, not an event,
so SSE clients ignore it; it exists only to keep the socket warm.

If the agent produces no events for `MaxSSEIdleSeconds` (default **300 seconds**), the
gateway closes the stream: it sends a terminal `task.error` with code `-32001`
(`"stream idle timeout"`) and issues a best-effort cancel to the agent. Tune the window
with `MaxSSEIdleSeconds` — see [Configuration](/aiway/a2a/configuration).

## Client disconnect [#client-disconnect]

If the caller closes the connection before a terminal event, the gateway detects the
disconnect and **cancels the work on the agent** rather than letting it run to completion.
It sends a `stream_cancel` query to the agent's virtual subscriber on
`_AGENTS_.agents/{agent_id}` (carrying `a2a_stream_id`, with a 10-second timeout), which
closes the relay's HTTP connection to the agent. This frees the
[concurrency slot](/aiway/a2a/guides/concurrency) promptly instead of waiting
for the task to finish.

<Callout type="info">
  The SSE stream endpoint omits the per-route timeout middleware — long-lived streams are
  bounded by `MaxSSEIdleSeconds`, not the 60-second request timeout that applies to
  `message/send`.
</Callout>

## Response size limit [#response-size-limit]

During relay, the virtual subscriber tracks the accumulated `data:` bytes of each SSE
event. If a single event exceeds `AgentMaxResponseBytes` (default **10 MB**), the relay is
aborted to protect the gateway from memory exhaustion. Keep individual artifact events
under this bound; for large results, chunk them across multiple `task.artifact` events.

## Related [#related]

<Cards>
  <Card title="SSE behavior" href="/aiway/a2a/guides/sse-behavior" description="The SSE wire format, keepalive, idle timeout, and reconnection in depth." />

  <Card title="Synchronous messaging" href="/aiway/a2a/sync-messaging" description="message/send for single-reply request/reply, context IDs, and the Executed flag." />

  <Card title="How it works" href="/aiway/a2a/architecture" description="The virtual subscriber, the temporary stream channel, and stream_cancel routing." />

  <Card title="Error handling" href="/aiway/a2a/error-handling" description="A2A error codes, the idle-timeout -32001 error, and retry strategy." />
</Cards>