In step 1 you started KubeMQ and confirmed the Aiway endpoints are live with an empty agent roster. Now you'll build the research agent the rest of this tutorial discovers, invokes, and streams — a plain HTTP server that speaks JSON-RPC 2.0, with no KubeMQ SDK. KubeMQ's per-agent Agent Bridge (a virtual subscriber) does all the broker work; your agent only answers HTTP POSTs.

This page keeps the agent minimal so you can follow the end-to-end story. For the full agent contract — every method, header forwarding, and the lifecycle endpoints — see Building agents.

The agent listens on its own port (18080 below). It registers with — and is invoked through — the Aiway gateway on the shared HTTP port 9090. Keep both running.

Write the research agent

The agent handles two JSON-RPC methods on a single POST / route:

message/send — returns one synchronous result (used in step 3).
message/stream — emits Server-Sent Events: three task.status updates (searching → reading → summarizing), then a task.artifact, then a terminal task.done (consumed in step 4).

The task.artifact envelope follows the wire format in SSE behavior — type: "artifact" with a payload carrying a name and data. Building agents shows only the status → done path, so the artifact shape comes from the SSE behavior reference.

Save this as research_agent.py. It uses only aiohttp (server) and httpx (registration) — install them with pip install aiohttp httpx.

"""A plain-HTTP research agent — no KubeMQ SDK."""

import asyncio
import json
import signal

import httpx
from aiohttp import web

KUBEMQ_URL = "http://localhost:9090"   # Aiway gateway (register + invoke)
AGENT_ID = "research-agent-01"
AGENT_PORT = 18080                      # the agent's own HTTP port


def extract_query(body: dict) -> str:
    """Pull the caller's text out of the JSON-RPC params."""
    parts = body.get("params", {}).get("message", {}).get("parts", [])
    return " ".join(p.get("text", "") for p in parts).strip() or "(empty query)"


async def handle_send(body: dict) -> web.Response:
    """message/send — return one synchronous JSON-RPC result."""
    query = extract_query(body)
    return web.json_response({
        "jsonrpc": "2.0",
        "id": body.get("id"),
        "result": {
            "summary": f"Research summary for: {query}",
            "sources": 3,
        },
    })


async def handle_stream(request: web.Request, body: dict) -> web.StreamResponse:
    """message/stream — emit status events, an artifact, then done over SSE."""
    query = extract_query(body)
    resp = web.StreamResponse(
        status=200,
        headers={"Content-Type": "text/event-stream", "Cache-Control": "no-cache"},
    )
    await resp.prepare(request)

    stages = ["searching", "reading", "summarizing"]
    for i, status in enumerate(stages, start=1):
        event = json.dumps({
            "type": "status_update",
            "payload": {"status": status, "progress": i, "total": len(stages)},
        })
        await resp.write(f"event: task.status\ndata: {event}\n\n".encode())
        await asyncio.sleep(0.5)

    # Intermediate artifact — envelope per the SSE behavior reference.
    artifact = json.dumps({
        "type": "artifact",
        "payload": {
            "name": "summary.json",
            "data": {"summary": f"Research summary for: {query}", "sources": 3},
        },
    })
    await resp.write(f"event: task.artifact\ndata: {artifact}\n\n".encode())

    # Terminal event — stop reading after this.
    done = json.dumps({"type": "done", "payload": {"final_result": "completed", "event_count": 4}})
    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.StreamResponse:
    body = await request.json()
    if body.get("method") == "message/stream":
        return await handle_stream(request, body)
    return await handle_send(body)


async def register_agent() -> None:
    card = {
        "agent_id": AGENT_ID,
        "name": "Research Agent",
        "description": "Searches, reads, and summarizes — a mock LLM research agent.",
        "version": "1.0.0",
        "url": f"http://localhost:{AGENT_PORT}/",
        "skills": [
            {
                "id": "research",
                "name": "Research",
                "description": "Searches sources and returns a summary.",
                "tags": ["research", "summarize"],
            }
        ],
        "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"Research agent listening on port {AGENT_PORT}")

    # Always start the server BEFORE registering — KubeMQ may route the moment
    # registration succeeds.
    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())

Run it in one terminal and leave it running:

python research_agent.py

It prints Research agent listening on port 18080 and then the registration response.

Register the Agent Card

The script above self-registers on startup, but you can register (or re-register) any agent with a plain curl — useful for a sidecar agent or a language not shown here. The body is the Agent Card: agent_id, url, name, and the skills array, with protocolVersions: ["1.0"] (this is the A2A card protocol version — not the MCP session version you'll see in step 5).

curl -X POST http://localhost:9090/agents/register \
  -H "Content-Type: application/json" \
  -d '{
    "agent_id": "research-agent-01",
    "name": "Research Agent",
    "description": "Searches, reads, and summarizes — a mock LLM research agent.",
    "version": "1.0.0",
    "url": "http://localhost:18080/",
    "skills": [
      {"id": "research", "name": "Research", "description": "Searches sources and returns a summary.", "tags": ["research", "summarize"]}
    ],
    "defaultInputModes": ["text"],
    "defaultOutputModes": ["text"],
    "protocolVersions": ["1.0"]
  }'

A 200 response echoes the stored card, enriched with server-set registered_at and last_seen timestamps. See the agent registry for the full card schema and the agent cards reference for every field.

Confirm it's in the roster

In step 1 GET /agents returned an empty list. Now it returns your agent:

curl http://localhost:9090/agents

[
  {
    "agent_id": "research-agent-01",
    "name": "Research Agent",
    "url": "http://localhost:18080/",
    "skills": [
      {"id": "research", "name": "Research", "tags": ["research", "summarize"]}
    ],
    "protocolVersions": ["1.0"],
    "registered_at": "2026-06-26T10:00:00Z",
    "last_seen": "2026-06-26T10:00:00Z"
  }
]

Keep it alive (heartbeat & TTL)

The registry expires agents that go silent. A background liveness checker removes any agent whose last_seen is older than the TTL (AgentTTLSeconds, default 300s). To stay registered, an agent must heartbeat (or re-register) within that window — a beat every 30–60s gives plenty of margin:

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

For this tutorial the agent stays up the whole time, so you won't hit the TTL — but a production agent should heartbeat on a timer and deregister on shutdown. The agent registry covers heartbeat, TTL, deregistration, and ownership in full.

Next step

Your research agent is live in the fabric. Next, find it by capability and call it synchronously.

3. Discover & invoke

Find your agent by skill tag and call it synchronously with message/send.

2. Build & register an agent