SSE Behavior

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;

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

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

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

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

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