Build a Self-Organizing Agent Swarm: No Central Scheduler

What We Are Building

The system has these properties:

• Any number of agents, one per machine or container, each with its own Pilot daemon and a role (summarizer, analyzer, reviewer, translator)
• A hub: one Pilot daemon whose built-in pub/sub broker carries a roster topic. It relays announcements and does nothing else.
• A roster that each agent builds for itself from those announcements, with entries that expire when an agent goes quiet
• Pairwise trust: every agent asks every peer for a trust handshake; when both sides have asked, the daemons approve it on their own
• Direct work requests with pilotctl send-message --wait, answered with --reply-to
• LLM integration for task execution (each agent calls an LLM to process work)
• Routing that adapts: each agent picks among peers with the right role, weighted toward the ones that answered fastest, and retries elsewhere when one does not answer

The architecture relies on three Pilot Protocol features: trust handshakes, the event stream (port 1002) for the roster, and data exchange (port 1001, used by send-message) for passing work between agents.

Prerequisites

• Python 3.10+
• Pilot Protocol on every machine (curl -fsSL https://pilotprotocol.network/install.sh | sh), with the daemon running (pilotctl daemon start)
• The public backbone (the default registry) or a registry you run yourself
• An OpenAI API key (or any LLM API) for task execution

Agent Architecture

Each agent is a Python script that drives pilotctl through subprocess calls. With --json, a command prints one JSON document of the form {"status":"ok","data":{...}} and exits non-zero on failure, so a small helper is all the integration you need (subscribe is the exception: it streams one JSON object per event). (There is also a Python SDK; the CLI keeps the example short and works from any language.)

The Helpers

import json
import os
import random
import subprocess
import sys
import threading
import time

HUB = os.environ["SWARM_HUB"]   # the hub daemon's address, e.g. 0:0000.0000.1A2B
ROSTER = "swarm.roster"          # exact topic name on the hub's broker
ANNOUNCE_EVERY = 30              # seconds between announcements
ROSTER_TTL = 90                  # forget peers not heard from for this long
ASK_AGAIN = 300                  # re-send an unanswered handshake after this long
TASK_FOR_ROLE = {
    "summarizer": "summarize",
    "analyzer": "analyze",
    "reviewer": "code_review",
    "translator": "translate",
}


def pilotctl(*args, timeout=30):
    """Run pilotctl --json and return the "data" part of its reply."""
    out = subprocess.run(["pilotctl", "--json", *args],
                         capture_output=True, text=True, timeout=timeout)
    if out.returncode != 0:
        raise RuntimeError(out.stderr.strip() or out.stdout.strip())
    return json.loads(out.stdout)["data"]


def node_id(address):
    """'0:0000.0000.1A2B' -> 0x1A2B (the last two groups are the node ID)."""
    return int(address.split(":", 1)[1].split(".", 1)[1].replace(".", ""), 16)


def trusted_ids():
    return {int(t["node_id"]) for t in pilotctl("trust")["trusted"]}

The Agent Class

class SwarmAgent:
    def __init__(self, role):
        self.role = role
        me = pilotctl("info")
        self.address = me["address"]
        self.node_id = int(me["node_id"])
        self.hostname = f"swarm-{role}-{self.node_id}"
        self.peers = {}        # address -> role, hostname, node_id, seen, latency, asked
        self.lock = threading.Lock()
        self.handled = set()   # message IDs of requests already answered

pilotctl --json info returns this daemon's address and node ID. The hostname is only a readable label; it includes the node ID because hostnames are unique on a registry.

The Hub

The hub is an ordinary Pilot daemon. Every daemon runs a small pub/sub broker on port 1002, so any machine all your agents can reach will do; a small VM is fine. Keep it private (the default): a private daemon only accepts tunnels from peers it trusts, so only agents you approved can read or write the roster.

# On the hub
pilotctl daemon start
pilotctl --json info        # note "address": the agents get it as SWARM_HUB

# As agents ask to join
pilotctl pending             # node IDs and their justification
pilotctl approve <node_id>

Approving by hand is the safe default. For a demo on machines you control, a short loop on the hub can approve requests that carry the swarm's join text:

# hub_approver.py: on the hub, approve handshakes from swarm workers
import json
import subprocess
import time

JOIN = "swarm worker, role "

while True:
    out = subprocess.run(["pilotctl", "--json", "pending"], capture_output=True, text=True)
    for req in json.loads(out.stdout)["data"]["pending"]:
        if req.get("justification", "").startswith(JOIN):
            subprocess.run(["pilotctl", "approve", str(int(req["node_id"]))])
    time.sleep(5)

Joining

    def join_hub(self):
        """Ask the hub for trust and wait until it is approved."""
        pilotctl("handshake", HUB, f"swarm worker, role {self.role}")
        while node_id(HUB) not in trusted_ids():
            time.sleep(5)
        print(f"[{self.hostname}] trusted by the hub")

A handshake is a request, not a connection: it waits on the hub until the hub approves it. Once approved, both daemons hold a trust record bound to each other's Ed25519 key, and the agent can open tunnels to the hub.

The Roster

Every agent publishes a small announcement to swarm.roster on the hub every 30 seconds and keeps a subscription to the same topic open. Topics match exactly, so the name just has to be the same everywhere.

    def announce_forever(self):
        msg = json.dumps({"address": self.address, "node_id": self.node_id,
                          "hostname": self.hostname, "role": self.role})
        while True:
            try:
                pilotctl("publish", HUB, ROSTER, "--data", msg)
            except RuntimeError as e:
                print(f"[{self.hostname}] announce failed: {e}")
            time.sleep(ANNOUNCE_EVERY)

    def watch_roster(self):
        """Keep a local roster from the announcements on the hub's broker."""
        while True:
            proc = subprocess.Popen(["pilotctl", "--json", "subscribe", HUB, ROSTER],
                                    stdout=subprocess.PIPE, text=True)
            for line in proc.stdout:
                try:
                    peer = json.loads(json.loads(line)["data"])
                    addr = peer["address"]
                except (ValueError, KeyError, TypeError):
                    continue
                if addr == self.address:
                    continue
                with self.lock:
                    entry = self.peers.setdefault(addr, {"latency": None, "asked": 0})
                    entry.update(role=peer["role"], hostname=peer["hostname"],
                                 node_id=int(peer["node_id"]), seen=time.time())
            proc.wait()
            time.sleep(5)   # subscription ended (hub restarted?): subscribe again

    def live_peers(self):
        now = time.time()
        with self.lock:
            return [(a, dict(p)) for a, p in self.peers.items()
                    if now - p.get("seen", 0) < ROSTER_TTL]

pilotctl --json subscribe prints one JSON object per event, with the payload in data. An agent that stops announcing drops out of everyone's roster after 90 seconds, and nobody has to tell the others. Each agent also sees its own announcements, which it skips.

Trust Between Agents

The roster tells an agent who exists. It does not let them talk: by default, agents are invisible, and a private daemon refuses tunnels from peers it does not trust. Each pair of agents needs its own trust record.

    def trust_peers(self):
        """Ask every live peer for trust. Every agent does the same, so each
        pair ends up with requests in both directions, and the daemons
        approve a mutual request on their own."""
        trusted = trusted_ids()
        for addr, p in self.live_peers():
            if p["node_id"] in trusted or time.time() - p["asked"] < ASK_AGAIN:
                continue
            try:
                pilotctl("handshake", addr, f"swarm peer, role {self.role}")
                with self.lock:
                    self.peers[addr]["asked"] = time.time()
            except RuntimeError as e:
                print(f"[{self.hostname}] handshake to {addr} failed: {e}")

This is where the swarm sets itself up without an operator. When agent X asks Y for trust, the request waits on Y. When Y asks X in turn, X's daemon sees that it has an open request to Y, treats the pair as mutual, approves on its own and tells Y, and both sides are trusted. Because every agent asks every peer it sees, each pair ends up mutual within one or two rounds, and nobody approves anything by hand after the hub.

The flip side is that trust follows the roster: any agent the hub approved can join every pair. That is why the hub's approval is the real admission control for the swarm.

The Work Handoff

Work goes directly from one agent to another over data exchange (port 1001). The flow is: send a task with send-message --wait, answer it with --reply-to, and the sender gets the answer inline.

Choosing the Right Peer

Each agent keeps a running average of how long each peer took to answer and picks among trusted peers with the right role, weighted toward the fastest. A peer that has not answered yet counts as one second. A failure is recorded as a 120-second answer, so a failing peer is still tried now and then and can earn its way back.

    def pick_peer(self, role, exclude=()):
        """Choose a trusted peer with the role, favouring fast responders."""
        trusted = trusted_ids()
        cands = [(a, p) for a, p in self.live_peers()
                 if p["role"] == role and p["node_id"] in trusted and a not in exclude]
        if not cands:
            return None
        weights = [1.0 / (p["latency"] or 1.0) for _, p in cands]
        return random.choices([a for a, _ in cands], weights=weights)[0]

    def record_latency(self, addr, seconds):
        with self.lock:
            old = self.peers[addr]["latency"]
            self.peers[addr]["latency"] = seconds if old is None else 0.7 * old + 0.3 * seconds

Sending Work to a Peer

    def request(self, role, task, attempts=3):
        """Send a task to a peer with the role and wait for its answer."""
        tried = set()
        for _ in range(attempts):
            addr = self.pick_peer(role, exclude=tried)
            if addr is None:
                return None
            tried.add(addr)
            start = time.time()
            try:
                out = pilotctl("send-message", addr, "--data", json.dumps(task),
                               "--wait", "120s", timeout=150)
                self.record_latency(addr, time.time() - start)
                return json.loads(out["reply"]["data"])
            except (RuntimeError, subprocess.TimeoutExpired, KeyError, ValueError) as e:
                print(f"[{self.hostname}] {addr} did not answer: {e}")
                self.record_latency(addr, 120.0)   # push it to the back
        return None

--wait 120s blocks until the peer's answer arrives and returns it inline under data.reply; its body is in data.reply.data. There is no need to read the inbox for it. If the peer does not answer in time, pilotctl exits non-zero and the agent tries another peer with the same role.

Answering Work

    def serve(self):
        """Answer work requests that arrive in the inbox."""
        while True:
            try:
                msgs = pilotctl("inbox", "--full", "--limit", "0")["messages"]
            except RuntimeError:
                time.sleep(5)
                continue
            for m in reversed(msgs):                       # oldest first
                mid = m.get("message_id")
                if not mid or mid in self.handled or m.get("reply_to"):
                    continue    # already answered, or an answer to one of our own requests
                self.handled.add(mid)
                try:
                    task = json.loads(m["data"])
                    if task.get("type") not in TASK_FOR_ROLE.values():
                        continue
                    result = execute_with_llm(self.role, task)
                    pilotctl("send-message", m["from"], "--data", json.dumps(result),
                             "--reply-to", mid)
                except Exception as e:
                    print(f"[{self.hostname}] could not answer {mid}: {e}")
            try:
                pilotctl("inbox", "--clear", "--before", "10m")   # keep the inbox small
            except RuntimeError:
                pass
            time.sleep(2)

Incoming requests land in the inbox. Every message carries a message_id, and answering with --reply-to is what lets the sender's --wait match the answer to its request. Messages that have a reply_to are answers to this agent's own requests; the serving loop leaves those alone.

Executing Tasks with LLMs

def execute_with_llm(role, task):
    """Call an LLM to process the task."""
    import openai

    client = openai.OpenAI(api_key=os.environ["OPENAI_API_KEY"])
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": f"You are a {role} agent. Task type: {task['type']}."},
            {"role": "user", "content": task["payload"]},
        ],
        max_tokens=1024,
    )
    return {"output": response.choices[0].message.content, "by": role}

