Natural Language Device Control: Turn Plain Requests into Validated MQTT Commands

Natural language device control lets people say what they want, such as "make the office warmer" or "turn off everything in the living room", instead of hunting through dashboards. In this tutorial you will build a Python service where an LLM translates requests into structured function calls, a validation layer checks every call against a device registry, and only valid commands are published over MQTT to devices that acknowledge them.

How the Pipeline Works

The key idea is that the language model never writes MQTT messages. It only chooses from a menu of functions you define, and your code turns those choices into commands after checking them.

1. User request. Free text from a chat box, a voice assistant transcript or a messaging bot.
2. Context. The service gives the model a compact list of devices, their friendly names and capabilities.
3. Function calling. The model returns zero or more tool calls, for example set_light(device="kitchen-ceiling", on=true, brightness=30).
4. Validation. Code checks that the device exists, supports the capability and that values are in range.
5. Publish. Valid commands go to .../<device>/set over MQTT with a unique command ID.
6. Acknowledge. The device applies the change and publishes its new state, which the service reports back to the user.

Prerequisites

• Python 3.10 or newer.
• Ollama running locally with a model that supports tool calling, such as qwen2.5:7b or llama3.1:8b. Any provider with function calling works the same way; only the API call changes.
• An MQTT broker. We use broker.connectmq.com for the demo.

ollama pull qwen2.5:7b
pip install ollama "paho-mqtt>=2.0"

Step 1: Define the Device Registry

The registry is the single source of truth for what exists and what it can do. The model sees a summary of it; the validator uses all of it.

# registry.py
DEVICES = {
    "kitchen-ceiling": {"name": "kitchen ceiling light", "room": "kitchen", "type": "light", "dimmable": True},
    "kitchen-under":   {"name": "kitchen under-cabinet light", "room": "kitchen", "type": "light", "dimmable": False},
    "living-lamp":     {"name": "living room floor lamp", "room": "living room", "type": "light", "dimmable": True},
    "living-blinds":   {"name": "living room blinds", "room": "living room", "type": "blinds"},
    "office-thermo":   {"name": "office thermostat", "room": "office", "type": "thermostat", "min_c": 16, "max_c": 26},
}


def describe_devices():
    lines = []
    for dev_id, d in DEVICES.items():
        extra = ""
        if d["type"] == "light":
            extra = " (dimmable)" if d["dimmable"] else " (on/off only)"
        if d["type"] == "thermostat":
            extra = f" (range {d['min_c']}-{d['max_c']} C)"
        lines.append(f"- {dev_id}: {d['name']}, room={d['room']}, type={d['type']}{extra}")
    return "\n".join(lines)

Step 2: Define the Functions

Each function corresponds to one capability. Enumerating device IDs in the schema helps the model choose valid targets.

# tools.py
from registry import DEVICES

LIGHTS = [k for k, d in DEVICES.items() if d["type"] == "light"]
BLINDS = [k for k, d in DEVICES.items() if d["type"] == "blinds"]
THERMOS = [k for k, d in DEVICES.items() if d["type"] == "thermostat"]

TOOLS = [
    {"type": "function", "function": {
        "name": "set_light",
        "description": "Turn a light on or off and optionally set brightness in percent (dimmable lights only).",
        "parameters": {"type": "object", "properties": {
            "device": {"type": "string", "enum": LIGHTS},
            "on": {"type": "boolean"},
            "brightness": {"type": "integer", "minimum": 1, "maximum": 100}},
            "required": ["device", "on"]}}},
    {"type": "function", "function": {
        "name": "set_blinds",
        "description": "Set blinds position in percent, 0 is closed and 100 is fully open.",
        "parameters": {"type": "object", "properties": {
            "device": {"type": "string", "enum": BLINDS},
            "position": {"type": "integer", "minimum": 0, "maximum": 100}},
            "required": ["device", "position"]}}},
    {"type": "function", "function": {
        "name": "set_thermostat",
        "description": "Set the target temperature in degrees Celsius.",
        "parameters": {"type": "object", "properties": {
            "device": {"type": "string", "enum": THERMOS},
            "target_c": {"type": "number"}},
            "required": ["device", "target_c"]}}},
]

