OPC UA to MQTT Bridge for Industrial Data

An OPC UA to MQTT bridge connects to an OPC UA server as a client, subscribes to the tags you care about, and republishes every data change as an MQTT message on a well-named topic. The plant keeps OPC UA where it is strong — rich models, security, machine integration — while dashboards, historians and cloud services get a lightweight publish/subscribe feed they can scale out freely.

This tutorial builds such a bridge in Python with asyncua and paho-mqtt, maps OPC UA node ids to a Unified Namespace topic tree, and shows how to test it with a simulated OPC UA server.

Why bridge instead of connect directly

OPC UA is client/server: every consumer opens a session to every server it needs. With five consumers and twenty servers, that is a hundred sessions, each with its own subscriptions, certificates and load on the server. A bridge turns this into a hub:

The bridge is a one-way street by default: OPC UA in, MQTT out. That keeps the OT side safe while IT gets the data. For a broader comparison of the two protocols, read MQTT vs OPC UA.

1. Start a test OPC UA server

Do not build against a production server. Microsoft's open-source OPC PLC simulator runs in Docker and exposes changing values:

docker run --rm -p 50000:50000 mcr.microsoft.com/iotedge/opc-plc:latest \
  --pn=50000 --autoaccept --unsecuretransport

The endpoint is opc.tcp://localhost:50000. Browse it with a GUI client such as UaExpert, or with the uals command that ships with asyncua, and note the node ids you want to bridge. The simulator exposes nodes such as StepUp, AlternatingBoolean and RandomSignedInt32 in its own namespace; check the namespace index in your browser, because it can differ between versions.

2. Map nodes to topics

The mapping is the real design work. A topic should say where the value lives in the plant, not which server or node id happens to provide it today. Keep the map in configuration:

BASE = "testmqtt/acme/plant1/packaging/line1"

NODES = {
    "ns=3;s=StepUp":             f"{BASE}/filler/counter",
    "ns=3;s=AlternatingBoolean": f"{BASE}/filler/running",
    "ns=3;s=RandomSignedInt32":  f"{BASE}/capper/torque",
}

When you later replace the simulated server with the real one — or move a tag to a different PLC — only the left side of the map changes. Every subscriber keeps working. This is the core idea of a Unified Namespace: the topic tree is the contract, sources are implementation details.

3. The bridge

asyncua's subscription API calls a handler every time a monitored item changes. The handler publishes the value, its OPC UA quality and the source timestamp:

import asyncio, json, os
import paho.mqtt.client as mqtt
from asyncua import Client
from nodes import BASE, NODES

OPC_URL = os.getenv("OPC_URL", "opc.tcp://localhost:50000")
STATUS = f"{BASE}/_bridge/status"

mq = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="opcua-bridge-line1")
mq.username_pw_set(os.getenv("MQTT_USER"), os.getenv("MQTT_PASS"))
mq.tls_set()
mq.will_set(STATUS, "offline", qos=1, retain=True)
mq.connect("broker.connectmq.com", 8883, keepalive=30)
mq.loop_start()

class Handler:
    def datachange_notification(self, node, val, data):
        dv = data.monitored_item.Value
        ts = dv.SourceTimestamp or dv.ServerTimestamp
        payload = {
            "v": val,
            "good": dv.StatusCode.is_good(),
            "ts": ts.isoformat() if ts else None,
        }
        topic = NODES[node.nodeid.to_string()]
        mq.publish(topic, json.dumps(payload, default=str), qos=1, retain=True)

async def run():
    while True:
        try:
            async with Client(url=OPC_URL, timeout=10) as opc:
                nodes = [opc.get_node(nid) for nid in NODES]
                sub = await opc.create_subscription(500, Handler())   # 500 ms publishing interval
                await sub.subscribe_data_change(nodes)
                mq.publish(STATUS, "online", qos=1, retain=True)
                print("bridging", len(nodes), "nodes")
                while True:
                    await asyncio.sleep(5)
                    await opc.check_connection()
        except Exception as exc:          # connection lost, server restart, bad node id
            print("OPC UA error:", exc)
            mq.publish(STATUS, "opcua-unreachable", qos=1, retain=True)
            await asyncio.sleep(5)

asyncio.run(run())

pip install "asyncua>=1.1" "paho-mqtt>=2.0"
python opcua_mqtt_bridge.py

What the design gives you:

• Change-driven traffic. OPC UA sends notifications only when values change, so the MQTT side carries no redundant samples. Tune the publishing interval (500 ms here) and per-item sampling interval to your process.
• Quality travels with the value. A bad or uncertain OPC UA status code is information; keep it in the payload instead of silently dropping the message.
• Source timestamps. Historians should store when the value changed at the source, not when the bridge forwarded it.
• Retained values and a status topic. New subscribers see the current state immediately, and the Last Will flips the bridge status to offline if the process dies. More on that in retained messages.
• Automatic recovery. If the OPC UA server restarts, the outer loop reconnects and recreates the subscription. paho reconnects to the broker on its own after loop_start().

4. Test the bridge

1. Start the simulator and the bridge. The bridge should print the number of bridged nodes.
2. Open the online MQTT client, connect to broker.connectmq.com and subscribe to testmqtt/acme/plant1/#. filler/counter should tick up, filler/running toggle, and _bridge/status show online.
3. Stop the simulator container. Within a few seconds the status should become opcua-unreachable; restart it and values should resume without restarting the bridge.
4. Kill the bridge with kill -9. After the keep-alive timeout the broker publishes the Last Will: offline.

Watch bridged OPC UA tags in your browser

Check that topic filters used by consumers match what the bridge publishes with the topic matcher, and keep large structured values (arrays, extension objects) in mind: the packet size calculator helps estimate what one notification costs on the wire.

Other ways to bridge

A small Python bridge is ideal for learning, for a handful of servers and for full control of the mapping. For larger estates, consider:

Whatever tool you choose, the same rules apply: topics describe the plant, payloads carry value + quality + timestamp, the bridge announces its own health, and writes back into OPC UA are a separate, deliberately secured path.

Production checklist

• Use OPC UA security (Basic256Sha256 sign and encrypt) and a dedicated application certificate for the bridge on real servers; the unsecured simulator setting is for local tests only.
• Read the server's limits on monitored items per subscription and split large tag lists across several subscriptions.
• Run one bridge per area or line so a failure stays local.
• Version the node-to-topic map and review it like code.
• Monitor the bridge status topic and alert when it is not online.

An OPC UA to MQTT bridge lets you keep OPC UA for what it does best while giving every other system a simple, scalable feed. Map nodes to plant-shaped topics, publish change notifications with quality and timestamps, test against a simulator and the online client, and your OT data becomes something the whole company can subscribe to.

Frequently asked questions