Comunicação entre Agentes de IA com MQTT: Padrões

A comunicação entre agentes de IA via MQTT dá aos sistemas multiagente as mesmas propriedades que tornaram o MQTT o padrão em IoT: emissores e receptores desacoplados, detecção de presença, mensagens pequenas e entrega confiável em redes imperfeitas. Este artigo explica por que o publish/subscribe se encaixa bem em agentes, quais convenções de topics fazem a descoberta e a troca de mensagens funcionarem e como recursos do MQTT 5, como response topic, correlation data e expiração de mensagens, se aplicam às interações entre agentes.

Por que os agentes precisam de uma camada de mensageria

Assim que você tem mais de um agente, por exemplo um agente de monitoramento que observa a telemetria de uma planta em Campinas, um agente de planejamento que propõe ações e um agente de relatórios que escreve resumos, eles precisam de uma forma de conversar. A primeira tentativa comum são chamadas HTTP diretas entre agentes. Funciona com dois agentes e vira um sofrimento com cinco:

Cada agente precisa saber o endereço de todos os outros.
Se um agente estiver ocupado ou reiniciando, as chamadas falham e cada chamador precisa implementar novas tentativas.
Transmitir um evento ("a linha 2 parou") significa chamar individualmente cada agente interessado.
Não existe um lugar único para observar o que os agentes estão dizendo uns aos outros.

Um broker resolve isso colocando topics entre os agentes. Os emissores publicam em um topic, os receptores assinam, e o broker MQTT cuida do roteamento, da distribuição para vários assinantes e do armazenamento para sessões persistentes. Os agentes viram serviços fracamente acoplados que podem ser adicionados, reiniciados ou escalados sem reconfigurar seus pares.

Por que MQTT e não outro protocolo?

Vários sistemas de mensageria poderiam transportar o tráfego entre agentes. O MQTT tem vantagens específicas quando os agentes convivem com dispositivos:

Necessidade	Recurso do MQTT
Agentes e dispositivos na mesma espinha dorsal	Os agentes leem telemetria e enviam comandos pelo mesmo broker que os dispositivos já usam
Saber quais agentes estão ativos	Last will and testament mais mensagens de status com retained message
Quem chega depois recebe o contexto atual	Retained messages em topics de recursos e de estado
Requisição/resposta entre agentes	Response topic e correlation data do MQTT 5
Descartar requisições obsoletas	Intervalo de expiração de mensagens do MQTT 5
Distribuir trabalho entre réplicas de agentes	Shared subscriptions do MQTT 5
Rodar em hosts limitados ou remotos	Baixo overhead de protocolo, uma única conexão de saída, funciona sobre TLS e WebSockets

A última linha importa para implantações na borda: um agente em um gateway de fábrica precisa apenas de uma conexão de saída na porta 8883, sem regras de firewall de entrada.

Convenções de topics para comunicação entre agentes de IA

Um namespace consistente é o que transforma um broker em uma rede de agentes. Uma convenção simples e funcional:

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

Descoberta com cards de agente retidos

Cada agente publica, ao iniciar, um "card" retido que o descreve:

{
  "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."
}

Qualquer agente novo assina agents/+/card e recebe imediatamente todos os cards, porque o broker entrega as retained messages no momento da assinatura. Isso é um registro de serviços completo, sem nenhuma infraestrutura extra. Para remover um agente do registro, publique uma retained message vazia no topic do card dele.

Presença com last will

Cada agente se conecta com um last will offline no seu topic de status e publica um online retido assim que se conecta. Se o agente travar ou perder a rede, o broker publica o will em nome dele. Os outros agentes que assinam agents/+/status ficam sabendo, dentro do período de keepalive, que um par sumiu e podem parar de enviar trabalho a ele.

Mensagens diretas e requisição/resposta

O MQTT 5 acrescenta duas propriedades que deixam a requisição/resposta limpa. Quem faz a requisição define um response topic (para onde enviar a resposta) e um correlation data (um ID opaco para associar a resposta à pergunta). Adicionar um intervalo de expiração de mensagem garante que uma requisição que ninguém atenda a tempo seja descartada, em vez de processada muito depois de deixar de importar.

O exemplo abaixo mostra um agente respondedor e um requisitante usando paho-mqtt 2.x com MQTT 5 no servidor MQTT público (gratuito) em broker.connectmq.com, adequado para testes e protótipos, não para dados de produção. Substitua 9e1f3b pela sua própria string aleatória.

# responder.py - um agente "summarizer" que responde às requisições da sua 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", "")
    # Um agente real chamaria um LLM aqui
    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 - outro agente pedindo ajuda ao summarizer
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   # segundos; descarta a requisição se não for atendida a tempo
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")

Execute o respondedor, depois o requisitante, e acompanhe toda a conversa com:

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

Pools de workers com shared subscriptions

