AI Agent Communication MQTT Patterns: Pub/Sub for Multi-Agent Systems

AI agent communication over MQTT gives multi-agent systems the same properties that made MQTT the default for IoT: decoupled senders and receivers, presence detection, small messages and reliable delivery over imperfect networks. This explainer shows why publish/subscribe fits agents well, which topic conventions make discovery and messaging work, and how MQTT 5 features such as response topics, correlation data and message expiry map onto agent interactions.

Why Agents Need a Messaging Layer

Once you have more than one agent, for example a monitoring agent that watches telemetry, a planning agent that proposes actions and a reporting agent that writes summaries, they need a way to talk. The common first attempt is direct HTTP calls between agents. It works for two agents and becomes painful at five:

• Every agent must know every other agent's address.
• If an agent is busy or restarting, calls fail and each caller has to implement retries.
• Broadcasting an event ("line 2 stopped") means calling every interested agent individually.
• Nobody has a single place to observe what agents are saying to each other.

A broker solves these by putting topics between agents. Senders publish to a topic, receivers subscribe, and the MQTT broker handles routing, fan-out and buffering for persistent sessions. Agents become loosely coupled services that can be added, restarted or scaled without reconfiguring their peers.

Why MQTT Rather Than Another Protocol?

Several messaging systems could carry agent traffic. MQTT has specific advantages when agents live alongside devices:

The last row matters for edge deployments: an agent on a factory gateway needs only an outbound connection on port 8883, with no inbound firewall rules.

Topic Conventions for AI Agent Communication

A consistent namespace is what turns a broker into an agent network. A simple, workable convention:

agents/<agent-id>/status        retained: online/offline (set as last will)
agents/<agent-id>/card          retained: capabilities, version, accepted message types
agents/<agent-id>/inbox         direct messages and requests to this agent
agents/<agent-id>/outbox/<n>    responses and results produced by this agent
agents/broadcast/<event>        events every agent may react to
tasks/<capability>              work queues, consumed by shared subscriptions

Discovery With Retained Agent Cards

Each agent publishes a retained "card" describing itself when it starts:

{
  "id": "planner-1",
  "role": "maintenance planner",
  "version": "0.4.2",
  "accepts": ["plan.request"],
  "emits": ["plan.proposal"],
  "description": "Proposes maintenance actions from alarms and work order history."
}

Any new agent subscribes to agents/+/card and immediately receives every card, because the broker delivers retained messages on subscription. That is a complete service registry with no extra infrastructure. To remove an agent from the registry, publish an empty retained message to its card topic.

Presence With Last Will

Each agent connects with a last will of offline on its status topic and publishes a retained online once connected. If the agent crashes or loses its network, the broker publishes the will for it. Other agents subscribed to agents/+/status know within the keepalive period that a peer is gone and can stop sending it work.

Direct Messages and Request/Response

MQTT 5 adds two properties that make request/response clean. A requester sets a response topic (where to send the answer) and correlation data (an opaque ID to match the answer to the question). Adding a message expiry interval ensures a request that nobody handles in time is discarded rather than processed long after it matters.

The example below is a responder agent and a requester using paho-mqtt 2.x with MQTT 5. Replace 9e1f3b with your own random string.

# responder.py - a "summarizer" agent that answers requests in its inbox
import json
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

BASE = "connectmq/demo/9e1f3b/agents"
ME = "summarizer-1"


def on_connect(client, userdata, flags, reason_code, properties):
    client.publish(f"{BASE}/{ME}/status", "online", qos=1, retain=True)
    client.publish(f"{BASE}/{ME}/card", json.dumps({"id": ME, "accepts": ["summarize.request"]}), qos=1, retain=True)
    client.subscribe(f"{BASE}/{ME}/inbox", qos=1)


def on_message(client, userdata, msg):
    props = msg.properties
    response_topic = getattr(props, "ResponseTopic", None)
    if not response_topic:
        return
    request = json.loads(msg.payload)
    text = request.get("text", "")
    # A real agent would call an LLM here
    result = {"summary": text[:80] + ("..." if len(text) > 80 else ""), "agent": ME}
    reply_props = Properties(PacketTypes.PUBLISH)
    if hasattr(props, "CorrelationData"):
        reply_props.CorrelationData = props.CorrelationData
    client.publish(response_topic, json.dumps(result), qos=1, properties=reply_props)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"{ME}-9e1f3b", protocol=mqtt.MQTTv5)
client.will_set(f"{BASE}/{ME}/status", "offline", qos=1, retain=True)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_forever()

# requester.py - another agent asking the summarizer for help
import json
import threading
import uuid
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

BASE = "connectmq/demo/9e1f3b/agents"
ME = "monitor-1"
REPLY_TOPIC = f"{BASE}/{ME}/outbox/replies"
pending = {}


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(REPLY_TOPIC, qos=1)
    userdata.set()


