Why Connect Industrial IoT Gateways to an AI Agent?
Industrial IoT (IIoT) gateways are the backbone of smart factories—they bridge legacy PLCs, Modbus sensors, and CAN bus machinery to the cloud. But managing them manually is a nightmare: you write custom scripts for each device, maintain cron jobs for data collection, and build dashboards from scratch. What if you could just talk to an AI agent and let it handle everything?
That’s exactly what ASI Biont does. Instead of spending weeks coding integrations, you describe your setup in natural language: “Connect to the Modbus TCP gateway at 192.168.1.100, read holding registers 0-10 every 30 seconds, and log them to a Google Sheet.” The AI agent writes the Python code, connects via your preferred protocol (MQTT, Modbus/TCP, OPC UA, BACnet, or even raw serial over a Hardware Bridge), and starts monitoring—all in seconds.
This article walks through 12 real-world scenarios where IIoT gateways meet ASI Biont’s AI agent. Each example includes a working code snippet, connection method, and the exact chat prompt you’d use. No fluff, no dashboards—just a conversation with an AI.
How ASI Biont Connects to IIoT Gateways
ASI Biont doesn’t require a custom SDK or a “device management panel.” Instead, it relies on a universal mechanism: execute_python. The AI agent runs Python scripts in a secure sandbox (on Railway) with a rich set of libraries—pymodbus, paho-mqtt, opcua-asyncio, snap7, bac0, pycomm3, python-can, paramiko, and more. You simply tell the AI what to do, and it generates the code.
For local serial devices (RS-232/RS-485), you run a lightweight bridge.py on your PC. It connects to ASI Biont via WebSocket and exposes COM ports to the AI. The agent sends commands like serial_write_and_read(data="48454c500a"), and the bridge handles the hardware.
Here’s the key: you never write the integration code yourself. The AI writes it for you, tailored to your exact device specs.
12 Real-World Integration Scenarios
1. Modbus/TCP Gateway → Read Temperature & Control Relay
Use case: A factory uses a Modbus RTU-to-TCP gateway (e.g., USR-W630) connected to temperature sensors and relay modules. You want to read sensor values and switch a cooling fan on/off.
Connection method: Modbus/TCP via pymodbus.
Chat prompt: “Connect to Modbus TCP gateway at 192.168.1.100:502. Read holding register 0 (temperature in °C). If temperature > 30°C, write coil 0 to 1 (turn on fan). Log every reading to console.”
AI-generated code (runs in execute_python):
from pymodbus.client import ModbusTcpClient
import time
client = ModbusTcpClient('192.168.1.100', port=502)
client.connect()
while True:
result = client.read_holding_registers(0, 1, unit=1)
temp = result.registers[0] / 10.0 # scale factor = 10
print(f"Temperature: {temp}°C")
if temp > 30:
client.write_coil(0, True, unit=1)
print("Fan ON")
else:
client.write_coil(0, False, unit=1)
print("Fan OFF")
time.sleep(10)
Pitfall to avoid: The sandbox has a 30-second timeout. Use time.sleep() with reasonable intervals, and avoid infinite loops without breaks.
2. MQTT Gateway → ESP32 Temperature/Humidity Sensor
Use case: An ESP32 with DHT22 publishes temperature and humidity via MQTT to a local broker (Mosquitto). You want ASI Biont to subscribe, analyze trends, and send Telegram alerts if humidity > 70%.
Connection method: MQTT via paho-mqtt.
Chat prompt: “Subscribe to MQTT broker at 192.168.1.50:1883, topic ‘factory/sensor/dht’. Parse JSON payload with temperature and humidity. If humidity > 70%, send a Telegram message to my chat ID using the token from environment variable TELEGRAM_TOKEN.”
AI-generated code:
import paho.mqtt.client as mqtt
import json
import requests
import os
def on_message(client, userdata, msg):
data = json.loads(msg.payload)
temp = data['temperature']
hum = data['humidity']
print(f"Temp: {temp}°C, Hum: {hum}%")
if hum > 70:
token = os.environ['TELEGRAM_TOKEN']
chat_id = '123456789'
text = f"Alert: Humidity {hum}% exceeded threshold!"
requests.post(f"https://api.telegram.org/bot{token}/sendMessage", json={'chat_id': chat_id, 'text': text})
client = mqtt.Client()
client.on_message = on_message
client.connect('192.168.1.50', 1883, 60)
client.subscribe('factory/sensor/dht')
client.loop_forever()
Pitfall to avoid: The sandbox can’t run loop_forever() indefinitely. Instead, use loop_start() and a time.sleep(300) to run for 5 minutes.
3. OPC UA Server → Predictive Maintenance for Motors
Use case: A factory OPC UA server exposes vibration, temperature, and current of three motors. You want the AI to detect anomalies and predict failures.
Connection method: OPC UA via opcua-asyncio.
Chat prompt: “Connect to OPC UA server at opc.tcp://192.168.1.200:4840. Read variables ns=2;i=100 (vibration), ns=2;i=101 (temperature), ns=2;i=102 (current). If vibration > 50 mm/s or temperature > 80°C, log a warning. Send a summary every hour.”
AI-generated code:
import asyncio
from opcua import Client
async def monitor():
client = Client("opc.tcp://192.168.1.200:4840")
await client.connect()
root = client.get_objects_node()
vib = await root.get_child(["2:Vibration"]).read_value()
temp = await root.get_child(["2:Temperature"]).read_value()
current = await root.get_child(["2:Current"]).read_value()
print(f"Vib: {vib}, Temp: {temp}, Current: {current}")
if vib > 50 or temp > 80:
print("WARNING: Anomaly detected")
await client.disconnect()
asyncio.run(monitor())
Pitfall to avoid: opcua-asyncio uses async/await. Ensure your script is inside async def and called with asyncio.run().
4. Siemens S7 PLC → Conveyor Belt Control
Use case: A Siemens S7-1200 PLC controls a conveyor belt. You want to read status (DB1.DBW0) and start/stop the belt via chat commands.
Connection method: Siemens S7 via snap7.
Chat prompt: “Connect to Siemens S7-1200 at 192.168.1.10, rack=0, slot=1. Read DB1.DBW0 (status word). If bit 0 = 1 (running), print ‘Belt is running’. If I say ‘stop’, write DB1.DBW0 bit 0 = 0.”
AI-generated code:
import snap7
plc = snap7.client.Client()
plc.connect('192.168.1.10', 0, 1)
# Read DB1.DBW0
data = plc.db_read(1, 0, 2) # DB1, start offset 0, length 2 bytes
status = snap7.util.get_bool(data, 0, 0) # bit 0 of byte 0
print(f"Belt running: {status}")
# Write to stop (set bit 0 = 0)
snap7.util.set_bool(data, 0, 0, False)
plc.db_write(1, 0, data)
plc.disconnect()
Pitfall to avoid: snap7 requires the IP address and rack/slot numbers. Get these from your PLC configuration.
5. BACnet Gateway → Building HVAC Monitoring
Use case: A BACnet/IP gateway connects to an HVAC system. You want to read zone temperature and adjust setpoints.
Connection method: BACnet via bac0.
Chat prompt: “Discover BACnet devices on 192.168.1.0/24. Connect to device 12345. Read analog input ‘ZoneTemp’. If temperature > 25°C, write analog output ‘Setpoint’ to 22°C.”
AI-generated code:
import bac0
bacnet = bac0.connect(ip='192.168.1.100')
devices = bacnet.whois()
print(f"Found devices: {devices}")
# Read property
temp = bacnet.read('analogInput:1', 'presentValue')
print(f"Zone Temp: {temp}")
if temp > 25:
bacnet.write('analogOutput:1', 'presentValue', 22.0)
print("Setpoint changed to 22°C")
bacnet.disconnect()
Pitfall to avoid: bac0 uses the local network interface. Ensure your bridge or sandbox can reach the BACnet broadcast.
6. EtherNet/IP → Rockwell PLC Tag Read/Write
Use case: A Rockwell CompactLogix PLC with a conveyor system. You want to read a tag ‘Conveyor_Speed’ and write to ‘Target_Speed’.
Connection method: EtherNet/IP via pycomm3.
Chat prompt: “Connect to Rockwell PLC at 192.168.1.50:44818. Read tag ‘Conveyor_Speed’ (REAL). If speed < 10, write tag ‘Target_Speed’ = 15.”
AI-generated code:
from pycomm3 import LogixDriver
with LogixDriver('192.168.1.50') as plc:
speed = plc.read('Conveyor_Speed')
print(f"Current speed: {speed.value}")
if speed.value < 10:
plc.write(('Target_Speed', 15.0))
print("Target speed updated to 15")
Pitfall to avoid: pycomm3 requires the PLC to have EtherNet/IP enabled. Check your Rockwell firmware.
7. CAN Bus Gateway → Robotic Arm Control
Use case: A CAN bus gateway connects to a robotic arm. You want to send a frame to move joint 1 to 90 degrees.
Connection method: CAN bus via python-can.
Chat prompt: “Connect to CAN interface ‘can0’ at 250 kbps. Send frame ID 0x100 with data [0x01, 0x5A, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00] to move joint 1 to 90 degrees.”
AI-generated code:
import can
bus = can.interface.Bus(channel='can0', bustype='socketcan', bitrate=250000)
msg = can.Message(arbitration_id=0x100, data=[0x01, 0x5A, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00], is_extended_id=False)
bus.send(msg)
print("Frame sent")
bus.shutdown()
Pitfall to avoid: python-can needs a physical CAN interface (e.g., USBtin, PCAN). The sandbox only works if the interface is available on the host running bridge.py (for local access) or via a remote gateway.
8. CoAP IoT Gateway → Sensor Data from ESP32
Use case: An ESP32 with CoAP server exposes temperature at path /temp. You want to GET it every 5 seconds.
Connection method: CoAP via aiocoap.
Chat prompt: “Send CoAP GET request to coap://192.168.1.80/temp. Parse response as float. If temperature > 40°C, send a POST to coap://192.168.1.80/actuator with payload ‘fan_on’.”
AI-generated code:
import asyncio
from aiocoap import *
async def coap_check():
protocol = await Context.create_client_context()
request = Message(code=GET, uri='coap://192.168.1.80/temp')
response = await protocol.request(request).response
temp = float(response.payload.decode())
print(f"Temp: {temp}")
if temp > 40:
post_request = Message(code=POST, uri='coap://192.168.1.80/actuator', payload=b'fan_on')
await protocol.request(post_request).response
print("Fan activated")
await protocol.shutdown()
asyncio.run(coap_check())
Pitfall to avoid: CoAP runs on UDP. Ensure your network allows UDP traffic on port 5683.
9. SSH into Raspberry Pi → Camera + OpenCV
Use case: A Raspberry Pi with a camera module runs a Python script. You want the AI to SSH in, capture an image, and detect objects.
Connection method: SSH via paramiko.
Chat prompt: “SSH into 192.168.1.90 with user pi, password raspberry. Run ‘python3 capture.py’ to take a photo. Then run ‘python3 detect.py’ to detect objects. Print the output.”
AI-generated code:
import paramiko
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('192.168.1.90', username='pi', password='raspberry')
stdin, stdout, stderr = ssh.exec_command('python3 capture.py')
print("Capture output:", stdout.read().decode())
stdin, stdout, stderr = ssh.exec_command('python3 detect.py')
print("Detection output:", stdout.read().decode())
ssh.close()
Pitfall to avoid: Hardcoding passwords is insecure. Use SSH keys and store them in environment variables via os.environ.
10. HTTP API Smart Plug → Energy Management
Use case: A TP-Link Kasa smart plug reports power consumption via HTTP API. You want to turn it off when consumption exceeds 2000W.
Connection method: HTTP API via requests.
Chat prompt: “Get power consumption from http://192.168.1.70?cmd=energy. Parse JSON field ‘power’. If power > 2000, send POST to http://192.168.1.70?cmd=off.”
AI-generated code:
import requests
import json
resp = requests.get('http://192.168.1.70?cmd=energy')
data = resp.json()
power = data['power']
print(f"Power: {power} W")
if power > 2000:
requests.post('http://192.168.1.70?cmd=off')
print("Plug turned off")
Pitfall to avoid: Some smart plugs require authentication (token). Include it in the URL or headers as needed.
11. Serial (RS-232) GPS Tracker → Real-Time Location
Use case: A GPS tracker sends NMEA sentences over RS-232 at 9600 baud. You want to parse coordinates and log them.
Connection method: COM port via Hardware Bridge (bridge.py).
Chat prompt: “Connect to COM3 at 9600 baud. Use the bridge to read NMEA sentences. Parse $GPGGA and extract latitude/longitude. Print them.”
AI-generated code (sent via industrial_command):
industrial_command(
protocol='serial',
command='serial_write_and_read',
data='' # just read
)
The bridge returns raw data. The AI then parses it with pynmea2 in a subsequent execute_python call.
Pitfall to avoid: The bridge uses a 5-second timeout by default. If the GPS sends data less frequently, increase the timeout via --timeout=10.
12. gRPC Microservice → Industrial Data Pipeline
Use case: A gRPC service exposes sensor data streaming. You want to subscribe and trigger alerts.
Connection method: gRPC via grpcio.
Chat prompt: “Connect to gRPC server at 192.168.1.100:50051. Call the ‘SubscribeSensorData’ RPC with sensor_id=5. Print each response. If value > 100, log warning.”
AI-generated code:
import grpc
import sensor_pb2
import sensor_pb2_grpc
channel = grpc.insecure_channel('192.168.1.100:50051')
stub = sensor_pb2_grpc.SensorServiceStub(channel)
request = sensor_pb2.SensorRequest(sensor_id=5)
for response in stub.SubscribeSensorData(request):
print(f"Value: {response.value}")
if response.value > 100:
print("WARNING: High value!")
Pitfall to avoid: gRPC requires .proto files compiled to Python. The sandbox doesn’t have them; you must include the generated stubs in the script or use a simpler HTTP API if possible.
Why This Changes Everything
Traditional IIoT integration requires a developer to write custom connectors, test them, and deploy them. With ASI Biont, you skip all that. You describe the task in plain English, and the AI agent writes the exact Python code needed—using the right library, the right protocol, and the right error handling.
No more waiting for vendor SDKs, no more debugging serial port timing, no more “device not supported” messages. ASI Biont connects to any industrial gateway because it uses open protocols (Modbus, MQTT, OPC UA, BACnet, etc.) and generates code on the fly.
Get Started Now
Ready to connect your IIoT gateway to an AI agent? Go to asibiont.com, create an API key, download bridge.py (if you need serial access), and start a chat with the AI. Tell it: “Connect to my Modbus gateway at 192.168.1.100, read holding register 0, and send me an email if it’s above 80.” The AI will handle the rest.
Your factory data is only a conversation away.
Comments