KI-Agenten-Kommunikation über MQTT: Pub/Sub-Muster für Multi-Agenten-Systeme

KI-Agenten-Kommunikation über MQTT verleiht Multi-Agenten-Systemen dieselben Eigenschaften, die MQTT zum Standard im IoT gemacht haben: entkoppelte Sender und Empfänger, Präsenzerkennung, kleine Nachrichten und zuverlässige Zustellung über unvollkommene Netze. Dieser Überblick zeigt, warum Publish/Subscribe gut zu Agenten passt, mit welchen Topic-Konventionen Discovery und Nachrichtenaustausch funktionieren und wie sich MQTT-5-Funktionen wie Response Topics, Correlation Data und Message Expiry auf Agenteninteraktionen abbilden lassen.

Warum Agenten eine Messaging-Schicht brauchen

Sobald Sie mehr als einen Agenten haben – etwa einen Überwachungsagenten, der Telemetrie beobachtet, einen Planungsagenten, der Aktionen vorschlägt, und einen Berichtsagenten, der Zusammenfassungen schreibt –, brauchen diese einen Kommunikationsweg. Der übliche erste Versuch sind direkte HTTP-Aufrufe zwischen Agenten. Das funktioniert bei zwei Agenten und wird bei fünf mühsam:

• Jeder Agent muss die Adresse jedes anderen Agenten kennen.
• Ist ein Agent beschäftigt oder startet neu, schlagen Aufrufe fehl, und jeder Aufrufer muss Wiederholungen selbst implementieren.
• Ein Ereignis zu verbreiten („Linie 2 steht“) bedeutet, jeden interessierten Agenten einzeln aufzurufen.
• Es gibt keinen zentralen Ort, an dem sich beobachten lässt, was die Agenten einander mitteilen.

Ein Broker löst diese Probleme, indem er Topics zwischen die Agenten schaltet. Sender veröffentlichen auf ein Topic, Empfänger abonnieren es, und der MQTT-Broker übernimmt Routing, Fan-out und Pufferung für persistente Sessions. Agenten werden zu lose gekoppelten Diensten, die sich hinzufügen, neu starten oder skalieren lassen, ohne ihre Gegenstellen umzukonfigurieren.

Warum MQTT und kein anderes Protokoll?

Mehrere Messaging-Systeme könnten Agentenverkehr transportieren. MQTT hat spezifische Vorteile, wenn Agenten neben Geräten leben:

Die letzte Zeile ist für Edge-Deployments wichtig: Ein Agent auf einem Fabrik-Gateway braucht nur eine ausgehende Verbindung auf Port 8883, ohne eingehende Firewall-Regeln. Betreibt das Werk bereits einen Unified Namespace (UNS), finden Agenten den Kontext dort, wo ihn auch MES und SCADA lesen: unter ISA-95-Pfaden wie acme/werk-stuttgart/montage/linie-1/zelle-3/temperatur. Die Agenten-Topics unten ergänzen diesen Baum, statt ihn zu ersetzen.

Topic-Konventionen für die KI-Agenten-Kommunikation

Erst ein konsistenter Namensraum macht aus einem Broker ein Agentennetz. Eine einfache, praxistaugliche Konvention:

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 mit retained Agent Cards

Jeder Agent veröffentlicht beim Start eine retained „Card“, die ihn beschreibt:

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

Jeder neue Agent abonniert agents/+/card und erhält sofort alle Cards, weil der Broker retained Messages beim Abonnieren ausliefert. Das ist eine vollständige Service-Registry ohne zusätzliche Infrastruktur. Um einen Agenten aus der Registry zu entfernen, veröffentlichen Sie eine leere retained Message auf seinem Card-Topic.

Präsenz mit Last Will

Jeder Agent verbindet sich mit einem Last Will offline auf seinem Status-Topic und veröffentlicht nach dem Verbinden ein retained online. Stürzt der Agent ab oder verliert er sein Netzwerk, veröffentlicht der Broker den Last Will für ihn. Andere Agenten, die agents/+/status abonniert haben, wissen innerhalb der Keepalive-Periode, dass ein Partner weg ist, und können aufhören, ihm Arbeit zu schicken.

