Python SDK

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

先看文件	用途
src/main.py	编写命令处理、状态更新和事件逻辑
device-spec.json	核对命令、属性和事件定义
.env.example	配置 MQTT 地址和设备身份
pyproject.toml / README.md	查看 Python 版本、依赖和运行方式

安装和启动

1. 下载并解压代码包，进入代码包根目录。
2. 将 .env.example 复制为 .env，填写连接信息。
3. 使用 uv 安装 pyproject.toml 中的依赖。
4. 在 src/main.py 中接入真实设备或业务服务。
5. 启动程序，回到设备智能体工作区验证。

Python 入口只读取 SDK 包目录下的 .env，不会读取父级工作区的 .env；启动前已经导出的环境变量优先。最小启动命令如下：

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

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

uv run python src/main.py

命令名、参数名、属性字段和事件名必须与 device-spec.json 保持一致。uv 会根据 pyproject.toml 安装运行依赖，包括 paho-mqtt、python-dotenv 和 websockets。

实现命令、状态和事件

默认代码会根据 device-spec.json 校验命令和参数，再调用 apply_command_to_state()。在这个函数中执行真实动作并返回最新状态：

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

三类数据的用途不同：

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

事件名和数据字段应已在 device-spec.json 中定义：

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

接入语音和视觉

语音

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

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

视觉

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 接入 中的检查项验证设备上线、状态、命令和事件。