MQTT Meets AI: How to Integrate Mosquitto and EMQX with ASI Biont for Smarter IoT Automation

Introduction

Imagine a factory floor where a temperature spike on an ESP32 sensor triggers an AI agent to shut down a cooling pump, log the event to a database, and send a Slack alert — all without a single line of manual code. That's the promise of connecting your MQTT infrastructure to an AI agent like ASI Biont. MQTT (Message Queuing Telemetry Transport) is the de facto standard for IoT communication, used by millions of devices from smart thermostats to industrial PLCs. Mosquitto and EMQX are two of the most popular brokers, with Mosquitto powering lightweight edge deployments (often on Raspberry Pi) and EMQX handling enterprise-scale clusters with millions of concurrent connections.

Traditionally, managing MQTT-based systems requires writing custom Python scripts (using paho-mqtt), setting up dashboards in Node-RED or Grafana, and manually coding business logic. ASI Biont changes the game: you describe your MQTT setup in natural language, and the AI agent writes the integration code, subscribes to topics, analyzes payloads, and executes actions — all through a chat interface. No dashboards. No 'add device' buttons. Just conversation and instant automation.

In this guide, you'll learn how ASI Biont connects to MQTT brokers (Mosquitto, EMQX) using paho-mqtt, with real code examples for monitoring sensors, controlling relays, and sending alerts. We'll cover five concrete use cases, from a single ESP32 temperature sensor to a multi-broker industrial deployment.

How ASI Biont Connects to MQTT

ASI Biont supports two methods for MQTT integration:

