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
Section titled “Supported features”- MQTT protocol connectivity
- Telemetry data streaming
- Attribute transmission and subscriptions
- Server-side RPC handling
- Client and shared attribute requests
- Device claiming
Installation
Section titled “Installation”Install using circup:
circup install thingsboard-circuitpython-client-sdkAlternative installation via Web-Workflow:
circup --host <your_device_ip> --password <your_password> install thingsboard-circuitpython-client-sdkGetting started
Section titled “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 = 1883TOKEN = "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
Section titled “API reference”connect / disconnect
Section titled “connect / disconnect”# Default connectingclient.connect()
# Connecting and waiting for the connection resultresponse = client.connect()client.connect()
# some tasks with ThingsBoard
client.disconnect()send_telemetry
Section titled “send_telemetry”# Sending telemetry data in dictionary formattelemetry = {"temperature": 25.5, "humidity": 60}client.send_telemetry(telemetry)
# Sending telemetry data grouped by timestampsfrom time import timetelemetry = [{"ts": 1451649600000, "values": {"temperature": 42.2, "humidity": 71}}, {"ts": 1451649601000, "values": {"temperature": 42.3, "humidity": 72}}]client.send_telemetry(telemetry)send_attributes
Section titled “send_attributes”attributes = {"sensorModel": "DHT-22", "attribute_2": "value"}client.send_attributes(attributes)request_attributes
Section titled “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 processeddeadline = time.monotonic() + TIMEOUTwhile time.monotonic() < deadline: client.check_for_msg() time.sleep(0.05)subscribe_to_attribute / unsubscribe
Section titled “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 processeddeadline = time.monotonic() + TIMEOUTwhile 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
Section titled “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
Section titled “claim_device”# Claiming a device with a claim code that will be valid indefinitelyclient.claim_device("my_claim_code")
# Claiming a device with a claim code that will be valid for 60 secondsclient.claim_device("my_claim_code", duration_ms=60000)Troubleshooting
Section titled “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-sdkMore examples
Section titled “More examples”Additional examples are available in the CircuitPython_thingsboard-client-sdk GitHub repository.