KubeMQ
AiwayAI Agents (A2A)Scenarios

Multi-Agent Gateway

Run several A2A agents behind one KubeMQ gateway — register agents by skill, discover them with skill-tag filtering, and route message/send calls by agent_id.

This scenario stands up three independent agents — echo, translate, and summarize — behind a single A2A gateway, then drives them from one client: discover the right agent by skill tag, and dispatch a message/send to it by agent_id. It ties together the registry and synchronous messaging.

The setup

Each agent is a plain HTTP server registered by URL — no KubeMQ SDK runs on any of them. The gateway keeps one virtual subscriber per agent, so a caller reaches any agent through the same POST /a2a/{agent_id} surface. Callers never address an agent's HTTP URL directly; they address its agent_id and the gateway routes the request.

One gateway fronts many agents; the client discovers by skill tag and routes by agent_id.

Step 1 — Register the agents

Each agent registers its own agent card with a unique agent_id, its HTTP url, and a skills list. The tags on each skill are what make the agent discoverable later. Agents that share a capability (here, translate and summarize both carry the nlp tag) can be found together.

# Register three agents with distinct skills.
# (Each agent process registers itself on start; shown here as explicit calls.)
curl -X POST http://localhost:9090/agents/register \
  -H "Content-Type: application/json" \
  -d '{
    "agent_id": "echo-agent-01",
    "name": "Echo Agent",
    "url": "http://localhost:18081/",
    "skills": [{"id": "echo", "name": "Echo", "tags": ["echo"]}]
  }'

curl -X POST http://localhost:9090/agents/register \
  -H "Content-Type: application/json" \
  -d '{
    "agent_id": "translate-agent-01",
    "name": "Translate Agent",
    "url": "http://localhost:18082/",
    "skills": [{"id": "translate", "name": "Translate", "tags": ["translate", "nlp"]}]
  }'

curl -X POST http://localhost:9090/agents/register \
  -H "Content-Type: application/json" \
  -d '{
    "agent_id": "summarize-agent-01",
    "name": "Summarize Agent",
    "url": "http://localhost:18083/",
    "skills": [{"id": "summarize", "name": "Summarize", "tags": ["summarize", "nlp"]}]
  }'
using System.Text;
using System.Text.Json.Nodes;

const string KubeMqUrl = "http://localhost:9090";
using var client = new HttpClient();

var agents = new[]
{
    ("echo-agent-01", "Echo Agent", 18081, "echo", new[] { "echo" }),
    ("translate-agent-01", "Translate Agent", 18082, "translate", new[] { "translate", "nlp" }),
    ("summarize-agent-01", "Summarize Agent", 18083, "summarize", new[] { "summarize", "nlp" }),
};

foreach (var (id, name, port, skillId, tags) in agents)
{
    var card = new JsonObject
    {
        ["agent_id"] = id,
        ["name"] = name,
        ["url"] = $"http://localhost:{port}/",
        ["skills"] = new JsonArray(new JsonObject
        {
            ["id"] = skillId,
            ["name"] = skillId,
            ["tags"] = new JsonArray(tags.Select(t => (JsonNode)t!).ToArray())
        })
    };
    var resp = await client.PostAsync(
        $"{KubeMqUrl}/agents/register",
        new StringContent(card.ToJsonString(), Encoding.UTF8, "application/json"));
    Console.WriteLine($"Registered {id}: {(int)resp.StatusCode}");
}
package main

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

const kubemqURL = "http://localhost:9090"

func main() {
	agents := []map[string]interface{}{
		{
			"agent_id": "echo-agent-01", "name": "Echo Agent",
			"url":    "http://localhost:18081/",
			"skills": []map[string]interface{}{{"id": "echo", "name": "Echo", "tags": []string{"echo"}}},
		},
		{
			"agent_id": "translate-agent-01", "name": "Translate Agent",
			"url":    "http://localhost:18082/",
			"skills": []map[string]interface{}{{"id": "translate", "name": "Translate", "tags": []string{"translate", "nlp"}}},
		},
		{
			"agent_id": "summarize-agent-01", "name": "Summarize Agent",
			"url":    "http://localhost:18083/",
			"skills": []map[string]interface{}{{"id": "summarize", "name": "Summarize", "tags": []string{"summarize", "nlp"}}},
		},
	}
	for _, card := range agents {
		data, _ := json.Marshal(card)
		resp, err := http.Post(kubemqURL+"/agents/register", "application/json", bytes.NewReader(data))
		if err != nil {
			fmt.Printf("Register %v failed: %v\n", card["agent_id"], err)
			continue
		}
		resp.Body.Close()
		fmt.Printf("Registered %v: %d\n", card["agent_id"], resp.StatusCode)
	}
}
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.util.List;
import java.util.Map;

