Python SDK

Python SDK 适合网关程序、验证脚本和已有 Python 服务，可以直接对接传感器、执行器或业务 API。生成包已经处理 MQTT 连接、命令校验、命令响应和数据上报，主要在 src/main.py 中补充设备逻辑。

先看文件	用途
src/main.py	编写命令处理、状态更新和事件逻辑
pyproject.toml / README.md	查看 Python 版本、依赖和运行方式

.env、runtime-model.json 和 device-spec.json 的用途与修改规则见 SDK 工程文件。

当前接入方式的工程

本页示例用于由 Device Agent 直接管理的设备。不同平台的扩展回调可能不同，使用包内 README 指定的接口。平台支持可写状态时，除命令处理外，还需实现状态更新回调，并在硬件应用成功后上报实际值。

安装和启动

1. 下载并解压代码包，进入代码包根目录。
2. 检查生成的 .env，确认连接地址和设备身份；文件缺失时才从 .env.example 创建并补齐配置。
3. 使用 uv 安装 pyproject.toml 中的依赖。
4. 在 src/main.py 中接入真实设备或业务服务。
5. 启动程序，回到设备智能体工作区验证。

生成工程要求 Python 3.11 或更高版本。Python 入口只读取 SDK 包目录下的 .env，不会读取父级工作区的 .env；启动前已经导出的环境变量优先。最小启动命令如下：

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

也可以直接运行入口文件：

uv run python src/main.py

uv 会根据 pyproject.toml 安装运行依赖，包括 paho-mqtt、python-dotenv 和 websockets。

实现命令、状态和事件

本例工程在命令校验后调用 apply_command_to_state()，用于准备动作后的状态。将其中的示例状态映射替换为硬件调用与实际读数，并保留上层校验、响应关联和上报流程。调用失败时，在命令处理入口返回失败结果，不要只写入一个模拟成功值。其他接入方式提供回调时，按生成包注册动作与可写状态处理。

三类数据的用途不同：

类型	作用	Python 代码
当前状态	保存设备的最新属性值	state 和 apply_command_to_state()
状态快照	连接成功或命令完成后，上报在线状态和最新属性	publish_state_snapshot()
事件	上报告警、按键等独立事件	在 main() 的业务回调中调用 publish_event()

在 MQTT 连接成功后，从 main() 内的业务回调调用 publish_event()。本例使用定义设备智能体中的 over_temperature 事件：

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

接入语音和视觉

语音

src/voice_client.py 提供异步语音客户端。使用前自行设置 VOICE_WS_URL，并将该地址显式传给 VoiceClient；再接入麦克风采集、扬声器播放和事件回调：

stop_listening() 只结束音频采集。保持事件循环运行，等到本轮回复完成、出错或连接断开后再清理连接：

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

视觉

src/main.py 包含生成包预置的单次拍照识别流程，具体触发命令见生成包 README。通过 VOICE_CHAT_HOST 配置服务地址，并在 capture_local_vision_image() 中读取真实摄像头、截图或图像文件：

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

函数返回一张图片；生成代码会把识别结果作为命令响应返回。

完整的语音和视觉配置见 语音交互 和 摄像头与视觉识别。启动后，按 SDK 接入 中的检查项验证设备上线、状态、命令和事件。