Build an Industrial IT-OT Convergence Agent Workflow with Cisco & Rockwell

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

For decades, the factory floor and the cloud spoke different languages. Programmable logic controllers broadcast dense, millisecond telemetry over OPC UA and MQTT on a private plant network, while enterprise systems reasoned about weeks and months in a cloud analytics lake. Bridging those worlds meant custom gateways, brittle adapters, and enough skepticism between the IT and OT teams to fill a meeting room. In August 2026, Cisco and Rockwell Automation announced a converged industrial networking and automation architecture built precisely to close that gap — Cisco bringing the secure network fabric and edge gateway, Rockwell bringing the control system and MES (manufacturing execution system) integration, and both agreeing that AI would sit on top of the joined stream. This dispatch builds the agent that runs on top of that convergence: a LangGraph workflow that ingests OT telemetry from PLCs and sensors, streams it through an edge gateway into a time-series store, runs anomaly detection and predictive maintenance rules, pushes alerts into the MES, enforces safety interlocks, and routes every machine-affecting action through a human approval gate.

Why Industrial AI Needs an Agent, Not a Dashboard

An industrial monitoring dashboard tells you that spindle vibration crossed a threshold at 14:32:07. An industrial agent tells you that the vibration pattern matches the bearing-wear signature that preceded last quarter's catastrophic spindle failure, that the maintenance window opens in four hours, and that the repair requires a lockout — which needs a human to confirm. The difference is reasoning across domains: OT data is high-frequency and low-semantics, while IT context is high-semantics and low-frequency. The agent's job is to fuse the two without pretending that a plant network and a public cloud are the same trust domain.

The Cisco–Rockwell convergence matters because it supplies the infrastructure the agent depends on. Cisco's industrial edge gateway terminates the OT protocols on-site, applies access control between the control zone and the enterprise zone, and forwards a filtered stream to the cloud; Rockwell's FactoryTalk and PlantPAx stack keeps the control system, historians, and MES coherent. The agent never talks to PLCs directly. It reads from the time-series store and the MES, and it writes back only through sanctioned, safety-checked commands. That separation of concerns is what makes the workflow below safe to run in a real plant.

Architecture at a Glance

The flow is a ladder: telemetry enters at the edge, is validated and stored, reasoned over, and only then — with safety and human gates in the way — allowed to affect machines.

 PLCs / Sensors / Drives
   (OPC UA, MQTT, EtherNet/IP)
           |
           v
+---------------------------+
|  Cisco Edge Gateway       |   terminate OT protocols
|  (OPC UA server / broker) |   filter + authenticate
+-------------+-------------+
              | MQTT / HTTPS (tunneled, encrypted)
              v
+---------------------------+
|  Time-Series Store        |   e.g. InfluxDB / AWS Timestream
|  (raw + windowed metrics) |
+-------------+-------------+
              |
              v
+---------------------------+
|  Rule Engine + Anomaly    |   thresholds + ML anomaly score
|  Detection                |
+-------------+-------------+
              |
              +-----------+-----------+
              |           |           |
              v           v           v
       Predictive    MES Update   Safety Interlock
       Alert (Pm)    (work order) (stop/lockout)
              \           |           /
               \          v          /
                +---------------------+
                |  Human Approval     |  machine-affecting actions
                +---------------------+
                        |
                        v
                Control System Command

Safety interlocks are never delivered to the control system through the cloud path. The edge gateway holds the safety rules and executes the fastest stop locally; the agent's job is to detect the condition, confirm the interlock fired, and document it. Anything that alters machine behavior goes through the human approval node first.

Environment Configuration

Configuration splits into edge-side and cloud-side concerns. Edge credentials never leave the plant; cloud credentials never enter the plant network.

EDGE_GATEWAY_URL=https://edge-gateway.example.com:8443
EDGE_OPCUA_ENDPOINT=opc.tcp://edge-gateway.example.com:4840
EDGE_MQTT_BROKER=ssl://edge-broker.example.com:8883
EDGE_MQTT_TOPIC=plant/line1/telemetry
EDGE_CERT_PATH=/etc/edge/client.pem

TIMESTREAM_DATABASE=plant_telemetry
TIMESTREAM_TABLE=line1_raw
INFLUX_TOKEN=xxxxxxxxxxxxxxxxxxxxxxxx

