EtherNet/IP to MQTT: Reading CIP Devices and Logix Tags with Python

EtherNet/IP to MQTT bridging gives you a way to get data out of Rockwell Logix controllers, drives and other CIP devices and into dashboards, historians and cloud applications without touching the control program. In this tutorial you will learn the difference between implicit and explicit messaging, then use the open-source pycomm3 library on a gateway to read Logix tags and generic CIP attributes and publish them over MQTT with clean, self-describing payloads.

EtherNet/IP and CIP in Five Minutes

EtherNet/IP is the Common Industrial Protocol (CIP) carried over standard Ethernet and TCP/UDP. CIP models every device as a set of objects with classes, instances and attributes. A drive, for example, has an Identity object (class 0x01) describing vendor, product code and serial number, plus application objects for speed, current and status.

CIP traffic comes in two flavours:

A monitoring gateway should use explicit messaging. It does not disturb the controller's I/O connections, needs no changes to the PLC project, and can be added or removed without downtime. The cost is that it is a polling protocol, so you choose how often to read.

Two Kinds of Targets

1. Logix controllers (ControlLogix, CompactLogix, Micro800 to a degree). These support symbolic tag access: you read Line1_Speed or Conveyor[3].Status by name. This is by far the most convenient source of data.
2. Generic CIP devices (drives, I/O adapters, scanners, third-party devices). These expose object attributes that you read by class, instance and attribute number from the device's documentation or EDS file.

pycomm3 handles both: LogixDriver for tag-based controllers and CIPDriver for generic explicit messages. pylogix is another popular open-source option for Logix tags. Siemens, Beckhoff and other platforms have their own access paths; for Rockwell-specific topics such as controller configuration see our Allen-Bradley MQTT guide.

Prerequisites

• Python 3.10 or newer on a gateway that can reach the controller on TCP 44818.
• pycomm3 and paho-mqtt 2.x.
• The controller's IP address and, for chassis-based controllers, the slot number (for example 10.0.0.20/0 for slot 0).
• A read-only list of tags agreed with the controls engineer.

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

Step 1: Explore the Controller

Before writing the bridge, confirm connectivity and list the tags available. This script prints controller information and the first controller-scoped tags.

# explore_logix.py - confirm connectivity and list tags
from pycomm3 import LogixDriver

with LogixDriver("10.0.0.20/0") as plc:          # IP/slot of the controller
    print(plc.info)                               # vendor, product, revision, name
    tags = plc.get_tag_list()                     # controller-scoped tags
    for t in tags[:30]:
        print(t["tag_name"], t["data_type_name"])

get_tag_list() uploads tag definitions, including user-defined types, so it can take a moment on large projects. Program-scoped tags are read with the Program:MainProgram.TagName syntax.

Step 2: Read a Generic CIP Attribute

For a drive or any CIP device without symbolic tags, use explicit messages. The example reads the product name from the Identity object, which every EtherNet/IP device implements.

# identity.py - read Identity object attribute 7 (Product Name) from any CIP device
from pycomm3 import CIPDriver, Services, ClassCode, SHORT_STRING

with CIPDriver("10.0.0.31") as dev:
    resp = dev.generic_message(
        service=Services.get_attribute_single,
        class_code=ClassCode.identity_object,   # 0x01
        instance=1,
        attribute=7,                             # Product Name
        data_type=SHORT_STRING,
        connected=False,
        name="product_name",
    )
    print(resp.value if not resp.error else resp.error)

Replace class, instance, attribute and data type with values from the device manual to read application data such as output frequency or motor current.

Step 3: The EtherNet/IP to MQTT Bridge

The bridge reads a list of tags in a single multi-tag request each cycle, publishes values that changed (or that cross a deadband), and reports read errors as bad quality instead of silently keeping old values.

# eip_mqtt_bridge.py - poll Logix tags over EtherNet/IP, publish to MQTT
import json, logging, time, uuid
import paho.mqtt.client as mqtt
from pycomm3 import LogixDriver, CommError

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

PLC_PATH = "10.0.0.20/0"
POLL_SECONDS = 1.0
TOPIC_ROOT = "connectmq/demo/9e6d/plant1/packaging/line2"   # your own random id
MQTT_HOST, MQTT_PORT = "broker.connectmq.com", 1883         # production: 8883 with TLS
MQTT_USER, MQTT_PASS = None, None                           # production: "YOUR_USERNAME", "YOUR_PASSWORD"
CLIENT_ID = f"eip-bridge-{uuid.uuid4().hex[:8]}"            # production: "YOUR_CLIENT_ID"

# PLC tag -> (topic leaf, unit, deadband)
TAGS = {
    "Line2_Speed":          ("speed", "ppm", 1.0),
    "Line2_Running":        ("running", "", 0),
    "Line2_GoodCount":      ("good_count", "count", 0),
    "Line2_RejectCount":    ("reject_count", "count", 0),
    "Filler.MotorCurrent":  ("filler_current", "A", 0.2),
}


def mqtt_client():
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if MQTT_USER:
        c.username_pw_set(MQTT_USER, MQTT_PASS)
        c.tls_set()
    c.will_set(f"{TOPIC_ROOT}/bridge", "offline", qos=1, retain=True)
    c.on_connect = lambda cl, u, f, rc, p: cl.publish(f"{TOPIC_ROOT}/bridge", "online", qos=1, retain=True)
    c.connect(MQTT_HOST, MQTT_PORT, keepalive=30)
    c.loop_start()
    return c


