Communication Layer of Cortex

Agent-to-Agent
Protocol Hub

Agents register, discover each other by capabilities, delegate tasks, and get results — all through a lightweight WebSocket protocol. The nervous system your agent network needs.

Get Started See Protocol

a2a-hub — zsh

# Start the hub
$ a2a-hub start
Hub listening on ws://localhost:8765

# Agents connect and register
Agent registered: code-reviewer [code-review, python]
Agent registered: test-runner [testing, pytest]
Agent registered: doc-writer [documentation, markdown]

# Discover agents by capability
$ a2a-hub agents
3 agents connected · 5 capabilities registered

# Delegate a task
task:a1b2c3 delegated to code-reviewer
task:a1b2c3 COMPLETED in 4.2s

Protocol

Everything connects through the Hub

A central WebSocket server manages agent registration, capability-based discovery, task delegation, and result routing. Any language can connect.

Connected Agents

code-reviewer

code-review, python

test-runner

testing, pytest

doc-writer

documentation, md

deployer

deploy, docker

↓

↑↓

DELEGATE

↑↓

RESULT

↑↓

HEARTBEAT

A2A Hub Server

WebSocket :8765

Agent Registry (id → AgentInfo)

Task Manager (task_id → TaskRecord)

Message Router

Task Cleanup Loop (30s)

WebSocket

MCP Bridge

FastMCP · 6 tools · JSON-RPC over stdio

stdio

Claude Code

MCP Client · Your AI

Task Lifecycle

Every delegated task follows a deterministic state machine.

PENDING

→

IN_PROGRESS

→

COMPLETED

FAILED

TIMEOUT

Protocol Messages

10 message types. Complete coverage.

All messages are Pydantic-validated, JSON-serialized, and routed through the hub. Clean, typed, and predictable.

Join the hub with a name, description, and list of capabilities. Receive an agent ID.

DEREGISTER Agent → Hub

Gracefully leave the hub. Removes agent from registry and cleans up state.

DISCOVER Agent → Hub

Find agents by capability keyword. Returns matching agent IDs with their full capability lists.

DELEGATE Agent → Hub → Agent

Assign a task to a specific agent. Hub creates a TaskRecord and routes to the target.

RESULT Agent → Hub → Agent

Return a task result back to the requester. Includes success status and payload data.

STATUS Agent ↔ Hub

Query the status of a specific task or the overall hub state. Bidirectional.

BROADCAST Agent → Hub → All

Send a message to every connected agent. Useful for announcements and coordination.

HEARTBEAT Agent → Hub

Keep-alive ping sent at 30-second intervals. Maintains connection liveness.

HEARTBEAT_ACK Hub → Agent

Hub acknowledges the heartbeat. Confirms the connection is healthy and active.

ERROR Hub → Agent

Error response for invalid requests, unknown agents, or failed operations.

Agent SDK

Two styles. Same power.

Build agents with the functional decorator style for quick scripts, or the class-based style for complex agents. Both connect to the hub automatically.

        agent_echo.py
        Functional Style
      
        # Simple, fast, decorator-based agents

        from a2a_hub import Agent

        agent = Agent("echo-agent", capabilities=["echo"])

        @agent.on_task("echo")

        async def handle_echo(message: str = "") -> dict:

            return {"echo": message}

        agent.run()

        # That's it. Agent connects, registers, handles tasks.

        # Auto-reconnect, heartbeat, cleanup — all built in.

        agent_reviewer.py
        Class-Based Style
      
        # Structured agents with multiple capabilities

        from a2a_hub import Agent, capability

        class CodeReviewAgent(Agent):

            name = "code-reviewer"

            @capability("code-review", description="Review Python code")

            async def review(self, code: str) -> dict:

                issues = analyze(code)

                return {"issues": issues, "score": 8.5}

            @capability("lint", description="Lint Python files")

            async def lint(self, path: str) -> dict:

                results = run_linter(path)

                return {"warnings": results}

        agent = CodeReviewAgent()

        agent.run()

        # Multiple capabilities, shared state, clean structure.

        # Both styles produce identical protocol behavior.

MCP Bridge

Claude Code meets the agent network

The MCP bridge exposes 6 tools that let Claude Code interact with the entire agent network. Discover, delegate, broadcast — all from your AI assistant.

▣

hub_status

Show all registered agents and total task count. Quick health check for the network.

⚲

discover_agents

Find agents by capability keyword. Returns matching agents with their full capability lists.

➔

delegate_task

Assign a task to a specific agent by ID. Creates tracked task with automatic timeout.

⏱

get_task_result

Poll for a task result with configurable timeout. Blocks until result or expiry.

⚙

broadcast_message

Send a message to every connected agent simultaneously. For coordination and announcements.

☰

list_tasks

