LLM Function Calling IoT Tutorial: Strict JSON Schemas and Validated MQTT Commands

LLM function calling for IoT turns a model's understanding of a request into a structured call like set_valve(valve_id="V-12", position_pct=40), which your code can check and convert into an MQTT command. This tutorial focuses on the part most demos skip: strict JSON schemas for device actions, independent validation of every call, a repair loop for invalid output and MQTT 5 acknowledgements, so nothing the model produces reaches a device unchecked.

Function Calling in One Paragraph

With function calling (also called tool calling), you send the model a list of functions with JSON schema parameter definitions. Instead of answering in prose, the model can return the name of a function and a JSON object of arguments. Your application decides whether to execute it. The model never executes anything itself, which is exactly why the pattern is useful for devices: the application is a choke point where you can enforce rules.

The catch is that a schema in the prompt is guidance, not enforcement. Models occasionally return arguments with the wrong type, values outside a range, extra fields or device IDs that do not exist. Some providers offer strict modes that constrain output to the schema, but semantic rules (is this valve on this line, is this change too large) always need your own code.

Step 1: Design Strict Schemas for Device Actions

Good schemas for physical actions share a few traits:

Here is a schema module shared by the command service and, ideally, the device firmware or gateway:

# schemas.py
DEVICE_ACTIONS = {
    "set_valve": {
        "description": "Set a control valve opening in percent.",
        "parameters": {
            "type": "object",
            "properties": {
                "valve_id": {"type": "string", "enum": ["V-11", "V-12", "V-21"]},
                "position_pct": {"type": "integer", "minimum": 0, "maximum": 100},
                "reason": {"type": "string", "minLength": 5, "maxLength": 200},
            },
            "required": ["valve_id", "position_pct", "reason"],
            "additionalProperties": False,
        },
    },
    "set_pump_state": {
        "description": "Start or stop a pump.",
        "parameters": {
            "type": "object",
            "properties": {
                "pump_id": {"type": "string", "enum": ["P-1", "P-2"]},
                "state": {"type": "string", "enum": ["running", "stopped"]},
                "reason": {"type": "string", "minLength": 5, "maxLength": 200},
            },
            "required": ["pump_id", "state", "reason"],
            "additionalProperties": False,
        },
    },
}


def as_tools():
    """Convert to the tool format used by OpenAI-compatible and Ollama chat APIs."""
    return [
        {"type": "function", "function": {"name": name, "description": spec["description"],
                                          "parameters": spec["parameters"]}}
        for name, spec in DEVICE_ACTIONS.items()
    ]

Step 2: Validate Twice: Schema, Then Semantics

The first check is structural and uses the jsonschema library against the exact same schema the model saw. The second check is semantic: rules that depend on live state or plant policy, which no schema can express.

# validator.py
from jsonschema import Draft202012Validator
from schemas import DEVICE_ACTIONS

VALIDATORS = {n: Draft202012Validator(s["parameters"]) for n, s in DEVICE_ACTIONS.items()}
MAX_VALVE_STEP = 25   # plant policy: no more than 25 points per command


class ValidationFailed(Exception):
    pass


def check(name, args, live_state):
    validator = VALIDATORS.get(name)
    if validator is None:
        raise ValidationFailed(f"Unknown function '{name}'. Allowed: {sorted(VALIDATORS)}")
    errors = sorted(validator.iter_errors(args), key=lambda e: list(e.path))
    if errors:
        details = "; ".join(f"{'/'.join(map(str, e.path)) or '(root)'}: {e.message}" for e in errors)
        raise ValidationFailed(f"Arguments for {name} are invalid: {details}")
    if name == "set_valve":
        current = live_state.get(args["valve_id"], {}).get("position_pct")
        if current is None:
            raise ValidationFailed(f"Current position of {args['valve_id']} is unknown; refusing to move it.")
        if abs(args["position_pct"] - current) > MAX_VALVE_STEP:
            raise ValidationFailed(
                f"Change from {current} to {args['position_pct']} exceeds the {MAX_VALVE_STEP}-point step limit. "
                f"Propose an intermediate value.")
    if name == "set_pump_state" and args["state"] == "stopped":
        if live_state.get(args["pump_id"], {}).get("duty") == "critical":
            raise ValidationFailed(f"{args['pump_id']} is on critical duty and cannot be stopped by the assistant.")
    return args