def changed(old, new, deadband):
    if old is None:
        return True
    if isinstance(new, (int, float)) and not isinstance(new, bool) and deadband:
        return abs(new - old) >= deadband
    return new != old


def main():
    mq = mqtt_client()
    last = {}
    while True:
        try:
            with LogixDriver(PLC_PATH) as plc:
                log.info("connected to %s", plc.info.get("product_name"))
                while True:
                    results = plc.read(*TAGS.keys())
                    if not isinstance(results, list):
                        results = [results]
                    now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
                    for r in results:
                        leaf, unit, deadband = TAGS[r.tag]
                        topic = f"{TOPIC_ROOT}/{leaf}"
                        if r.error:
                            if last.get(r.tag, {}).get("quality") != "bad":
                                mq.publish(topic, json.dumps({"value": None, "quality": "bad",
                                                              "error": r.error, "ts": now}), qos=1, retain=True)
                                last[r.tag] = {"value": None, "quality": "bad"}
                            continue
                        prev = last.get(r.tag, {})
                        if prev.get("quality") == "bad" or changed(prev.get("value"), r.value, deadband):
                            payload = {"value": r.value, "unit": unit, "type": r.type,
                                       "quality": "good", "ts": now}
                            mq.publish(topic, json.dumps(payload), qos=1, retain=True)
                            last[r.tag] = {"value": r.value, "quality": "good"}
                    time.sleep(POLL_SECONDS)
        except CommError as exc:
            log.error("EtherNet/IP connection lost: %s - retrying in 5 s", exc)
            time.sleep(5)


if __name__ == "__main__":
    main()

plc.read() with several tag names packs them into as few requests as the connection size allows, which is much lighter on the controller than one request per tag. Each result is a Tag with tag, value, type and error fields, so the bridge can publish the data type alongside the value.

Step 4: Run and Verify

python eip_mqtt_bridge.py
mosquitto_sub -h broker.connectmq.com -p 1883 -i eip-viewer-9e6d -t "connectmq/demo/9e6d/#" -v

connectmq/demo/9e6d/plant1/packaging/line2/speed {"value": 118.0, "unit": "ppm", "type": "REAL", "quality": "good", "ts": "2026-10-16T12:00:01Z"}
connectmq/demo/9e6d/plant1/packaging/line2/running {"value": true, "unit": "", "type": "BOOL", "quality": "good", "ts": "2026-10-16T12:00:01Z"}

The topic tree follows an ISA-95 style layout (site, area, line, point) so the same consumers work whether data comes from EtherNet/IP, Modbus or OPC UA. Our ISA-95 MQTT topics guide explains the conventions.

Choosing Tags and Poll Rates

Not every tag in a Logix project deserves a topic. Start from the questions consumers will ask: is the line running, how fast, how many good and rejected units, which alarms are active, and how much energy the main drives draw. That usually means a few dozen values per line, not thousands. Group them by how quickly they change. States and counters can be read every second and published on change; slow analog values such as tank levels or temperatures can be read every five or ten seconds with a deadband. If a value is only useful as a shift total, consider letting the PLC accumulate it and publishing it on a slower cycle instead.

Good Practices for Production

1. Read-only by design. Do not add write support to a monitoring bridge. If you ever need commands, build a separate service with explicit approval logic.
2. Use an interface tag set. Ask the controls engineer to copy the values you need into a dedicated UDT or array. You then read one structure instead of dozens of scattered tags, and program changes do not break the bridge.
3. Mind connection limits. Controllers support a limited number of CIP connections, shared with HMIs, SCADA and engineering tools. One long-lived connection per bridge is ideal; avoid reconnecting every cycle.
4. Poll at the speed you need. One second is plenty for production counts and states. Faster polling adds load without adding insight.
5. Network segmentation. Place the gateway in the OT network or a DMZ with access only to port 44818 on the controllers and outbound 8883 to the broker. Our MQTT DMZ article covers the layout.
6. Private broker for real data. The public test broker is readable by everyone. Move to a fully managed MQTT service with TLS and per-broker credentials before connecting a real line.

Troubleshooting

Where ConnectMQ Fits

ConnectMQ is the cloud side of this pipeline: a managed MQTT broker with TLS on port 8883 and WebSockets for browser dashboards. It detects ISA-95 topic structures, so the devices view groups lines and machines automatically, and threshold alert rules can flag a reject count or motor current outside its limits. If you are new to managed brokers, see how a cloud MQTT broker compares with self-hosting, and for the full architecture read the industrial IoT MQTT guide.

Get Started with ConnectMQ

Connect your first EtherNet/IP line today. Sign up at app.connectmq.com, choose the free Serverless plan with no credit card, create a broker and paste the generated username, password and client ID into the bridge. Your Logix tags will appear in the ConnectMQ topic explorer within seconds.

Frequently Asked Questions

Does reading tags over EtherNet/IP affect the PLC program?

Explicit messaging uses the controller's communication resources, not its I/O connections, and does not change the program. Keep polling moderate and connections few to avoid affecting HMIs and SCADA.

Can I use implicit I/O messaging for an MQTT gateway?

It is possible if the gateway is configured as an adapter in the PLC's I/O tree, but that requires a project change. Explicit tag reads are simpler and non-intrusive for monitoring.

Does pycomm3 work with Micro800 and older PLC-5 or SLC controllers?

pycomm3 targets Logix controllers and has partial Micro800 support; older PLC-5 and SLC families use different addressing. Check the library documentation for your specific controller.

Which port does EtherNet/IP explicit messaging use?

TCP port 44818. Implicit I/O traffic uses UDP port 2222.