MQTT

Name: MQTT
Author: nagarjunr

本地服务

github.com/nagarjunr开发工具06 天前更新

代码仓库

关于 MQTT

MQTT 是通过标准线协议连接 MQTT 5.0 Broker 的 MCP 服务。它支持发布消息、管理缓冲订阅、查看连接状态，并通过保留消息近似发现主题。

主要功能

01发布带 MQTT 5 属性的消息
02启动和管理主题订阅
03缓冲并获取订阅消息
04查看 Broker 连接状态
05通过保留消息近似发现主题

适用场景

01向 MQTT Broker 发布遥测数据
02读取符合主题筛选器的消息
03检查保留消息以发现主题

MCP 相关

该服务向 MCP 客户端提供 Broker 连接、消息发布、订阅、消息获取和主题发现工具。它使用标准 MQTT 5.0 协议，并需要配置 Broker 连接参数。

配置

本地服务的配置在各客户端中通用，复制后粘贴到客户端的 MCP 配置文件即可。

mcp.json · 通用配置

{
  "mcpServers": {
    "nagarjunr-mqtt": {
      "command": "uvx",
      "args": [
        "mqtt5-mcp-server"
      ],
      "env": {
        "MQTT_HOST": "<MQTT_HOST>",
        "MQTT_PORT": "<MQTT_PORT>",
        "MQTT_USERNAME": "<MQTT_USERNAME>",
        "MQTT_PASSWORD": "<MQTT_PASSWORD>",
        "MQTT_USE_TLS": "<MQTT_USE_TLS>"
      }
    }
  }
}

启动命令uvx mqtt5-mcp-server

需要配置的环境变量

MQTT_HOSTBroker hostname or IP

MQTT_PORTBroker port, default 1883 (or 8883 for TLS)

MQTT_USERNAMEBroker username

MQTT_PASSWORD密钥Broker password

MQTT_USE_TLStrue/false, default false

本地服务需要在你的电脑上安装运行环境（如 Node.js 18+）。各客户端的配置文件位置不同，请参考客户端文档。

安装包

pypimqtt5-mcp-server0.1.0stdio

工具

还没有获取到工具列表

本地服务需要在你的电脑上运行后才能读取工具，维护者暂未提供工具列表。

详细介绍

MQTT MCP 服务

一个适用于 MQTT 的 Model Context Protocol（MCP）服务。MQTT 是标准的 IoT 发布/订阅消息协议。可连接任意 MQTT 代理（Eclipse Mosquitto、EMQX、HiveMQ、AWS IoT Core、Azure IoT Hub 的 MQTT 端点等）：仅使用标准 MQTT 5.0 线协议，不调用特定代理的管理 API。

基于 FastMCP 和 paho-mqtt 构建。

工具

工具	描述
`get_connection_status()`	返回此连接器与代理的连接状态。
`publish_message(topic, payload, qos, retain, ...)`	发布消息。支持 MQTT5 属性（内容类型、消息过期时间、响应主题、关联数据、用户属性），并可选择阻塞等待代理确认送达（QoS 1/2）。
`start_subscription(topic_filter, qos, ...)`	开始接收与主题筛选器匹配的消息（支持 `+`/`#` 通配符）。返回一个 `session_id`。
`get_messages(session_id, max_messages, timeout_seconds)`	从订阅会话中读取缓冲的消息；可以短暂等待一条消息到达。
`stop_subscription(session_id)`	停止会话并取消订阅（如果没有其他会话需要相同筛选器）。
`list_subscriptions()`	列出所有活跃的订阅会话。
`discover_topics(topic_filter, duration_seconds, max_topics)`	通过保留消息近似实现“列出主题”操作——MQTT 没有原生的主题列表操作。

为什么订阅采用会话模式

MQTT 是持续的发布/订阅流，而不是请求/响应 API。单次 MCP 工具调用无法“永远等待”消息，因此订阅分两步进行：start_subscription 注册订阅意向，并在后台缓冲消息（共用一个代理连接，在进程内按主题筛选器分发）；随后可通过任意次数的工具调用使用 get_messages 读取缓冲区，并可选择在限定时间内等待新数据。完成后调用 stop_subscription。

配置

环境变量	必填	描述
`MQTT_HOST`	是	代理主机名或 IP。
`MQTT_PORT`	否	默认值为 `1883`（TLS 使用 `8883`，需显式设置）。
`MQTT_USERNAME` / `MQTT_PASSWORD`	否	代理凭据。
`MQTT_CLIENT_ID`	否	默认使用代理分配的 ID。
`MQTT_KEEPALIVE`	否	秒数，默认值为 `60`。
`MQTT_CONNECT_TIMEOUT`	否	等待初始连接的秒数，默认值为 `10`。
`MQTT_TRANSPORT`	否	`tcp`（默认）或 `websockets`。
`MQTT_WS_PATH`	否	WebSocket 传输的路径，默认值为 `/mqtt`。
`MQTT_USE_TLS`	否	`true`/`false`，默认值为 `false`。
`MQTT_TLS_CA_CERTS`	否	CA 证书包的路径。省略时使用系统信任存储。
`MQTT_TLS_CLIENT_CERT` / `MQTT_TLS_CLIENT_KEY`	否	用于双向 TLS。
`MQTT_TLS_INSECURE`	否	跳过主机名验证——仅用于自签名测试证书。
`MQTT_WILL_TOPIC` / `MQTT_WILL_PAYLOAD` / `MQTT_WILL_QOS` / `MQTT_WILL_RETAIN`	否	遗嘱消息，在连接时设置一次。

在本地运行

uv sync
export MQTT_HOST=localhost
export MQTT_PORT=1883
export MQTT_USERNAME=mqtt
export MQTT_PASSWORD=mqtt
uv run mqttmcpserver.py

可查看 mcp.json 获取可直接使用的客户端配置（stdio 和 Docker）。

使用真实代理进行测试

本仓库的设计原则要求使用真实实例而非模拟对象进行测试。docker-compose.yml 会启动 Eclipse Mosquitto（2.x，支持 MQTT5），启用密码身份验证，并提供用于测试 TLS/mTLS 代码路径的自签名证书 TLS 监听器。

docker compose up -d       # start Mosquitto on localhost:1883 (plain, user: mqtt / pass: mqtt) and :8883 (TLS)
uv run pytest -v           # tests/conftest.py also starts/stops the stack automatically
docker compose down        # stop it manually if you started it yourself