Error messages are written for the model to read. That is deliberate: they feed the repair loop in the next step.

Step 3: The LLM Function Calling Loop With Repair

When a call fails validation, send the error back and let the model try again, once or twice at most. Many failures are fixable ("propose an intermediate value"), and a bounded loop avoids endless retries. This example uses Ollama with a tool-capable local model; the same structure works with any provider's function calling API.

# command_service.py
import json
import time
import uuid

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

from schemas import as_tools
from validator import ValidationFailed, check

MODEL = "qwen2.5:7b"
PREFIX = "connectmq/demo/d07c5a/plant"     # use your own random id
ACK_TOPIC = f"{PREFIX}/svc/acks"
MAX_REPAIRS = 2

live_state = {}   # filled from retained state topics
acks = {}


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


def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    if msg.topic == ACK_TOPIC:
        acks[data["id"]] = data
    else:
        live_state[msg.topic.split("/")[-2]] = data


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="fc-svc-d07c5a", protocol=mqtt.MQTTv5)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()
time.sleep(2)   # let retained state arrive


def publish_command(name, args):
    device = args.get("valve_id") or args.get("pump_id")
    cmd = {"id": str(uuid.uuid4()), "action": name, "args": args, "ts": int(time.time() * 1000)}
    props = Properties(PacketTypes.PUBLISH)
    props.ResponseTopic = ACK_TOPIC
    props.CorrelationData = cmd["id"].encode()
    props.MessageExpiryInterval = 15   # never execute a stale command
    client.publish(f"{PREFIX}/{device}/cmd", json.dumps(cmd), qos=1, properties=props)
    deadline = time.time() + 5
    while time.time() < deadline:
        if cmd["id"] in acks:
            return acks.pop(cmd["id"])
        time.sleep(0.1)
    return {"id": cmd["id"], "result": "no_ack"}


def run(request):
    messages = [
        {"role": "system", "content": "You operate plant equipment only through the provided functions. "
                                      "If a request cannot be done safely with them, explain why."},
        {"role": "user", "content": request},
    ]
    for attempt in range(MAX_REPAIRS + 1):
        resp = ollama.chat(model=MODEL, messages=messages, tools=as_tools(), options={"temperature": 0})
        calls = resp.message.tool_calls or []
        if not calls:
            return resp.message.content
        errors, results = [], []
        for call in calls:
            name, args = call.function.name, dict(call.function.arguments)
            try:
                check(name, args, live_state)
            except ValidationFailed as err:
                errors.append(str(err))
                continue
            results.append((name, args, publish_command(name, args)))
        if not errors:
            return results
        messages.append(resp.message)
        messages.append({"role": "user", "content": "Some function calls were rejected:\n- " + "\n- ".join(errors)
                         + "\nCorrect them or explain why the request cannot be done."})
    return {"rejected": errors}


if __name__ == "__main__":
    print(run("Open valve V-12 to 90 percent, the tank is filling too slowly."))

Notice what happens with that sample request. If V-12 is currently at 40 percent, the semantic check rejects the 50-point jump and tells the model to propose an intermediate value. The model typically retries with 65, which passes and is published. Any valid calls in a batch go out immediately, and only the rejected ones are sent back for repair.

Strict Modes and Constrained Decoding

Several model providers and local runtimes can constrain generation so that tool arguments always match the declared JSON schema. When available, turn it on: it removes most type errors and missing fields before they reach your validator. It does not remove the need for validation, for three reasons. First, constrained decoding only covers the subset of JSON Schema the runtime supports, and keywords such as numeric ranges or string patterns may be ignored. Second, a perfectly formed call can still be wrong for the current situation, such as a valve step that is too large or a pump on critical duty. Third, you may switch models or providers later, and your safety must not depend on a feature of one runtime. Treat constrained output as a quality improvement and code validation as the guarantee.

Step 4: Validate Again on the Device Side

The command service is not the last line of defense. The gateway or device that receives .../cmd should validate the same schema, check the message has not expired and reply with an acknowledgement on the response topic:

# valve_gateway.py (simulated device side)
import json
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties
from jsonschema import Draft202012Validator
from schemas import DEVICE_ACTIONS