Each agent is a specialist: the role goes into the system prompt, and the task type tells it what to do with the payload.

Putting It Together

    def run(self):
        pilotctl("set-hostname", self.hostname)
        self.join_hub()
        for job in (self.watch_roster, self.announce_forever, self.serve):
            threading.Thread(target=job, daemon=True).start()

        while True:
            self.trust_peers()
            role = random.choice([r for r in TASK_FOR_ROLE if r != self.role])
            task = {"type": TASK_FOR_ROLE[role], "payload": f"Sample input for a {role}."}
            result = self.request(role, task)
            if result:
                print(f"[{self.hostname}] {role} answered: {result['output'][:80]}")
            time.sleep(random.uniform(10, 30))

Start one copy per machine, each with its own role and the hub's address:

SWARM_HUB=0:0000.0000.1A2B python swarm_agent.py summarizer   # machine 1
SWARM_HUB=0:0000.0000.1A2B python swarm_agent.py analyzer     # machine 2
SWARM_HUB=0:0000.0000.1A2B python swarm_agent.py reviewer     # machine 3
SWARM_HUB=0:0000.0000.1A2B python swarm_agent.py translator   # machine 4

Add more agents with the same roles and the work spreads across them. Replace 0:0000.0000.1A2B with your hub's address.

What Organizes Itself, and What Does Not

