MQTT 接入

MQTT 接入用于把已有固件、网关或后端服务连接到所选设备记录，使设备智能体可以控制和查询它。设备端不需要使用设备端 SDK，也不需要重写原有驱动或业务逻辑，只需要增加一层 MQTT 适配：订阅命令、调用原有设备能力、返回执行结果，并按当前原生模型上报状态和事件。

确认设备协议

MQTT 接入使用当前 IoT 平台适配提供的主题、消息体和设备认证方式。已经在原平台运行的设备继续使用原有链路；完成IoT 平台集成后即可让 Agent 查询和控制，无需更换设备协议。

需要自行适配设备端时，从所选设备的 已有设备 入口获取协议说明。命令执行、期望状态更新、事件上报和回执关联按当前接入方式实现。物模型决定字段和约束，平台协议决定消息如何发送。

下面的固定主题、JSON 结构和 MQTTX 示例适用于由 Device Agent 直接管理的设备。经 IoT 平台接入的设备使用控制台或生成包显示的协议。

本页示例使用DeviceSpec 模型示例：current_temperature、target_temperature 两个属性，set_target_temperature(value) 命令，以及仅含 current_temperature 字段的 over_temperature 事件。适配自己的设备时，替换为实际模型已声明的字段。

获取接入信息

进入设备智能体工作区，创建或选择一条设备记录，点击该记录的 下载接入包，再选择 已有设备。控制台会给出所选记录的接入信息：

productId：当前设备智能体标识。
namespace：当前运行空间，默认是 default。
deviceId：所选设备记录的唯一标识，MQTT 客户端必须使用同一值。
MQTT 服务地址、用户名和密码状态。
命令、响应、遥测、事件和时间同步主题。
命令、响应、上线状态、状态遥测和设备事件的消息体示例。
当前设备规格中的命令、属性和事件字段。

如果运行配置中修改过 MQTT 主题模板，以控制台显示的主题为准。

选择适配位置

适配层放在哪里，取决于哪里最容易读取状态并执行动作：

设备固件：设备本身支持 MQTT，直接在固件中订阅命令并上报状态。
边缘网关：网关通过 Modbus、BLE、串口或私有协议连接设备，再转换为 MQTT。
后端服务：已有云端或本地服务可以控制设备，新增一个 MQTT 适配进程，把命令转成原有 API 调用。

同一种接入方式下，适配层可以部署在这些不同位置；协议仍以当前接入方式为准。

改造步骤

先把设备规格和已有设备能力对齐：

设备规格	映射到已有设备	MQTT 动作
命令	驱动调用、HTTP API、串口指令或业务函数	订阅命令主题，执行后发布响应
属性	当前状态、传感器读数、运行模式	发布上线状态和 `state` 遥测
事件	告警、异常、任务完成、按键回调	发布 `event` 事件

最小适配逻辑如下：

text

启动:
  连接 MQTT
  订阅命令主题
  发布上线状态和当前状态快照

收到命令:
  根据 cmd 和 params 调用原有设备能力
  发布命令响应，requestId 保持一致
  如果状态变化，发布最新状态快照

设备状态变化:
  发布 state 遥测

设备发生告警、异常或任务完成:
  发布 event 事件

设备智能体会从主题路径解析 productId 和 deviceId。消息体里的 metadata.productId 也建议保留，便于绑定、调试和自定义主题模板下的路由。如果需要强绑定 MQTT Client ID 和设备 ID，应在 MQTT 服务侧配置认证或 ACL。

主题

默认主题如下。实际接入时复制控制台中的主题。

方向	用途	默认主题
对话端发布	产品级对话请求	`device-agent/{productId}/in`
对话端订阅	产品级对话回复	`device-agent/{productId}/out`
对话端发布	指定设备的对话请求	`device-agent/{productId}/device/{deviceId}/in`
对话端订阅	指定设备的对话回复	`device-agent/{productId}/device/{deviceId}/out`
设备订阅	接收命令	`device-agent/{productId}/device/{deviceId}/commands`
设备发布	返回命令响应	`device-agent/{productId}/device/{deviceId}/responses`
设备发布	上报在线状态和遥测	`v1/{productId}/{deviceId}/telemetry`
设备发布	上报原始设备事件	`v1/{productId}/{deviceId}/event`
Gateway 发布	发布归一化实时事件	`v1/{productId}/{deviceId}/event`
设备发布	发起时间同步请求	`device-agent/{productId}/device/{deviceId}/ntp/request`
设备订阅	接收时间同步响应	`device-agent/{productId}/device/{deviceId}/ntp/response`

