Multi-Agent Communication in Go

The Problem

As agent systems grow, agents need to react to each other’s outputs: a planner publishes a task, an executor carries it out, a reviewer audits the result. If agents call each other directly, the graph of dependencies becomes a tangle — changing one agent’s interface breaks all its callers. Testing any single agent requires wiring up all its dependencies.

The Solution

A MessageBus decouples agents through typed Message values. Each agent subscribes under its own name and receives a dedicated buffered channel. To communicate, an agent calls bus.Publish(msg) — the bus routes the message to the addressed recipient (or broadcasts if To is empty). Agents never import each other; they only import the shared message type. In Go, channels are the natural implementation: the bus wraps a map[string]chan Message protected by a RWMutex.

Structure

Multi-Agent Communication Pattern

Step 1 of 4

Planner Publishes a Task

The PlannerAgent publishes a Message with Type=task.assigned and To=executor. The bus looks up the executor's channel and delivers the message — no direct reference to ExecutorAgent.

flowchart LR
Planner["PlannerAgent"]
Bus["MessageBus
Publish() / Subscribe()"]
Executor["ExecutorAgent"]
Reviewer["ReviewerAgent"]

Planner -->|"task.assigned → executor"| Bus
Bus -->|"inbox"| Executor
Executor -->|"task.completed → reviewer"| Bus
Bus -->|"inbox"| Reviewer
Reviewer -->|"review.passed → broadcast"| Bus

Implementation

package main

import "context"

// MessageType categorises a bus message so subscribers can filter.
type MessageType string

const (
	MsgTaskAssigned  MessageType = "task.assigned"
	MsgTaskCompleted MessageType = "task.completed"
	MsgReviewPassed  MessageType = "review.passed"
	MsgReviewFailed  MessageType = "review.failed"
)

// Message is the unit of communication between agents on the bus.
type Message struct {
	From    string
	To      string // empty means broadcast
	Type    MessageType
	Payload string
}

// Publisher sends messages to other agents.
type Publisher interface {
	Publish(msg Message)
}

// Agent is any actor that can send and receive messages via a bus.
type Agent interface {
	Name() string
	Run(ctx context.Context, pub Publisher, inbox <-chan Message) error
}

package main

import (
	"fmt"
	"sync"
)

// MessageBus is an in-process pub/sub bus for agent-to-agent communication.
type MessageBus struct {
	mu          sync.RWMutex
	subscribers map[string]chan Message
	bufSize     int
}

func NewMessageBus(bufSize int) *MessageBus {
	return &MessageBus{
		subscribers: make(map[string]chan Message),
		bufSize:     bufSize,
	}
}

// Subscribe registers an agent and returns its dedicated message channel.
func (b *MessageBus) Subscribe(agentID string) (<-chan Message, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if _, exists := b.subscribers[agentID]; exists {
		return nil, fmt.Errorf("agent %q already subscribed", agentID)
	}
	ch := make(chan Message, b.bufSize)
	b.subscribers[agentID] = ch
	return ch, nil
}

// Publish sends a message to its addressed recipient (or broadcasts if To is empty).
func (b *MessageBus) Publish(msg Message) {
	b.mu.RLock()
	defer b.mu.RUnlock()
	if msg.To == "" {
		for id, ch := range b.subscribers {
			if id != msg.From {
				select {
				case ch <- msg:
				default:
					// Drop on full buffer; production should use a ring buffer or back-pressure.
				}
			}
		}
		return
	}
	if ch, ok := b.subscribers[msg.To]; ok {
		select {
		case ch <- msg:
		default:
		}
	}
}

// Close drains and closes all subscriber channels.
func (b *MessageBus) Close() {
	b.mu.Lock()
	defer b.mu.Unlock()
	for _, ch := range b.subscribers {
		close(ch)
	}
	b.subscribers = make(map[string]chan Message)
}

package main

import (
	"context"
	"fmt"
	"log"
	"sync"
)

// plannerAgent decomposes work and assigns it to the executor.
type plannerAgent struct{}

func (p *plannerAgent) Name() string { return "planner" }
func (p *plannerAgent) Run(ctx context.Context, pub Publisher, inbox <-chan Message) error {
	pub.Publish(Message{
		From:    p.Name(),
		To:      "executor",
		Type:    MsgTaskAssigned,
		Payload: "implement feature X",
	})
	return nil
}

// executorAgent carries out the assigned task.
type executorAgent struct{}

func (e *executorAgent) Name() string { return "executor" }
func (e *executorAgent) Run(ctx context.Context, pub Publisher, inbox <-chan Message) error {
	msg := <-inbox
	fmt.Printf("[executor] received: %s — %s\n", msg.Type, msg.Payload)
	pub.Publish(Message{
		From:    e.Name(),
		To:      "reviewer",
		Type:    MsgTaskCompleted,
		Payload: msg.Payload + " [done]",
	})
	return nil
}

// reviewerAgent audits completed work.
type reviewerAgent struct{}

func (r *reviewerAgent) Name() string { return "reviewer" }
func (r *reviewerAgent) Run(ctx context.Context, pub Publisher, inbox <-chan Message) error {
	msg := <-inbox
	fmt.Printf("[reviewer] reviewing: %s\n", msg.Payload)
	pub.Publish(Message{
		From:    r.Name(),
		Type:    MsgReviewPassed,
		Payload: msg.Payload + " [approved]",
	})
	return nil
}

func main() {
	bus := NewMessageBus(10)

	agents := []Agent{&plannerAgent{}, &executorAgent{}, &reviewerAgent{}}
	inboxes := make(map[string]<-chan Message)
	for _, a := range agents {
		ch, err := bus.Subscribe(a.Name())
		if err != nil {
			log.Fatalf("subscribe: %v", err)
		}
		inboxes[a.Name()] = ch
	}

	ctx := context.Background()
	var wg sync.WaitGroup
	for _, a := range agents {
		wg.Add(1)
		go func(agent Agent) {
			defer wg.Done()
			if err := agent.Run(ctx, bus, inboxes[agent.Name()]); err != nil {
				log.Printf("%s error: %v", agent.Name(), err)
			}
		}(a)
	}

	wg.Wait()
	bus.Close()
	fmt.Println("All agents completed.")
	// Output:
	// [executor] received: task.assigned — implement feature X
	// [reviewer] reviewing: implement feature X [done]
	// All agents completed.
}

Real-World Analogy

An airport control tower: each aircraft (agent) communicates only with the tower (bus), never directly with other aircraft. The tower routes messages — “cleared for takeoff”, “hold position” — to the addressed party. Adding a new aircraft to the airspace means tuning to the tower frequency, not negotiating a direct radio link with every other plane.

Pros and Cons

Pros	Cons
Agents are fully decoupled — adding one never requires editing others	Bus is a central point of failure; must be robust and non-blocking
Go channels make the implementation idiomatic and race-free with `RWMutex`	Buffered channels drop messages silently on overflow — requires monitoring
Typed `MessageType` constants enable exhaustive switch handling	Debugging message flows requires tracing across multiple goroutines
Broadcast messages simplify fan-out notifications	Message ordering is per-agent-inbox, not globally guaranteed

Best Practices

Use typed MessageType constants and document every message type — undocumented message formats become tribal knowledge.
Monitor channel buffer occupancy in production; a full inbox channel is the first sign of a stuck agent.
Give every agent a context.Context so it can exit cleanly when the bus closes — avoid goroutine leaks on shutdown.
Log every message publication with From, To, Type, and a correlation ID — multi-agent traces are impossible to debug without it.
Test agents in isolation by providing a mock Publisher and a pre-loaded inbox channel; never require a live bus in unit tests.

When to Use

Systems with three or more agents that react to each other’s outputs.
Event-driven workflows where the sequence of agent activations is not fully predetermined.
Any architecture where independent deployability of agent logic is required.

When NOT to Use

Two-agent systems — a direct function call or a shared channel is simpler.
Strictly sequential pipelines where the next step is always the same — use a pipeline instead of a bus.
Systems that require guaranteed message delivery — an in-process channel bus has no persistence.

Mediator in Go Centralize communication between collaborating objects so they depend on a coordinator instead of referencing each other directly. Observer in Go Notify dependent subscribers when state changes so event-driven reactions stay decoupled from the publisher. Orchestrator-Subagent in Go Decompose a complex task by having a root agent spawn and coordinate specialized subagents, each responsible for one concern, then aggregate their results. Agent Loop (ReAct) in Go Drive an agent by alternating reasoning and acting steps inside a structured loop that terminates when the model signals a final answer.