An A2A agent is a plain HTTP server that speaks JSON-RPC 2.0. It runs anywhere, needs no KubeMQ SDK, and joins the platform by registering its URL with the registry. The gateway's per-agent virtual subscriber does all the broker work — your agent just answers HTTP POSTs.

Overview

A compliant agent has to do three things:

Listen on a URL, e.g. http://localhost:18080/.
Answer POST requests whose body is a JSON-RPC 2.0 request, returning a JSON-RPC 2.0 response (or an SSE stream for message/stream).
Register its agent card — including the absolute url — with KubeMQ so the gateway can route to it.

That is the entire contract. Your agent never connects to the broker, never imports a KubeMQ client, and never deals with protobuf. KubeMQ's virtual subscriber subscribes to Queries on _AGENTS_.agents/<agent_id> on your behalf, POSTs each request to your registered URL, and relays the reply back through the broker.

This replaces the previous SDK-based model. Agents no longer subscribe to Queries through the gRPC SDK — the AgentCard.URL field is now required and must be an absolute http:// or https:// URL. See Migration.

How it works

When a caller targets your agent, the request flows through the gateway and your agent's virtual subscriber, which turns it into an ordinary HTTP POST to your server.

Your agent is a plain HTTP server; the gateway's virtual subscriber bridges the broker to HTTP POST.

The virtual subscriber always injects an X-KubeMQ-Caller-ID header carrying the originating caller's identity, and forwards caller X-* headers (stripping hop-by-hop and sensitive ones). Your agent reads them like any other HTTP header.

Build the agent server

The minimal agent is an HTTP server with one POST / handler that parses the JSON-RPC body, processes it, and returns a JSON-RPC response with the same id. The example below is an echo agent that starts its server, then registers its card with KubeMQ.

Always start the HTTP server before registering. KubeMQ may route a request to your agent the moment registration succeeds.

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

public static class Agent
{
    private const string KubeMqUrl = "http://localhost:9090";
    private const string AgentId = "echo-agent-01";
    private const int AgentPort = 18080;

    public static async Task RunAsync()
    {
        var builder = WebApplication.CreateBuilder();
        builder.Logging.ClearProviders();
        var app = builder.Build();

        app.MapPost("/", async (HttpContext context) =>
        {
            var body = (await JsonSerializer.DeserializeAsync<JsonNode>(context.Request.Body))!;
            var response = new JsonObject
            {
                ["jsonrpc"] = "2.0",
                ["id"] = body["id"]?.DeepClone(),
                ["result"] = new JsonObject { ["echo"] = body.DeepClone() }
            };
            context.Response.ContentType = "application/json";
            await context.Response.WriteAsync(response.ToJsonString());
        });

        app.Urls.Add($"http://0.0.0.0:{AgentPort}");
        await app.StartAsync();
        Console.WriteLine($"Agent listening on port {AgentPort}");

        await RegisterAgentAsync();

        await app.WaitForShutdownAsync();
    }

    private static async Task RegisterAgentAsync()
    {
        var card = new JsonObject
        {
            ["agent_id"] = AgentId,
            ["name"] = "Echo Agent",
            ["description"] = "A simple echo agent for testing",
            ["version"] = "1.0.0",
            ["url"] = $"http://localhost:{AgentPort}/",
            ["skills"] = new JsonArray(new JsonObject
            {
                ["id"] = "echo", ["name"] = "Echo",
                ["description"] = "Echoes back the received message",
                ["tags"] = new JsonArray("test", "echo")
            }),
            ["defaultInputModes"] = new JsonArray("text"),
            ["defaultOutputModes"] = new JsonArray("text"),
            ["protocolVersions"] = new JsonArray("1.0")
        };

        using var client = new HttpClient();
        var resp = await client.PostAsync(
            $"{KubeMqUrl}/agents/register",
            new StringContent(card.ToJsonString(), System.Text.Encoding.UTF8, "application/json"));
        Console.WriteLine($"Registered: {(int)resp.StatusCode}");
        var body = await resp.Content.ReadAsStringAsync();
        Console.WriteLine(JsonSerializer.Serialize(JsonNode.Parse(body), new JsonSerializerOptions { WriteIndented = true }));
    }
}

package main

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

const (
	kubemqURL = "http://localhost:9090"
	agentID   = "echo-agent-01"
	agentPort = 18080
)