Step 3: Validate Every Call

Schemas guide the model but do not bind it. Small local models in particular sometimes return numbers as strings, invent device IDs or ignore ranges. The validator normalizes what is safe to normalize and rejects everything else with a reason.

# validate.py
from registry import DEVICES


class Invalid(Exception):
    pass


def as_int(value, field):
    try:
        return int(round(float(value)))
    except (TypeError, ValueError):
        raise Invalid(f"{field} must be a number")


def validate(name, args):
    dev_id = args.get("device")
    dev = DEVICES.get(dev_id)
    if dev is None:
        raise Invalid(f"unknown device {dev_id!r}")
    if name == "set_light":
        if dev["type"] != "light":
            raise Invalid(f"{dev_id} is not a light")
        on = args.get("on")
        if isinstance(on, str):
            on = on.lower() in ("true", "on", "1")
        cmd = {"on": bool(on)}
        if "brightness" in args and args["brightness"] is not None:
            if not dev["dimmable"]:
                raise Invalid(f"{dev['name']} is not dimmable")
            b = as_int(args["brightness"], "brightness")
            if not 1 <= b <= 100:
                raise Invalid("brightness must be 1-100")
            cmd["brightness"] = b
        return dev_id, cmd
    if name == "set_blinds":
        if dev["type"] != "blinds":
            raise Invalid(f"{dev_id} is not blinds")
        p = as_int(args.get("position"), "position")
        if not 0 <= p <= 100:
            raise Invalid("position must be 0-100")
        return dev_id, {"position": p}
    if name == "set_thermostat":
        if dev["type"] != "thermostat":
            raise Invalid(f"{dev_id} is not a thermostat")
        try:
            t = round(float(args.get("target_c")), 1)
        except (TypeError, ValueError):
            raise Invalid("target_c must be a number")
        if not dev["min_c"] <= t <= dev["max_c"]:
            raise Invalid(f"target must be {dev['min_c']}-{dev['max_c']} C")
        return dev_id, {"target_c": t}
    raise Invalid(f"unknown function {name}")

Step 4: The Natural Language Device Control Service

Now connect everything. The service reads a request, asks the model for tool calls, validates them and publishes commands.

# nl_control.py
import json
import time
import uuid

import ollama
import paho.mqtt.client as mqtt

from registry import describe_devices
from tools import TOOLS
from validate import Invalid, validate

MODEL = "qwen2.5:7b"
PREFIX = "connectmq/demo/e3b90c/home"   # use your own random id

SYSTEM = (
    "You control smart home devices by calling functions. "
    "Only use device IDs from this list:\n" + describe_devices() + "\n"
    "If a request is ambiguous or no device matches, ask a short clarifying question instead of calling a function. "
    "Never guess temperatures outside the allowed range."
)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="nl-control-e3b90c")
client.on_message = lambda c, u, m: print(f"  state {m.topic.split('/')[-2]}: {m.payload.decode()}")
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.subscribe(f"{PREFIX}/+/state", qos=1)
client.loop_start()


def handle(request):
    resp = ollama.chat(
        model=MODEL,
        messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": request}],
        tools=TOOLS,
        options={"temperature": 0},
    )
    calls = resp.message.tool_calls or []
    if not calls:
        print("assistant:", resp.message.content)
        return
    for call in calls:
        name, args = call.function.name, dict(call.function.arguments)
        try:
            dev_id, cmd = validate(name, args)
        except Invalid as err:
            print(f"rejected {name}({args}): {err}")
            continue
        cmd.update({"id": str(uuid.uuid4()), "ts": int(time.time() * 1000), "source": "nl"})
        client.publish(f"{PREFIX}/{dev_id}/set", json.dumps(cmd), qos=1)
        print(f"sent {dev_id}: {cmd}")


if __name__ == "__main__":
    while True:
        text = input("\nyou: ").strip()
        if text:
            handle(text)

Setting temperature to 0 makes tool selection more repeatable, which you want for control tasks.

