OPC UA MQTT Gateway: SPS-Daten in die Cloud bringen

Q: Kann das Gateway Werte in die SPS zurückschreiben?

Technisch ja, indem es ein Command-Topic abonniert und write_value aufruft, aber behandeln Sie das als separate, sorgfältig abgesicherte Funktion. Die meisten Installationen betreiben das Gateway nur lesend.

Ein OPC UA MQTT Gateway liest Tags von SPSen und OPC-UA-Servern im Werksnetz und veröffentlicht sie als MQTT-Nachrichten an einen Broker, den Dashboards, Historians und Cloud-Anwendungen abonnieren können. In diesem Tutorial bauen Sie eines in Python mit Open-Source-Bibliotheken: Es abonniert OPC-UA-Nodes, bildet sie auf Topics im ISA-95-Stil ab und veröffentlicht sauberes JSON an einen Cloud-Broker, der erste Baustein eines Unified Namespace (UNS). Das Ganze läuft in etwa fünfzehn Minuten auf einem Laptop.

Warum OPC UA überhaupt mit MQTT verbinden?

OPC UA ist in der Fertigung hervorragend und im Referenzarchitekturmodell RAMI 4.0 der Plattform Industrie 4.0 als Kommunikationsstandard verankert. Es hat ein reichhaltiges Informationsmodell, eingebaute Sicherheit und breite Unterstützung durch SPS-Hersteller. Es ist aber ein Client/Server-Protokoll: Jeder Konsument öffnet eine eigene Session zu jedem Server. Das funktioniert für ein SCADA-System mit zehn SPSen, wird aber unhandlich, wenn zwanzig Anwendungen, einige davon in der Cloud, dieselben Daten wollen.

MQTT kehrt diese Beziehung um. Das Gateway hält eine OPC-UA-Session pro Server und eine ausgehende MQTT-Verbindung, und beliebig viele Konsumenten abonnieren beim Broker. Die Werks-Firewall braucht nur eine ausgehende Verbindung auf Port 8883, und die SPSen sehen nie Cloud-Datenverkehr. Wenn Sie die beiden Protokolle grundsätzlicher abwägen, behandelt unser Vergleich OPC UA vs. MQTT die Kompromisse; kurz gesagt nutzen die meisten Architekturen beide.

Architektur des Gateways

 PLC / OPC UA server            Edge gateway (this tutorial)              MQTT broker
 opc.tcp://plc:4840   ---->  asyncua subscription  ---> paho-mqtt  ---->  broker.connectmq.com:8883
 (tags, monitored items)     map node -> ISA-95 topic   JSON payload       dashboards, historian, apps

Das Gateway hat vier Aufgaben:

Abonnieren statt pollen. OPC-UA-Subscriptions lassen den Server Wertänderungen in einem von Ihnen gewählten Abtastintervall pushen, was weitaus effizienter ist, als jeden Tag per Timer zu lesen.
Nodes auf Topics abbilden. Node IDs wie ns=2;s=Line1.CNC1.SpindleTemp sagen einem Cloud-Entwickler nichts. Das Gateway übersetzt sie in eine lesbare Hierarchie.
Payloads normalisieren. Jede Nachricht enthält den Wert, den OPC-UA-Status, einen Quellzeitstempel und eine physikalische Einheit.
Verbindungsabbrüche überstehen. Beide Seiten werden irgendwann wegbrechen; das Gateway muss sich neu verbinden, ohne seine Konfiguration zu verlieren.

Voraussetzungen

Python 3.10 oder neuer.
asyncua (die Open-Source-Clientbibliothek von FreeOpcUa) und paho-mqtt 2.x.
Ein OPC-UA-Server. Falls Sie keine SPS zur Hand haben, enthält das Tutorial einen kleinen simulierten Server.
Ein MQTT Broker. Zum Testen nutzen Sie den öffentlichen MQTT-Broker unter broker.connectmq.com; für echte Werksdaten verwenden Sie private Zugangsdaten eines verwalteten MQTT-Brokers.

python -m venv .venv
source .venv/bin/activate        # unter Windows: .venv\Scripts\activate
pip install asyncua "paho-mqtt>=2.0"

Schritt 1: Ein simulierter OPC-UA-Server (optional)

Haben Sie bereits einen OPC-UA-Server, überspringen Sie diesen Schritt und notieren Sie dessen Endpunkt-URL und Node IDs. Andernfalls stellt dieses Skript drei sich ändernde Tags unter opc.tcp://localhost:4840 bereit.

# sim_server.py - ein winziger OPC-UA-Server mit sich ändernden CNC-Werten
import asyncio
import math
import random
from asyncua import Server, ua

