Skip to content
Stand with Ukraine flag

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.

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

Install using circup:

Terminal window
circup install thingsboard-circuitpython-client-sdk

Alternative installation via Web-Workflow:

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

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)
# Default connecting
client.connect()
# Connecting and waiting for the connection result
response = client.connect()
client.connect()
# some tasks with ThingsBoard
client.disconnect()
# 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)
attributes = {"sensorModel": "DHT-22", "attribute_2": "value"}
client.send_attributes(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)
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)

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)
# 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)

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

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

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

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