Step 5: Add a Simulated Device and Test

A tiny simulator subscribes to all set topics, applies commands and publishes retained state:

# sim_home.py
import json
import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/e3b90c/home"
state = {}


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/+/set", qos=1)


def on_message(client, userdata, msg):
    dev = msg.topic.split("/")[-2]
    cmd = json.loads(msg.payload)
    state.setdefault(dev, {}).update({k: v for k, v in cmd.items() if k not in ("id", "ts", "source")})
    client.publish(f"{PREFIX}/{dev}/state", json.dumps({**state[dev], "lastCommand": cmd["id"]}), qos=1, retain=True)


c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="sim-home-e3b90c")
c.on_connect = on_connect
c.on_message = on_message
c.connect("broker.connectmq.com", 1883, keepalive=60)
c.loop_forever()

Run the simulator, then the service, and try these requests:

The interesting cases are the failures. Each one is caught either by the model asking for clarification or by the validator, and nothing invalid reaches the broker.

Why the Model Never Writes MQTT Payloads

It is tempting to skip the function layer and ask the model to output a topic and a JSON payload directly. That approach breaks in subtle ways. The model may invent a topic that looks right but does not exist, mix up key names between device types or produce valid JSON with the wrong units. Every one of those errors reaches the broker, and some devices will silently ignore malformed commands, leaving the user believing the change happened.

With function calling, the model chooses among a few named operations whose parameters you defined. Your code owns the mapping from operation to topic and payload format, so changing a device's firmware or topic layout never requires changing prompts. The validator then becomes the only place where commands are created, which makes it easy to test with ordinary unit tests and no model at all.

Handling Follow-Up Requests

Real conversations are rarely one-shot. People say "a bit brighter" or "the other one too". To support this, keep the last few user and assistant messages in the messages list and include the current state of recently touched devices in the system prompt, read from the retained state topics. The model can then resolve "a bit brighter" to the light it just changed and compute a new brightness from the current value, while the validator still checks the result. Limit the history to a handful of turns so stale context does not leak into new requests.

Making It Production Ready

When you move from the demo to a real home, building or plant, keep these points in mind:

1. Use private credentials and TLS. The public MQTT test broker is readable by anyone. For real devices, connect on port 8883 with credentials generated in the ConnectMQ portal.
2. Confirm destructive or unusual actions. Turning off all lights is fine; opening a gate or disabling an alarm should require explicit confirmation.
3. Wait for state, not just publish. Report success only after the device state confirms the command ID.
4. Keep the registry generated, not hand-written. Build it from retained device announcement topics so new devices appear automatically.
5. Log everything. Store the request, tool calls, validation results and acknowledgements for troubleshooting.

The same validation discipline scales to industrial equipment; our guides on LLM function calling for IoT and LLM agents for IoT devices extend it with schema libraries and human approval. If you want the whole stack offline, see running Ollama with MQTT on a gateway. For context on where language interfaces fit among other AI techniques, read our AI and IoT guide.

Running the control channel on a hosted MQTT broker means your devices, the language service and dashboards all reach one secure endpoint over TCP, TLS or WebSockets without opening ports at home or in the plant.

Get Started with ConnectMQ

ConnectMQ gives you a fully managed broker with an in-browser MQTT client so you can watch commands and state changes live while you test prompts. Create your free Serverless broker, no credit card required, and let people control devices in their own words.

Frequently Asked Questions

Do I need a large cloud model for natural language device control?

No. Small local models with tool-calling support handle simple device commands well when the device list is short and the functions are clear. Validation in code covers the cases where they get it wrong.

How do I handle ambiguous requests like "turn on the light"?

Tell the model in the system prompt to ask a clarifying question when several devices match. Your service shows the question to the user and sends the answer back in the next turn.

Can I use voice instead of text?

Yes. Run speech-to-text first and feed the transcript into the same pipeline. Keep confirmations for risky actions, because transcription errors add another source of mistakes.

Why publish commands to a set topic and read a separate state topic?

Separating commands from state means the state topic always reflects what the device actually did, and retained state gives every client the current value immediately.