func handler(w http.ResponseWriter, r *http.Request) {
	var body map[string]interface{}
	if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	resp := map[string]interface{}{
		"jsonrpc": "2.0",
		"id":      body["id"],
		"result":  map[string]interface{}{"echo": body},
	}
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(resp)
}

func registerAgent() {
	card := map[string]interface{}{
		"agent_id":    agentID,
		"name":        "Echo Agent",
		"description": "A simple echo agent for testing",
		"version":     "1.0.0",
		"url":         fmt.Sprintf("http://localhost:%d/", agentPort),
		"skills": []map[string]interface{}{
			{
				"id":          "echo",
				"name":        "Echo",
				"description": "Echoes back the received message",
				"tags":        []string{"test", "echo"},
			},
		},
		"defaultInputModes":  []string{"text"},
		"defaultOutputModes": []string{"text"},
		"protocolVersions":   []string{"1.0"},
	}
	data, _ := json.Marshal(card)
	resp, err := http.Post(kubemqURL+"/agents/register", "application/json", bytes.NewReader(data))
	if err != nil {
		fmt.Fprintf(os.Stderr, "Registration failed: %v\n", err)
		return
	}
	defer resp.Body.Close()
	body, _ := io.ReadAll(resp.Body)
	fmt.Printf("Registered: %d\n", resp.StatusCode)
	var pretty bytes.Buffer
	json.Indent(&pretty, body, "", "  ")
	fmt.Println(pretty.String())
}

func main() {
	http.HandleFunc("/", handler)

	ln, err := net.Listen("tcp", fmt.Sprintf(":%d", agentPort))
	if err != nil {
		fmt.Fprintf(os.Stderr, "Listen failed: %v\n", err)
		os.Exit(1)
	}
	fmt.Printf("Agent listening on port %d\n", agentPort)

	go http.Serve(ln, nil)

	registerAgent()

	sig := make(chan os.Signal, 1)
	signal.Notify(sig, os.Interrupt)
	<-sig
	fmt.Println("\nShutting down")
}

import com.fasterxml.jackson.databind.ObjectMapper;
import com.sun.net.httpserver.HttpServer;

import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.util.List;
import java.util.Map;

public class Agent {

    static final String KUBEMQ_URL = "http://localhost:9090";
    static final String AGENT_ID = "echo-agent-01";
    static final int AGENT_PORT = 18080;
    static final ObjectMapper MAPPER = new ObjectMapper();

    public static void main(String[] args) throws Exception {
        HttpServer server = HttpServer.create(new InetSocketAddress(AGENT_PORT), 0);
        server.createContext("/", exchange -> {
            if (!"POST".equals(exchange.getRequestMethod())) {
                exchange.sendResponseHeaders(405, -1);
                return;
            }
            var body = MAPPER.readTree(exchange.getRequestBody());
            var response = MAPPER.createObjectNode();
            response.put("jsonrpc", "2.0");
            response.set("id", body.get("id"));
            response.putObject("result").set("echo", body);

            byte[] out = MAPPER.writeValueAsBytes(response);
            exchange.getResponseHeaders().set("Content-Type", "application/json");
            exchange.sendResponseHeaders(200, out.length);
            try (OutputStream os = exchange.getResponseBody()) { os.write(out); }
        });
        server.start();
        System.out.println("Agent listening on port " + AGENT_PORT);

        var card = Map.of(
            "agent_id", AGENT_ID,
            "name", "Echo Agent",
            "description", "A simple echo agent for testing",
            "version", "1.0.0",
            "url", "http://localhost:" + AGENT_PORT + "/",
            "skills", List.of(Map.of(
                "id", "echo", "name", "Echo",
                "description", "Echoes back the received message",
                "tags", List.of("test", "echo")
            )),
            "defaultInputModes", List.of("text"),
            "defaultOutputModes", List.of("text"),
            "protocolVersions", List.of("1.0")
        );

        var client = HttpClient.newHttpClient();
        var req = HttpRequest.newBuilder()
            .uri(URI.create(KUBEMQ_URL + "/agents/register"))
            .header("Content-Type", "application/json")
            .POST(HttpRequest.BodyPublishers.ofString(MAPPER.writeValueAsString(card)))
            .build();
        var resp = client.send(req, HttpResponse.BodyHandlers.ofString());
        System.out.println("Registered: " + resp.statusCode());
        System.out.println(MAPPER.writerWithDefaultPrettyPrinter()
            .writeValueAsString(MAPPER.readTree(resp.body())));

        Thread.currentThread().join();
    }
}

