Agent Pub/Sub in Twelve Lines of Go, With the Broker Built In

The Problem with Brokers for Agents

Message brokers are built for microservices running in the same data center. They assume:

• All producers and consumers can reach the broker over TCP
• The broker runs on infrastructure you control
• You can manage topics, partitions, and consumer groups
• You need persistence, replay, and exactly-once delivery

Agent networks break most of these assumptions. Agents sit behind NAT. They run on laptops, VMs, and edge devices. They join and leave dynamically. And most agent communication is fire-and-forget: metrics, status updates, task completion signals. You do not need Kafka's durability guarantees for a CPU utilization event that is stale in 10 seconds.

What you need is topic routing between agents that can reach each other, over connections that work through NAT, with encryption that does not require certificate management. Port 1002 gives you that, with one condition: one of the daemons acts as the broker.

How the Built-In Broker Works

There is still a broker. It is just not a separate service: it is part of the Pilot daemon you already run. The rules are short, and all of them come from the broker's source (service.go in the eventstream repository):

• The first event on a connection is its subscription. Its topic is either an exact topic name or * for every topic. Every later event on that connection is a publication.
• Topics match exactly. A subscriber to metrics.cpu gets metrics.cpu and nothing else. There are no prefix wildcards: metrics.* is just a topic called metrics.*. Only * on its own means "everything".
• Fan-out, not queues. A publication goes to every subscriber of its topic plus every * subscriber, except the connection that sent it.
• Fire-and-forget. Nothing is stored. A subscriber that is not connected when an event is published never sees it.
• Limits. Each publishing connection may send 100 events per second, with bursts of up to 200; events over the limit are dropped. A topic can have up to 1,000 subscribers. Topic names are capped at 1,024 bytes and payloads at 16 MiB.
• Slow subscribers. A failed write is retried once. After three failed or timed-out (5 s) writes in a row, the broker closes that subscriber's connection.

Who can connect depends on the hub daemon's visibility. Port 1002 is reached over a Pilot tunnel, and a private daemon (the default) only accepts tunnels from peers it trusts or shares a network with. So a private hub serves only agents that have completed a trust handshake with it. A public daemon accepts tunnels from anyone, and the broker's default topic policy allows any topic, so keep a hub private unless you want an open channel.

The Twelve-Line Subscriber

Here is a complete subscriber. The body of main is twelve lines plus a closing brace, with the error checks written on one line each (gofmt would spread them out).

package main

import (
    "fmt"
    "log"
    "os"

    "github.com/pilot-protocol/common/driver"
    "github.com/pilot-protocol/common/protocol"
    "github.com/pilot-protocol/eventstream"
)

func main() {
    d, err := driver.Connect("")                       // local daemon's IPC socket
    if err != nil { log.Fatal(err) }
    defer d.Close()
    hub, err := protocol.ParseAddr(os.Args[1])         // e.g. 0:0000.0000.1A2B
    if err != nil { log.Fatal(err) }
    sub, err := eventstream.Subscribe(d, hub, "metrics.cpu")
    if err != nil { log.Fatal(err) }
    defer sub.Close()
    for {
        evt, err := sub.Recv()
        if err != nil { log.Fatal(err) }
        fmt.Printf("[%s] %s\n", evt.Topic, evt.Payload)
    }
}

driver.Connect("") connects to the local Pilot daemon over its IPC socket (the default path when the argument is empty). eventstream.Subscribe opens a stream to port 1002 on the hub and sends the subscription. Recv blocks until the next event and returns its Topic and Payload.

The publisher is the same connection used the other way. A connection must subscribe before it publishes, so a publish-only process subscribes to a topic of its own (here a control topic for that agent) and then calls Publish with any topic it likes:

    c, err := eventstream.Subscribe(d, hub, "control.agent-a")
    if err != nil { log.Fatal(err) }
    defer c.Close()
    for {
        c.Publish("metrics.cpu", []byte(`{"percent":42.0}`))
        time.Sleep(5 * time.Second)
    }

Build a Monitoring Pipeline

Let us build something concrete. Three agents and a hub:

• Hub: any Pilot daemon all three agents trust, for example a small VM that only runs pilotctl daemon start. Its address is passed to each agent as the first argument.
• Agent A (us-east): publishes CPU metrics every 5 seconds to metrics.cpu
• Agent B (eu-west): publishes memory metrics every 5 seconds to metrics.memory
• Agent C (us-central): subscribes to both topics, aggregates, and publishes alerts to alerts.cpu

