Allen-Bradley MQTT: Publishing ControlLogix and CompactLogix Tags

Allen-Bradley MQTT integration is the fastest way to get ControlLogix and CompactLogix tag data into dashboards, historians and cloud analytics without adding load to your SCADA server. In this tutorial you will build a small Python gateway that reads Logix tags over EtherNet/IP with pycomm3, maps them to ISA-95 topics, and publishes only meaningful changes to an MQTT broker. You will test against a public broker first and then switch to a private one with TLS.

How Allen-Bradley PLCs Fit with MQTT

Logix controllers communicate natively over EtherNet/IP using CIP (Common Industrial Protocol). Studio 5000 projects organize data as named tags, either controller-scoped or program-scoped, with atomic types (BOOL, DINT, REAL), arrays and user-defined types (UDTs). That named-tag model is a gift for MQTT integration: unlike Modbus, you do not need a register map, because the tag name already carries meaning.

There are three common ways to get those tags onto MQTT:

This tutorial takes the first approach because it shows every moving part. The same topic and payload design applies if you later move to a commercial driver or to Ignition with MQTT.

Prerequisites

• A ControlLogix or CompactLogix controller reachable from the gateway on TCP port 44818.
• A Linux or Windows machine with Python 3.9 or newer.
• A few tags to read, for example Line1_Temp (REAL), Line1_Speed (DINT), Line1_Running (BOOL) and a program-scoped tag Program:MainProgram.Pump1_Pressure (REAL).
• The mosquitto clients for testing.

Install the libraries:

python -m venv abmqtt
source abmqtt/bin/activate
pip install pycomm3 "paho-mqtt>=2.0"

Step 1: Discover the Tag List

Before writing the gateway, confirm connectivity and see exactly how tags are named. pycomm3 uploads the controller's tag database when it connects:

from pycomm3 import LogixDriver

# CompactLogix: just the IP. ControlLogix with the CPU in slot 3: "192.168.1.20/3"
PLC_PATH = "192.168.1.20"

with LogixDriver(PLC_PATH) as plc:
    print(plc.info)                       # controller name, revision, serial
    for name, definition in plc.tags.items():
        print(name, definition["data_type_name"], definition.get("dim", 0))

Program-scoped tags are not in the controller-scoped list by default; address them as Program:MainProgram.TagName when reading. For ControlLogix chassis, the path after the slash is the slot of the controller in the backplane. Get that wrong and the connection opens to the Ethernet module but tag reads fail.

The pycomm3 documentation covers path syntax, arrays and UDT handling in detail.

Step 2: Design ISA-95 Topics

Map each tag to a topic that describes the equipment, not the controller. A simple mapping table in the gateway keeps PLC naming conventions out of your MQTT namespace:

With a site and area prefix, a full topic becomes plant1/assembly/line1/oven/temperature. Consistent paths like this are what make a plant-wide namespace work; see ISA-95 MQTT topics for a deeper treatment of levels and naming.

Step 3: The Allen-Bradley MQTT Gateway

The gateway reads all mapped tags in a single multi-tag request each second, applies per-tag deadbands, publishes changes with QoS 1, refreshes everything every five minutes, and maintains a retained online/offline status with a Last Will.

import json
import time
import uuid
from datetime import datetime, timezone

from pycomm3 import LogixDriver
import paho.mqtt.client as mqtt

PLC_PATH = "192.168.1.20"

# Public test broker first. The public area is readable by everyone: test data only.
BROKER, PORT, USE_TLS = "broker.connectmq.com", 1883, False
USERNAME, PASSWORD = None, None                  # production: "YOUR_USERNAME", "YOUR_PASSWORD"
CLIENT_ID = f"ab-gw-{uuid.uuid4().hex[:8]}"      # production: "YOUR_CLIENT_ID"
BASE = f"connectmq/demo/{uuid.uuid4().hex[:6]}/plant1/assembly/line1"

TAGS = {
    "Line1_Temp": ("oven/temperature", "degC", 0.5),
    "Line1_Speed": ("conveyor/speed", "rpm", 5),
    "Line1_Running": ("line/running", None, 0),
    "Program:MainProgram.Pump1_Pressure": ("pump1/pressure", "bar", 0.05),
}


def on_connect(client, userdata, flags, reason_code, properties):
    print("MQTT connected:", reason_code)
    client.publish(f"{BASE}/gateway/status", "online", qos=1, retain=True)


def should_publish(tag, value, last, deadband):
    if tag not in last:
        return True
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return value != last[tag]
    return abs(value - last[tag]) >= deadband


def main():
    mq = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        mq.username_pw_set(USERNAME, PASSWORD)
    if USE_TLS:
        mq.tls_set()
    mq.will_set(f"{BASE}/gateway/status", "offline", qos=1, retain=True)
    mq.on_connect = on_connect
    mq.connect(BROKER, PORT, keepalive=30)
    mq.loop_start()
    print("Publishing under", BASE)

    plc = LogixDriver(PLC_PATH)
    last, last_full = {}, 0.0

    while True:
        try:
            if not plc.connected:
                plc.open()
            results = plc.read(*TAGS.keys())
            now = time.time()
            force = now - last_full > 300
            ts = datetime.now(timezone.utc).isoformat(timespec="seconds")

            for result in results:
                suffix, unit, deadband = TAGS[result.tag]
                if result.error:
                    mq.publish(f"{BASE}/{suffix}", json.dumps(
                        {"value": None, "quality": "bad", "error": result.error, "ts": ts}), qos=1)
                    continue
                value = result.value
                if isinstance(value, float):
                    value = round(value, 3)
                if force or should_publish(result.tag, value, last, deadband):
                    payload = {"value": value, "quality": "good", "ts": ts}
                    if unit:
                        payload["unit"] = unit
                    mq.publish(f"{BASE}/{suffix}", json.dumps(payload), qos=1)
                    last[result.tag] = value
            if force:
                last_full = now
        except Exception as exc:
            print("PLC communication failed:", exc)
            try:
                plc.close()
            except Exception:
                pass
            time.sleep(5)
        time.sleep(1)


