Streaming (SSE)

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"}]}
    }
  }'

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!");

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!")
}

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!");
    }
}

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())

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);