Safe AI Control of IoT Devices: Guardrails for AI-Issued MQTT Commands

Safe AI control of IoT devices means that a language model can propose actions, but only validated, rate-limited and logged commands ever reach a relay, valve or motor. In this guide you will design a command gateway that sits between an AI agent and your MQTT devices, and implement allowlists, range checks, rate limits, human confirmation and an audit topic in about 150 lines of Python.

Why AI Commands Need Their Own Safety Layer

Language models are good at turning "it's too warm in the meeting room" into "set the room 3 setpoint to 21 degrees". They are also capable of misreading a room number, repeating a command in a loop, or following an instruction hidden in a sensor label or an email. None of that is unusual behaviour for a probabilistic text generator, and prompts alone cannot rule it out.

The fix is the same one control engineers have always used: assume the upstream source can be wrong and validate everything at a boundary you control. The AI becomes one more untrusted client, and a deterministic program decides what is allowed.

There is a second reason. Even a perfectly correct model can act on stale information. If it read a tank level that is ten minutes old and opens a fill valve, the command was "reasonable" and the outcome is still a flood. The gateway is the place to check freshness and current state before acting.

The Command Gateway Pattern

The architecture has one rule: the AI never holds broker credentials that can reach devices. Instead:

1. The agent calls a tool such as request_action(device, action, value, reason), as described in our guide on LLM function calling for IoT.
2. The tool hands the request to the gateway, a small service you own.
3. The gateway validates the request, applies limits, optionally asks a human, and only then publishes to the device command topic.
4. Every decision, accepted or rejected, is published to an audit topic and written to a log.
5. The device acknowledges on a state topic, and the gateway reports the real outcome back to the agent.

This matters because an MQTT client with credentials can usually publish to any topic its account can reach. ConnectMQ binds client IDs to the credentials it issues, but it does not offer per-topic ACLs, so the honest way to enforce "the AI cannot touch the pump directly" is to keep the device credentials inside the gateway process and out of the agent entirely. If the agent runs as an MCP server for MQTT, expose only gateway-backed tools, never a raw publish.

Guardrails for Safe AI Control of IoT Devices

Each guardrail addresses a specific failure mode:

Interlocks that protect people and equipment stay in the PLC or device firmware. The gateway reduces nuisance and risk; it is not a substitute for a safety-rated system.

Topic Layout for Guarded Commands

Keep AI-originated traffic visibly separate from normal traffic. A layout that works well:

<prefix>/ai/requests            AI proposals (JSON), written by the agent tool
<prefix>/ai/pending/<id>        retained while waiting for a human
<prefix>/ai/confirm/<id>        human publishes "approve" or "reject"
<prefix>/ai/audit               every decision, never retained
<prefix>/ai/enabled             retained "true"/"false" kill switch
<prefix>/home/<room>/<device>/set     command topic, only the gateway publishes
<prefix>/home/<room>/<device>/state   retained device state

If you already follow a room/device/property scheme as described in our article on smart home MQTT topic structure, the gateway simply adds the ai/ subtree next to it.

Implementing the Gateway in Python

The gateway below uses paho-mqtt 2.x and MQTT 5. It runs against the public test broker so you can try it; change the prefix to your own random ID.

import json
import time
import uuid
from collections import defaultdict, deque

import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

PREFIX = "connectmq/demo/r8x2m4"   # use your own random id
ALLOW = {
    "livingroom/thermostat": {"set_temp": {"min": 16, "max": 24, "confirm": False}},
    "garage/door":           {"open": {"confirm": True}, "close": {"confirm": False}},
    "garden/valve":          {"open_minutes": {"min": 1, "max": 15, "confirm": True}},
}
RATE_LIMIT = (3, 600)          # max 3 commands per device per 10 minutes
MAX_STATE_AGE = 120            # seconds
history = defaultdict(deque)
last_state = {}
pending = {}
ai_enabled = True

def audit(client, decision, req, reason):
    rec = {"ts": time.time(), "decision": decision, "reason": reason, "request": req}
    client.publish(f"{PREFIX}/ai/audit", json.dumps(rec), qos=1)
    with open("ai_audit.log", "a", encoding="utf-8") as f:
        f.write(json.dumps(rec) + "\n")

def validate(req):
    if not ai_enabled:
        return "AI control disabled"
    rule = ALLOW.get(req.get("device"), {}).get(req.get("action"))
    if rule is None:
        return "device/action not allowlisted"
    if "min" in rule:
        v = req.get("value")
        if not isinstance(v, (int, float)) or not rule["min"] <= v <= rule["max"]:
            return f"value outside {rule['min']}..{rule['max']}"
    seen = last_state.get(req["device"])
    if seen is None or time.time() - seen > MAX_STATE_AGE:
        return "device state unknown or stale"
    window = history[req["device"]]
    while window and time.time() - window[0] > RATE_LIMIT[1]:
        window.popleft()
    if len(window) >= RATE_LIMIT[0]:
        return "rate limit reached"
    return None

def execute(client, req):
    props = Properties(PacketTypes.PUBLISH)
    props.MessageExpiryInterval = 15   # drop if not delivered within 15 s
    payload = {"action": req["action"], "value": req.get("value"), "source": "ai", "id": req["id"]}
    client.publish(f"{PREFIX}/home/{req['device']}/set", json.dumps(payload), qos=1, properties=props)
    history[req["device"]].append(time.time())
    audit(client, "executed", req, "passed all checks")

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe([(f"{PREFIX}/ai/requests", 1), (f"{PREFIX}/ai/confirm/+", 1),
                      (f"{PREFIX}/ai/enabled", 1), (f"{PREFIX}/home/+/+/state", 1)])