async def main():
    server = Server()
    await server.init()
    server.set_endpoint("opc.tcp://0.0.0.0:4840/sim/")
    idx = await server.register_namespace("urn:demo:cnc")

    cnc = await server.nodes.objects.add_object(ua.NodeId("Line1.CNC1", idx), "CNC1")
    temp = await cnc.add_variable(ua.NodeId("Line1.CNC1.SpindleTemp", idx), "SpindleTemp", 40.0)
    speed = await cnc.add_variable(ua.NodeId("Line1.CNC1.SpindleSpeed", idx), "SpindleSpeed", 0.0)
    count = await cnc.add_variable(ua.NodeId("Line1.CNC1.PartCount", idx), "PartCount", 0)

    async with server:
        t = 0
        while True:
            t += 1
            await temp.write_value(round(45 + 5 * math.sin(t / 20) + random.random(), 2))
            await speed.write_value(float(random.choice([0, 8000, 12000])))
            if t % 10 == 0:
                await count.write_value(await count.read_value() + 1)
            await asyncio.sleep(1)

if __name__ == "__main__":
    asyncio.run(main())

Da die Node IDs String-Bezeichner im registrierten Namespace (hier Index 2) verwenden, erscheinen sie als ns=2;s=Line1.CNC1.SpindleTemp und so weiter. Bei einer echten SPS durchsuchen Sie den Adressraum mit einem Client-Tool wie UaExpert, um die genauen Node IDs zu finden.

Schritt 2: Die Tag-Zuordnung

Halten Sie die Zuordnung in der Konfiguration, nicht im Code. Jeder Eintrag verknüpft einen OPC-UA-Node mit dem Topic-Pfad und der Einheit, die Konsumenten sehen sollen. Der Topic-Aufbau folgt einer Hierarchie im ISA-95-Stil (Standort, Bereich, Linie, Zelle), die unser Leitfaden zu ISA-95-MQTT-Topics ausführlich erklärt. Diese Hierarchie ist das Rückgrat des UNS: Die ISA-95-Ebenen Enterprise → Site → Area → Line → Cell werden zur Topic-Struktur auf dem Broker, im Produktivbetrieb etwa acme/werk-stuttgart/zerspanung/linie-1/cnc-1/spindle_temp. Jede weitere SPS, die ein Gateway anbindet, erweitert denselben Baum, statt eine neue Punkt-zu-Punkt-Schnittstelle zu erzeugen; wie das mit Sparkplug B zusammenspielt, zeigt MQTT, Sparkplug B und der Unified Namespace.

{
  "opcua_endpoint": "opc.tcp://localhost:4840/sim/",
  "topic_root": "connectmq/demo/a91c/plant1",
  "tags": [
    { "node": "ns=2;s=Line1.CNC1.SpindleTemp",  "path": "machining/line1/cnc1/spindle_temp",  "unit": "degC" },
    { "node": "ns=2;s=Line1.CNC1.SpindleSpeed", "path": "machining/line1/cnc1/spindle_speed", "unit": "rpm" },
    { "node": "ns=2;s=Line1.CNC1.PartCount",    "path": "machining/line1/cnc1/part_count",    "unit": "count" }
  ]
}

Speichern Sie sie als tags.json. Ersetzen Sie a91c durch eine eigene Zufallszeichenfolge, damit Ihre Topics auf dem öffentlichen Broker nicht mit denen anderer kollidieren. Die resultierenden Topics sind acht Ebenen tief, das Maximum, das der öffentliche Broker erlaubt; auf einem privaten Broker haben Sie mehr Spielraum.

Schritt 3: Der Code des OPC UA MQTT Gateways

# gateway.py - OPC-UA-Subscription -> MQTT-Publish
import asyncio
import json
import logging
import ssl
import uuid
from datetime import timezone

import paho.mqtt.client as mqtt
from asyncua import Client

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("gateway")

CONFIG = json.load(open("tags.json"))
MQTT_HOST = "broker.connectmq.com"
MQTT_PORT = 8883
MQTT_USER = None          # Produktion: "YOUR_USERNAME"
MQTT_PASS = None          # Produktion: "YOUR_PASSWORD"
CLIENT_ID = f"opcua-gw-{uuid.uuid4().hex[:8]}"   # Produktion: "YOUR_CLIENT_ID"
STATUS_TOPIC = f"{CONFIG['topic_root']}/gateway/status"