Run the swarm for a while and this is what the code does on its own:

1. Membership. New agents appear in every roster within one announcement interval of being approved by the hub. Agents that stop announcing disappear after 90 seconds.
2. Trust. Each pair of agents becomes mutually trusted once both have asked, with no operator involved.
3. Routing. Work for a role spreads across the agents that have it, and shifts toward the ones that answer fastest. An agent that stops answering gets less work until it recovers.

What it does not do: there is no queue, so a request that fails three times is dropped and the caller decides what to do. There is no global view of load; each agent only knows the latencies it has measured itself. And the hub is a shared dependency for membership. If it goes down, agents keep their trust records and keep working with the peers in their rosters until those entries expire; new agents cannot join until it is back. Raise ROSTER_TTL to ride out longer hub outages, or run two hubs and subscribe to both.

Monitoring the Swarm

Any daemon the hub trusts can watch the roster directly, for example from one of the agents:

pilotctl subscribe 0:0000.0000.1A2B swarm.roster

Each announcement prints as [swarm.roster] {...}. pilotctl trust on an agent shows which peers it trusts, and pilotctl inbox shows the requests it has received. For a fuller monitoring setup that works across NAT boundaries without a centralized Prometheus stack, see Distributed Monitoring Without Prometheus or Grafana.

If You Run Your Own Registry

