MicroPython Client

Overview

The MicroPython client (lua_device.py) runs on microcontrollers with as little as 264KB of RAM. It uses MQTT natively and supports the full device protocol: commands, triggers, heartbeats, reconnection, and idempotency dedup.

Hardware Requirements

Component	Minimum	Recommended
Board	Raspberry Pi Pico W	Any MicroPython board with WiFi
RAM	264 KB	512 KB+
Flash	2 MB	2 MB
MicroPython	v1.20+	Latest stable
Network	WiFi (2.4 GHz)	WiFi with stable connection

The client depends on umqtt.robust (preferred) or umqtt.simple, both included in standard MicroPython firmware for the Pico W.

Setup

Flash MicroPython firmware

Download the latest MicroPython UF2 from micropython.org and flash it to your Pico W.

Copy the client library

Copy lua_device.py to your Pico W. You can use Thonny, mpremote, or rshell:

mpremote cp lua_device.py :lua_device.py

Connect to WiFi

Create a main.py that connects to WiFi before creating the device:

import network
import time

wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect("YOUR_SSID", "YOUR_PASSWORD")

while not wlan.isconnected():
    time.sleep(0.5)

print("WiFi connected:", wlan.ifconfig()[0])

Create your device

Add the device setup after WiFi connection:

from lua_device import LuaDevice

device = LuaDevice(
    agent_id="your-agent-id",
    api_key="your-api-key",
    device_name="pico-sensor",
    server="mqtt.heylua.ai",
)

@device.command("read_sensor")
def read_sensor(payload):
    return {"temperature": 22.5, "humidity": 60}

device.connect()
device.run()  # blocks forever, handles commands

LuaDevice Class Reference

Constructor

LuaDevice(agent_id, api_key, device_name, server, port=443, group=None, use_ssl=True)

Parameter	Type	Default	Description
`agent_id`	str	required	Agent ID to connect to
`api_key`	str	required	API key for authentication
`device_name`	str	required	Unique name for this device
`server`	str	required	MQTT broker hostname (e.g., `mqtt.heylua.ai`)
`port`	int	`443`	MQTT broker port
`group`	str	`None`	Optional device group name
`use_ssl`	bool	`True`	Enable TLS encryption

Methods

Method	Description
`connect()`	Connect to the MQTT broker. Sets up LWT and subscriptions.
`disconnect()`	Gracefully disconnect. Publishes offline status first.
`run(check_interval_ms=100)`	Main loop. Blocks forever, checks for messages and sends heartbeats.
`trigger(name, payload=None)`	Fire a trigger event to the agent.
`on_command(name, handler)`	Register a command handler (non-decorator style).

The `@device.command` Decorator

The preferred way to register command handlers:

@device.command("led_on")
def led_on(payload):
    # payload is a dict with whatever the agent sent
    pin = machine.Pin("LED", machine.Pin.OUT)
    pin.on()
    return {"status": "on"}

The handler receives a payload dict and must return a dict (or None). If the handler raises an exception, the error message is sent back to the agent.

Complete Example: LED + DHT22

A Pico W that controls an onboard LED and reads a DHT22 temperature/humidity sensor:

import network
import machine
import time
import dht

# -- WiFi --
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect("YOUR_SSID", "YOUR_PASSWORD")

while not wlan.isconnected():
    time.sleep(0.5)

print("WiFi connected:", wlan.ifconfig()[0])

# -- Sensor setup --
led = machine.Pin("LED", machine.Pin.OUT)
dht_sensor = dht.DHT22(machine.Pin(15))

# -- Device --
from lua_device import LuaDevice

device = LuaDevice(
    agent_id="your-agent-id",
    api_key="your-api-key",
    device_name="pico-env-sensor",
    server="mqtt.heylua.ai",
    group="office-sensors",
)

@device.command("led_on")
def led_on(payload):
    led.on()
    return {"status": "on"}

@device.command("led_off")
def led_off(payload):
    led.off()
    return {"status": "off"}

@device.command("read_environment")
def read_environment(payload):
    dht_sensor.measure()
    return {
        "temperature": dht_sensor.temperature(),
        "humidity": dht_sensor.humidity(),
        "led": "on" if led.value() else "off",
    }

@device.command("blink")
def blink(payload):
    count = payload.get("count", 3)
    delay = payload.get("delay_ms", 200)
    for _ in range(count):
        led.on()
        time.sleep_ms(delay)
        led.off()
        time.sleep_ms(delay)
    return {"blinked": count}

# -- Connect and run --
device.connect()

# Optional: fire a trigger every 60s if temperature is high
last_check = time.time()

while True:
    try:
        device._client.check_msg()

        now = time.time()
        if now - device._last_heartbeat >= device._heartbeat_interval:
            device._client.publish(device._topic_prefix + "heartbeat", b"", qos=0)
            device._last_heartbeat = now

        # Periodic temperature check
        if now - last_check >= 60:
            dht_sensor.measure()
            temp = dht_sensor.temperature()
            if temp > 30:
                device.trigger("high_temperature", {
                    "temperature": temp,
                    "threshold": 30,
                })
            last_check = now

        time.sleep_ms(100)

    except OSError as e:
        print("Connection lost:", e)
        device._reconnect()
    except Exception as e:
        print("Error:", e)
        time.sleep(1)

Troubleshooting

OSError: [Errno 110] ETIMEDOUT

The Pico W cannot reach the MQTT broker. Check:

WiFi is connected (wlan.isconnected() returns True)
DNS resolution works (try socket.getaddrinfo("mqtt.heylua.ai", 443))
No firewall blocking port 443 (WebSocket)

MQTTException: 5 (Not authorized)

Authentication failed. Verify:

agent_id matches your agent exactly
api_key is valid and not expired
device_name contains only lowercase letters, numbers, and hyphens

MemoryError or device freezes

The Pico W has limited RAM. Try:

Reduce _dedup_ttl (default 300 seconds) if you are processing many commands
Avoid large payloads in command responses
Use gc.collect() periodically in your main loop
Compile lua_device.py to .mpy bytecode with mpy-cross to save RAM

Commands not received

Ensure:

device.connect() completed without error
You are calling device.run() or manually calling device._client.check_msg() in a loop
The command name in your handler matches the name the agent is using

SSL handshake fails

Some MicroPython builds have limited TLS support. Try:

Update to the latest MicroPython firmware
Set use_ssl=False temporarily for debugging (not recommended in production)

Next Steps

Pico W Setup Guide

Step-by-step hardware setup with photos and wiring

Industrial Sensor Example

Complete factory monitoring example on Pico W

MQTT Transport

MQTT topic structure and QoS details

Triggers

Send events from your device to the agent

Getting Started

Device Client

Agent Integration

Examples

Advanced

CLI Commands

MicroPython Client

Overview

Hardware Requirements

Setup

LuaDevice Class Reference

Constructor

Methods

The `@device.command` Decorator

Complete Example: LED + DHT22

Troubleshooting

Next Steps

Pico W Setup Guide

Industrial Sensor Example

MQTT Transport

Triggers

​Overview

​Hardware Requirements

​Setup

​LuaDevice Class Reference

​Constructor

​Methods

​The @device.command Decorator

​Complete Example: LED + DHT22

​Troubleshooting

​Next Steps

Pico W Setup Guide

Industrial Sensor Example

MQTT Transport

Triggers

Overview

Hardware Requirements

Setup

LuaDevice Class Reference

Constructor

Methods

The `@device.command` Decorator

Complete Example: LED + DHT22

Troubleshooting

Next Steps