Work with OPC UA PubSub in Python

The code

import asyncio, json
from asyncua import Client
import paho.mqtt.client as mqtt

BROKER = "localhost"
TOPIC  = "plant/line1/machinetool1/state"

class Republish:
    def __init__(self, mq):
        self.mq = mq

    def datachange_notification(self, node, val, data):
        dv = data.monitored_item.Value
        payload = {
            "value": val,
            "status": str(dv.StatusCode),
            "sourceTimestamp": dv.SourceTimestamp.isoformat(),
        }
        self.mq.publish(TOPIC, json.dumps(payload))

async def main():
    mq = mqtt.Client()
    mq.connect(BROKER, 1883)
    mq.loop_start()

    async with Client(url="opc.tcp://opcua.umati.app:4840") as client:
        sub = await client.create_subscription(1000, Republish(mq))
        await sub.subscribe_data_change(client.get_node("i=2258"))
        await asyncio.sleep(30)

asyncio.run(main())

What it prints

# in MQTT Explorer, subscribed to plant/line1/machinetool1/state:
{"value": "2026-09-08T03:45:00.102000", "status": "StatusCode(Good)", "sourceTimestamp": "2026-09-08T03:45:00.102000"}

Notes

• This is a client-server subscription republished to MQTT, not OPC UA PubSub. Be precise about that in an architecture document, because they secure and behave differently.
• Carry the status code and SourceTimestamp into the payload. Publishing a bare value throws away the two fields that let you audit the data later.
• Derive the topic from the browse path, not from a hand-written list. Hand-written topic maps stop scaling at about the second machine.
• If you need genuine OPC UA PubSub, use a stack that implements Part 14 (the.NET reference implementation or open62541) and treat this pattern as the pragmatic alternative.

What breaks first

Last verified 7 September 2026 against OPC UA 1.05.07. OPC UA Connect is an educational resource and is not affiliated with the OPC Foundation.