MCP-Server-MQTT-Bridge: KI-Assistenten Ihre Geräte lesen und steuern lassen

Ein MCP-Server für MQTT gibt KI-Assistenten einen strukturierten, berechtigungsgesteuerten Weg, Live-Gerätedaten einzusehen und – wenn Sie es erlauben – Befehle zu senden. In diesem Tutorial bauen Sie in Python einen Model-Context-Protocol-Server, der einen MQTT-Broker abonniert, die letzte Nachricht pro Topic zwischenspeichert und einen kleinen Satz Tools bereitstellt. Anschließend verbinden Sie ihn mit einem MCP-fähigen Assistenten und stellen Fragen zu Ihren Geräten in natürlicher Sprache.

Was MCP einem MQTT-System hinzufügt

Das Model Context Protocol ist ein offenes Protokoll, um KI-Anwendungen mit externen Tools und Daten zu verbinden. Ein MCP-Server deklariert Tools (Funktionen mit typisierten Parametern und Beschreibungen), und der Client – typischerweise ein Desktop-Assistent oder ein IDE-Agent – lässt das Modell sie aufrufen. Was tatsächlich möglich ist, entscheidet der Server.

MQTT und MCP lösen unterschiedliche Probleme und ergänzen sich gut:

Der MCP-Server sitzt dazwischen: Auf der einen Seite ist er ein gewöhnlicher MQTT-Client, auf der anderen ein MCP-Tool-Anbieter. Da MQTT Push-basiert und MCP Pull-basiert arbeitet, hält der Server einen Cache mit der letzten Nachricht pro Topic, damit Tool-Aufrufe sofort antworten, statt auf den nächsten Publish zu warten.

Architektur

AI assistant  <--stdio / MCP-->  mqtt_mcp_server.py  <--MQTT-->  broker  <--MQTT-->  devices
                                  |- background MQTT loop
                                  |- latest-value cache
                                  |- tools: list_topics, get_latest, send_command

Der Server abonniert nur ein einziges Präfix, und Befehle können nur an .../devices/<id>/cmd mit einer Aktion aus einer Allowlist veröffentlicht werden. Das Modell erhält nie rohen Publish-Zugriff auf beliebige Topics.

Schritt 1: Abhängigkeiten installieren

Legen Sie eine virtuelle Umgebung an und installieren Sie das offizielle MCP-Python-SDK sowie paho-mqtt 2.x:

python3 -m venv .venv
source .venv/bin/activate        # unter Windows: .venv\Scripts\activate
pip install "mcp[cli]" "paho-mqtt>=2.0"

Schritt 2: Einige Geräte simulieren

Damit es etwas abzufragen gibt, starten Sie einen kleinen Simulator, der alle fünf Sekunden retained Zustände für drei Geräte veröffentlicht und Befehle bestätigt. Ersetzen Sie 5d81fa überall in diesem Tutorial durch Ihre eigene zufällige Zeichenfolge.

# simulator.py
import json
import random
import time
import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/5d81fa"
devices = {
    "pump-1": {"running": True, "flow_lpm": 42.0, "pressure_bar": 2.1},
    "tank-1": {"level_pct": 63.0, "temp_c": 21.5},
    "fan-1": {"running": True, "speed_pct": 50},
}


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/devices/+/cmd", qos=1)


def on_message(client, userdata, msg):
    device = msg.topic.split("/")[-2]
    cmd = json.loads(msg.payload)
    state = devices.get(device)
    result = "unknown_device"
    if state is not None:
        if cmd["action"] in ("start", "stop") and "running" in state:
            state["running"] = cmd["action"] == "start"
            result = "applied"
        elif cmd["action"] == "set_speed" and "speed_pct" in state:
            state["speed_pct"] = max(0, min(100, int(cmd["value"])))
            result = "applied"
        else:
            result = "unsupported"
    ack = {"id": cmd.get("id"), "result": result, "ts": int(time.time() * 1000)}
    client.publish(f"{PREFIX}/devices/{device}/ack", json.dumps(ack), qos=1)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="sim-5d81fa")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()