PREFIX = "connectmq/demo/d07c5a/plant"
state = {"V-12": {"position_pct": 40}, "P-1": {"state": "running", "duty": "normal"}}


def on_connect(client, userdata, flags, reason_code, properties):
    for dev, s in state.items():
        client.publish(f"{PREFIX}/{dev}/state", json.dumps(s), qos=1, retain=True)
    client.subscribe(f"{PREFIX}/+/cmd", qos=1)


def on_message(client, userdata, msg):
    cmd = json.loads(msg.payload)
    spec = DEVICE_ACTIONS.get(cmd.get("action"))
    ok = spec is not None and not list(Draft202012Validator(spec["parameters"]).iter_errors(cmd.get("args", {})))
    device = msg.topic.split("/")[-2]
    if ok and cmd["action"] == "set_valve":
        state[device]["position_pct"] = cmd["args"]["position_pct"]
    elif ok and cmd["action"] == "set_pump_state":
        state[device]["state"] = cmd["args"]["state"]
    if ok:
        client.publish(f"{PREFIX}/{device}/state", json.dumps(state[device]), qos=1, retain=True)
    reply = Properties(PacketTypes.PUBLISH)
    if hasattr(msg.properties, "CorrelationData"):
        reply.CorrelationData = msg.properties.CorrelationData
    ack = {"id": cmd.get("id"), "result": "applied" if ok else "rejected_invalid"}
    client.publish(msg.properties.ResponseTopic, json.dumps(ack), qos=1, properties=reply)


c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="valve-gw-d07c5a", protocol=mqtt.MQTTv5)
c.on_connect = on_connect
c.on_message = on_message
c.connect("broker.connectmq.com", 1883, keepalive=60)
c.loop_forever()

Install the dependencies with pip install ollama jsonschema "paho-mqtt>=2.0", start the gateway, then the command service.

Testing Function Definitions Before Deployment

Function definitions deserve the same testing discipline as any other interface to physical equipment. Build a test suite in two layers:

1. Validator unit tests without a model. Feed hand-written argument sets, valid and invalid, into check() and assert the outcome. Include boundary values, wrong types, unknown IDs, extra fields and stale or missing live state.
2. Model behavior tests. Keep a list of realistic requests with the expected function and arguments, such as "close V-11 completely" or "stop the main pump". Run them against the model with temperature 0 and record how often the first attempt is valid, how often the repair loop fixes it and how often it ends in a rejection.

Rerun the behavior tests whenever you change a description, add a function or switch models. Small wording changes in a description can shift which function a model picks, and you want to discover that in testing rather than on the plant floor. Logging the rejected calls from production gives you new test cases over time.

Checklist for Production LLM Function Calling in IoT

1. Schemas are versioned and shared between the command service and devices.
2. additionalProperties: false, enums for IDs and bounds for every number.
3. Structural validation with a JSON Schema library, then semantic checks against live state.
4. A bounded repair loop with clear, model-readable error messages.
5. Commands carry an ID, expiry and response topic; success is reported only after acknowledgement.
6. High-impact functions require human approval; safety functions are never exposed.
7. Every call, rejection and acknowledgement is logged.

The demo uses the public MQTT test broker. For real equipment, switch to port 8883 with TLS and the username, password and client ID generated for your broker; a managed MQTT broker like ConnectMQ handles certificates, MQTT 5 support and connection quotas for you. For the overall agent design see LLM agents for IoT devices, for a friendlier front end see natural language device control, and for broader context our AI and IoT guide. The JSON Schema specification documents every keyword used above.

Get Started with ConnectMQ

ConnectMQ supports the MQTT 5 response topics, correlation data and message expiry this pattern relies on. Sign up for the free Serverless plan, no credit card required, and run your first validated function call against a real broker today.

Frequently Asked Questions

Is the JSON schema I send to the model enough to keep commands safe?

No. It guides the model but does not guarantee compliant output. Always validate the arguments in your own code, and again on the device side.

What should happen when the model's function call is invalid?

Return a clear error message to the model and allow one or two repair attempts. If it still fails, report the rejection to the user instead of guessing.

Why use MQTT 5 for commands?

Response topics and correlation data give a standard way to match acknowledgements to commands, and message expiry prevents devices from executing stale commands.

Can one request produce several function calls?

Yes. Models can return multiple tool calls in one response. Validate each one independently and publish only those that pass.