普通 Gateway MQTT 发布和订阅当前使用 QoS 0、retain: false。生成的设备端 SDK 使用 QoS 1，但实际交付等级仍取决于发布端、Broker 和 Gateway 订阅，不应依赖端到端 QoS 1。

消息体规则

所有消息体都是 UTF-8 JSON 对象。

字段	说明
`ts`	设备端生成的 Unix 毫秒时间戳。平台部署中的设备上报必须包含 `ts` 或 `eventId`；重试同一条上报时应复用原值。
`eventId`	无法提供时间戳时可使用的稳定设备端事件 ID。重试同一条上报时应复用原值。
`metadata`	可选对象。建议至少包含 `productId`，例如 `{ "productId": "thermostat", "source": "existing-device" }`。
`data`	上报数据对象。状态和属性字段必须来自当前设备规格。

平台部署会在更新设备状态或运行 Workflow 前，拒绝早于 MQTT 去重窗口的设备端 ts。生产窗口至少为 48 小时，并且至少覆盖 MQTT session expiry 再加 24 小时。仅使用 eventId 的消息（例如离线遗嘱）会从平台接收时开始使用同一个有界去重窗口。新上报使用新的 ts 或 eventId，重试同一条消息保留原值。下面 JSON 中的时间戳仅用于展示结构，发送时替换为当前 Unix 毫秒时间；不要把示例中的旧时间用于实际接入。

当前校验规则如下：

命令名和参数必须匹配设备规格中的命令定义。
type: "state" 遥测按设备规格严格校验；字段未知或类型错误时不会更新当前状态。
type: "status" 中的 data.state 当前不会按设备规格严格校验。只上报已定义且类型正确的属性，避免把无效字段写入当前状态。
type: "event" 的事件名和字段按设备规格严格校验；失败时会发布 validation_error 归一化事件。

上线状态

设备连接 MQTT 后，先向遥测主题发布上线状态。首次成功上报后，同一条设备记录会显示为在线，并更新当前数据。

json

{
  "type": "status",
  "data": {
    "status": "online",
    "state": {
      "current_temperature": 26.5,
      "target_temperature": 24
    }
  },
  "ts": 1710000000000,
  "metadata": {
    "productId": "thermostat",
    "source": "existing-device"
  }
}

status 使用 online、offline 或 error。state 建议上报完整状态快照。设备断开前可以主动上报 offline；如果支持 MQTT Will Message，也可以把离线状态配置为遗嘱消息。每次新的 MQTT 连接周期都应为遗嘱生成新的 eventId，不带 ts，避免连接时设置的时间戳在断开时已过期；同一条遗嘱被 Broker 重投时则必须保留原身份。

状态遥测

设备运行中向遥测主题上报状态变化。常用 type 是 state，data 中只放设备规格中定义过的属性字段。

json

{
  "type": "state",
  "data": {
    "current_temperature": 27.1,
    "target_temperature": 24
  },
  "ts": 1710000005000,
  "metadata": {
    "productId": "thermostat",
    "source": "existing-device"
  }
}

建议在启动后、命令执行后和关键状态变化后上报完整快照。只上报部分字段时，控制台只能更新这些字段。

命令响应

当用户通过对话控制设备时，设备智能体会向命令主题发布命令。

json

{
  "cmd": "set_target_temperature",
  "params": {
    "value": 24
  },
  "requestId": "req-001",
  "ts": 1710000010000
}

字段	类型	必填	说明
`cmd`	string	是	设备规格中定义的命令名。
`params`	object	否	命令参数，必须匹配该命令的参数定义。无参数时使用 `{}` 或省略。
`requestId`	string	是	本次命令请求 ID。
`ts`	number	否	Unix 毫秒时间戳。

本例使用定义设备智能体中的 set_target_temperature(value)。参数名必须与模型一致；上报属性 target_temperature 是另一个字段。

设备执行后向响应主题发布结果。requestId 必须和收到的命令一致。

json