On a registry you operate, a private network can replace the hub. Creating a network needs that registry's admin token, which is why this only works on your own registry; on the public backbone only its operators can create networks. Start every daemon with --registry pointing at your registry, then:

# Once, with the registry's admin token
PILOT_ADMIN_TOKEN=... pilotctl network create --name swarm --join-rule token --token <join-secret>

# On each agent
pilotctl network join <network_id> --token <join-secret>
pilotctl --json network members <network_id>

Three things change. The member list is the roster: it returns every member's node ID and hostname, so put the role in the hostname (as the script already does) and read it from there. Hostnames of members resolve for each other. And a handshake between two members of the same network is approved automatically, so the pairwise trust step still runs but never waits. See the networks documentation for join rules and roles.

Scaling Up

A few things grow with the size of the swarm:

• Trust pairs. Full mesh trust is n(n-1)/2 pairs: 45 for 10 agents, 4,950 for 100. For a large swarm, only ask for trust from peers in roles you exchange work with.
• Roster traffic. Every announcement goes to every subscriber, so the hub's fan-out grows with the square of the swarm size. Announce less often as the swarm grows.
• Broker limits. A topic can have up to 1,000 subscribers, and each publishing connection may send 100 events per second (bursts of 200). One announcement every 30 seconds per agent is far below that.

Narrowing trust to the roles you use is a small change in trust_peers:

    def trust_peers(self, roles):
        """Ask for trust only from peers in the given roles."""
        trusted = trusted_ids()
        for addr, p in self.live_peers():
            if p["role"] not in roles or p["node_id"] in trusted:
                continue
            ...

Both sides of a pair must ask for the request to become mutual, so pick the roles symmetrically: if summarizers send work to analyzers, analyzers should also ask summarizers.

The Complete Agent Script

Here is the full script. Save it as swarm_agent.py and run it with the role as its argument and the hub's address in SWARM_HUB.

#!/usr/bin/env python3
"""Pilot Protocol swarm agent: a roster over pub/sub, pairwise trust, direct work requests."""

import json
import os
import random
import subprocess
import sys
import threading
import time