"""Echo agent that registers with KubeMQ."""

import asyncio
import json
import signal

import httpx
from aiohttp import web

KUBEMQ_URL = "http://localhost:9090"
AGENT_ID = "echo-agent-01"
AGENT_PORT = 18080


async def handle_request(request: web.Request) -> web.Response:
    body = await request.json()
    return web.json_response({
        "jsonrpc": "2.0",
        "id": body.get("id"),
        "result": {"echo": body},
    })


async def register_agent() -> None:
    card = {
        "agent_id": AGENT_ID,
        "name": "Echo Agent",
        "description": "A simple echo agent for testing",
        "version": "1.0.0",
        "url": f"http://localhost:{AGENT_PORT}/",
        "skills": [
            {
                "id": "echo",
                "name": "Echo",
                "description": "Echoes back the received message",
                "tags": ["test", "echo"],
            }
        ],
        "defaultInputModes": ["text"],
        "defaultOutputModes": ["text"],
        "protocolVersions": ["1.0"],
    }
    async with httpx.AsyncClient() as client:
        resp = await client.post(f"{KUBEMQ_URL}/agents/register", json=card)
        print(f"Registered: {resp.status_code}")
        print(json.dumps(resp.json(), indent=2))


async def main() -> None:
    app = web.Application()
    app.router.add_post("/", handle_request)

    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "0.0.0.0", AGENT_PORT)
    await site.start()
    print(f"Agent listening on port {AGENT_PORT}")

    await register_agent()

    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, stop.set)

    await stop.wait()
    await runner.cleanup()


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

import express from "express";

const KUBEMQ_URL = "http://localhost:9090";
const AGENT_ID = "echo-agent-01";
const AGENT_PORT = 18080;

const app = express();
app.use(express.json());

app.post("/", (req, res) => {
  const body = req.body;
  console.log("Received request:", JSON.stringify(body));
  res.json({
    jsonrpc: "2.0",
    id: body.id,
    result: { echo: body },
  });
});

app.listen(AGENT_PORT, async () => {
  console.log(`Agent listening on port ${AGENT_PORT}`);

  const card = {
    agent_id: AGENT_ID,
    name: "Echo Agent",
    description: "A simple echo agent for testing",
    version: "1.0.0",
    url: `http://localhost:${AGENT_PORT}/`,
    skills: [
      {
        id: "echo",
        name: "Echo",
        description: "Echoes back the received message",
        tags: ["test", "echo"],
      },
    ],
    defaultInputModes: ["text"],
    defaultOutputModes: ["text"],
    protocolVersions: ["1.0"],
  };

  const resp = await fetch(`${KUBEMQ_URL}/agents/register`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify(card),
  });

  const data = await resp.json();
  console.log("Registered:", JSON.stringify(data, null, 2));
});

You can register the same way with plain curl — useful for a sidecar agent or a language not shown above:

curl -X POST http://localhost:9090/agents/register \
  -H "Content-Type: application/json" \
  -d '{
    "agent_id": "echo-agent-01",
    "name": "Echo Agent",
    "description": "A simple echo agent for testing",
    "version": "1.0.0",
    "url": "http://localhost:18080/",
    "skills": [{"id": "echo", "name": "Echo", "tags": ["test", "echo"]}],
    "defaultInputModes": ["text"],
    "defaultOutputModes": ["text"],
    "protocolVersions": ["1.0"]
  }'

Handle streaming (message/stream)

If your agent produces incremental output, support the message/stream method. Inspect the request's method; when it is message/stream, respond with Content-Type: text/event-stream and write SSE events instead of a single JSON body. Each event uses an event: name and a JSON data: line, and the stream ends with a terminal task.done (or task.error) event.

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

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

The handler below extends the echo agent: it emits five task.status events, then a task.done, for message/stream requests, and falls back to a plain JSON-RPC reply for everything else.

