CircuitPython Client SDK

The ThingsBoard CircuitPython Client SDK enables CircuitPython devices to connect to ThingsBoard via MQTT. It supports telemetry upload, attribute management, server-side RPC, and device claiming.

Supported features

MQTT protocol connectivity
Telemetry data streaming
Attribute transmission and subscriptions
Server-side RPC handling
Client and shared attribute requests
Device claiming

Installation

Install using circup:

circup install thingsboard-circuitpython-client-sdk

Alternative installation via Web-Workflow:

circup --host <your_device_ip> --password <your_password> install thingsboard-circuitpython-client-sdk

Getting started

Connect to ThingsBoard and run a non-blocking main loop:

import time

import wifi

from tb_device_mqtt import TBDeviceMqttClient

print("WiFi connected:", wifi.radio.connected)
print("IP:", wifi.radio.ipv4_address)

HOST = "127.0.0.1"
PORT = 1883
TOKEN = "YOUR_ACCESS_TOKEN"

client = TBDeviceMqttClient(host=HOST, port=PORT, access_token=TOKEN)
print("Connecting...")
client.connect()
time.sleep(1)

while True:
  client.check_for_msg()
  client.send_telemetry({"CPU": 12.0})
  client.send_attributes({"status": "ok"})
  time.sleep(0.05)

API reference

connect / disconnect

# Default connecting
client.connect()

# Connecting and waiting for the connection result
response = client.connect()

client.connect()

# some tasks with ThingsBoard

client.disconnect()

send_telemetry

# Sending telemetry data in dictionary format
telemetry = {"temperature": 25.5, "humidity": 60}
client.send_telemetry(telemetry)

# Sending telemetry data grouped by timestamps
from time import time
telemetry = [{"ts": 1451649600000, "values": {"temperature": 42.2, "humidity": 71}},
             {"ts": 1451649601000, "values": {"temperature": 42.3, "humidity": 72}}]
client.send_telemetry(telemetry)

send_attributes

attributes = {"sensorModel": "DHT-22", "attribute_2": "value"}
client.send_attributes(attributes)

request_attributes

TIMEOUT = 20

def on_attributes_change(result, exception=None):
    if exception is not None:
        print("Exception:", exception)
    else:
        print("Attributes response:", result)


client.request_attributes(client_keys=["atr1", "atr2"], callback=on_attributes_change)

# Keep looping so incoming MQTT messages are processed
deadline = time.monotonic() + TIMEOUT
while time.monotonic() < deadline:
    client.check_for_msg()
    time.sleep(0.05)

subscribe_to_attribute / unsubscribe

TIMEOUT = 20

def callback(result, *args):
    print("Received data:", result)


sub_id = client.subscribe_to_attribute("frequency", callback)

# Keep looping so incoming MQTT messages are processed
deadline = time.monotonic() + TIMEOUT
while time.monotonic() < deadline:
    client.check_for_msg()
    time.sleep(0.05)

Subscribe to all attributes:

def callback(result, *args):
    print("Received data: %r", result)


sub_id = client.subscribe_to_all_attributes(callback)

Unsubscribe using the returned subscription ID:

def callback(result, *args):
    print("Received data: %r", result)

sub_id = client.subscribe_to_attribute("frequency", callback)
client.unsubscribe_from_attribute(sub_id)

RPC handling

Set a handler for server-side RPC requests:

def on_server_side_rpc_request(request_id, request_body):
    print("[RPC] id:", request_id, "body:", request_body)
    client.send_rpc_reply(request_id, "ok")


client.set_server_side_rpc_request_handler(on_server_side_rpc_request)

while True:
    client.check_for_msg()
    time.sleep(0.1)

claim_device

# Claiming a device with a claim code that will be valid indefinitely
client.claim_device("my_claim_code")

# Claiming a device with a claim code that will be valid for 60 seconds
client.claim_device("my_claim_code", duration_ms=60000)

Troubleshooting

Memory constraints — reduce allocations and consult CircuitPython memory-saving guides.

Installation failures — use the Web-Workflow method as an alternative:

circup --host <your_device_ip> --password <your_password> install thingsboard-circuitpython-client-sdk

More examples

Additional examples are available in the CircuitPython_thingsboard-client-sdk GitHub repository.

Was this helpful?