public class Client {

    static final String KUBEMQ_URL = "http://localhost:9090";
    static final ObjectMapper MAPPER = new ObjectMapper();

    public static void main(String[] args) throws Exception {
        var client = HttpClient.newHttpClient();

        var agents = List.of(
            Map.of("agent_id", "echo-agent-01", "name", "Echo Agent",
                "url", "http://localhost:18081/",
                "skills", List.of(Map.of("id", "echo", "name", "Echo", "tags", List.of("echo")))),
            Map.of("agent_id", "translate-agent-01", "name", "Translate Agent",
                "url", "http://localhost:18082/",
                "skills", List.of(Map.of("id", "translate", "name", "Translate", "tags", List.of("translate", "nlp")))),
            Map.of("agent_id", "summarize-agent-01", "name", "Summarize Agent",
                "url", "http://localhost:18083/",
                "skills", List.of(Map.of("id", "summarize", "name", "Summarize", "tags", List.of("summarize", "nlp"))))
        );

        for (var card : agents) {
            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 " + card.get("agent_id") + ": " + resp.statusCode());
        }
    }
}
import asyncio

import httpx

KUBEMQ_URL = "http://localhost:9090"

AGENTS = [
    {"agent_id": "echo-agent-01", "name": "Echo Agent", "url": "http://localhost:18081/",
     "skills": [{"id": "echo", "name": "Echo", "tags": ["echo"]}]},
    {"agent_id": "translate-agent-01", "name": "Translate Agent", "url": "http://localhost:18082/",
     "skills": [{"id": "translate", "name": "Translate", "tags": ["translate", "nlp"]}]},
    {"agent_id": "summarize-agent-01", "name": "Summarize Agent", "url": "http://localhost:18083/",
     "skills": [{"id": "summarize", "name": "Summarize", "tags": ["summarize", "nlp"]}]},
]


async def main() -> None:
    async with httpx.AsyncClient() as client:
        for card in AGENTS:
            resp = await client.post(f"{KUBEMQ_URL}/agents/register", json=card)
            print(f"Registered {card['agent_id']}: {resp.status_code}")


if __name__ == "__main__":
    asyncio.run(main())
const KUBEMQ_URL = "http://localhost:9090";

const AGENTS = [
  { agent_id: "echo-agent-01", name: "Echo Agent", url: "http://localhost:18081/",
    skills: [{ id: "echo", name: "Echo", tags: ["echo"] }] },
  { agent_id: "translate-agent-01", name: "Translate Agent", url: "http://localhost:18082/",
    skills: [{ id: "translate", name: "Translate", tags: ["translate", "nlp"] }] },
  { agent_id: "summarize-agent-01", name: "Summarize Agent", url: "http://localhost:18083/",
    skills: [{ id: "summarize", name: "Summarize", tags: ["summarize", "nlp"] }] },
];

async function main() {
  for (const card of AGENTS) {
    const resp = await fetch(`${KUBEMQ_URL}/agents/register`, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify(card),
    });
    console.log(`Registered ${card.agent_id}: ${resp.status}`);
  }
}

main().catch(console.error);

Step 2 — Discover agents by skill

The router doesn't hardcode agent_ids — it asks the registry which agents have a needed skill. GET /agents?skill_tags=... returns only agents whose skills carry all of the requested tags (comma-separated). Filtering nlp returns both the translate and summarize agents; filtering echo returns just one.

# Agents that can do NLP work
curl "http://localhost:9090/agents?skill_tags=nlp"

# Agents that can echo
curl "http://localhost:9090/agents?skill_tags=echo"
using System.Text.Json.Nodes;

const string KubeMqUrl = "http://localhost:9090";
using var client = new HttpClient();

async Task<List<string>> Discover(string tag)
{
    var resp = await client.GetAsync($"{KubeMqUrl}/agents?skill_tags={tag}");
    var root = JsonNode.Parse(await resp.Content.ReadAsStringAsync())!;
    var agents = root is JsonArray arr ? arr : root["agents"]!.AsArray();
    return agents.Select(a => a!["agent_id"]!.GetValue<string>()).ToList();
}

Console.WriteLine($"nlp:  [{string.Join(", ", await Discover("nlp"))}]");
Console.WriteLine($"echo: [{string.Join(", ", await Discover("echo"))}]");
package main

import (
	"encoding/json"
	"fmt"
	"io"
	"net/http"
)

const kubemqURL = "http://localhost:9090"

