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
pyproject.toml / README.md	Check the Python version, dependencies, and run commands

For .env, runtime-model.json, and device-spec.json, follow the shared project file and model rules.

Projects for the current adapter

The examples below use an SDK project for devices managed directly by Device Agent. Other adapters may use different callbacks; follow the package README. Where writable state is supported, implement its update callback as well as command handling, and report actual values after applying them to hardware.

Install and Start

1. Download and extract the package, then open its root directory.
2. Check the generated .env. Copy .env.example only when it is absent, then complete 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 generated project requires Python 3.11 or later. 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:

test -f .env || cp .env.example .env
uv run device-agent-toolkit

You can also run the entry file directly:

uv run python src/main.py

uv installs the runtime dependencies from pyproject.toml, including paho-mqtt, python-dotenv, and websockets.

Implement Commands, State, and Events

The example project calls apply_command_to_state() after validation to prepare state after an action. Replace example mappings with hardware calls and actual readings while preserving validation, response correlation, and reporting. On failure, return an error from the command entry rather than writing simulated success values. If another adapter offers callbacks, register action and writable-state handling as described in its package.

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

Call publish_event() from a business callback inside main(), after MQTT connects. This example uses the over_temperature event from Define a Device Agent:

publish_event("over_temperature", {"current_temperature": 32.5})

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:

stop_listening() only ends audio capture. Keep the event loop running until the reply completes, an error occurs, or the connection closes, then clean up the connection:

import asyncio
import os

from voice_client import VoiceClient

async def main(pcm_chunk: bytes):
    voice = VoiceClient(
        ws_url=os.environ["VOICE_WS_URL"],
        device_id=os.environ["DEVICE_ID"],
        product_id=os.environ["PRODUCT_ID"],
    )
    finished = asyncio.Event()
    voice.on("tts_complete", lambda _task_id: finished.set())
    voice.on("error", lambda _error: finished.set())
    voice.on("disconnected", lambda _reason: finished.set())
    try:
        await voice.connect()
        await voice.start_listening("manual")
        await voice.send_audio(pcm_chunk)
        await voice.stop_listening()
        await finished.wait()
    finally:
        await voice.disconnect()

asyncio.run(main(pcm_chunk))

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.