Direktnachrichten und Request/Response

MQTT 5 bringt zwei Eigenschaften mit, die Request/Response sauber machen. Ein Anfragender setzt ein Response Topic (wohin die Antwort gesendet werden soll) und Correlation Data (eine undurchsichtige ID, um die Antwort der Frage zuzuordnen). Ein zusätzliches Message Expiry Interval stellt sicher, dass eine Anfrage, die niemand rechtzeitig bearbeitet, verworfen wird, statt lange nach ihrer Relevanz verarbeitet zu werden.

Das folgende Beispiel zeigt einen antwortenden Agenten und einen anfragenden Agenten mit paho-mqtt 2.x und MQTT 5. Ersetzen Sie 9e1f3b durch Ihre eigene zufällige Zeichenfolge.

# responder.py - ein "Summarizer"-Agent, der Anfragen in seiner Inbox beantwortet
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", "")
    # Ein echter Agent würde hier ein LLM aufrufen
    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 - ein weiterer Agent, der den Summarizer um Hilfe bittet
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   # Sekunden; Anfrage verwerfen, wenn nicht rechtzeitig bearbeitet
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")

Starten Sie den Responder, dann den Requester, und verfolgen Sie die gesamte Unterhaltung mit:

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

Worker-Pools mit Shared Subscriptions

Wenn ein Aufgabentyp stark ausgelastet ist, betreiben Sie mehrere Replikate desselben Agenten und lassen Sie sie ein Abonnement wie $share/planners/tasks/plan teilen. Der Broker liefert jede Nachricht nur an ein Mitglied der Gruppe aus, was Ihnen Lastverteilung ohne separates Queue-System bietet. Bei ConnectMQ sind Shared Subscriptions ab dem kostenpflichtigen Serverless-Tarif verfügbar.

Nachrichtendesign für Agenten

Agenten tauschen natürliche Sprache aus, doch der Umschlag sollte strukturiert sein. Eine gute Nachricht enthält:

1. Typ (plan.request, plan.proposal), damit Empfänger weiterleiten können, ohne Fließtext zu parsen.
2. ID und Korrelation, um Antworten und Folgenachrichten miteinander zu verknüpfen.
3. Absender und Zeitstempel für Audit und Reihenfolge.
4. Payload mit natürlicher Sprache in einem klar gekennzeichneten Feld, getrennt von strukturierten Daten.
5. Ablaufzeit oder Frist, damit veraltete Anweisungen nicht ausgeführt werden.

Halten Sie Nachrichten klein. Übergeben Sie Referenzen wie einen Topic-Namen oder eine Dokument-ID, statt große Dokumente einzubetten. Der kostenlose öffentliche MQTT-Broker, der für Tests und Prototypen und nicht für Produktionsdaten gedacht ist, begrenzt Nachrichten auf 64 KB, und kleinere Nachrichten sind ohnehin für jeden Subscriber schneller.

Sicherheit in der Agent-zu-Agent-Kommunikation

Multi-Agenten-Systeme schaffen ein neues Risiko: Die Ausgabe eines Agenten wird zur Eingabe eines anderen. Fasst ein Agent eine Wartungsnotiz zusammen, die eingeschleuste Anweisungen enthält, behandelt ein nachgelagerter Agent diese womöglich als Befehle. Praktische Abwehrmaßnahmen:

• Behandeln Sie jede eingehende Nachricht als Daten, niemals als Anweisungen, die den eigenen System-Prompt eines Agenten überschreiben.
• Geben Sie jedem Agenten eigene Broker-Zugangsdaten und eine eigene Client-ID, und überlassen Sie die Zugangsdaten, mit denen Gerätebefehle veröffentlicht werden können, einem einzigen validierenden Dispatcher. Unser Leitfaden zu LLM-Agenten für IoT-Geräte zeigt dieses Policy-Gate im Detail.
• Protokollieren Sie den gesamten Agentenverkehr, indem Sie einen Audit-Client auf den Agenten-Namensraum abonnieren lassen.
• Begrenzen Sie die Rate der Agenten, damit eine Schleife zwischen zwei Agenten den Broker nicht fluten kann. Broker mit Flood-Schutz und Kontingenten pro Tarif, etwa ein verwalteter MQTT-Broker, bieten eine Absicherung, doch die eigentliche Lösung gehört in die Agenten.

Häufige Anti-Patterns, die Sie vermeiden sollten

Einige Designs wirken anfangs bequem und verursachen später Probleme:

• Ein riesiges Topic für alles. Wird der gesamte Agentenverkehr auf ein einziges Topic veröffentlicht, muss jeder Agent jede Nachricht parsen. Verwenden Sie stattdessen Inboxen pro Agent und typisierte Ereignis-Topics.
• Geschwätzige Agentenschleifen. Zwei Agenten, die auf die Broadcasts des jeweils anderen antworten, können eine endlose Unterhaltung erzeugen. Geben Sie jeder Nachricht eine ID, ignorieren Sie eigene Nachrichten und begrenzen Sie die Antworttiefe.
• Große Payloads. Ganze Dokumente oder Bilder in Agentennachrichten einzubetten verlangsamt jeden Subscriber. Veröffentlichen Sie Referenzen und lassen Sie Agenten Inhalte bei Bedarf abrufen.
• Das Falsche retainen. Retainen Sie Status und Capability Cards, nicht Anfragen oder Ergebnisse, sonst erhalten neue Subscriber alte Arbeitsaufträge.

Wie das mit Agentenprotokollen zusammenhängt

Agentenspezifische Protokolle legen fest, wie ein Assistent Tools aufruft oder wie Agenten sich selbst beschreiben, während MQTT ein Transport ist, der Nachrichten zwischen vielen Beteiligten bewegt. Beides ergänzt sich: Ein MCP-Server für MQTT lässt einen KI-Assistenten über Tools Geräte und Agenten am Broker erreichen, und die obigen Topic-Konventionen ermöglichen es autonomen Agenten, sich untereinander abzustimmen. Das Gesamtbild von KI in vernetzten Systemen zeigt unser Leitfaden zu KI und IoT, und mehr über die hier genutzten Protokollfunktionen erfahren Sie im Beitrag zu den Funktionen von MQTT 5.

Starten Sie mit ConnectMQ

Geben Sie Ihren Agenten ein gemeinsames Backbone mit ConnectMQ, einem Cloud-MQTT-Broker mit Unterstützung für MQTT 5 Response Topics, Message Expiry, Retained Messages und Session Expiry. Erstellen Sie einen kostenlosen Serverless-Broker, ohne Kreditkarte, und verbinden Sie Ihre ersten Agenten in wenigen Minuten.

Häufig gestellte Fragen

Ist MQTT schnell genug für KI-Agenten?

Ja. Die Nachrichtenzustellung über einen Broker ist weit schneller als LLM-Inferenz, daher ist der Transport selten der Engpass. Die eigentliche Designarbeit steckt in Topic-Struktur und Nachrichtenformaten.

Wie finden sich Agenten über MQTT?

Jeder Agent veröffentlicht eine retained Capability Card auf einem bekannten Topic-Muster. Neue Agenten abonnieren per Wildcard und erhalten sofort alle Cards.

Kann ich mit MQTT Request/Response umsetzen?

Ja. MQTT 5 Response Topics und Correlation Data bieten ein standardisiertes Request/Response-Muster, und Message Expiry verwirft Anfragen, die nicht rechtzeitig bearbeitet werden.

Was passiert, wenn ein Agent abstürzt?

Der Broker veröffentlicht den Last Will des Agenten, typischerweise einen Offline-Status, sodass andere Agenten ihm keine Arbeit mehr schicken, bis er sich wieder verbindet.