func discover(tag string) []string {
	resp, err := http.Get(kubemqURL + "/agents?skill_tags=" + tag)
	if err != nil {
		return nil
	}
	defer resp.Body.Close()
	body, _ := io.ReadAll(resp.Body)

	var wrapper map[string]interface{}
	json.Unmarshal(body, &wrapper)
	agents, _ := wrapper["agents"].([]interface{})
	ids := make([]string, 0, len(agents))
	for _, a := range agents {
		ids = append(ids, a.(map[string]interface{})["agent_id"].(string))
	}
	return ids
}

func main() {
	fmt.Printf("nlp:  %v\n", discover("nlp"))
	fmt.Printf("echo: %v\n", discover("echo"))
}
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.util.ArrayList;
import java.util.List;

public class Client {

    static final String KUBEMQ_URL = "http://localhost:9090";
    static final ObjectMapper MAPPER = new ObjectMapper();
    static final HttpClient CLIENT = HttpClient.newHttpClient();

    static List<String> discover(String tag) throws Exception {
        var req = HttpRequest.newBuilder()
            .uri(URI.create(KUBEMQ_URL + "/agents?skill_tags=" + tag))
            .GET().build();
        var resp = CLIENT.send(req, HttpResponse.BodyHandlers.ofString());
        var root = MAPPER.readTree(resp.body());
        var agents = root.isArray() ? root : root.get("agents");
        var ids = new ArrayList<String>();
        for (var agent : agents) ids.add(agent.get("agent_id").asText());
        return ids;
    }

    public static void main(String[] args) throws Exception {
        System.out.println("nlp:  " + discover("nlp"));
        System.out.println("echo: " + discover("echo"));
    }
}
import asyncio

import httpx

KUBEMQ_URL = "http://localhost:9090"


async def discover(client: httpx.AsyncClient, tag: str) -> list[str]:
    resp = await client.get(f"{KUBEMQ_URL}/agents", params={"skill_tags": tag})
    data = resp.json()
    agents = data.get("agents", data) if isinstance(data, dict) else data
    return [a["agent_id"] for a in agents]


async def main() -> None:
    async with httpx.AsyncClient() as client:
        print(f"nlp:  {await discover(client, 'nlp')}")
        print(f"echo: {await discover(client, 'echo')}")


if __name__ == "__main__":
    asyncio.run(main())
const KUBEMQ_URL = "http://localhost:9090";

async function discover(tag: string): Promise<string[]> {
  const resp = await fetch(`${KUBEMQ_URL}/agents?skill_tags=${tag}`);
  const data = await resp.json();
  const agents = Array.isArray(data) ? data : (data.agents || []);
  return agents.map((a: { agent_id: string }) => a.agent_id);
}

async function main() {
  console.log("nlp: ", await discover("nlp"));
  console.log("echo:", await discover("echo"));
}

main().catch(console.error);

Step 3 — Route a request by agent_id

Once the router has picked an agent, it sends a normal message/send to POST /a2a/{agent_id}. The same client can fan a workload across agents by choosing a different agent_id per call — the gateway routes each request to the matching agent's virtual subscriber.

# Route to the translate agent
curl -X POST http://localhost:9090/a2a/translate-agent-01 \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "message/send",
    "params": {"message": {"parts": [{"text": "Translate: hello"}]}}
  }'

# Route to the summarize agent
curl -X POST http://localhost:9090/a2a/summarize-agent-01 \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 2,
    "method": "message/send",
    "params": {"message": {"parts": [{"text": "Summarize this report..."}]}}
  }'
using System.Text;
using System.Text.Json.Nodes;

const string KubeMqUrl = "http://localhost:9090";
using var client = new HttpClient();

async Task RouteTo(string agentId, string text)
{
    var payload = new JsonObject
    {
        ["jsonrpc"] = "2.0",
        ["id"] = 1,
        ["method"] = "message/send",
        ["params"] = new JsonObject
        {
            ["message"] = new JsonObject
            {
                ["parts"] = new JsonArray(new JsonObject { ["text"] = text })
            }
        }
    };
    var resp = await client.PostAsync(
        $"{KubeMqUrl}/a2a/{agentId}",
        new StringContent(payload.ToJsonString(), Encoding.UTF8, "application/json"));
    Console.WriteLine($"{agentId} -> {(int)resp.StatusCode}");
}

await RouteTo("translate-agent-01", "Translate: hello");
await RouteTo("summarize-agent-01", "Summarize this report...");
package main

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

const kubemqURL = "http://localhost:9090"