1. execute_python (sandbox): The AI writes a Python script using the paho-mqtt library and runs it in a secure sandbox environment on the ASI Biont server (Railway). The script connects to your MQTT broker (e.g., mqtt://192.168.1.100:1883), subscribes to topics, receives callbacks, and publishes commands. This is the most flexible method — AI can parse JSON payloads, compute averages, trigger HTTP calls, or write to databases.

2. industrial_command tool: For quick, atomic operations (like publishing a single command), AI can use the industrial_command tool with the publish command. Example: industrial_command(protocol='mqtt', command='publish', params={"topic": "relay/1", "payload": "ON"}). This is useful for toggling relays or sending one-off messages without writing a full script.

Both methods require the broker's IP address or hostname, port (default 1883 for MQTT, 8883 for MQTTS), and optional username/password. The user simply describes these parameters in the chat.

Use Case 1: ESP32 Temperature Sensor → MQTT → AI Alerting

Scenario: An ESP32 with a DHT22 sensor publishes temperature and humidity every 10 seconds to sensor/room1. You want the AI agent to monitor this topic, detect when temperature exceeds 30°C, and send a Telegram alert.

AI prompt: "Connect to MQTT broker at 192.168.1.100:1883, subscribe to 'sensor/room1', parse JSON payload, and if temperature > 30°C, send a Telegram message using my bot token."

AI-generated code (runs in execute_python sandbox):

import paho.mqtt.client as mqtt
import json
import asyncio
import aiohttp

# Configuration from user
broker = "192.168.1.100"
port = 1883
topic = "sensor/room1"
telegram_token = "YOUR_BOT_TOKEN"
telegram_chat_id = "YOUR_CHAT_ID"

def on_message(client, userdata, msg):
    try:
        data = json.loads(msg.payload.decode())
        temp = data.get("temperature")
        if temp and temp > 30:
            asyncio.run(send_telegram(f"Alert: Temperature {temp}°C exceeds 30°C!"))
    except Exception as e:
        print(f"Parse error: {e}")

async def send_telegram(text):
    async with aiohttp.ClientSession() as session:
        url = f"https://api.telegram.org/bot{telegram_token}/sendMessage"
        payload = {"chat_id": telegram_chat_id, "text": text}
        async with session.post(url, json=payload) as resp:
            print(f"Telegram sent: {resp.status}")

client = mqtt.Client()
client.on_message = on_message
client.connect(broker, port, 60)
client.subscribe(topic)
client.loop_start()
# Keep running for 30 seconds (sandbox limit)
import time
time.sleep(30)
client.loop_stop()

Step-by-step:
1. User provides broker IP, port, and Telegram credentials in chat.
2. AI writes the script, connects to broker, subscribes to sensor/room1.
3. When temperature exceeds threshold, AI calls Telegram API asynchronously.
4. Script runs for 30 seconds (sandbox timeout), then stops. For continuous monitoring, AI can be scheduled to run the script periodically (e.g., every 5 minutes via a cron-like trigger).

Use Case 2: Two-Way Relay Control via MQTT

Scenario: An ESP32 controls a relay connected to GPIO 5. The ESP32 subscribes to relay/1 and toggles the relay on receiving "ON" or "OFF". You want to control the relay from the ASI Biont chat.

AI prompt: "Publish 'ON' to topic 'relay/1' on MQTT broker at 10.0.0.5:1883."

AI action (uses industrial_command tool):

industrial_command(
    protocol='mqtt',
    command='publish',
    params={
        "broker": "10.0.0.5",
        "port": 1883,
        "topic": "relay/1",
        "payload": "ON"
    }
)

Step-by-step:
1. User says: "Turn on the relay via MQTT."
2. AI identifies the device and uses the publish command.
3. Broker delivers the message to the ESP32, which sets GPIO high.
4. The user can also ask: "Read the relay state" — AI subscribes to relay/1/status and returns the value.

Use Case 3: Multi-Sensor Dashboard with EMQX

Scenario: An industrial facility uses EMQX to aggregate data from 20 PLCs publishing to topics like factory/line1/temperature, factory/line1/pressure. You want the AI agent to calculate the average temperature across all sensors every minute and log it to a PostgreSQL database.

AI prompt: "Connect to EMQX at mqtts://factory.emqx.cloud:8883 with username 'admin' and password 'secret'. Subscribe to 'factory/+/temperature'. Every 60 seconds, calculate average temperature and insert into PostgreSQL table 'sensor_avg' with columns 'timestamp' and 'avg_temp'."

AI-generated code:

import paho.mqtt.client as mqtt
import json
import psycopg2
from datetime import datetime

broker = "factory.emqx.cloud"
port = 8883
username = "admin"
password = "secret"
topic = "factory/+/temperature"

db_config = {
    "dbname": "iot_data",
    "user": "postgres",
    "password": "postgres",
    "host": "localhost",
    "port": 5432
}

temperatures = []

def on_message(client, userdata, msg):
    try:
        temp = float(msg.payload.decode())
        temperatures.append(temp)
        print(f"Received {temp} from {msg.topic}")
    except:
        pass

client = mqtt.Client()
client.username_pw_set(username, password)
client.tls_set()  # For MQTTS
client.on_message = on_message
client.connect(broker, port, 60)
client.subscribe(topic)
client.loop_start()

import time
for _ in range(60):  # Collect for ~60 seconds
    time.sleep(1)

client.loop_stop()

if temperatures:
    avg = sum(temperatures) / len(temperatures)
    conn = psycopg2.connect(**db_config)
    cur = conn.cursor()
    cur.execute("INSERT INTO sensor_avg (timestamp, avg_temp) VALUES (%s, %s)", 
                (datetime.now(), avg))
    conn.commit()
    cur.close()
    conn.close()
    print(f"Inserted average: {avg}")

Step-by-step:
1. User provides EMQX credentials, database connection string.
2. AI script subscribes to wildcard topic factory/+/temperature (MQTT wildcard + matches any single level).
3. Over 60 seconds, it collects readings, computes average, writes to PostgreSQL.
4. This can be scheduled as a recurring task.

Use Case 4: Bridging MQTT and Modbus via AI

Scenario: A Modbus/TCP PLC controls a motor. You want to expose PLC registers over MQTT — when an MQTT topic receives a command, AI reads/writes Modbus registers.

AI prompt: "Listen to MQTT topic 'motor/command' at broker 192.168.1.10:1883. When payload is 'START', write value 1 to Modbus register 40001 at 10.0.0.100:502. When 'STOP', write 0. Also publish current motor status from register 40002 to 'motor/status' every 10 seconds."

AI-generated code:

import paho.mqtt.client as mqtt
from pymodbus.client import ModbusTcpClient
import time

mqtt_broker = "192.168.1.10"
modbus_host = "10.0.0.100"
modbus_port = 502

modbus_client = ModbusTcpClient(modbus_host, modbus_port)
modbus_client.connect()

def on_message(client, userdata, msg):
    command = msg.payload.decode()
    if command == "START":
        modbus_client.write_register(40001, 1)
        print("Motor started")
    elif command == "STOP":
        modbus_client.write_register(40001, 0)
        print("Motor stopped")

mqtt_client = mqtt.Client()
mqtt_client.on_message = on_message
mqtt_client.connect(mqtt_broker, 1883, 60)
mqtt_client.subscribe("motor/command")
mqtt_client.loop_start()

for _ in range(30):
    status = modbus_client.read_holding_registers(40002, 1).registers[0]
    mqtt_client.publish("motor/status", str(status))
    time.sleep(10)

mqtt_client.loop_stop()
modbus_client.close()

Step-by-step:
1. User provides both MQTT and Modbus parameters.
2. AI creates a bidirectional bridge: MQTT commands → Modbus writes; Modbus reads → MQTT publishes.
3. The script runs for 30 seconds (sandbox limit) — for persistent bridging, user can run it on a local PC via Hardware Bridge or as a scheduled task.

Use Case 5: EMQX Shared Subscription for High Availability

Scenario: You have two identical ESP32 sensors publishing to sensor/temp. You want to use EMQX's shared subscription feature (e.g., $share/group1/sensor/temp) to distribute messages across two AI agent instances for load balancing.

AI prompt: "Subscribe to shared topic '$share/group1/sensor/temp' on EMQX at 10.0.0.50:1883. Process each message by calculating a moving average of the last 10 readings and publish to 'sensor/temp/avg'."

AI-generated code (simplified):

import paho.mqtt.client as mqtt
import json
from collections import deque

broker = "10.0.0.50"
port = 1883
topic = "$share/group1/sensor/temp"

window = deque(maxlen=10)

def on_message(client, userdata, msg):
    try:
        temp = float(msg.payload.decode())
        window.append(temp)
        avg = sum(window) / len(window)
        client.publish("sensor/temp/avg", f"{avg:.2f}")
        print(f"Average: {avg}")
    except Exception as e:
        print(f"Error: {e}")

client = mqtt.Client()
client.on_message = on_message
client.connect(broker, port, 60)
client.subscribe(topic)
client.loop_start()

import time
time.sleep(30)
client.loop_stop()

Step-by-step:
1. User specifies shared subscription syntax ($share/<group>/<topic>).
2. AI subscribes, maintains a sliding window of 10 readings, publishes the average.
3. If a second AI instance subscribes with the same group, EMQX distributes messages round-robin.

Comparison with Traditional MQTT Dashboards

Why This Matters

Traditional MQTT monitoring requires writing boilerplate connection code, parsing payloads, and maintaining alerting logic. With ASI Biont, you just talk to the AI: "Monitor temperature sensors and alert me if any exceed 35°C." The AI generates production-ready Python code using paho-mqtt, connects to your broker, and executes the logic — all in seconds. This is especially powerful for:

• Rapid prototyping: Test MQTT scenarios without writing code.
• Edge cases: Quickly create one-off scripts for debugging or data collection.
• Multi-protocol environments: Bridge MQTT with Modbus, OPC UA, or HTTP without manual coding.

Getting Started

To try this integration:
1. Go to asibiont.com and sign up.
2. In the chat, say: "Connect to my MQTT broker at 192.168.1.100:1883, subscribe to 'sensor/#' and tell me the latest temperature."
3. The AI will respond, write the code, and execute it — you'll see the data in seconds.

No dashboards. No plugins. Just AI-powered IoT automation.

Conclusion

MQTT is the heartbeat of IoT, and ASI Biont amplifies its power by adding an AI brain that understands your data and acts on it. Whether you're using Mosquitto on a Raspberry Pi or EMQX in a cloud cluster, the AI agent connects in seconds, writes robust Python code, and automates your workflows. From simple temperature alerts to multi-protocol bridges, the integration is limited only by your imagination — and your chat prompt.

Stop building dashboards. Start conversing with your data. Try ASI Biont today at asibiont.com.