# 2. Build & register an agent (/aiway/tutorial/build-and-register-an-agent)



In [step 1](/aiway/tutorial/setup) 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](/aiway/a2a/architecture) (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](/aiway/a2a/guides/building-agents).

<Callout type="info">
  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.
</Callout>

<Steps>
  <Step>
    ## Write the research agent [#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](/aiway/tutorial/discover-and-invoke)).
    * **`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](/aiway/tutorial/stream-live-results)).

    The `task.artifact` envelope follows the wire format in
    [SSE behavior](/aiway/a2a/guides/sse-behavior) — `type: "artifact"` with a
    `payload` carrying a `name` and `data`. [Building agents](/aiway/a2a/guides/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`.

    ```python
    """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:

    ```bash
    python research_agent.py
    ```

    It prints `Research agent listening on port 18080` and then the registration response.
  </Step>

  <Step>
    ## Register the Agent Card [#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](/aiway/tutorial/orchestrate-from-an-llm)).

    ```bash
    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](/aiway/a2a/registry) for the full
    card schema and the [agent cards](/aiway/a2a/agent-cards) reference for every field.
  </Step>

  <Step>
    ## Confirm it's in the roster [#confirm-its-in-the-roster]

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

    ```bash
    curl http://localhost:9090/agents
    ```

    ```json
    [
      {
        "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"
      }
    ]
    ```
  </Step>

  <Step>
    ## Keep it alive (heartbeat & TTL) [#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:

    ```bash
    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](/aiway/a2a/registry) covers heartbeat, TTL, deregistration, and
    ownership in full.
  </Step>
</Steps>

## Next step [#next-step]

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

<Cards>
  <Card title="3. Discover & invoke" href="/aiway/tutorial/discover-and-invoke" description="Find your agent by skill tag and call it synchronously with message/send." />
</Cards>