HUB = os.environ["SWARM_HUB"]   # the hub daemon's address, e.g. 0:0000.0000.1A2B
ROSTER = "swarm.roster"          # exact topic name on the hub's broker
ANNOUNCE_EVERY = 30              # seconds between announcements
ROSTER_TTL = 90                  # forget peers not heard from for this long
ASK_AGAIN = 300                  # re-send an unanswered handshake after this long
TASK_FOR_ROLE = {
    "summarizer": "summarize",
    "analyzer": "analyze",
    "reviewer": "code_review",
    "translator": "translate",
}


def pilotctl(*args, timeout=30):
    """Run pilotctl --json and return the "data" part of its reply."""
    out = subprocess.run(["pilotctl", "--json", *args],
                         capture_output=True, text=True, timeout=timeout)
    if out.returncode != 0:
        raise RuntimeError(out.stderr.strip() or out.stdout.strip())
    return json.loads(out.stdout)["data"]


def node_id(address):
    """'0:0000.0000.1A2B' -> 0x1A2B (the last two groups are the node ID)."""
    return int(address.split(":", 1)[1].split(".", 1)[1].replace(".", ""), 16)


def trusted_ids():
    return {int(t["node_id"]) for t in pilotctl("trust")["trusted"]}


def execute_with_llm(role, task):
    """Call an LLM to process the task."""
    import openai

    client = openai.OpenAI(api_key=os.environ["OPENAI_API_KEY"])
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": f"You are a {role} agent. Task type: {task['type']}."},
            {"role": "user", "content": task["payload"]},
        ],
        max_tokens=1024,
    )
    return {"output": response.choices[0].message.content, "by": role}


