OPC UA to MQTT Gateway: Bridge PLC Data to the Cloud

An OPC UA to MQTT gateway reads tags from PLCs and OPC UA servers on the plant network and publishes them as MQTT messages to a broker that dashboards, historians and cloud apps can subscribe to. In this tutorial you will build one in Python with open-source libraries: it subscribes to OPC UA nodes, maps them to ISA-95 style topics and publishes clean JSON to a cloud broker. You can run the whole thing on a laptop in about fifteen minutes.

Why Bridge OPC UA to MQTT at All?

OPC UA is excellent on the plant floor. It has a rich information model, built-in security and broad support across PLC vendors. But it is a client/server protocol: every consumer opens its own session to every server. That works for a SCADA system talking to ten PLCs, and becomes awkward when twenty applications, some of them in the cloud, want the same data.

MQTT inverts that relationship. The gateway holds one OPC UA session per server and one outbound MQTT connection, and any number of consumers subscribe to the broker. The plant firewall only needs an outbound connection on port 8883, and the PLCs never see cloud traffic. If you are weighing the two protocols more broadly, our OPC UA vs MQTT comparison covers the trade-offs; the short version is that most architectures use both.

Architecture of the Gateway

 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

The gateway has four responsibilities:

Subscribe, do not poll. OPC UA subscriptions let the server push value changes at a sampling interval you choose, which is far more efficient than reading every tag on a timer.
Map nodes to topics. Node IDs like ns=2;s=Line1.CNC1.SpindleTemp mean nothing to a cloud developer. The gateway translates them into a readable hierarchy.
Normalise payloads. Every message gets the value, the OPC UA status, a source timestamp and an engineering unit.
Survive disconnects. Both sides will drop at some point; the gateway must reconnect without losing its configuration.

Prerequisites

Python 3.10 or newer.
asyncua (the open-source FreeOpcUa client library) and paho-mqtt 2.x.
An OPC UA server. If you do not have a PLC handy, the tutorial includes a small simulated server.
An MQTT broker. For testing, use the public MQTT test broker at broker.connectmq.com; for real plant data use private credentials from a hosted MQTT broker.

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

Step 1: A Simulated OPC UA Server (Optional)

If you already have an OPC UA server, skip this step and note its endpoint URL and node IDs. Otherwise, this script exposes three changing tags on opc.tcp://localhost:4840.

# sim_server.py - a tiny OPC UA server with changing CNC values
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())

Because the node IDs use string identifiers in the registered namespace (index 2 here), they will appear as ns=2;s=Line1.CNC1.SpindleTemp and so on. On a real PLC, browse the address space with a client tool such as UaExpert to find the exact node IDs.

Step 2: The Tag Map

Keep the mapping in configuration, not code. Each entry pairs an OPC UA node with the topic path and unit you want consumers to see. The topic layout follows an ISA-95 style hierarchy (site, area, line, cell), which our guide on ISA-95 MQTT topics explains in depth.

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

Save it as tags.json. Replace a91c with your own random string so your topics do not collide with anyone else's on the public broker. The resulting topics are eight levels deep, which is the maximum the public broker allows; on a private broker you have more room.

Step 3: The OPC UA to MQTT Gateway Code

# 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          # production: "YOUR_USERNAME"
MQTT_PASS = None          # production: "YOUR_PASSWORD"
CLIENT_ID = f"opcua-gw-{uuid.uuid4().hex[:8]}"   # production: "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 calls datachange_notification for every monitored item change."""

    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 interval ms
                nodes = [opc.get_node(n) for n in node_map]
                await sub.subscribe_data_change(nodes)
                while True:                       # keep the session alive, detect failures
                    await asyncio.sleep(5)
                    await opc.check_connection()
        except Exception as exc:                  # server restart, network loss, etc.
            log.error("OPC UA error: %s - retrying in 5 s", exc)
            await asyncio.sleep(5)


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

A few design decisions are worth calling out:

Retained messages mean a dashboard that connects later immediately sees the last value of every tag.
The Last Will on the status topic flips to {"online": false} if the gateway dies without a clean disconnect, so consumers can tell stale data from live data.
The OPC UA status code travels with each value. A Good reading and a BadCommunicationError reading should never look the same downstream.
check_connection() raises when the session is lost, which drops us into the retry loop and rebuilds the subscription.

Step 4: Run and Verify

Start the simulated server, then the gateway, in two terminals:

python sim_server.py
python gateway.py

In a third terminal, subscribe to everything under your root:

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

You should see lines such as:

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}

Moving to Production

The public broker is readable by anyone, so it is only for sample data. For real PLC data, create a private broker, then set MQTT_USER, MQTT_PASS and CLIENT_ID to the credentials from the portal. ConnectMQ binds client IDs to the issued credentials, so use the generated client ID rather than a random one.

Checklist before you put the gateway on a plant network:

Item	Why it matters
OPC UA security mode `SignAndEncrypt` with a client certificate	Anonymous, unencrypted sessions are fine for a lab, not for a PLC
Dedicated read-only OPC UA user	The gateway should never be able to write setpoints
Sampling interval per tag class	Fast for vibration, slow for tank levels; avoid flooding
Deadband on analog tags	Reduces noise-driven messages, see report by exception
Run as a service with restart	systemd or a Windows service, with log rotation
Buffering during broker outages	paho queues while disconnected, but for long outages consider store and forward
Outbound-only firewall rule to port 8883	No inbound ports opened to the OT network

On ConnectMQ, the portal's live topic explorer shows your ISA-95 tree as it fills, and the devices view groups machines by ISA-95 levels, which is a quick way to confirm your topic mapping is consistent across lines. Threshold alert rules can then flag, for example, a spindle temperature above a limit without writing any extra code. For the bigger picture of where this gateway sits, see our industrial IoT MQTT guide.

Troubleshooting

BadIdentityTokenRejected or BadUserAccessDenied: the server does not allow anonymous access. Pass a user with Client.set_user() and set_password(), or configure certificates.
BadNodeIdUnknown: the node ID string is wrong, often the namespace index. Namespace indexes can change between server versions; resolve them by namespace URI with get_namespace_index() at startup.
Values arrive but timestamps are null: some servers do not set a source timestamp. The code falls back to the server timestamp.
MQTT connects then immediately disconnects: on the public broker, check that your client ID is unique and non-empty. On a private broker, check that the client ID matches the issued credentials.
Topics rejected on the public broker: you exceeded eight topic levels. Shorten the root or the path.

Run Your OPC UA Gateway on ConnectMQ

Create a free broker at app.connectmq.com on the Serverless plan, no credit card required. Copy the generated username, password and client ID into gateway.py, and within minutes your PLC tags appear in a private, TLS-protected cloud MQTT broker ready for dashboards and historians.

Frequently Asked Questions

Do I need a commercial product for an OPC UA to MQTT gateway?

No. Open-source libraries such as asyncua and paho-mqtt are enough for many projects. Commercial gateways add GUIs, store and forward and vendor support, which can matter at scale.

Should the gateway poll OPC UA tags or use subscriptions?

Use subscriptions. The server samples values and pushes only changes, which reduces load on the PLC and the network compared with cyclic reads.

Can the gateway write values back to the PLC?

Technically yes, by subscribing to a command topic and calling write_value, but treat it as a separate, carefully secured feature. Most deployments keep the gateway read-only.

Which MQTT QoS should I use for PLC tag data?

QoS 1 is the usual choice: delivery is guaranteed at least once and overhead is low. Combine it with retained messages so new consumers get the current state immediately.

OPC UA to MQTT Gateway: Bridging PLC Data to the Cloud