{
  "code": 0,
  "msg": "ok",
  "requestId": "req-001",
  "data": {
    "target_temperature": 24
  },
  "ts": 1710000011000,
  "metadata": {
    "productId": "thermostat",
    "source": "existing-device"
  }
}

字段	类型	必填	说明
`code`	number	是	`0` 表示成功，非 `0` 表示失败。
`msg`	string	是	执行结果说明。
`requestId`	string	是	必须等于命令中的 `requestId`。
`data`	any	否	返回给设备智能体的执行结果。
`ts`	number	否	Unix 毫秒时间戳。

如果命令改变了设备状态，响应后立即再上报一次最新 status 或 state，用户才能在状态面板中看到执行结果。

设备事件

事件表示设备主动发生的一次性变化，例如告警、异常、阈值触发、任务完成或用户按键。设备向事件主题发布 type: "event" 的消息体。

json

{
  "type": "event",
  "data": {
    "event": "over_temperature",
    "current_temperature": 38.5
  },
  "ts": 1710000020000,
  "metadata": {
    "productId": "thermostat",
    "source": "existing-device"
  }
}

data.event 必须是设备规格中定义过的事件名，其余字段只能使用该事件定义的输出字段。事件不会自动写入当前数据，但会出现在最近上报事件中。阈值类事件建议做边沿触发或冷却时间，避免重复上报。

Gateway 处理设备上报后，会把归一化实时事件发布到同一个事件主题。设备上行消息使用 type 和 data；归一化消息使用 eventId、kind、namespace、deviceId、occurredAt 和 summary。订阅该主题时应按消息结构区分方向，不要把归一化消息再次作为设备事件上报。

通过 MQTT 对话

向产品级输入主题发布请求；需要固定设备时使用带 deviceId 的输入主题。回复从对应输出主题接收。

json

{
  "id": "req-001",
  "sessionId": "terminal-001",
  "prompt": "把目标温度设为 24 度",
  "metadata": {
    "source": "mqtt-client"
  }
}

每个请求使用唯一 id；相同 sessionId 延续上下文。

一次对话会返回多条消息。过程消息的顶层 text 为空，进度位于 metadata.status：

`metadata.status.type`	关键字段
`start`、`thinking`	当前执行阶段。
`text_block`	增量文本位于 `metadata.status.text`。
`tool_start`、`tool_end`	工具名、输入、输出或错误。
`complete`、`error`	当前执行结果；完成文本位于 `metadata.status.text`。

状态消息的 metadata.replyTo 对应请求 id。Gateway 还会发送顶层 text 最终回复：

json

{
  "sessionId": "terminal-001",
  "text": "目标温度已设为 24 度。",
  "metadata": {
    "timestamp": "2026-05-11T10:00:00.000Z"
  },
  "timestamp": "2026-05-11T10:00:00.000Z"
}

客户端应选择增量状态或最终回复，不要同时追加两者。最终回复当前不保证携带 replyTo。同一 sessionId 应在本轮完成后再发送下一条请求；并行对话使用不同的 sessionId。

中断或清空会话时，向原输入主题发送相同 sessionId、空 prompt，并分别设置 metadata.interrupt: true 或 metadata.clearSession: true。清空操作会先归档已保存的对话，再重置上下文；后续请求可以继续使用相同的 sessionId。归档失败或仍有回复在执行时，清空操作会返回错误且不会重置对话。成功时通过 metadata.status.type: complete 返回，失败时返回 error；metadata.replyTo 对应请求 id。

用 MQTTX 快速验证

在改造真实设备前，可以先用 MQTTX CLI 模拟一台设备。下面示例假设：

MQTT 服务：127.0.0.1:1883
productId：thermostat
deviceId：thermostat-001

如果 MQTT 服务没有启用用户名和密码，去掉命令中的 -u 和 -P。

先打开一个终端订阅命令主题，用来观察设备智能体下发的命令：

每次新的测试上报都替换示例 eventId；重试同一条消息则保留它。示例使用本页所列的 MQTT 协议，并假设模型已经声明相同字段。

bash

mqttx sub \
  -h 127.0.0.1 \
  -p 1883 \
  -u your-username \
  -P 'your-password' \
  -q 0 \
  -t 'device-agent/thermostat/device/thermostat-001/commands'

再用另一个终端模拟设备上线：

bash