Each agent runs on its own daemon and has completed a handshake with the hub (pilotctl handshake <hub-address> on the agent, pilotctl approve <node_id> on the hub).

Agent A: CPU Publisher

package main

import (
    "encoding/json"
    "log"
    "os"
    "os/exec"
    "strconv"
    "strings"
    "time"

    "github.com/pilot-protocol/common/driver"
    "github.com/pilot-protocol/common/protocol"
    "github.com/pilot-protocol/eventstream"
)

type CPUMetric struct {
    Host    string  `json:"host"`
    Percent float64 `json:"percent"`
    Ts      int64   `json:"ts"`
}

func main() {
    d, err := driver.Connect("")
    if err != nil {
        log.Fatal(err)
    }
    defer d.Close()

    hub, err := protocol.ParseAddr(os.Args[1])
    if err != nil {
        log.Fatal(err)
    }
    host, _ := os.Hostname()

    // Subscribe to this agent's own control topic, then publish metrics.
    c, err := eventstream.Subscribe(d, hub, "control."+host)
    if err != nil {
        log.Fatal(err)
    }
    defer c.Close()

    for {
        data, _ := json.Marshal(CPUMetric{
            Host:    host,
            Percent: readCPU(),
            Ts:      time.Now().Unix(),
        })
        if err := c.Publish("metrics.cpu", data); err != nil {
            log.Fatal(err) // connection closed; restart and resubscribe
        }
        time.Sleep(5 * time.Second)
    }
}

func readCPU() float64 {
    // Linux: user CPU percentage from top
    out, _ := exec.Command("sh", "-c",
        "top -bn1 | grep 'Cpu(s)' | awk '{print $2}'").Output()
    val, _ := strconv.ParseFloat(strings.TrimSpace(string(out)), 64)
    return val
}

Agent B: Memory Publisher

package main

import (
    "encoding/json"
    "log"
    "os"
    "runtime"
    "time"

    "github.com/pilot-protocol/common/driver"
    "github.com/pilot-protocol/common/protocol"
    "github.com/pilot-protocol/eventstream"
)

type MemMetric struct {
    Host    string `json:"host"`
    AllocMB uint64 `json:"alloc_mb"`
    TotalMB uint64 `json:"total_mb"`
    Ts      int64  `json:"ts"`
}

func main() {
    d, err := driver.Connect("")
    if err != nil {
        log.Fatal(err)
    }
    defer d.Close()

    hub, err := protocol.ParseAddr(os.Args[1])
    if err != nil {
        log.Fatal(err)
    }
    host, _ := os.Hostname()

    c, err := eventstream.Subscribe(d, hub, "control."+host)
    if err != nil {
        log.Fatal(err)
    }
    defer c.Close()

    for {
        var m runtime.MemStats
        runtime.ReadMemStats(&m)

        data, _ := json.Marshal(MemMetric{
            Host:    host,
            AllocMB: m.Alloc / 1024 / 1024,
            TotalMB: m.TotalAlloc / 1024 / 1024,
            Ts:      time.Now().Unix(),
        })
        if err := c.Publish("metrics.memory", data); err != nil {
            log.Fatal(err)
        }
        time.Sleep(5 * time.Second)
    }
}

Agent C: Aggregator and Alerter

Topics match exactly, so Agent C opens one subscription per topic and merges them into one channel.

package main

import (
    "encoding/json"
    "fmt"
    "log"
    "os"
    "time"

    "github.com/pilot-protocol/common/driver"
    "github.com/pilot-protocol/common/protocol"
    "github.com/pilot-protocol/eventstream"
)

type Alert struct {
    Level   string `json:"level"`
    Message string `json:"message"`
    Source  string `json:"source"`
    Ts      int64  `json:"ts"`
}