Quando um tipo de tarefa fica movimentado, execute várias réplicas do mesmo agente e deixe que elas compartilhem uma assinatura como $share/planners/tasks/plan. O broker entrega cada mensagem a apenas um membro do grupo, o que dá balanceamento de carga sem um sistema de filas separado. No ConnectMQ, as shared subscriptions estão disponíveis a partir do plano Serverless (Paid).

Design de mensagens para agentes

Os agentes trocam linguagem natural, mas o envelope deve ser estruturado. Uma boa mensagem carrega:

Tipo (plan.request, plan.proposal), para que os receptores possam rotear sem interpretar texto.
ID e correlação, para ligar respostas e continuações.
Remetente e timestamp, para auditoria e ordenação.
Payload com a linguagem natural em um campo claramente identificado, separado dos dados estruturados.
Expiração ou prazo, para que instruções obsoletas não sejam executadas.

Mantenha as mensagens pequenas. Passe referências, como o nome de um topic ou o ID de um documento, em vez de embutir documentos grandes. O servidor MQTT público limita as mensagens a 64 KB, e mensagens menores são mais rápidas para todos os assinantes de qualquer forma.

Segurança na comunicação entre agentes

Sistemas multiagente criam um novo risco: a saída de um agente vira a entrada de outro. Se um agente resumir uma nota de manutenção que contém instruções injetadas, um agente seguinte pode tratá-las como ordens. Defesas práticas:

Trate toda mensagem recebida como dado, nunca como instrução que se sobreponha ao prompt de sistema do próprio agente.
Dê a cada agente suas próprias credenciais de broker e seu próprio client ID, e deixe as credenciais capazes de publicar comandos para dispositivos com um único dispatcher que faça a validação. Nosso guia sobre agentes LLM para dispositivos IoT mostra esse gate de políticas em detalhes.
Registre todo o tráfego dos agentes assinando o namespace de agentes com um client de auditoria.
Limite a taxa dos agentes para que um loop entre dois deles não inunde o broker. Brokers com proteção contra flood e cotas por plano, como um broker MQTT gerenciado, oferecem uma rede de segurança, mas a correção pertence aos agentes.

Antipadrões comuns a evitar

Alguns designs parecem convenientes no início e causam problemas depois:

Um topic gigante para tudo. Publicar todo o tráfego de agentes em um único topic obriga cada agente a interpretar cada mensagem. Use inboxes por agente e topics de eventos tipados.
Loops de agentes tagarelas. Dois agentes que respondem aos broadcasts um do outro podem criar uma conversa sem fim. Dê um ID a cada mensagem, ignore as suas próprias mensagens e limite a profundidade das respostas.
Payloads grandes. Embutir documentos inteiros ou imagens nas mensagens dos agentes deixa todos os assinantes mais lentos. Publique referências e deixe os agentes buscarem o conteúdo quando precisarem.
Reter as coisas erradas. Retenha status e cards de recursos, não requisições ou resultados; caso contrário, novos assinantes recebem itens de trabalho antigos.

Como isso se relaciona com protocolos de agentes

Protocolos específicos para agentes definem como um assistente chama ferramentas ou como os agentes se descrevem, enquanto o MQTT é um transporte que move mensagens entre muitas partes. Eles se complementam: um servidor MCP para MQTT permite que um assistente de IA alcance dispositivos e agentes no broker por meio de ferramentas, e as convenções de topics acima permitem que agentes autônomos se coordenem entre si. Para uma visão geral da IA em sistemas conectados, veja nosso guia de IA e IoT e, para saber mais sobre os recursos do protocolo usados aqui, leia sobre os recursos do MQTT 5.

Comece com o ConnectMQ

Dê aos seus agentes uma espinha dorsal compartilhada com o ConnectMQ, um broker MQTT na nuvem e broker MQTT gerenciado com suporte a response topics do MQTT 5, expiração de mensagens, retained messages e expiração de sessão. Crie um broker Serverless gratuito, sem cartão de crédito, e conecte seus primeiros agentes em minutos.

Perguntas frequentes

O MQTT é rápido o suficiente para agentes de IA?

Sim. A entrega de mensagens por um broker é muito mais rápida do que a inferência de um LLM, então o transporte raramente é o gargalo. O principal trabalho de design está na estrutura de topics e nos formatos de mensagem.

Como os agentes se encontram via MQTT?

Cada agente publica um card de recursos retido em um padrão de topic conhecido. Novos agentes assinam com um curinga e recebem todos os cards imediatamente.

Dá para fazer requisição/resposta com MQTT?

Sim. Os response topics e o correlation data do MQTT 5 oferecem um padrão de requisição/resposta, e a expiração de mensagens descarta requisições que não forem atendidas a tempo.

O que acontece se um agente travar?

O broker publica o last will do agente, normalmente um status offline, para que os outros agentes parem de enviar trabalho a ele até que se reconecte.

Comunicação entre Agentes de IA com MQTT: Padrões Pub/Sub para Sistemas Multiagente