SDK Reference

Overview

The robotops SDK exposes a single public class — RobotClient — that manages the full lifecycle of a robot's connection to the platform. Under the hood it composes three internal pieces:

• ConnectionManager — paho-MQTT lifecycle, LWT, exponential back-off reconnect, thread-safe publish queue.
• HeartbeatLoop — background thread that pings the platform on a configurable interval.
• Broker config resolution — on start() the SDK calls POST /auth/broker-config on the backend to retrieve real MQTT credentials. No broker host or password ever appears in robot code.

Installation

Requires Python 3.9+ and a single runtime dependency (paho-mqtt).

# Install from PyPI
pip install robotops

# Or install in development mode from source
git clone https://github.com/element-robotics/robotops-sdk
pip install -e ".[dev]"

Quick start

Five lines to connect a robot and start streaming data to your fleet dashboard.

from robotops import RobotClient

robot = RobotClient(
    robot_id="arm-042",
    org_id="acme-corp",
    api_key="sk-••••••••••••••••",
    backend_url="https://api.elementrobotics.xyz",
)

robot.start()
robot.report_status("online")

# Your robot's work loop
while running:
    robot.send_telemetry("temp_c", value=read_temp(), unit="C")
    time.sleep(5)

robot.stop()   # publishes "offline", disconnects cleanly

RobotClient

start()

Resolves MQTT broker credentials from the backend, connects to the broker, and starts the heartbeat background thread. Must be called before any publish method.

Calling start() a second time is a no-op (logged as a warning).

# Block until connected (default)
robot.start()

# Non-blocking — connect in background, messages will queue
robot.start(wait_for_connection=False)

# Extend timeout for slow networks
robot.start(connection_timeout_s=30.0)

stop()

Publishes an "offline" status message, stops the heartbeat thread, and disconnects cleanly from the broker. Because the status is published before disconnecting, the dashboard reflects the robot's offline state immediately rather than waiting for the LWT timeout.

import signal

def shutdown(sig, frame):
    robot.stop()

signal.signal(signal.SIGINT, shutdown)
signal.signal(signal.SIGTERM, shutdown)

report_status()

Publishes the robot's current operational status to the platform. The message is sent with QoS 1 and retain=True so any subscriber (including a freshly loaded dashboard) always receives the latest known state immediately.

See Status values for the full list and their meanings.

robot.report_status("online")   # robot is active and healthy
robot.report_status("idle")     # connected but not doing work
robot.report_status("fault")    # use report_fault() instead — it sets this automatically
robot.report_status("offline")  # use stop() instead — it sets this and disconnects cleanly

report_fault()

Reports a discrete fault event to the platform. Each call generates a UUID fault_id that the backend uses to deduplicate QoS 1 redeliveries — the same fault is never recorded twice even if the network retries the publish.

Calling this method automatically transitions the robot's status to "fault", so operators see the alert on the dashboard immediately without a separate report_status("fault") call.

# Basic fault
fault_id = robot.report_fault("LOW_BATTERY", severity="warning")

# With contextual data
fault_id = robot.report_fault(
    "MOTOR_OVERHEAT",
    severity="critical",
    data={"motor": 3, "temp_c": 94.2, "load_pct": 87},
)
print(f"Fault recorded: {fault_id}")

send_telemetry()

Streams a sensor reading to the platform. Published with QoS 0 (fire-and-forget) — suitable for high-frequency sensors where occasional loss is acceptable.

Three payload shapes are supported and can be combined. At least one of value, lat/lng, or data must be provided.

Telemetry shapes

Numeric metric — temperature, battery %, motor current, RPM, etc.

robot.send_telemetry("temp_c",       value=87.3,   unit="C")
robot.send_telemetry("battery_pct",  value=42.0,   unit="%")
robot.send_telemetry("motor_rpm",    value=1450.0, unit="rpm")
robot.send_telemetry("current_a",    value=4.2,    unit="A")

GPS / location — latitude, longitude, and optional heading.

robot.send_telemetry(
    "location",
    lat=37.7749, lng=-122.4194, heading=45.0,
)

Structured JSON — free-form dict for complex or multi-field readings.

robot.send_telemetry(
    "motor_state",
    data={"motor": 3, "rpm": 1200, "current_a": 4.2, "temp_c": 71.5},
)

Combined — shapes can be mixed in a single call.

robot.send_telemetry(
    "motor_temp",
    value=87.3, unit="C",
    data={"motor_id": 3, "warning_threshold": 85.0},
)

Properties

robot.is_connected → bool

True if the MQTT connection is currently established. Safe to poll from any thread.

robot.status → str | None

The last status string passed to report_status(), or None if report_status() has never been called. This reflects the value most recently sent by the client, not necessarily what the server has stored.

if robot.is_connected:
    robot.send_telemetry("rpm", value=1200)

print(f"Last reported status: {robot.status}")

Status values

Severity levels

MQTT topic schema

All topics are prefixed with robotops/<org_id>/<robot_id>/. The org slug is embedded in every topic to enable broker-level ACL tenant isolation.

Error handling

try:
    robot.start()
except ConnectionError as e:
    print(f"Backend unreachable or bad API key: {e}")
    sys.exit(1)
except TimeoutError:
    print("Broker did not respond in time — check network")
    sys.exit(1)

Logging

The SDK uses Python's standard logging module under the logger name robotops. By default no handler is attached, so output is suppressed. To enable SDK logs in your application:

import logging

# Enable all robotops logs at DEBUG level
logging.basicConfig(level=logging.DEBUG)

# Or scope it to the SDK only
logging.getLogger("robotops").setLevel(logging.INFO)

Key log events and their levels: