Streaming Task Pipeline

# -N disables curl buffering so events print as they arrive.
curl -N -X POST http://localhost:9090/a2a/task-events-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": "Show me all event types"}]}}
  }'

using System.Text;
using System.Text.Json.Nodes;

const string KubeMqUrl = "http://localhost:9090";
const string AgentId = "task-events-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"] = "Show me all event types" })
        }
    }
};

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);
// ...continue reading events in Step 2

package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "net/http"
    "os"
)

const (
    kubemqURL = "http://localhost:9090"
    agentID   = "task-events-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": "Show me all event types"}},
            },
        },
    }

    data, _ := json.Marshal(payload)
    resp, err := http.Post(kubemqURL+"/a2a/"+agentID, "application/json", bytes.NewReader(data))
    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...")
    // ...continue reading events in Step 2
}

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 = "task-events-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", "Show me all event types")))
            )
        );

        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());
        // ...continue reading events in Step 2
    }
}

"""Stream a long-running task and consume its task.* envelopes."""

import asyncio
import json
from collections import Counter

import httpx
from httpx_sse import aconnect_sse

KUBEMQ_URL = "http://localhost:9090"
AGENT_ID = "task-events-agent-01"


async def main() -> None:
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "message/stream",
        "params": {"message": {"parts": [{"text": "Show me all event types"}]}},
    }

    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
        ) as event_source:
            ...  # consume events in Step 2


if __name__ == "__main__":
    asyncio.run(main())

const KUBEMQ_URL = "http://localhost:9090";
const AGENT_ID = "task-events-agent-01";

async function main() {
  const request = {
    jsonrpc: "2.0",
    id: 1,
    method: "message/stream",
    params: { message: { parts: [{ text: "Send me task events" }] } },
  };

  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();
  // ...continue reading events in Step 2
}

main().catch(console.error);

event: task.status
data: {"stream_id":"...","type":"status_update","payload":{"status":"working","progress":1,"total":3}}

event: task.status
data: {"stream_id":"...","type":"status_update","payload":{"status":"working","progress":2,"total":3}}

event: task.artifact
data: {"stream_id":"...","type":"artifact","payload":{"name":"result.json","data":{"key":"value"}}}

event: task.done
data: {"stream_id":"...","type":"done","payload":{"final_result":"completed"}}

var eventTypes = new Dictionary<string, int>();
string? eventType = null;

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)
    {
        eventTypes[eventType] = eventTypes.GetValueOrDefault(eventType) + 1;
        var data = JsonNode.Parse(line[6..])!;

        if (eventType == "task.status")
        {
            var p = data["payload"]!["payload"]!;
            Console.WriteLine($"  [STATUS]   progress={p["progress"]}/{p["total"]}");
        }
        else if (eventType == "task.artifact")
            Console.WriteLine($"  [ARTIFACT] name={data["payload"]!["payload"]!["name"]}");
        else if (eventType == "task.done")
            Console.WriteLine($"  [DONE]     result={data["payload"]!["payload"]!["final_result"]}");
        else if (eventType == "task.error")
            Console.WriteLine($"  [ERROR]    {data["payload"]!["payload"]}");

        if (eventType is "task.done" or "task.error")
            break;
    }
    else if (line.Length == 0)
        eventType = null;
}

Console.WriteLine($"\n--- Event Summary ---");
foreach (var (et, count) in eventTypes.OrderBy(kv => kv.Key))
    Console.WriteLine($"  {et}: {count}");
Console.WriteLine($"  Total: {eventTypes.Values.Sum()}");

scanner := bufio.NewScanner(resp.Body)
eventType := ""
counts := map[string]int{}

for scanner.Scan() {
    line := scanner.Text()
    if strings.HasPrefix(line, "event: ") {
        eventType = strings.TrimPrefix(line, "event: ")
    } else if strings.HasPrefix(line, "data: ") {
        counts[eventType]++
        dataStr := strings.TrimPrefix(line, "data: ")

        var d map[string]interface{}
        json.Unmarshal([]byte(dataStr), &d)
        payload, _ := d["payload"].(map[string]interface{})

        inner, _ := payload["payload"].(map[string]interface{})
        if inner == nil {
            inner = payload
        }
        switch eventType {
        case "task.status":
            fmt.Printf("  [STATUS]   progress=%.0f/%.0f\n", inner["progress"], inner["total"])
        case "task.artifact":
            fmt.Printf("  [ARTIFACT] name=%v\n", inner["name"])
        case "task.done":
            fmt.Printf("  [DONE]     result=%v\n", inner["final_result"])
        case "task.error":
            fmt.Printf("  [ERROR]    %v\n", inner)
        }

        if eventType == "task.done" || eventType == "task.error" {
            break
        }
    }
}