func routeTo(agentID, text string) {
	payload := map[string]interface{}{
		"jsonrpc": "2.0", "id": 1, "method": "message/send",
		"params": map[string]interface{}{
			"message": map[string]interface{}{
				"parts": []map[string]interface{}{{"text": text}},
			},
		},
	}
	data, _ := json.Marshal(payload)
	resp, err := http.Post(kubemqURL+"/a2a/"+agentID, "application/json", bytes.NewReader(data))
	if err != nil {
		fmt.Printf("%s -> error: %v\n", agentID, err)
		return
	}
	defer resp.Body.Close()
	fmt.Printf("%s -> %d\n", agentID, resp.StatusCode)
}

func main() {
	routeTo("translate-agent-01", "Translate: hello")
	routeTo("summarize-agent-01", "Summarize this report...")
}
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.util.List;
import java.util.Map;

public class Client {

    static final String KUBEMQ_URL = "http://localhost:9090";
    static final ObjectMapper MAPPER = new ObjectMapper();
    static final HttpClient CLIENT = HttpClient.newHttpClient();

    static void routeTo(String agentId, String text) throws Exception {
        var payload = Map.of(
            "jsonrpc", "2.0", "id", 1, "method", "message/send",
            "params", Map.of("message", Map.of("parts", List.of(Map.of("text", text))))
        );
        var req = HttpRequest.newBuilder()
            .uri(URI.create(KUBEMQ_URL + "/a2a/" + agentId))
            .header("Content-Type", "application/json")
            .POST(HttpRequest.BodyPublishers.ofString(MAPPER.writeValueAsString(payload)))
            .build();
        var resp = CLIENT.send(req, HttpResponse.BodyHandlers.ofString());
        System.out.println(agentId + " -> " + resp.statusCode());
    }

    public static void main(String[] args) throws Exception {
        routeTo("translate-agent-01", "Translate: hello");
        routeTo("summarize-agent-01", "Summarize this report...");
    }
}
import asyncio

import httpx

KUBEMQ_URL = "http://localhost:9090"


async def route_to(client: httpx.AsyncClient, agent_id: str, text: str) -> None:
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "message/send",
        "params": {"message": {"parts": [{"text": text}]}},
    }
    resp = await client.post(f"{KUBEMQ_URL}/a2a/{agent_id}", json=payload)
    print(f"{agent_id} -> {resp.status_code}")


async def main() -> None:
    async with httpx.AsyncClient() as client:
        await route_to(client, "translate-agent-01", "Translate: hello")
        await route_to(client, "summarize-agent-01", "Summarize this report...")


if __name__ == "__main__":
    asyncio.run(main())
const KUBEMQ_URL = "http://localhost:9090";

async function routeTo(agentId: string, text: string) {
  const request = {
    jsonrpc: "2.0",
    id: 1,
    method: "message/send",
    params: { message: { parts: [{ text }] } },
  };
  const resp = await fetch(`${KUBEMQ_URL}/a2a/${agentId}`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify(request),
  });
  console.log(`${agentId} -> ${resp.status}`);
}

async function main() {
  await routeTo("translate-agent-01", "Translate: hello");
  await routeTo("summarize-agent-01", "Summarize this report...");
}

main().catch(console.error);

How routing works

The gateway is a thin router, not a load balancer — agent_id selects exactly one agent.

  • Addressing. The {agent_id} path segment names the target. The gateway validates it against the registry, then issues a Query on _AGENTS_.agents/<agent_id>, which only that agent's virtual subscriber consumes.
  • Discovery vs. routing. Skill tags are a discovery convenience for picking an agent_id; they never auto-route. The caller (or its own routing logic) decides which agent_id to send to.
  • Isolation. Each agent has its own concurrency cap (AgentMaxConcurrency, default 100) and its own liveness/TTL — one busy or expired agent does not affect the others.
  • Scaling out. Want two interchangeable translators? Register them under different agent_ids with the same translate tag, then let your router pick between the matches returned by GET /agents?skill_tags=translate.

Skill-tag filtering matches agents that carry all requested tags, and the filter runs in memory after the registry fetch. To group agents for discovery, give them a shared tag (like nlp above) in addition to their specific skill.

Agent registry

Register, list, heartbeat, and deregister agents — the source of truth for routing.

Synchronous messaging

The message/send call each routed request uses.

How it works

Virtual subscribers and the _AGENTS_.agents internal channels behind routing.

Streaming task pipeline

Drive a long-running task across the gateway with message/stream.

Was this page helpful?

SSE Behavior

Reference for the A2A Server-Sent Events wire protocol — event types, the data envelope, keepalive comments, idle timeout, and client-disconnect cancellation.

Streaming Task Pipeline

Drive a long-running A2A agent task with message/stream — consume task.status, task.artifact, and task.done envelopes over SSE, then cancel by disconnecting.

On this page

The setupStep 1 — Register the agentsStep 2 — Discover agents by skillStep 3 — Route a request by agent_idHow routing worksRelated