def build_mqtt() -> mqtt.Client:
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID,
                         protocol=mqtt.MQTTv5)
    if MQTT_USER:
        client.username_pw_set(MQTT_USER, MQTT_PASS)
    client.tls_set(cert_reqs=ssl.CERT_REQUIRED)
    client.will_set(STATUS_TOPIC, json.dumps({"online": False}), qos=1, retain=True)

    def on_connect(client, userdata, flags, reason_code, properties):
        log.info("MQTT connected: %s", reason_code)
        client.publish(STATUS_TOPIC, json.dumps({"online": True}), qos=1, retain=True)

    def on_disconnect(client, userdata, flags, reason_code, properties):
        log.warning("MQTT disconnected: %s", reason_code)

    client.on_connect = on_connect
    client.on_disconnect = on_disconnect
    client.reconnect_delay_set(min_delay=1, max_delay=30)
    client.connect(MQTT_HOST, MQTT_PORT, keepalive=30)
    client.loop_start()
    return client


class DataChangeHandler:
    """asyncua ruft datachange_notification bei jeder Änderung eines Monitored Items auf."""

    def __init__(self, mqtt_client, node_map):
        self.mqtt = mqtt_client
        self.node_map = node_map

    def datachange_notification(self, node, val, data):
        tag = self.node_map[node.nodeid.to_string()]
        dv = data.monitored_item.Value
        ts = dv.SourceTimestamp or dv.ServerTimestamp
        payload = {
            "value": val,
            "unit": tag["unit"],
            "quality": dv.StatusCode.name,
            "ts": ts.replace(tzinfo=timezone.utc).isoformat() if ts else None,
        }
        topic = f"{CONFIG['topic_root']}/{tag['path']}"
        self.mqtt.publish(topic, json.dumps(payload), qos=1, retain=True)
        log.info("%s -> %s", topic, payload["value"])


async def run_opcua(mqtt_client):
    node_map = {t["node"]: t for t in CONFIG["tags"]}
    while True:
        try:
            async with Client(url=CONFIG["opcua_endpoint"]) as opc:
                log.info("OPC UA connected to %s", CONFIG["opcua_endpoint"])
                handler = DataChangeHandler(mqtt_client, node_map)
                sub = await opc.create_subscription(500, handler)  # Publishing-Intervall in ms
                nodes = [opc.get_node(n) for n in node_map]
                await sub.subscribe_data_change(nodes)
                while True:                       # Session am Leben halten, Ausfälle erkennen
                    await asyncio.sleep(5)
                    await opc.check_connection()
        except Exception as exc:                  # Server-Neustart, Netzwerkverlust usw.
            log.error("OPC UA error: %s - retrying in 5 s", exc)
            await asyncio.sleep(5)


if __name__ == "__main__":
    asyncio.run(run_opcua(build_mqtt()))

Einige Designentscheidungen verdienen Erwähnung:

Retained Messages sorgen dafür, dass ein später verbundenes Dashboard sofort den letzten Wert jedes Tags sieht.
Der Last Will auf dem Status-Topic wechselt auf {"online": false}, wenn das Gateway ohne sauberen Disconnect ausfällt, sodass Konsumenten veraltete von aktuellen Daten unterscheiden können.
Der OPC-UA-Statuscode reist mit jedem Wert. Ein Good-Messwert und ein BadCommunicationError-Messwert dürfen nachgelagert nie gleich aussehen.
check_connection() löst eine Ausnahme aus, wenn die Session verloren geht, was uns in die Wiederholungsschleife führt und die Subscription neu aufbaut.

Schritt 4: Ausführen und prüfen

Starten Sie in zwei Terminals zuerst den simulierten Server, dann das Gateway:

python sim_server.py
python gateway.py

Abonnieren Sie in einem dritten Terminal alles unter Ihrem Root:

mosquitto_sub -h broker.connectmq.com -p 1883 -i opcua-viewer-a91c \
  -t "connectmq/demo/a91c/plant1/#" -v

Sie sollten Zeilen wie diese sehen:

connectmq/demo/a91c/plant1/machining/line1/cnc1/spindle_temp {"value": 47.31, "unit": "degC", "quality": "Good", "ts": "2026-10-16T09:12:04.118000+00:00"}
connectmq/demo/a91c/plant1/gateway/status {"online": true}

Der Weg in die Produktion

Der öffentliche Broker ist für jeden lesbar und daher nur für Beispieldaten geeignet. Für echte SPS-Daten legen Sie einen privaten Broker an und setzen MQTT_USER, MQTT_PASS und CLIENT_ID auf die Zugangsdaten aus dem Portal. ConnectMQ bindet Client IDs an die ausgegebenen Zugangsdaten, verwenden Sie also die generierte Client ID statt einer zufälligen.

Checkliste, bevor Sie das Gateway ins Werksnetz bringen:

