Python SDK

The Python SDK is for gateway programs, validation scripts, and existing Python services that connect sensors, actuators, or business APIs. The generated package handles MQTT connectivity, command validation, command responses, and data reporting; add device behavior in src/main.py.

Start with	Purpose
src/main.py	Implement command handling, state updates, and event logic
device-spec.json	Check command, property, and event definitions
.env.example	Configure the MQTT endpoint and device identity
pyproject.toml / README.md	Check the Python version, dependencies, and run commands

Install and Start

1. Download and extract the package, then open its root directory.
2. Copy .env.example to .env and fill in the connection settings.
3. Use uv to install the dependencies from pyproject.toml.
4. Connect the real device or business service in src/main.py.
5. Start the program and verify it in the Device Agent workspace.

The Python entry point only loads .env from the SDK package directory; it does not load a parent workspace .env. Environment variables exported before startup take precedence. The minimum start commands are:

cp .env.example .env
uv run device-agent-toolkit

You can also run the entry file directly:

uv run python src/main.py

Keep command names, parameter names, property fields, and event names aligned with device-spec.json. uv installs the runtime dependencies from pyproject.toml, including paho-mqtt, python-dotenv, and websockets.

Implement Commands, State, and Events

The generated code validates commands and parameters against device-spec.json, then calls apply_command_to_state(). Perform the real action and return the latest state from this function:

def apply_command_to_state(device_spec, state, command, params):
    next_state = deepcopy(state)

    if command == "set_temperature":
        target = params["target_temperature"]
        call_thermostat_service(target)
        next_state["target_temperature"] = target

    if "updated_at" in next_state:
        next_state["updated_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())

    return next_state

The three data types serve different purposes:

Type	Purpose	Python code
Current state	Keep the latest device property values	state and apply_command_to_state()
State snapshot	Report online status and the latest properties after connection or a command	publish_state_snapshot()
Event	Report a discrete occurrence such as an alert or button press	Call publish_event() from a business callback inside main()

Event names and data fields must be defined in device-spec.json:

publish_event("temperature_alarm", {
    "current_temperature": 32.5,
    "level": "warning",
})

Add Voice and Vision

Voice

src/voice_client.py provides an asynchronous voice client. Set VOICE_WS_URL yourself and pass it explicitly to VoiceClient, then connect microphone capture, speaker playback, and event callbacks:

import os

from voice_client import VoiceClient

voice = VoiceClient(
    ws_url=os.environ["VOICE_WS_URL"],
    device_id=os.environ["DEVICE_ID"],
    product_id=os.environ["PRODUCT_ID"],
)

await voice.connect()
await voice.start_listening("manual")
await voice.send_audio(pcm_chunk)
await voice.stop_listening()

Vision

src/main.py includes the generated package's preset single-photo recognition flow; see the package README for its trigger commands. Set the service host with VOICE_CHAT_HOST, then read a real camera, screenshot, or image file in capture_local_vision_image():

def capture_local_vision_image():
    return {
        "mimeType": "image/jpeg",
        "imageBase64": read_camera_frame_as_base64(),
        "source": "sdk-camera",
    }

The function returns one image, and the generated code returns the recognition result as the command response.

See Voice Interaction and Camera and Vision for complete media settings. After startup, use the checks in SDK Access to verify online state, commands, state reports, and events.