while True:
    devices["pump-1"]["flow_lpm"] = round(40 + random.uniform(-3, 3), 1) if devices["pump-1"]["running"] else 0.0
    devices["tank-1"]["level_pct"] = round(min(100, max(0, devices["tank-1"]["level_pct"] + random.uniform(-1, 1))), 1)
    for name, state in devices.items():
        payload = {**state, "ts": int(time.time() * 1000)}
        client.publish(f"{PREFIX}/devices/{name}/state", json.dumps(payload), qos=1, retain=True)
    time.sleep(5)

Schritt 3: Den MCP-Server für MQTT schreiben

Das ist der Kern des Tutorials. FastMCP aus dem offiziellen SDK macht aus typisierten Python-Funktionen MCP-Tools und nutzt die Docstrings als Beschreibungen, die das Modell liest.

# mqtt_mcp_server.py
import json
import logging
import sys
import threading
import time
import uuid

import paho.mqtt.client as mqtt
from mcp.server.fastmcp import FastMCP

# stdio-Transport nutzt stdout für Protokollnachrichten: nur nach stderr loggen
logging.basicConfig(stream=sys.stderr, level=logging.INFO)
log = logging.getLogger("mqtt-mcp")

BROKER = "broker.connectmq.com"
PORT = 1883
PREFIX = "connectmq/demo/5d81fa"
ALLOWED_ACTIONS = {"start", "stop", "set_speed"}

cache: dict[str, dict] = {}
acks: dict[str, dict] = {}
lock = threading.Lock()


def on_connect(client, userdata, flags, reason_code, properties):
    log.info("MQTT connected: %s", reason_code)
    client.subscribe(f"{PREFIX}/#", qos=1)


def on_message(client, userdata, msg):
    text = msg.payload.decode("utf-8", errors="replace")
    with lock:
        cache[msg.topic] = {"payload": text, "retained": msg.retain, "receivedAt": time.time()}
        if msg.topic.endswith("/ack"):
            try:
                ack = json.loads(text)
                if ack.get("id"):
                    acks[ack["id"]] = ack
            except json.JSONDecodeError:
                pass


mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="mcp-bridge-5d81fa")
mqttc.on_connect = on_connect
mqttc.on_message = on_message
mqttc.connect(BROKER, PORT, keepalive=60)
mqttc.loop_start()

mcp = FastMCP("mqtt-devices")


@mcp.tool()
def list_topics(contains: str = "") -> list[str]:
    """List MQTT topics that have received data, optionally filtered by a substring."""
    with lock:
        return sorted(t for t in cache if contains in t)


@mcp.tool()
def get_latest(device_id: str) -> dict:
    """Return the latest state message of a device (for example pump-1) and its age in seconds."""
    topic = f"{PREFIX}/devices/{device_id}/state"
    with lock:
        entry = cache.get(topic)
    if entry is None:
        return {"error": f"no data for {device_id}"}
    try:
        data = json.loads(entry["payload"])
    except json.JSONDecodeError:
        data = entry["payload"]
    return {"device": device_id, "state": data,
            "ageSeconds": round(time.time() - entry["receivedAt"], 1)}


@mcp.tool()
def send_command(device_id: str, action: str, value: float | None = None) -> dict:
    """Send a command to a device. action must be start, stop or set_speed (value 0-100).
    Waits up to 5 seconds for the device acknowledgement."""
    if action not in ALLOWED_ACTIONS:
        return {"error": f"action must be one of {sorted(ALLOWED_ACTIONS)}"}
    if action == "set_speed" and (value is None or not 0 <= value <= 100):
        return {"error": "set_speed requires a value between 0 and 100"}
    with lock:
        known = f"{PREFIX}/devices/{device_id}/state" in cache
    if not known:
        return {"error": f"unknown device {device_id}"}
    cmd_id = str(uuid.uuid4())
    cmd = {"id": cmd_id, "action": action, "value": value, "source": "mcp"}
    mqttc.publish(f"{PREFIX}/devices/{device_id}/cmd", json.dumps(cmd), qos=1)
    deadline = time.time() + 5
    while time.time() < deadline:
        with lock:
            if cmd_id in acks:
                return {"sent": cmd, "ack": acks.pop(cmd_id)}
        time.sleep(0.1)
    return {"sent": cmd, "ack": None, "note": "no acknowledgement within 5 seconds"}