Punkt	Warum er wichtig ist
OPC-UA-Sicherheitsmodus `SignAndEncrypt` mit Client-Zertifikat	Anonyme, unverschlüsselte Sessions sind fürs Labor in Ordnung, nicht für eine SPS
Eigener OPC-UA-Benutzer nur mit Leserechten	Das Gateway sollte niemals Sollwerte schreiben können
Abtastintervall je Tag-Klasse	Schnell für Vibration, langsam für Tankfüllstände; Überflutung vermeiden
Totband bei analogen Tags	Reduziert rauschbedingte Nachrichten, siehe Report by Exception
Betrieb als Dienst mit Neustart	systemd oder ein Windows-Dienst, mit Log-Rotation
Pufferung bei Broker-Ausfällen	paho puffert während der Trennung, bei langen Ausfällen sollten Sie aber Store and Forward in Betracht ziehen
Rein ausgehende Firewall-Regel zu Port 8883	Keine eingehenden Ports zum OT-Netz geöffnet

Die rein ausgehende Verbindung aus dem OT-Netz passt zum Zonen-und-Conduits-Modell von IEC 62443. Bei ConnectMQ zeigt der Live-Topic-Explorer im Portal, wie sich Ihr ISA-95-Baum im UNS füllt, und die Geräteansicht gruppiert Maschinen nach ISA-95-Ebenen, was eine schnelle Möglichkeit ist, die Konsistenz Ihrer Topic-Zuordnung über Linien hinweg zu prüfen. Regeln für Schwellwert-Alarme können dann zum Beispiel eine Spindeltemperatur über einem Grenzwert melden, ohne zusätzlichen Code zu schreiben. Für das größere Bild, wo dieses Gateway angesiedelt ist, lesen Sie unseren Leitfaden zu Industrial IoT mit MQTT.

Fehlerbehebung

BadIdentityTokenRejected oder BadUserAccessDenied: Der Server erlaubt keinen anonymen Zugriff. Übergeben Sie einen Benutzer mit Client.set_user() und set_password() oder konfigurieren Sie Zertifikate.
BadNodeIdUnknown: Der Node-ID-String ist falsch, oft der Namespace-Index. Namespace-Indizes können sich zwischen Serverversionen ändern; lösen Sie sie beim Start mit get_namespace_index() über die Namespace-URI auf.
Werte kommen an, aber Zeitstempel sind null: Manche Server setzen keinen Quellzeitstempel. Der Code greift dann auf den Server-Zeitstempel zurück.
MQTT verbindet sich und trennt sofort wieder: Prüfen Sie auf dem öffentlichen Broker, ob Ihre Client ID eindeutig und nicht leer ist. Prüfen Sie auf einem privaten Broker, ob die Client ID zu den ausgegebenen Zugangsdaten passt.
Topics werden auf dem öffentlichen Broker abgelehnt: Sie haben acht Topic-Ebenen überschritten. Kürzen Sie den Root oder den Pfad.

Betreiben Sie Ihr OPC-UA-Gateway mit ConnectMQ

Erstellen Sie unter app.connectmq.com einen kostenlosen Broker im Serverless-Plan, ganz ohne Kreditkarte. Kopieren Sie den generierten Benutzernamen, das Passwort und die Client ID in gateway.py, und innerhalb weniger Minuten erscheinen Ihre SPS-Tags in einem privaten, TLS-geschützten Cloud-MQTT-Broker, bereit für Dashboards und Historians.

Häufig gestellte Fragen

Brauche ich ein kommerzielles Produkt für ein OPC UA MQTT Gateway?

Nein. Open-Source-Bibliotheken wie asyncua und paho-mqtt reichen für viele Projekte aus. Kommerzielle Gateways bieten zusätzlich grafische Oberflächen, Store and Forward und Herstellersupport, was im großen Maßstab wichtig sein kann.

Sollte das Gateway OPC-UA-Tags pollen oder Subscriptions nutzen?

Nutzen Sie Subscriptions. Der Server tastet die Werte ab und pusht nur Änderungen, was SPS und Netzwerk im Vergleich zu zyklischen Lesezugriffen entlastet.

Kann das Gateway Werte in die SPS zurückschreiben?

Technisch ja, indem es ein Command-Topic abonniert und write_value aufruft, aber behandeln Sie das als separate, sorgfältig abgesicherte Funktion. Die meisten Installationen betreiben das Gateway nur lesend.

Welche MQTT-QoS sollte ich für SPS-Tag-Daten verwenden?

QoS 1 ist die übliche Wahl: Die Zustellung ist mindestens einmal garantiert, und der Overhead ist gering. Kombinieren Sie sie mit Retained Messages, damit neue Konsumenten sofort den aktuellen Zustand erhalten.