Map<String, Integer> eventTypes = new LinkedHashMap<>();
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: ") && currentEvent != null) {
        var data = MAPPER.readTree(line.substring(6));
        eventTypes.merge(currentEvent, 1, Integer::sum);

        var inner = data.path("payload").path("payload");
        switch (currentEvent) {
            case "task.status" ->
                System.out.println("  [STATUS]   progress=" + inner.path("progress") + "/" + inner.path("total"));
            case "task.artifact" ->
                System.out.println("  [ARTIFACT] name=" + inner.path("name").asText());
            case "task.done" ->
                System.out.println("  [DONE]     result=" + inner.path("final_result").asText());
            case "task.error" ->
                System.out.println("  [ERROR]    " + inner);
        }

        if ("task.done".equals(currentEvent) || "task.error".equals(currentEvent)) break;
    }
}

event_types: Counter[str] = Counter()

async for event in event_source.aiter_sse():
    data = json.loads(event.data)
    event_types[event.event] += 1

    if event.event == "task.status":
        inner = data["payload"]["payload"]
        print(f"  [STATUS]   progress={inner['progress']}/{inner['total']}")
    elif event.event == "task.artifact":
        inner = data["payload"]["payload"]
        print(f"  [ARTIFACT] name={inner['name']}")
    elif event.event == "task.done":
        inner = data["payload"]["payload"]
        print(f"  [DONE]     result={inner['final_result']}")
    elif event.event == "task.error":
        print(f"  [ERROR]    {data['payload']['payload']}")

    if event.event in ("task.done", "task.error"):
        break

print(f"\n--- Event Summary ---")
for event_type, count in sorted(event_types.items()):
    print(f"  {event_type}: {count}")
print(f"  Total: {sum(event_types.values())}")

function parseSSE(chunk: string): Array<{ event: string; data: string }> {
  const events: Array<{ event: string; data: string }> = [];
  let currentEvent = "";
  let currentData = "";

  for (const line of chunk.split("\n")) {
    if (line.startsWith("event: ")) {
      currentEvent = line.slice(7).trim();
    } else if (line.startsWith("data: ")) {
      currentData = line.slice(6).trim();
    } else if (line === "" && currentEvent) {
      events.push({ event: currentEvent, data: currentData });
      currentEvent = "";
      currentData = "";
    }
  }
  return events;
}

const decoder = new TextDecoder();
const counts: Record<string, number> = {};

while (true) {
  const { done, value } = await reader.read();
  if (done) break;

  const events = parseSSE(decoder.decode(value, { stream: true }));
  for (const evt of events) {
    counts[evt.event] = (counts[evt.event] || 0) + 1;
    const payload = JSON.parse(evt.data);
    const inner = payload.payload?.payload ?? payload.payload;

    switch (evt.event) {
      case "task.status":
        console.log(`[STATUS]   progress=${inner.progress}/${inner.total} status=${inner.status}`);
        break;
      case "task.artifact":
        console.log(`[ARTIFACT] name=${inner.name} data=${JSON.stringify(inner.data)}`);
        break;
      case "task.done":
        console.log(`[DONE]     result=${inner.final_result}`);
        break;
      case "task.error":
        console.log(`[ERROR]    ${inner.message}`);
        break;
    }

    if (evt.event === "task.done" || evt.event === "task.error") {
      console.log("\n=== Event Summary ===");
      for (const [type, count] of Object.entries(counts)) {
        console.log(`  ${type}: ${count}`);
      }
      reader.cancel();
      return;
    }
  }
}

# Read only the first events, then Ctrl-C (or pipe through head) to disconnect.
# The gateway sends stream_cancel to the agent on disconnect.
curl -N -X POST http://localhost:9090/a2a/slow-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":"I will disconnect early"}]}}}' \
  | head -n 6

// Break out of the read loop early and close the body — the gateway
// detects the disconnect and cancels the agent's stream.
resp, _ := http.Post(kubemqURL+"/a2a/"+agentID, "application/json", bytes.NewReader(data))

count := 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: ") {
        count++
        var d map[string]interface{}
        json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &d)
        p, _ := d["payload"].(map[string]interface{})
        fmt.Printf("  Event %d: [%s] progress=%.0f\n", count, eventType, p["progress"])

        if count >= maxEvents {
            fmt.Printf("\nDisconnecting after %d events...\n", maxEvents)
            break
        }
    }
}
resp.Body.Close() // closing the connection triggers stream_cancel
fmt.Println("KubeMQ will detect the disconnect and clean up the stream.")

"""Disconnect from the SSE stream after a couple of events to cancel the task."""
MAX_EVENTS = 2

async with httpx.AsyncClient(timeout=30) as client:
    async with aconnect_sse(
        client, "POST", f"{KUBEMQ_URL}/a2a/{AGENT_ID}", json=payload
    ) as event_source:
        count = 0
        async for event in event_source.aiter_sse():
            count += 1
            data = json.loads(event.data)
            print(f"  Event {count}: [{event.event}] progress={data['payload'].get('progress')}")
            if count >= MAX_EVENTS:
                print(f"\nDisconnecting after {MAX_EVENTS} events...")
                break  # leaving the context closes the connection -> stream_cancel

print("Client disconnected.")
print("KubeMQ will detect the disconnect and clean up the stream.")