mcpField
M
mqtt

mqtt

github.com/alphaonedev开发52 次安装
代码仓库

关于 mqtt

供 AI 使用 MQTT 发布/订阅消息的技能，面向 IoT 通信场景。它涵盖遥测发布和设备更新订阅，并说明常见的连接安全与可靠性选项。

主要功能

  1. 01按 QoS 等级发布主题消息
  2. 02异步订阅并处理消息
  3. 03保留消息与通配符主题
  4. 04TLS 和用户名密码认证

适用场景

  1. 01向 IoT 仪表板发送传感器读数
  2. 02监听设备控制命令

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/alphaonedev/openclaw-graph --skill mqtt

安装后在 Claude Code 中输入/mqtt确认已加载。

详细介绍

mqtt

用途

此技能可让 AI 与轻量级发布/订阅协议 MQTT 交互，以实现高效的 IoT 通信。可用于发布遥测数据或订阅设备更新，从而在受限环境中实现实时交互。

适用场景

当场景需要低带宽实时消息传递时，可使用此技能，例如从 IoT 设备发送传感器数据、远程控制设备或监控遥测数据。对于文件传输等高吞吐量需求，请避免使用此技能，因为 HTTP 更适合。

主要功能

  • 向主题发布消息，并可配置服务质量（QoS）级别（0、1 或 2）。
  • 订阅主题并异步处理收到的消息。
  • 支持保留消息、遗嘱消息和通配符订阅（例如，使用 "#" 订阅所有子主题）。
  • 通过 TLS 和用户名/密码等身份验证方式建立安全连接。

使用模式

使用 MQTT 时，首先通过客户端库连接到代理。使用凭据建立连接，然后发布或订阅。例如，在循环中依次连接、订阅、处理消息并断开连接。使用环境变量存储敏感数据，例如用 $MQTT_BROKER_URL 设置代理地址。对于不稳定的网络，务必处理重新连接。

常用命令/API

使用 Paho MQTT 库通过 Python 进行交互。按如下方式导入：

import paho.mqtt.client as mqtt

连接到代理：

  • 命令：client.connect(broker_address, port=1883)
  • 身份验证示例：client.username_pw_set(username='$MQTT_USERNAME', password='$MQTT_PASSWORD')

发布消息：

  • API：client.publish(topic, payload, qos=1)
  • 示例：client.publish('sensors/temp', '22.5', qos=1)

订阅：

  • API：client.subscribe(topic, qos=1)
  • 示例：client.subscribe('commands/device1', qos=1)

配置格式：使用 JSON 文件设置配置，例如 {"broker": "$MQTT_BROKER_URL", "topic": "sensors/temp"}。可通过子进程调用 mosquitto_pub 等 CLI 工具：subprocess.run(['mosquitto_pub', '-h', '$MQTT_BROKER_URL', '-t', 'sensors/temp', '-m', '22.5'])。

集成说明

将 MQTT 封装到 AI Agent 代码中即可完成集成。为密钥设置环境变量，例如通过 export $MQTT_API_KEY 配置基于令牌的身份验证。对于多服务配置，可使用 HiveMQ 或 Mosquitto 等代理；通过 client.connect('$MQTT_BROKER_URL') 连接。请为安全端点启用 TLS（例如，调用 client.tls_set()）。如果与其他技能（例如数据库）结合使用，可在操作成功后发布数据：例如，插入数据库后，运行 client.publish('db/updates', json.dumps(data))。

错误处理

明确处理常见错误。对于连接问题，捕获 ConnectionRefusedError 并使用指数退避重试，例如调用 time.sleep(5)。对于发布失败，检查返回码（例如，如果 client.publish() 返回值大于 0，则记录日志并重试）。订阅错误可能由无效主题引起；请使用 if '/' not in topic: raise ValueError('Invalid topic') 进行验证。使用回调处理消息错误，例如，在 on_message 处理程序中使用 if error: print('Error: ', error)。务必在 finally 块中调用 client.disconnect() 进行清理。

具体使用示例

示例 1：发布传感器数据 要将温度读数发布到 MQTT 主题，首先设置环境变量：export MQTT_BROKER_URL='broker.example.com'。然后，在代码中：

client = mqtt.Client()
client.connect(os.environ['MQTT_BROKER_URL'])
client.publish('sensors/temperature', '25.0', qos=1)
client.disconnect()

这会向代理发送消息，以更新 IoT 仪表盘。

示例 2：订阅设备命令 要订阅命令并作出响应，请使用：

def on_message(client, userdata, message):
    print(f"Received: {message.payload.decode()}")
client = mqtt.Client()
client.on_message = on_message
client.connect(os.environ['MQTT_BROKER_URL'])
client.subscribe('commands/device1', qos=1)
client.loop_forever()

这样客户端便会持续监听命令，以控制 IoT 设备。

图关系

  • 关联的集群：iot
  • 关联的标签：iot、mqtt
  • 关联：可与其他 iot 技能集成，例如用于数据处理或设备管理。

评论

支持 Markdown