if __name__ == "__main__":
    mcp.run()  # standardmäßig stdio-Transport

Drei Details sind hier entscheidend:

1. Niemals nach stdout schreiben. Beim stdio-Transport transportiert stdout das JSON-RPC-Protokoll. Ein verirrtes print() beschädigt den Datenstrom, und der Client trennt die Verbindung mit einem Parse-Fehler. Loggen Sie nach stderr.
2. Die MQTT-Schleife läuft in einem eigenen Thread. loop_start() hält den Cache aktuell, während der MCP-Server Tool-Aufrufe bearbeitet.
3. Die Validierung liegt im Server. Der Docstring sagt dem Modell, was erlaubt ist, aber der Code setzt es durch – auch bei Werten, die das Modell erfindet.

Schritt 4: Mit dem MCP Inspector testen

Starten Sie den Simulator in einem Terminal und führen Sie den Server dann unter dem Inspector aus. Dieser öffnet eine Web-Oberfläche, in der Sie Tools von Hand aufrufen können (Node.js ist erforderlich):

python simulator.py
mcp dev mqtt_mcp_server.py

Rufen Sie list_topics auf, dann get_latest mit pump-1 und anschließend send_command mit fan-1, set_speed, 70. Sie sollten eine Bestätigung mit "result": "applied" sehen. Um gleichzeitig den rohen Datenverkehr zu beobachten:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-5d81fa -t "connectmq/demo/5d81fa/#" -v

Schritt 5: Einen KI-Assistenten verbinden

MCP-Clients starten stdio-Server aus einer Konfigurationsdatei. Für Desktop-Assistenten, die das gängige mcpServers-Format verwenden, fügen Sie einen Eintrag mit absoluten Pfaden zum Python Ihrer virtuellen Umgebung und zum Skript hinzu:

{
  "mcpServers": {
    "mqtt-devices": {
      "command": "/home/you/mqtt-mcp/.venv/bin/python",
      "args": ["/home/you/mqtt-mcp/mqtt_mcp_server.py"]
    }
  }
}

Starten Sie den Assistenten neu und fragen Sie: „Wie hoch ist der Tankfüllstand gerade, und läuft die Pumpe?“ Das Modell ruft get_latest zweimal auf und antwortet mit Live-Werten. Versuchen Sie dann „Stelle fan-1 auf 30 Prozent“: Sie sehen den Tool-Aufruf, die Bestätigung und die Antwort. Die meisten Clients bitten Sie, Tool-Aufrufe zu bestätigen; lassen Sie diese Bestätigung für jedes Tool aktiviert, das den Gerätezustand verändert.

Tool-Beschreibungen, die das Modell versteht

Bei FastMCP sind Funktionsname, Parameternamen, Typ-Hinweise und Docstring alles, was das Modell über ein Tool weiß. Vage Beschreibungen führen zu vager Tool-Nutzung, behandeln Sie sie also als Teil der Schnittstelle:

1. Sagen Sie, was das Tool zurückgibt, nicht nur, was es tut. „Return the latest state message of a device and its age in seconds“ verrät dem Modell, dass es über die Aktualität der Daten nachdenken kann.
2. Nennen Sie die gültigen Werte. Erlaubte Aktionen und Wertebereiche im Docstring aufzulisten reduziert fehlgeschlagene Aufrufe, auch wenn der Code sie weiterhin durchsetzt.
3. Verwenden Sie Domänennamen, keine Topic-Pfade. Das Modell sollte nach pump-1 fragen, statt einen vollständigen Topic-String zusammenzubauen. Wenn Sie die Topic-Struktur hinter Geräte-IDs verbergen, muss das Modell keine Pfade raten, und Sie können Topics später umorganisieren.
4. Geben Sie Fehler als Daten zurück. Ein Rückgabewert wie {"error": "unknown device fan-9"} lässt das Modell sich selbst korrigieren oder dem Benutzer das Problem erklären, während eine Exception den Austausch oft mit einem generischen Fehler beendet.
5. Halten Sie die Tool-Liste kurz. Drei bis sechs fokussierte Tools funktionieren besser als ein Dutzend sich überschneidender. Wenn Sie mehr brauchen, verteilen Sie sie nach Zweck auf mehrere Server.