List all tracked tasks with their current status, age, and assigned agents.

        Claude Code Session
        Using MCP Bridge
      
        # Claude Code can now talk to your agent network:

        # "Find me an agent that can review code"

        discover_agents(capability="code-review")

        # → [{id: "agent-1", name: "code-reviewer", capabilities: [...]}]

        # "Ask it to review this file"

        task = delegate_task(

            agent_id="agent-1",

            capability="code-review",

            payload={"code": "def hello(): ..."}

        )

        # "Get the result"

        result = get_task_result(task_id=task["task_id"], timeout=30)

        # → {issues: [...], score: 8.5}

Features

Built for real agent networks

Not a toy protocol. Production-ready features for reliability, discoverability, and seamless integration.

Capability-Based Discovery

Agents register what they can do, not who they are. Need code review? Search by capability, not by name. The right agent surfaces automatically.

DISCOVER → match capabilities → return agents

Task Lifecycle Management

Every task has a deterministic lifecycle: PENDING, IN_PROGRESS, COMPLETED, FAILED, or TIMEOUT. Full state tracking with automatic transitions.

TaskRecord with created_at, updated_at, TTL

TTL-Based Cleanup

Tasks auto-expire after their time-to-live. A background cleanup loop runs every 30 seconds, preventing stale task accumulation.

Default TTL: 300s · Cleanup: 30s interval

Auto-Reconnect

Agents automatically reconnect on disconnect with exponential backoff. No manual intervention needed — the network self-heals.

5s × min(attempt, 5) · max 10 attempts

Heartbeat Keep-Alive

30-second interval heartbeats maintain connection liveness. Hub detects stale connections and cleans up disconnected agents.

HEARTBEAT → HEARTBEAT_ACK · 30s cycle

MCP Bridge Integration

Claude Code interacts with the agent network through 6 MCP tools. Discover agents, delegate tasks, get results — all from your AI assistant.

FastMCP · JSON-RPC over stdio

Broadcast Messaging

Send a message to every connected agent simultaneously. Perfect for coordination signals, configuration updates, or shutdown notices.

BROADCAST → Hub → All connected agents

Language Agnostic

Any language that speaks WebSocket can connect. Python SDK included, but the protocol is simple JSON — build clients in any language.

WebSocket + JSON · No language lock-in

Architecture

Clean module structure

Six focused modules. Each does one thing well. Pydantic types flow through everything.

protocol.py

Pydantic message types for all 10 protocol messages. Type-safe serialization, validation, and helper constructors.

server.py

HubServer class. WebSocket listener, agent registry, task manager, message router, and background cleanup loop.

agent.py

Agent SDK with both class-based and functional decorator styles. Auto-reconnect, heartbeat, and task handler registration.

mcp_bridge.py

FastMCP bridge exposing 6 tools for Claude Code integration. Connects to the hub as a special client.

cli.py

Click-based CLI with 5 commands: start, bridge, status, agents, tasks. Full hub management from the terminal.

utils.py

Shared constants, logger configuration, and utility functions. Default ports, timeouts, and retry parameters.

CLI Reference

Five commands. Full control.

Manage the hub, start the bridge, and inspect state — all from your terminal.

a2a-hub start [--host HOST] [--port PORT]

Start the WebSocket hub server. Default: localhost:8765.

a2a-hub bridge [--host HOST] [--port PORT]

Start the MCP bridge server. Connects to the hub and exposes tools via stdio.

a2a-hub status [--host HOST] [--port PORT]

Query hub status: connected agents count, active tasks, uptime.

a2a-hub agents [--host HOST] [--port PORT]

List all connected agents with their IDs, names, and capabilities.

a2a-hub tasks [--host HOST] [--port PORT]

List all tracked tasks with status, assigned agent, and age.

Quick Start

Up and running in 60 seconds

Install, start the hub, create an agent, connect from Claude Code.

1 — Install Python 3.11+

pip install a2a-hub

# Or from source:
git clone https://github.com/zzhiyuann/a2a-hub.git
cd a2a-hub && pip install -e .

2 — Start the Hub

a2a-hub start

Hub listening on ws://localhost:8765
Task cleanup loop started (30s interval)
Waiting for agents...

3 — Create an Agent my_agent.py

from a2a_hub import Agent

agent = Agent("my-agent", capabilities=["greet"])

@agent.on_task("greet")
async def handle(name: str = "world") -> dict:
return {"greeting": f"Hello, {name}!"}

agent.run()

# In another terminal:
python my_agent.py
Connected to hub · Registered as my-agent

4 — Connect from Claude Code

# Add to your MCP config (~/.claude/settings.json):
{
  "mcpServers": {
    "a2a-hub": {
      "command": "a2a-hub",
      "args": ["bridge"]
    }
  }
}

# Now Claude Code can use:
hub_status() → 1 agent, 0 tasks
discover_agents("greet") → [{name: "my-agent", ...}]
delegate_task("agent-id", "greet", {"name": "Claude"})
get_task_result("task-id")
→ {"greeting": "Hello, Claude!"}

Agent-to-Agent Protocol Hub