func main() {
    d, err := driver.Connect("")
    if err != nil {
        log.Fatal(err)
    }
    defer d.Close()

    hub, err := protocol.ParseAddr(os.Args[1])
    if err != nil {
        log.Fatal(err)
    }

    // One connection per exact topic, merged into one channel.
    events := make(chan *eventstream.Event)
    var cpu *eventstream.Client
    for _, topic := range []string{"metrics.cpu", "metrics.memory"} {
        c, err := eventstream.Subscribe(d, hub, topic)
        if err != nil {
            log.Fatal(err)
        }
        defer c.Close()
        if topic == "metrics.cpu" {
            cpu = c
        }
        go func(c *eventstream.Client) {
            for {
                evt, err := c.Recv()
                if err != nil {
                    log.Fatal(err)
                }
                events <- evt
            }
        }(c)
    }

    fmt.Println("Aggregator listening on metrics.cpu and metrics.memory")

    for evt := range events {
        switch evt.Topic {
        case "metrics.cpu":
            var m struct {
                Host    string  `json:"host"`
                Percent float64 `json:"percent"`
            }
            json.Unmarshal(evt.Payload, &m)
            fmt.Printf("[CPU] %s: %.1f%%\n", m.Host, m.Percent)

            if m.Percent > 90.0 {
                alert := Alert{
                    Level:   "critical",
                    Message: fmt.Sprintf("CPU at %.1f%% on %s", m.Percent, m.Host),
                    Source:  m.Host,
                    Ts:      time.Now().Unix(),
                }
                data, _ := json.Marshal(alert)
                // Any connection can publish to any topic.
                cpu.Publish("alerts.cpu", data)
                fmt.Printf("[ALERT] %s\n", alert.Message)
            }

        case "metrics.memory":
            var m struct {
                Host    string `json:"host"`
                AllocMB uint64 `json:"alloc_mb"`
            }
            json.Unmarshal(evt.Payload, &m)
            fmt.Printf("[MEM] %s: %d MB\n", m.Host, m.AllocMB)
        }
    }
}

When CPU exceeds 90%, Agent C publishes an alert to alerts.cpu. Another agent can subscribe to alerts.cpu on the same hub to trigger remediation: scale down a workload, notify a human, or restart the offending process. If the hub carries nothing else, Agent C could instead subscribe once to * and switch on evt.Topic; with * it also receives every other topic on that hub.

Build a Workflow Pipeline

Event streams are not just for metrics. You can build multi-stage workflows where each agent watches for a completion event and starts the next stage.

The Pipeline

1. Agent 1 (Ingestion): Receives raw data, normalizes it, publishes pipeline.stage1.complete
2. Agent 2 (Analysis): Subscribes to pipeline.stage1.complete, runs analysis, publishes pipeline.stage2.complete
3. Agent 3 (Summary): Subscribes to pipeline.stage2.complete, generates a summary with an LLM, publishes pipeline.stage3.complete
4. Agent 4 (Output): Subscribes to pipeline.stage3.complete, writes the final report

Here is Agent 2 as an example. The others follow the same pattern.

package main

import (
    "encoding/json"
    "fmt"
    "log"
    "os"

    "github.com/pilot-protocol/common/driver"
    "github.com/pilot-protocol/common/protocol"
    "github.com/pilot-protocol/eventstream"
)

type StageEvent struct {
    Stage   int             `json:"stage"`
    JobID   string          `json:"job_id"`
    Payload json.RawMessage `json:"payload"`
}

func main() {
    d, err := driver.Connect("")
    if err != nil {
        log.Fatal(err)
    }
    defer d.Close()
    hub, err := protocol.ParseAddr(os.Args[1])
    if err != nil {
        log.Fatal(err)
    }

    // Wait for stage 1 to complete
    c, err := eventstream.Subscribe(d, hub, "pipeline.stage1.complete")
    if err != nil {
        log.Fatal(err)
    }
    defer c.Close()

    fmt.Println("Agent 2 (Analysis) waiting for stage 1...")

    for {
        evt, err := c.Recv()
        if err != nil {
            log.Fatal(err)
        }
        var prev StageEvent
        json.Unmarshal(evt.Payload, &prev)

        fmt.Printf("Received job %s from stage 1\n", prev.JobID)

        // Do analysis work
        result := analyze(prev.Payload)

        // Signal the next stage on the same connection
        data, _ := json.Marshal(StageEvent{
            Stage:   2,
            JobID:   prev.JobID,
            Payload: result,
        })
        c.Publish("pipeline.stage2.complete", data)

        fmt.Printf("Job %s: stage 2 complete\n", prev.JobID)
    }
}