Hilfreich ist außerdem, in Leseergebnissen das Alter jedes Werts mitzuliefern. Ein Modell, das "ageSeconds": 900 sieht, weist meist darauf hin, dass der Messwert veraltet ist – genau die Vorsicht, die Sie wollen, wenn das Gerät dahinter offline sein könnte.

Den MCP-Server für die Produktion härten

Das Tutorial läuft auf dem kostenlosen öffentlichen MQTT-Broker, der ideal für Experimente, aber für jeden lesbar ist und daher keine Produktionsdaten transportieren sollte. Bevor Sie einen Assistenten auf echte Anlagen ansetzen:

In Industrie-4.0-Umgebungen sind die Gerätedaten häufig bereits in einem Unified Namespace (UNS) organisiert, dessen Topic-Baum der ISA-95-Hierarchie folgt (Enterprise → Site → Area → Line → Cell). Dann richten Sie den MCP-Server gezielt auf einen Teilbaum wie acme/werk-muenchen/abfuellung/#, statt den gesamten UNS freizugeben, und die Geräte-IDs der Tools entsprechen den Zellen oder Maschinen dieser Ebene. Befehlspfade in die Produktion gehören zudem in das OT-Sicherheitskonzept, etwa nach IEC 62443.

Ein praxisnahes Muster sind zwei MCP-Server: ein reiner Leseserver, den Sie bedenkenlos aktivieren können, und ein Befehlsserver, den Sie nur bei Bedarf einschalten. In einem verwalteten MQTT-Broker wie ConnectMQ erhält jeder Server im Portal seinen eigenen generierten Benutzernamen, sein Passwort und seine Client-ID, und Sie können beide Clients beim Testen im Live-Topic-Explorer beobachten. Denken Sie daran, dass der Broker keine Berechtigungen pro Topic bietet; die Trennung zwischen Lesen und Befehlen muss daher in Ihren Servern durchgesetzt werden.

Die Designprinzipien hinter sicheren Tool-Oberflächen erläutert unser Leitfaden zum Aufbau eines LLM-Agenten für IoT-Geräte, und Muster zur Schema-Validierung finden Sie unter LLM Function Calling für IoT. Die offizielle Model Context Protocol-Dokumentation beschreibt weitere Transporte wie Streamable HTTP, falls Sie den Server entfernt hosten möchten. Für das Gesamtbild verbindet unser Grundlagenleitfaden zu KI und IoT diese Bausteine.

Starten Sie mit ConnectMQ

Richten Sie Ihren MCP-Server auf einen Broker, den Sie kontrollieren. ConnectMQ bietet einen Cloud-MQTT-Broker mit MQTT 5, TLS auf Port 8883 und einem Client im Browser zum Testen. Starten Sie mit dem kostenlosen Serverless-Tarif, ohne Kreditkarte, und geben Sie Ihrem Assistenten in wenigen Minuten Live-Kontext zu Ihren Geräten.

Häufig gestellte Fragen

Warum braucht der MCP-Server einen Cache?

MQTT schiebt Nachrichten, sobald Geräte veröffentlichen, während MCP-Tool-Aufrufe Daten bei Bedarf anfordern. Der Cache speichert die letzte Nachricht pro Topic, damit ein Tool-Aufruf sofort antworten kann, statt auf den nächsten Publish zu warten.

Kann der KI-Assistent Topics direkt abonnieren?

Nicht im Sinne eines Datenstroms über MCP-Tools. Der Server abonniert stellvertretend für den Assistenten und stellt Momentaufnahmen bereit, was zugleich die Kontrolle darüber behält, welche Topics sichtbar sind.

Ist es sicher, einen Assistenten MQTT-Befehle senden zu lassen?

Nur über eng gefasste Tools mit Validierung, geräteseitigen Grenzwerten, Bestätigungen und Benutzerbestätigung. Stellen Sie einem Modell niemals ein generisches Publish-Tool zur Verfügung.

Warum zeigt mein Client beim Start des Servers einen JSON-Parse-Fehler?

Irgendetwas schreibt nach stdout, das der stdio-Transport für Protokollnachrichten reserviert. Leiten Sie sämtliche Log- und Debug-Ausgaben nach stderr.