if __name__ == "__main__":
    main()

A few design points worth calling out:

• One multi-tag read. pycomm3 packs several tags into as few CIP requests as possible, which is far kinder to the controller than reading each tag separately.
• Per-tag errors. A typo in one tag name returns an error for that tag only. Publishing quality: bad makes the problem visible instead of silently freezing a value.
• UDTs work too. Reading a UDT returns a Python dict, which json.dumps turns into a structured payload. Keep UDT payloads small and avoid publishing large arrays every second.

Step 4: Verify on the Public Broker

Start the gateway and subscribe to its prefix:

mosquitto_sub -h broker.connectmq.com -p 1883 -i ab-watch-01 \
  -t 'connectmq/demo/+/plant1/assembly/line1/#' -v

You should immediately see the retained gateway status and the first full set of values. Toggle Line1_Running from Studio 5000 and the change arrives within a second. Using the public MQTT test broker for this step is fine as long as the values are not sensitive.

Step 5: Production on a Private Broker

For production, create a private broker in the ConnectMQ portal and change four lines: port 8883, USE_TLS = True, the issued username and password, and the issued client ID. Remove the connectmq/demo/... prefix so topics start at the enterprise or site level. Because client IDs are bound to their credentials, give each gateway its own set rather than sharing one across sites.

Run the gateway as a service so it restarts automatically:

[Unit]
Description=Allen-Bradley MQTT gateway
After=network-online.target

[Service]
ExecStart=/opt/abmqtt/bin/python /opt/abmqtt/gateway.py
Restart=always
RestartSec=5
User=abmqtt

[Install]
WantedBy=multi-user.target

Once data flows, the ConnectMQ portal's devices view groups the ISA-95 topics by site, area and line, and threshold alert rules can flag a temperature or pressure outside its limits. Historians and dashboards subscribe like any other client of the MQTT broker.

Tuning Read Rates and Deadbands

The read interval and the deadbands together decide how much traffic the gateway creates, both on the controls network and on the broker. A practical way to set them:

• Start from the consumer. A dashboard refreshed by an operator every few seconds does not need 100 ms data. A historian used for energy reports may be happy with one-minute averages.
• Size deadbands from sensor noise. Watch a value for a few minutes in Studio 5000 trend view. A deadband slightly larger than the noise band suppresses chatter without hiding real changes.
• Never deadband states and alarms. Booleans, mode words and alarm bits should publish on every change, and their topics are good candidates for retained messages so a new subscriber sees the current state immediately.
• Keep the periodic refresh. Without it, a value that genuinely stays constant for hours looks identical to a gateway that has silently stopped.

If you later outgrow a single gateway, run one per area or line rather than one giant process for the whole plant. Failures stay contained and each instance has a small, reviewable tag list.

Security Considerations

CIP over EtherNet/IP has no authentication in most installations, so the gateway is effectively a trusted engineering client. Put it on the controls network behind a firewall, allow only outbound TLS to the broker from it, and never let MQTT commands write to PLC tags without an explicit allow-list, range checks and an interlock in the PLC logic. Reading is safe to automate; writing deserves a design review. The EtherNet/IP to MQTT guide covers network segmentation for CIP traffic in more depth, and the industrial IoT with MQTT guide explains where gateways sit in a complete architecture.

Troubleshooting

1. Connection times out. Check routing and that TCP 44818 is open between gateway and controller.
2. Connected but every tag fails. Wrong slot in a ControlLogix path, or the tag is program-scoped and needs the Program: prefix.
3. REAL values with long decimals. That is normal float representation; round before publishing.
4. MQTT disconnects every few seconds. Another client is using the same client ID.
5. High controller communication load. Increase the read interval or reduce the tag list; most dashboards do not need sub-second data.

Connect Your Logix Controllers to ConnectMQ

Your gateway is ready; give it a secure cloud MQTT broker to publish into. Sign up at app.connectmq.com for a free Serverless plan with no credit card, paste the generated credentials into the script and watch your Allen-Bradley tags appear live.

Frequently Asked Questions

Can a ControlLogix PLC publish MQTT natively?

Logix controllers do not include a general-purpose MQTT client in the controller itself. The usual approach is an edge gateway, connectivity server or SCADA module that reads tags over EtherNet/IP and publishes them.

Does pycomm3 work with Micro800 or SLC 500?

pycomm3's LogixDriver targets ControlLogix, CompactLogix and Micro800 with some limitations; older PLC-5 and SLC 500 use a separate driver. Check the library documentation for supported features per family.

How often should I read tags?

One second is plenty for most monitoring. Use deadbands so only real changes are published, plus a periodic full refresh.

Should I use Sparkplug B instead of JSON?

Sparkplug B adds birth and death certificates and compact Protobuf encoding, which helps with large tag counts and SCADA integration. Plain JSON is simpler to start with and easy to inspect.