func analyze(raw json.RawMessage) json.RawMessage {
    // Your analysis logic here
    result := map[string]any{
        "analyzed": true,
        "findings": []string{"pattern_a detected", "anomaly in sector 7"},
    }
    data, _ := json.Marshal(result)
    return data
}

Each agent is independent. Agent 2 does not know about Agent 3. It just publishes a completion event. If Agent 3 is down, the event is lost: this is fire-and-forget pub/sub, not a durable queue. When Agent 3 comes back and subscribes again, it picks up new events. If a lost job matters, have the stage that owns it re-publish until the next stage confirms, or keep jobs in a store and use events only as wake-up signals.

Branching Pipelines

Suppose stage 2 can produce two kinds of output: normal results and anomalies. Publish them to different topics:

// Normal result
c.Publish("pipeline.stage2.result.normal", data)

// Anomaly detected
c.Publish("pipeline.stage2.result.anomaly", data)

An agent that handles only anomalies subscribes to pipeline.stage2.result.anomaly. An agent that handles both opens one subscription per topic, as Agent C does above. Because there are no prefix wildcards, the routing table is the explicit list of topics each agent subscribes to, which also makes it easy to read.

CLI Equivalent for Quick Testing

You do not need to write Go to test the event stream. pilotctl has publish and subscribe commands that speak the same protocol.

# Terminal 1: subscribe to CPU metrics on the hub
pilotctl subscribe <hub-address> metrics.cpu

# Terminal 2: publish a CPU metric
pilotctl publish <hub-address> metrics.cpu --data '{"host":"agent-a","percent":73.2}'

# Terminal 3: publish a memory metric (terminal 1 does not see this one)
pilotctl publish <hub-address> metrics.memory --data '{"host":"agent-b","alloc_mb":256}'

Terminal 1 prints each matching event as [topic] payload:

[metrics.cpu] {"host":"agent-a","percent":73.2}

# Subscribe to everything on the hub (useful for debugging)
pilotctl subscribe <hub-address> '*'

# Stop after 10 events, or after 60 s
pilotctl subscribe <hub-address> pipeline.stage2.result.anomaly --count 10 --timeout 60s

# Machine-readable: one JSON object per event
pilotctl --json subscribe <hub-address> metrics.cpu

The hub can be given as an address or, once you trust it, as its hostname. Use the CLI to try out your topic names before writing Go code.

When You Still Need Kafka

Pilot's event stream is not a replacement for Kafka in all scenarios. You still need a real broker when:

• Persistence: Events must survive restarts. Pilot's stream keeps nothing. If the subscriber is offline when an event is published, that event is gone.
• Replay: You need to re-process events from a specific offset. There is no event log.
• Consumer groups: Multiple instances of the same subscriber need to split the workload. Pilot delivers every event to every subscriber of its topic, not to one of many.
• Delivery guarantees: The broker gives at-most-once delivery. Events published during a disconnect, over the rate limit, or to a subscriber whose writes keep failing are dropped.
• Volume: Each publishing connection is limited to 100 events per second (bursts of 200). Workloads far beyond that belong on infrastructure built for them.
• Pattern subscriptions: If you depend on topic wildcards or hierarchies, use a broker that has them.

But here is the question to ask yourself: does your agent system actually need any of those? Most agent event traffic is:

• Status updates (fire-and-forget)
• Task completion signals (at-most-once is fine when the application handles retries)
• Metrics (stale data is replaced by fresh data)
• Coordination signals ("I am ready", "start next phase")

For these patterns, the broker already in your Pilot daemon is often enough. Save Kafka for when you actually need Kafka.

Putting It Together

Start with the CLI to try out your topic names. Move to Go when you need programmatic control. Pick a hub daemon every agent can trust, open one subscription per exact topic, and publish completion events to chain stages. When a topic outgrows the rate limit or needs durability, move that topic to a dedicated broker and keep Pilot for the coordination traffic.

The event stream is one of the daemon's built-in services. Port 1001 carries send-message and send-file (data exchange), port 1002 carries pub/sub, and the gateway can carry ordinary HTTP over the overlay.

For the reference, see the pub/sub documentation. The Go sources are the eventstream package (github.com/pilot-protocol/eventstream) and the driver in common/driver (github.com/pilot-protocol/common).