MES_ENDPOINT=https://mes.example.com/api/v1
MES_API_KEY=xxxxxxxxxxxxxxxxxxxxxxxx

ANOMALY_THRESHOLD=0.85
PREDICTIVE_MAINTENANCE_LOOKAHEAD_HOURS=24
SAFETY_INTERLOCK_AUTO_CONFIRM=false
RETRY_MAX_ATTEMPTS=3
RETRY_BASE_BACKOFF=2
RETRY_MAX_BACKOFF=30

SAFETY_INTERLOCK_AUTO_CONFIRM=false is the most important line in the file. When an interlock fires, the workflow can auto-confirm the event in the MES log — that is documentation, not action. Turning the flag on would let the agent also clear the interlock and restart the machine, which is a decision that belongs to the human at the line, not the cloud.

Domain Schemas

The state model separates raw signal data from interpreted events. Raw samples are high-volume and ephemeral; interpreted events are low-volume and persistent.

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional


class SignalKind(str, Enum):
    VIBRATION = "vibration"
    TEMPERATURE = "temperature"
    CURRENT = "current"
    PRESSURE = "pressure"
    ROTATION = "rotation"


class AlertSeverity(str, Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"


class InterlockState(str, Enum):
    ARMED = "armed"
    FIRED = "fired"
    CONFIRMED = "confirmed"
    RESET = "reset"


@dataclass
class TelemetrySample:
    device_id: str
    ts: datetime
    kind: SignalKind
    value: float
    quality: int  # OPC UA quality byte


@dataclass
class AnomalyEvent:
    device_id: str
    ts: datetime
    kind: SignalKind
    score: float
    window: tuple[datetime, datetime]


@dataclass
class MaintenanceAlert:
    device_id: str
    severity: AlertSeverity
    predicted_failure_hours: float
    recommended_action: str
    work_order_id: Optional[str] = None


@dataclass
class SafetyInterlock:
    device_id: str
    interlock_id: str
    state: InterlockState
    fired_at: Optional[datetime] = None
    confirmed_by: Optional[str] = None


@dataclass
class PlantState:
    samples: list[TelemetrySample] = field(default_factory=list)
    anomalies: list[AnomalyEvent] = field(default_factory=list)
    alerts: list[MaintenanceAlert] = field(default_factory=list)
    interlocks: dict[str, SafetyInterlock] = field(default_factory=dict)
    pending_actions: list[dict] = field(default_factory=list)

Keeping pending_actions as an explicit list is what makes the human gate auditable: nothing touches a machine unless it appears in that list, is approved, and is executed through the sanctioned command path.

Edge Ingestion, Time-Series, and Rule Tools

The tool layer hides the three infrastructure surfaces behind small, testable functions: an MQTT/OPC UA consumer at the edge, a time-series writer and reader, and a rule engine plus MES client.

import json
import os
import paho.mqtt.client as mqtt
from schemas import AnomalyEvent, PlantState, TelemetrySample
from influxdb_client import InfluxDBClient, Point


class EdgeTelemetryClient:
    def __init__(self):
        self.broker = os.environ["EDGE_MQTT_BROKER"]
        self.topic = os.environ["EDGE_MQTT_TOPIC"]
        self.cert = os.environ["EDGE_CERT_PATH"]

    def start(self, on_sample):
        client = mqtt.Client(transport="websockets")
        client.tls_set(self.cert)
        client.on_message = lambda c, u, m: on_sample(self._decode(m.payload))
        client.connect(self.broker, port=8883)
        client.subscribe(self.topic, qos=1)
        client.loop_forever()

    def _decode(self, payload: bytes) -> TelemetrySample:
        msg = json.loads(payload)
        return TelemetrySample(**msg)


class TimeSeriesStore:
    def __init__(self):
        self.client = InfluxDBClient(token=os.environ["INFLUX_TOKEN"])

    def write_batch(self, samples: list[TelemetrySample]) -> None:
        points = [
            Point("telemetry")
            .tag("device", s.device_id)
            .tag("kind", s.kind.value)
            .field("value", s.value)
            .time(s.ts)
            for s in samples
        ]
        self.client.write_api().write(bucket="line1", record=points)

    def window(self, device_id: str, kind: str, minutes: int) -> list[float]:
        query = f'from(bucket:"line1") |> range(start: -{minutes}m) |> filter(fn: (r) => r.device == "{device_id}" and r.kind == "{kind}")'
        tables = self.client.query_api().query(query)
        return [float(rec["_value"]) for table in tables for rec in table.records]


class RuleEngine:
    def __init__(self):
        self.threshold = float(os.environ["ANOMALY_THRESHOLD"])

    def detect(self, samples: list[TelemetrySample]) -> list[AnomalyEvent]:
        events = []
        for s in samples:
            score = _ml_anomaly_score(s)  # trained model, or statistical z-score
            if score >= self.threshold:
                events.append(
                    AnomalyEvent(
                        device_id=s.device_id,
                        ts=s.ts,
                        kind=s.kind,
                        score=score,
                        window=(s.ts, s.ts),
                    )
                )
        return events

The rule engine is deliberately pluggable: a threshold check, a z-score, or a trained model all satisfy the same detect interface. Plants move from statistical rules to ML scoring without any change to the graph.

The LangGraph State Machine

graph.py composes ingestion, reasoning, and action into a loop that runs per telemetry batch, with safety and human gates before any machine-affecting action.

import os
from langgraph.graph import END, START, StateGraph
from schemas import InterlockState, MaintenanceAlert, PlantState
from tools import EdgeTelemetryClient, RuleEngine, TimeSeriesStore, MesClient


def ingest(state: PlantState) -> PlantState:
    store = TimeSeriesStore()
    store.write_batch(state["samples"])
    return state


def detect_anomalies(state: PlantState) -> PlantState:
    store = TimeSeriesStore()
    engine = RuleEngine()
    state["anomalies"] = []
    for s in state["samples"]:
        window = store.window(s.device_id, s.kind.value, minutes=5)
        if (score := engine.threshold_check(window)) >= engine.threshold:
            state["anomalies"].append(AnomalyEvent(device_id=s.device_id, ts=s.ts, kind=s.kind, score=score, window=(s.ts, s.ts)))
    return state


def predictive_maintenance(state: PlantState) -> PlantState:
    for anomaly in state["anomalies"]:
        alert = forecast_failure(anomaly)  # RUL model, e.g. Random Forest on trend
        if alert.predicted_failure_hours <= float(os.environ["PREDICTIVE_MAINTENANCE_LOOKAHEAD_HOURS"]):
            state["alerts"].append(alert)
    return state


def update_mes(state: PlantState) -> PlantState:
    mes = MesClient()
    for alert in state["alerts"]:
        if alert.work_order_id is None:
            alert.work_order_id = mes.create_work_order(alert)
    return state


def safety_interlock(state: PlantState) -> PlantState:
    for anomaly in state["anomalies"]:
        if anomaly.score >= 0.95:  # severe: local stop required
            interlock = state["interlocks"].get(anomaly.device_id)
            if interlock and interlock.state == InterlockState.ARMED:
                # Edge gateway executes the stop locally; we confirm + document.
                confirm_interlock(interlock)
    return state


def human_approval(state: PlantState) -> PlantState:
    # Interrupted for machine-affecting actions: restart, resume, new setpoint.
    return state


builder = StateGraph(PlantState)
builder.add_node("ingest", ingest)
builder.add_node("detect_anomalies", detect_anomalies)
builder.add_node("predictive_maintenance", predictive_maintenance)
builder.add_node("update_mes", update_mes)
builder.add_node("safety_interlock", safety_interlock)
builder.add_node("human_approval", human_approval)
builder.add_node("apply_actions", apply_approved_actions)

builder.add_edge(START, "ingest")
builder.add_edge("ingest", "detect_anomalies")
builder.add_edge("detect_anomalies", "predictive_maintenance")
builder.add_edge("predictive_maintenance", "update_mes")
builder.add_edge("update_mes", "safety_interlock")
builder.add_edge("safety_interlock", "human_approval")
builder.add_edge("human_approval", "apply_actions")
builder.add_edge("apply_actions", END)

graph = builder.compile(interrupt_before=["human_approval"])

Notice what is automatic versus what is gated. Writing a work order into the MES is automatic — it is a plan, not an action. Confirming a fired interlock is automatic. Restarting a line, changing a setpoint, or resuming production is not; each one stops at human_approval and waits for a named operator to approve with their badge-scoped identity.

Running the Workflow

main.py runs the edge consumer and the graph loop. Telemetry batches flow continuously; the graph only emits state transitions that need attention.

import os
from collections import deque
from graph import graph, PlantState
from tools import EdgeTelemetryClient, TimeSeriesStore


def main():
    batch: deque = deque(maxlen=200)
    store = TimeSeriesStore()

    def on_sample(sample):
        batch.append(sample)
        if len(batch) >= 200:
            process_batch(list(batch))
            batch.clear()

    def process_batch(samples):
        state = PlantState(samples=samples)
        config = {"configurable": {"thread_id": f"batch-{samples[-1].ts}"}}
        result = graph.invoke(state, config)
        if result.get("pending_actions"):
            # Operator approves on the shop-floor terminal or the CMS.
            graph.invoke(
                config,
                input={"pending_actions": await_operator_approval(result["pending_actions"])},
            )

    EdgeTelemetryClient().start(on_sample)


if __name__ == "__main__":
    main()

Retry Rules

Industrial telemetry has different failure semantics than web traffic, so the retry policy is tuned per layer.

• Bounded retries. Time-series writes retry 3 times, then spill the batch to a local spool file at the edge for replay — never drop telemetry silently. MES calls retry 3 times with backoff; if they still fail, the alert stays in a pending queue and the MES is reconciled on the next tick.
• Exponential backoff. Network retries use base * (2 ** attempt) with jitter, capped at RETRY_MAX_BACKOFF. Telemetry older than 60 seconds when it finally lands is flagged with a stale quality tag so downstream anomaly scoring can discount it.
• Re-queue. Failed MES work orders and un-acknowledged alerts are re-queued with a monotonic attempt counter. When attempts exceed the cap, the alert is escalated to the shift supervisor's terminal instead of being retried forever.
• Safety rules are never retried. An interlock condition is evaluated by the edge gateway on every sample, locally and independently of the agent. If the agent's cloud path is down, the interlock still fires — the retry policy governs the agent's documentation of the event, never the safety action itself.

Edge-Gateway Model vs. Direct-to-Cloud Telemetry

The Cisco–Rockwell architecture is the answer to a question that every IIoT project eventually faces: should OT data go straight to the cloud, or terminate at the edge first?

The agent workflow in this dispatch only becomes safe and reliable in the right-hand column. If you are planning an industrial AI rollout, treat the edge gateway as a prerequisite, not an option. For more agent design patterns that follow this same state-machine discipline, see the workflows library, and for tool wiring across OT and IT systems, browse the MCP directory. Keep the AI news desk on your radar as Cisco, Rockwell, and the rest of the industrial stack ship their next convergence layers.

Frequently Asked Questions

Does the agent talk to PLCs directly?

No. The agent reads from the time-series store and the MES, and writes commands only through the sanctioned control path. PLC access is terminated at the Cisco edge gateway; the agent never holds OPC UA client credentials for the live control network.

What happens if the cloud connection drops mid-shift?

The edge gateway keeps running: telemetry spools locally, safety interlocks still evaluate on every sample, and the time-series store is replayed when connectivity returns. The agent's retry policy covers the replay; it never covers the safety decision.

What counts as a machine-affecting action that needs approval?

Restarting a line, resuming production, changing a setpoint, or overriding an interlock. Work order creation in the MES and interlock documentation do not need approval — they are records, not actions. The graph's pending_actions list is exactly the set of actions waiting on the operator.

How does the predictive maintenance alert decide severity?

A failure-forecast model (for example, a Random Forest on vibration and temperature trend features) predicts remaining useful life in hours. Anything under the PREDICTIVE_MAINTENANCE_LOOKAHEAD_HOURS window becomes a work order; anything over it stays a monitoring signal. The ANOMALY_THRESHOLD controls what enters the model at all.

Can the same workflow manage multiple plants?

Yes, if each plant has its own edge gateway and MES scope. Thread every graph invocation by plant_id + device_id, keep credentials per plant, and never share the pending_actions list across plants — an approval intended for Line 1 in Ohio must never apply to Line 2 in Gujarat.