MCP clients (Claude Code / Codex / any MCP client)
      │
  MCP Server (FastMCP, stdio)
      │
  Control Plane (FastAPI :8100)
      │
     NATS (:4222)
   ┌──┬──┬──┬── … ──┐        ← any number of agents can join/leave
 agent  agent  agent  …       (echo, researcher, writer, your own…)
      │
  Webhooks → n8n → your stack

# 1. Start NATS (native binary — no Docker needed)
nats-server -js -m 8222 --name swarm-nats
# (Alternative, if you prefer Docker:)
#   docker run -d --name nats -p 4222:4222 -p 8222:8222 nats:latest -js -m 8222

# 2. Create the Python env (system Python 3.9 is too old for `mcp`; use 3.11+)
#    This repo ships a ready-to-use venv at .venv313 (Python 3.13).
#    To recreate it:  uv venv --python 3.13 .venv313 && .venv313/bin/pip install -r requirements.txt

# 3. Start the control plane (FastAPI on :8100)
export PYTHONPATH=.
.venv313/bin/python -m core.control_plane

# 4. Start agents (each in its own terminal or as a background process)
.venv313/bin/python -m agents.echo_agent
#    Team example (two cooperating agents):
#      .venv313/bin/python -m agents.team_example researcher
#      .venv313/bin/python -m agents.team_example writer

# 5. Use the MCP server from Claude Code (see mcp-config.snippet.json)
#    Add it to your Claude Code / claude_desktop_config.json mcpServers block.

./run.sh nats        # start NATS
./run.sh control     # start the control plane
./run.sh echo        # start the echo agent
./run.sh bridge <id> # join a turn-based AI (Claude Code/GPT) — see JOINING-AI-AGENT.md
./run.sh test        # run the unit tests
make nats / make control / make echo / make test

from core.base_agent import BaseAgent
from core.models import SwarmMessage

class MyAgent(BaseAgent):
    def __init__(self):
        super().__init__(agent_id="my-agent", agent_type="worker", capabilities=["analyze"])

    async def handle_message(self, subject: str, message: SwarmMessage) -> None:
        if message.payload.get("action") == "analyze":
            result = {"ok": True}
            # For a request, reply on the one-shot reply subject:
            await self.reply_to_request(message, result)

if __name__ == "__main__":
    import asyncio
    asyncio.run(MyAgent().start())