class SwarmAgent:
    def __init__(self, role):
        self.role = role
        me = pilotctl("info")
        self.address = me["address"]
        self.node_id = int(me["node_id"])
        self.hostname = f"swarm-{role}-{self.node_id}"
        self.peers = {}        # address -> role, hostname, node_id, seen, latency, asked
        self.lock = threading.Lock()
        self.handled = set()   # message IDs of requests already answered

    # --- joining -----------------------------------------------------

    def join_hub(self):
        """Ask the hub for trust and wait until it is approved."""
        pilotctl("handshake", HUB, f"swarm worker, role {self.role}")
        while node_id(HUB) not in trusted_ids():
            time.sleep(5)
        print(f"[{self.hostname}] trusted by the hub")

    # --- the roster ----------------------------------------------------

    def announce_forever(self):
        msg = json.dumps({"address": self.address, "node_id": self.node_id,
                          "hostname": self.hostname, "role": self.role})
        while True:
            try:
                pilotctl("publish", HUB, ROSTER, "--data", msg)
            except RuntimeError as e:
                print(f"[{self.hostname}] announce failed: {e}")
            time.sleep(ANNOUNCE_EVERY)

    def watch_roster(self):
        """Keep a local roster from the announcements on the hub's broker."""
        while True:
            proc = subprocess.Popen(["pilotctl", "--json", "subscribe", HUB, ROSTER],
                                    stdout=subprocess.PIPE, text=True)
            for line in proc.stdout:
                try:
                    peer = json.loads(json.loads(line)["data"])
                    addr = peer["address"]
                except (ValueError, KeyError, TypeError):
                    continue
                if addr == self.address:
                    continue
                with self.lock:
                    entry = self.peers.setdefault(addr, {"latency": None, "asked": 0})
                    entry.update(role=peer["role"], hostname=peer["hostname"],
                                 node_id=int(peer["node_id"]), seen=time.time())
            proc.wait()
            time.sleep(5)   # subscription ended (hub restarted?): subscribe again

    def live_peers(self):
        now = time.time()
        with self.lock:
            return [(a, dict(p)) for a, p in self.peers.items()
                    if now - p.get("seen", 0) < ROSTER_TTL]

    # --- trust ---------------------------------------------------------

    def trust_peers(self):
        """Ask every live peer for trust. Every agent does the same, so each
        pair ends up with requests in both directions, and the daemons
        approve a mutual request on their own."""
        trusted = trusted_ids()
        for addr, p in self.live_peers():
            if p["node_id"] in trusted or time.time() - p["asked"] < ASK_AGAIN:
                continue
            try:
                pilotctl("handshake", addr, f"swarm peer, role {self.role}")
                with self.lock:
                    self.peers[addr]["asked"] = time.time()
            except RuntimeError as e:
                print(f"[{self.hostname}] handshake to {addr} failed: {e}")

    # --- sending work --------------------------------------------------

    def pick_peer(self, role, exclude=()):
        """Choose a trusted peer with the role, favouring fast responders."""
        trusted = trusted_ids()
        cands = [(a, p) for a, p in self.live_peers()
                 if p["role"] == role and p["node_id"] in trusted and a not in exclude]
        if not cands:
            return None
        weights = [1.0 / (p["latency"] or 1.0) for _, p in cands]
        return random.choices([a for a, _ in cands], weights=weights)[0]

    def record_latency(self, addr, seconds):
        with self.lock:
            old = self.peers[addr]["latency"]
            self.peers[addr]["latency"] = seconds if old is None else 0.7 * old + 0.3 * seconds

    def request(self, role, task, attempts=3):
        """Send a task to a peer with the role and wait for its answer."""
        tried = set()
        for _ in range(attempts):
            addr = self.pick_peer(role, exclude=tried)
            if addr is None:
                return None
            tried.add(addr)
            start = time.time()
            try:
                out = pilotctl("send-message", addr, "--data", json.dumps(task),
                               "--wait", "120s", timeout=150)
                self.record_latency(addr, time.time() - start)
                return json.loads(out["reply"]["data"])
            except (RuntimeError, subprocess.TimeoutExpired, KeyError, ValueError) as e:
                print(f"[{self.hostname}] {addr} did not answer: {e}")
                self.record_latency(addr, 120.0)   # push it to the back
        return None

    # --- answering work ------------------------------------------------

    def serve(self):
        """Answer work requests that arrive in the inbox."""
        while True:
            try:
                msgs = pilotctl("inbox", "--full", "--limit", "0")["messages"]
            except RuntimeError:
                time.sleep(5)
                continue
            for m in reversed(msgs):                       # oldest first
                mid = m.get("message_id")
                if not mid or mid in self.handled or m.get("reply_to"):
                    continue    # already answered, or an answer to one of our own requests
                self.handled.add(mid)
                try:
                    task = json.loads(m["data"])
                    if task.get("type") not in TASK_FOR_ROLE.values():
                        continue
                    result = execute_with_llm(self.role, task)
                    pilotctl("send-message", m["from"], "--data", json.dumps(result),
                             "--reply-to", mid)
                except Exception as e:
                    print(f"[{self.hostname}] could not answer {mid}: {e}")
            try:
                pilotctl("inbox", "--clear", "--before", "10m")   # keep the inbox small
            except RuntimeError:
                pass
            time.sleep(2)

    # --- main loop -----------------------------------------------------

    def run(self):
        pilotctl("set-hostname", self.hostname)
        self.join_hub()
        for job in (self.watch_roster, self.announce_forever, self.serve):
            threading.Thread(target=job, daemon=True).start()

        while True:
            self.trust_peers()
            role = random.choice([r for r in TASK_FOR_ROLE if r != self.role])
            task = {"type": TASK_FOR_ROLE[role], "payload": f"Sample input for a {role}."}
            result = self.request(role, task)
            if result:
                print(f"[{self.hostname}] {role} answered: {result['output'][:80]}")
            time.sleep(random.uniform(10, 30))


if __name__ == "__main__":
    SwarmAgent(sys.argv[1]).run()

What You Have Built

The patterns here (a shared announcement topic, pairwise trust, direct requests with replies, and routing that learns from latency) are building blocks for larger multi-agent systems. What is missing is just as important: no scheduler, no work queue, no service mesh and no load balancer. For a simpler introduction to the same concepts, see Build a Multi-Agent Network in Five Minutes.

For the event stream itself, see agent pub/sub in Go. For a marketplace-style take on finding agent services, see building an agent marketplace on Pilot. For chaining specialized models across agent roles, see Chain AI Models Across Machines With Pilot Protocol. For the network fundamentals, see How Pilot Protocol Works.