mqttx pub \
  -h 127.0.0.1 \
  -p 1883 \
  -u your-username \
  -P 'your-password' \
  -q 0 \
  -t 'v1/thermostat/thermostat-001/telemetry' \
  -m '{"type":"status","data":{"status":"online","state":{"current_temperature":26.5,"target_temperature":24}},"metadata":{"productId":"thermostat","source":"mqttx"},"eventId":"mqttx-online-001"}'

已有的 thermostat-001 记录显示为在线后，可以继续模拟状态变化：

bash

mqttx pub \
  -h 127.0.0.1 \
  -p 1883 \
  -u your-username \
  -P 'your-password' \
  -q 0 \
  -t 'v1/thermostat/thermostat-001/telemetry' \
  -m '{"type":"state","data":{"current_temperature":27.1,"target_temperature":24},"metadata":{"productId":"thermostat","source":"mqttx"},"eventId":"mqttx-state-001"}'

也可以模拟设备主动上报事件：

bash

mqttx pub \
  -h 127.0.0.1 \
  -p 1883 \
  -u your-username \
  -P 'your-password' \
  -q 0 \
  -t 'v1/thermostat/thermostat-001/event' \
  -m '{"type":"event","data":{"event":"over_temperature","current_temperature":38.5},"metadata":{"productId":"thermostat","source":"mqttx"},"eventId":"mqttx-event-001"}'

最后在对话中让设备智能体控制这台设备。订阅命令的终端会收到一条带 requestId 的命令，把这个 requestId 复制到响应消息中：

bash

mqttx pub \
  -h 127.0.0.1 \
  -p 1883 \
  -u your-username \
  -P 'your-password' \
  -q 0 \
  -t 'device-agent/thermostat/device/thermostat-001/responses' \
  -m '{"code":0,"msg":"ok","requestId":"替换为收到的 requestId","data":{"target_temperature":24},"metadata":{"productId":"thermostat","source":"mqttx"},"eventId":"mqttx-response-001"}'

这组命令可以验证完整链路：设备上线、状态上报、事件上报、命令接收和命令响应。真实设备接入时，把这些 MQTTX 命令替换为固件、网关或后端服务中的 MQTT 发布订阅逻辑即可。

时间同步

如果设备需要估算本地时间与网关服务的时间差，可以使用时间同步主题。

设备向请求主题发布：

json

{
  "deviceSendTime": 1710000030000
}

设备从响应主题收到：

json

{
  "deviceSendTime": 1710000030000,
  "serverRecvTime": 1710000030120,
  "serverSendTime": 1710000030122
}

deviceSendTime 是设备发送请求时的本地 Unix 毫秒时间戳。serverRecvTime 和 serverSendTime 分别是网关服务收到请求和发出响应的时间。

验证接入

完成设备端适配后，按下面顺序检查：

设备已连接 MQTT，命令主题已订阅成功。
遥测主题收到第一条 status 或 state 后，同一条设备记录显示为在线。
当前数据能看到设备规格中定义的属性字段。
对话中下发命令后，设备端能收到命令，并向响应主题返回同一个 requestId。
命令改变状态后，设备端有新的状态或遥测上报。
主动上报事件后，最近上报事件中能看到记录。

如果记录没有上线，先检查 MQTT 服务地址和认证信息，再检查 productId、deviceId、主题路径和 metadata.productId。如果记录已在线但当前数据不更新，优先检查属性字段名和类型是否与设备规格一致。运行配置见配置。

下一步

阅读 SDK 接入，使用生成的设备端工程接入真实硬件。
阅读设备模拟器，在没有真实硬件时验证设备规格。
阅读使用设备智能体，了解如何选择设备并通过对话控制设备。

MQTT 接入 ​

确认设备协议 ​

获取接入信息 ​

选择适配位置 ​

改造步骤 ​

主题 ​

消息体规则 ​

上线状态 ​

状态遥测 ​

命令响应 ​

设备事件 ​

通过 MQTT 对话 ​

用 MQTTX 快速验证 ​

时间同步 ​

验证接入 ​

下一步 ​

MQTT 接入

确认设备协议

获取接入信息

选择适配位置

改造步骤

主题

消息体规则

上线状态

状态遥测

命令响应

设备事件

通过 MQTT 对话

用 MQTTX 快速验证

时间同步

验证接入

下一步