def on_message(client, userdata, msg):
    global ai_enabled
    t = msg.topic
    if t.endswith("/state"):
        device = "/".join(t.split("/")[-3:-1])
        last_state[device] = time.time()
    elif t == f"{PREFIX}/ai/enabled":
        ai_enabled = msg.payload.decode().strip().lower() == "true"
    elif t == f"{PREFIX}/ai/requests":
        try:
            req = json.loads(msg.payload)
        except ValueError:
            return
        req["id"] = str(uuid.uuid4())[:8]
        problem = validate(req)
        if problem:
            audit(client, "rejected", req, problem)
            return
        if ALLOW[req["device"]][req["action"]].get("confirm"):
            pending[req["id"]] = req
            client.publish(f"{PREFIX}/ai/pending/{req['id']}", json.dumps(req), qos=1, retain=True)
            audit(client, "awaiting_confirmation", req, "high-impact action")
        else:
            execute(client, req)
    elif "/ai/confirm/" in t:
        req_id = t.rsplit("/", 1)[-1]
        req = pending.pop(req_id, None)
        if req is None:
            return
        client.publish(f"{PREFIX}/ai/pending/{req_id}", b"", qos=1, retain=True)  # clear
        if msg.payload.decode().strip() == "approve":
            problem = validate(req)   # re-check: state may have changed
            if problem:
                audit(client, "rejected", req, "on confirm: " + problem)
            else:
                execute(client, req)
        else:
            audit(client, "rejected", req, "human rejected")

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="ai-gateway-r8x2m4",
                     protocol=mqtt.MQTTv5)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()

Test it from a terminal. First publish a fresh device state, then a request that should run, then one that should be blocked:

P=connectmq/demo/r8x2m4
mosquitto_pub -h broker.connectmq.com -p 1883 -i t1-r8x2m4 -r -t $P/home/livingroom/thermostat/state -m '{"temp":22}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i t1-r8x2m4 -t $P/ai/requests \
  -m '{"device":"livingroom/thermostat","action":"set_temp","value":21,"reason":"user feels warm"}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i t1-r8x2m4 -t $P/ai/requests \
  -m '{"device":"livingroom/thermostat","action":"set_temp","value":35,"reason":"test"}'
mosquitto_sub -h broker.connectmq.com -p 1883 -i t2-r8x2m4 -t "$P/ai/#" -v

The second request shows up on the audit topic as rejected with value outside 16..24. Because the public area is readable by anyone, use it only for experiments like this; never put real device control on it.

Design Notes That Matter in Practice

Re-validate on confirmation. A human might approve a request five minutes after it was made. The code calls validate again, so a rate limit or stale state that appeared in the meantime still blocks it.

Expire commands. The MQTT 5 message expiry interval means a command queued for an offline device is discarded rather than delivered an hour later when the garage door reconnects. Our overview of MQTT 5 features covers expiry and the response topic pattern you can use for richer acknowledgements.

Report the real outcome. The agent should tell the user "the thermostat confirmed 21 degrees", not "I set it". Wait for the state topic to change before answering.

Make the kill switch boring. A retained false on ai/enabled stops all AI actions immediately and survives gateway restarts because the broker replays it on subscribe.

Treat descriptions as untrusted input. Device names, notes and sensor labels that the model reads can contain instructions. The gateway ignores the reason field for decisions; it is logged for humans only.

Safe AI Control Checklist for Production

1. The agent has no device credentials; only the gateway does.
2. Every device/action pair is explicitly allowlisted, with ranges.
3. Rate limits exist per device and globally.
4. High-impact actions require human confirmation, and approval re-runs validation.
5. Commands carry an expiry and a unique ID.
6. Every decision is published to an audit topic and stored durably by a separate subscriber.
7. A retained kill switch is documented and tested.
8. Physical interlocks remain in firmware or the PLC.
9. Production traffic runs over TLS on port 8883 with credentials from the portal, as recommended in our MQTT security best practices.

Moving From Demo to a Private Broker

Switch the connection to your own broker and credentials:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID", protocol=mqtt.MQTTv5)
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()
client.connect("broker.connectmq.com", 8883, keepalive=60)

A hosted MQTT broker such as ConnectMQ gives you TLS, generated credentials, flood protection and a live topic explorer for watching the audit stream, while the guardrail logic stays in code you control. To see where command gateways fit among other AI patterns, read the AI and IoT complete guide, and for background on hosting choices see our page on the cloud MQTT broker. For the natural-language side of the same system, see natural language device control.

Build Guarded AI Automation on ConnectMQ

Start with the public test broker, then move the gateway to a private broker with TLS and its own credentials. Create a free Serverless broker at app.connectmq.com, no credit card needed, and keep every AI command visible on an audit topic from day one.

Frequently Asked Questions

Can I rely on the system prompt to keep the AI safe?

No. Prompts reduce mistakes but cannot guarantee behaviour. Enforce rules in deterministic code that the model cannot bypass, and keep device credentials out of the agent.

Does ConnectMQ validate AI commands for me?

No. ConnectMQ delivers messages securely and reliably; validation, confirmation and auditing live in your gateway service, which connects to ConnectMQ as an ordinary MQTT client.

Which actions should always need human confirmation?

Anything that moves mechanisms, opens doors or valves, unlocks access, or could cause damage if repeated. Comfort adjustments within a narrow range are usually safe to automate.

How do I store the audit trail long-term?

Run a separate subscriber on the audit topic that writes to a file or database. Audit messages are not retained on the broker, so the subscriber must be running to capture them.