def on_message(client, userdata, msg):
    corr = getattr(msg.properties, "CorrelationData", b"").decode()
    if corr in pending:
        print("reply for", pending.pop(corr), "->", msg.payload.decode())


ready = threading.Event()
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"{ME}-9e1f3b",
                     protocol=mqtt.MQTTv5, userdata=ready)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_start()
ready.wait(10)

corr_id = str(uuid.uuid4())
pending[corr_id] = "alarm digest"
props = Properties(PacketTypes.PUBLISH)
props.ResponseTopic = REPLY_TOPIC
props.CorrelationData = corr_id.encode()
props.MessageExpiryInterval = 30   # seconds; drop the request if not handled in time
request = {"type": "summarize.request", "text": "Line 2 conveyor stopped three times this morning due to jam sensor trips at station 4."}
client.publish(f"{BASE}/summarizer-1/inbox", json.dumps(request), qos=1, properties=props)

input("waiting for replies, press Enter to quit\n")

Run the responder, then the requester, and watch the whole conversation with:

mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 -i spy-9e1f3b -t "connectmq/demo/9e1f3b/#" -v

Worker Pools With Shared Subscriptions

When a task type gets busy, run several replicas of the same agent and let them share a subscription such as $share/planners/tasks/plan. The broker delivers each message to only one member of the group, giving you load balancing without a separate queue system. On ConnectMQ, shared subscriptions are available on the paid Serverless plan and above.

Message Design for Agents

Agents exchange natural language, but the envelope should be structured. A good message carries:

1. Type (plan.request, plan.proposal) so receivers can route without parsing prose.
2. ID and correlation to tie responses and follow-ups together.
3. Sender and timestamp for audit and ordering.
4. Payload with natural language in a clearly marked field, separate from structured data.
5. Expiry or deadline so stale instructions are not acted on.

Keep messages small. Pass references, such as a topic name or document ID, instead of embedding large documents. The free public MQTT broker caps messages at 64 KB, and smaller messages are faster for every subscriber anyway.

Safety in Agent-to-Agent Communication

Multi-agent systems create a new risk: one agent's output becomes another agent's input. If an agent summarizes a maintenance note that contains injected instructions, a downstream agent may treat them as orders. Practical defenses:

• Treat every incoming message as data, never as instructions that override an agent's own system prompt.
• Give each agent its own broker credentials and client ID, and keep the credentials that can publish device commands with a single, validating dispatcher. Our guide to LLM agents for IoT devices shows this policy gate in detail.
• Log all agent traffic by subscribing an audit client to the agent namespace.
• Rate-limit agents so a loop between two agents cannot flood the broker. Brokers with flood protection and per-plan quotas, such as a fully managed MQTT service, provide a backstop, but the fix belongs in the agents.

Common Anti-Patterns to Avoid

A few designs look convenient at first and cause trouble later:

• One giant topic for everything. Publishing all agent traffic to a single topic forces every agent to parse every message. Use per-agent inboxes and typed event topics instead.
• Chatty agent loops. Two agents that reply to each other's broadcasts can create an endless conversation. Give every message an ID, ignore your own messages and cap reply depth.
• Large payloads. Embedding whole documents or images in agent messages slows every subscriber. Publish references and let agents fetch content when they need it.
• Retaining the wrong things. Retain status and capability cards, not requests or results, otherwise new subscribers receive old work items.

How This Relates to Agent Protocols

Agent-specific protocols define how an assistant calls tools or how agents describe themselves, while MQTT is a transport that moves messages between many parties. They complement each other: an MCP server for MQTT lets an AI assistant reach devices and agents on the broker through tools, and the topic conventions above let autonomous agents coordinate among themselves. For the bigger picture of AI on connected systems, see our AI and IoT guide, and to learn more about the protocol features used here, read about MQTT 5 features.

Get Started with ConnectMQ

Give your agents a shared backbone with ConnectMQ, a cloud MQTT broker supporting MQTT 5 response topics, message expiry, retained messages and session expiry. Create a free Serverless broker, no credit card required, and connect your first agents in minutes.

Frequently Asked Questions

Is MQTT fast enough for AI agents?

Yes. Message delivery through a broker is far faster than LLM inference, so the transport is rarely the bottleneck. The main design work is in topic structure and message formats.

How do agents find each other over MQTT?

Each agent publishes a retained capability card to a known topic pattern. New agents subscribe with a wildcard and receive all cards immediately.

Can I do request/response with MQTT?

Yes. MQTT 5 response topics and correlation data provide a standard request/response pattern, and message expiry discards requests that are not handled in time.

What happens if an agent crashes?

The broker publishes the agent's last will, typically an offline status, so other agents stop sending it work until it reconnects.