app.MapPost("/", async (HttpContext context) =>
{
    var body = (await JsonSerializer.DeserializeAsync<JsonNode>(context.Request.Body))!;
    var method = body["method"]?.GetValue<string>() ?? "";

    if (method == "message/stream")
    {
        context.Response.ContentType = "text/event-stream";
        context.Response.Headers.CacheControl = "no-cache";

        for (int i = 1; i <= 5; i++)
        {
            var ev = JsonSerializer.Serialize(new { type = "status_update", payload = new { status = "working", progress = i, total = 5 } });
            await context.Response.WriteAsync($"event: task.status\ndata: {ev}\n\n");
            await context.Response.Body.FlushAsync();
            await Task.Delay(500);
        }

        var done = JsonSerializer.Serialize(new { type = "done", payload = new { final_result = "completed", event_count = 5 } });
        await context.Response.WriteAsync($"event: task.done\ndata: {done}\n\n");
        await context.Response.Body.FlushAsync();
        return;
    }

    var response = new JsonObject
    {
        ["jsonrpc"] = "2.0",
        ["id"] = body["id"]?.DeepClone(),
        ["result"] = new JsonObject { ["echo"] = body.DeepClone() }
    };
    context.Response.ContentType = "application/json";
    await context.Response.WriteAsync(response.ToJsonString());
});

func handleStream(w http.ResponseWriter, r *http.Request) {
	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "streaming not supported", http.StatusInternalServerError)
		return
	}

	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")

	for i := 1; i <= 5; i++ {
		event, _ := json.Marshal(map[string]interface{}{
			"type": "status_update",
			"payload": map[string]interface{}{
				"status": "working", "progress": i, "total": 5,
			},
		})
		fmt.Fprintf(w, "event: task.status\ndata: %s\n\n", event)
		flusher.Flush()
		time.Sleep(500 * time.Millisecond)
	}

	done, _ := json.Marshal(map[string]interface{}{
		"type":    "done",
		"payload": map[string]interface{}{"final_result": "completed", "event_count": 5},
	})
	fmt.Fprintf(w, "event: task.done\ndata: %s\n\n", done)
	flusher.Flush()
}

func handler(w http.ResponseWriter, r *http.Request) {
	var body map[string]interface{}
	json.NewDecoder(r.Body).Decode(&body)

	if method, _ := body["method"].(string); method == "message/stream" {
		handleStream(w, r)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(map[string]interface{}{
		"jsonrpc": "2.0",
		"id":      body["id"],
		"result":  map[string]interface{}{"echo": body},
	})
}

server.createContext("/", exchange -> {
    if (!"POST".equals(exchange.getRequestMethod())) {
        exchange.sendResponseHeaders(405, -1);
        return;
    }
    var body = MAPPER.readTree(exchange.getRequestBody());
    var method = body.path("method").asText("");

    if ("message/stream".equals(method)) {
        exchange.getResponseHeaders().set("Content-Type", "text/event-stream");
        exchange.getResponseHeaders().set("Cache-Control", "no-cache");
        exchange.sendResponseHeaders(200, 0);
        try (OutputStream os = exchange.getResponseBody()) {
            for (int i = 1; i <= 5; i++) {
                var event = MAPPER.writeValueAsString(Map.of(
                    "type", "status_update",
                    "payload", Map.of("status", "working", "progress", i, "total", 5)
                ));
                os.write(("event: task.status\ndata: " + event + "\n\n").getBytes());
                os.flush();
                try { Thread.sleep(500); } catch (InterruptedException ie) { break; }
            }
            var done = MAPPER.writeValueAsString(Map.of(
                "type", "done",
                "payload", Map.of("final_result", "completed", "event_count", 5)
            ));
            os.write(("event: task.done\ndata: " + done + "\n\n").getBytes());
            os.flush();
        }
        return;
    }

    var response = MAPPER.createObjectNode();
    response.put("jsonrpc", "2.0");
    response.set("id", body.get("id"));
    response.putObject("result").set("echo", body);
    byte[] out = MAPPER.writeValueAsBytes(response);
    exchange.getResponseHeaders().set("Content-Type", "application/json");
    exchange.sendResponseHeaders(200, out.length);
    try (OutputStream os = exchange.getResponseBody()) { os.write(out); }
});

async def handle_stream(request: web.Request) -> web.StreamResponse:
    resp = web.StreamResponse(
        status=200,
        headers={"Content-Type": "text/event-stream", "Cache-Control": "no-cache"},
    )
    await resp.prepare(request)

    for i in range(1, 6):
        event = json.dumps({"type": "status_update", "payload": {"status": "working", "progress": i, "total": 5}})
        await resp.write(f"event: task.status\ndata: {event}\n\n".encode())
        await asyncio.sleep(0.5)

    done = json.dumps({"type": "done", "payload": {"final_result": "completed", "event_count": 5}})
    await resp.write(f"event: task.done\ndata: {done}\n\n".encode())
    await resp.write_eof()
    return resp


async def handle_request(request: web.Request) -> web.Response:
    body = await request.json()
    method = body.get("method", "")
    if method == "message/stream":
        return await handle_stream(request)
    return web.json_response({"jsonrpc": "2.0", "id": body.get("id"), "result": {"echo": body}})

app.post("/", async (req, res) => {
  if (req.body.method === "message/stream") {
    res.setHeader("Content-Type", "text/event-stream");
    res.setHeader("Cache-Control", "no-cache");
    res.setHeader("Connection", "keep-alive");
    res.flushHeaders();

    for (let i = 1; i <= 5; i++) {
      const event = {
        type: "status_update",
        payload: { status: "working", progress: i, total: 5 },
      };
      res.write(`event: task.status\ndata: ${JSON.stringify(event)}\n\n`);
      await new Promise((r) => setTimeout(r, 500));
    }

    const done = { type: "done", payload: { final_result: "completed", event_count: 5 } };
    res.write(`event: task.done\ndata: ${JSON.stringify(done)}\n\n`);
    res.end();
  } else {
    res.json({ jsonrpc: "2.0", id: req.body.id, result: { echo: req.body } });
  }
});

The gateway maps your SSE event names to caller-facing envelopes: task.status, task.artifact, task.done, and task.error. See SSE behavior for the full wire format, keepalive, and cancel-on-disconnect rules.

Agent lifecycle

Beyond serving requests, an agent manages its presence in the registry:

Stage	Endpoint	When
Register	`POST /agents/register`	After the HTTP server is listening, on startup.
Heartbeat	`POST /agents/heartbeat`	Every 30–60s to refresh `last_seen` and beat the TTL.
Deregister	`POST /agents/deregister`	On graceful shutdown, before stopping the server.

If an agent stops sending heartbeats, the registry expires it after AgentTTLSeconds (default 300) and tears down its virtual subscriber. Refresh liveness with a periodic heartbeat:

curl -X POST http://localhost:9090/agents/heartbeat \
  -H "Content-Type: application/json" \
  -d '{"agent_id": "echo-agent-01"}'

On shutdown, deregister first so callers stop being routed to the agent, then drain in-flight requests and stop the server:

curl -X POST http://localhost:9090/agents/deregister \
  -H "Content-Type: application/json" \
  -d '{"agent_id": "echo-agent-01"}'

When authentication is enabled, only the principal that registered an agent can heartbeat or deregister it. See Authentication.

Migration from the SDK model

Earlier KubeMQ versions required receiving agents to use the gRPC SDK to subscribe to Queries on _AGENTS_.agents/<agent_id>. That path is gone — replaced by the virtual-subscriber architecture, where agents are plain HTTP servers.

What changed:

Agents no longer connect to KubeMQ via the gRPC SDK to receive tasks, and no longer call SubscribeToQueries() on agent channels.
The AgentCard.URL field is now required and must be an absolute http:// or https:// URL. Relative paths (e.g. /a2a/<agent_id>) are rejected.
The old default URL auto-assignment (/a2a/<agent_id> when url was empty) has been removed.

To migrate an existing agent:

Deploy it as a standard HTTP server that accepts POST requests with Content-Type: application/json.
Implement a JSON-RPC 2.0 handler at the root URL (or any path).
Update registration to include the full absolute URL, e.g. "url": "http://my-agent:8080/".
Remove all KubeMQ SDK dependencies from the agent.
For streaming agents, implement SSE responses for message/stream.

The caller side is unchanged — callers can keep using any KubeMQ transport (gRPC, REST, the A2A HTTP gateway, or the MCP bridge) to reach agents.

Building Agents

Overview

How it works

Build the agent server

Handle streaming (message/stream)

Agent lifecycle

Migration from the SDK model

Agent registry

Agent cards

SSE behavior

Synchronous messaging

On this page

Building Agents

Overview

How it works

Build the agent server

Handle streaming (message/stream)

Agent lifecycle

Migration from the SDK model

Related

Agent registry

Agent cards

SSE behavior

Synchronous messaging

On this page