AI导哈

MQTT MCP服务器

一个连接LLM驱动的工作流与IoT智能时钟的模型上下文协议服务器，支持通过REST和MQTT协议控制时钟设备。

  • by PaulKnauer
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 PaulKnauer/mqtt-mcp-server

使用说明

MQTT MCP服务器

A. 模型上下文协议（MCP） 桥接服务器 LLM驱动的工作流程与物联网智能时钟。AI代理可以设置警报、显示消息， 并通过该服务器控制基于ESP32的时钟的亮度。

服务器使用 按方向划分的混合运输：命令（写入）分派到 clock-server的REST API，并且设备遥测（读取）通过MQTT到达。 clock-server 拥有面向MQTT发布、身份验证、速率限制和写路径审计的设备。

建筑学

                                   REST writes (commands)
                              ┌──────────────────────────────►┌──────────────┐
┌─────────────┐    stdio    ┌─┴────────────────────┐          │ clock-server │──► MQTT ──► ┌──────────────┐
│  AI Agent   │ ◄─────────► │  clock-mcp-server    │          │  (REST API)  │            │  Smart Clock │
│  (Claude,   │             │  (FastMCP + httpx +  │          └──────────────┘            │  (ESP32)     │
│   Codex)    │             │   paho)              │◄── MQTT reads (state/events) ────────┤              │
└─────────────┘             └──────────────────────┘                                      └──────────────┘
  • 写入路径（REST）： 命令工具调度到 clock-server 通过HTTP；一 202 Accepted 确认命令已被接受（而不是设备已动作）。
  • 读取路径（MQTT）： 服务器订阅设备状态/事件主题并缓存 读取端工具的最新观察结果。

六边形架构：

图层 位置 责任
域名 src/clock_mcp/domain/ 验证规则、类型异常（零基础设施依赖项）
配置 src/clock_mcp/config/ 来自env/.env的Pydantic验证设置
认证 src/clock_mcp/auth.py 承载令牌验证+设备范围授权
适配器 src/clock_mcp/adapters/ rest_adapter.py （httpx写入路径）， mqtt_subscription_adapter.py （paho读取路径）
服务项目 src/clock_mcp/services/ 命令调度（REST），读取端状态缓存
工具 src/clock_mcp/tools/ MCP工具注册（set_alarm、display_message、set_brightness、读取端查询）

快速开始

先决条件

  • Python 3.12+
  • 紫外线 （Python包管理器）
  • MQTT代理（例如Mosquitto）——请参阅 docker-compose.yml 在......下面

设置

git clone https://github.com/PaulKnauer/mqtt-mcp-server.git
cd mqtt-mcp-server

# Create config from template
cp .env.example .env
# Edit .env with your broker URL and credentials

# Run
uv run python -m clock_mcp

码头工人

docker build -t mqtt-mcp-server .
docker run --rm \
  -e MQTT_MCP_BROKER_URL=mqtt://host:1883 \
  -e MQTT_MCP_REST_BASE_URL=http://host:8080 \
  mqtt-mcp-server

Docker Compose（本地MQTT代理）

version: "3.8"
services:
  mosquitto:
    image: eclipse-mosquitto:2
    ports: ["1883:1883"]
  mqtt-mcp:
    build: .
    depends_on: [mosquitto]
    environment:
      MQTT_MCP_BROKER_URL: mqtt://mosquitto:1883

此组合文件仅支持MQTT代理，因此它执行 读取路径 （设备状态/事件）。命令分派（写路径）还需要一个可访问的 clock-server REST端点通过 MQTT_MCP_REST_BASE_URL没有它，命令工具就会失效 关闭，直到可以访问时钟服务器。

配置

MQTT读取路径（设备状态/事件遥测）：

变量 默认值 描述
MQTT_MCP_BROKER_URL mqtt://localhost:1883 MQTT代理URL(mqtt:// 或 mqtts://)
MQTT_MCP_BROKER_USERNAME -- MQTT代理用户名
MQTT_MCP_BROKER_PASSWORD -- MQTT代理密码
MQTT_MCP_TOPIC_PREFIX clocks/commands MQTT命名空间前缀；读取端订阅筛选器从中派生（多段值必须以结尾 /commands)
MQTT_MCP_AUTH_MODE none 身份验证模式(none 或 static)
MQTT_MCP_AUTH_TOKEN -- 传统承载令牌（通配符范围 *)
MQTT_MCP_AUTH_CREDENTIALS -- 多凭证格式： id|token|scope1,scope2;id2|token2|*
MQTT_MCP_LOG_LEVEL INFO 日志级别(DEBUG, INFO, WARNING, ERROR)

时钟服务器REST写入路径（命令调度）：

变量 默认值 描述
MQTT_MCP_REST_BASE_URL http://localhost:8080 时钟服务器REST基本URL(http:// 或 https://)
MQTT_MCP_REST_AUTH_TOKEN -- 时钟服务器REST请求的承载令牌
MQTT_MCP_REST_FORWARD_PROTO_HTTPS true 发送 X-Forwarded-Proto: https 关于REST请求
MQTT_MCP_REST_TIMEOUT_SECONDS 5.0 REST请求超时（秒）

工具

set_alarm

在智能时钟设备上设置闹钟。

参数：

  • device_id （str，必填）：目标设备（例如“clock-1”）
  • alarm_time （str，必填）：RFC3339 UTC时间（例如，“2030-01-01T07:00:00Z”）
  • label （str，可选）：报警标签（例如“唤醒”）
  • token （str，可选）：承载令牌（启用身份验证时需要）

display_message

在智能时钟设备上显示消息。

参数：

  • device_id （str，必填）：目标设备
  • message （str，必填）：消息文本
  • duration_seconds （int，必填）：显示持续时间（1-3600）
  • token （str，可选）：承载令牌

set_brightness

设置屏幕亮度。

参数：

  • device_id （str，必填）：目标设备
  • level （int，必填）：亮度0-100
  • token （str，可选）：承载令牌

ping

活体检查。退货 {"status": "ok"}.

server_info

返回服务器元数据，包括读取路径MQTT连接状态和 clock_server_ready （写路径时钟服务器就绪）。

命令契约（REST写入路径）

命令工具发送到 clock-server的REST API。域安全验证运行 在每次请求之前；无效输入在没有网络调用的情况下返回安全错误形状。 clock-server (internal/api/handlers.go)是本合同的真实来源。

工具 方法 路径 正文字段
set_alarm POST /commands/alarms deviceId, alarmTime, label
display_message POST /commands/messages deviceId, message, durationSeconds
set_brightness PUT /commands/brightness deviceId, level

请求体是带有camelCase字段的JSON，例如：

// POST /commands/alarms
{"deviceId": "clock-1", "alarmTime": "2030-01-01T07:00:00Z", "label": "wake up"}

// POST /commands/messages
{"deviceId": "clock-1", "message": "Meeting in 5", "durationSeconds": 30}

// PUT /commands/brightness
{"deviceId": "clock-1", "level": 75}

所需标题：

  • Authorization: Bearer <token>
  • X-Forwarded-Proto: https （通过切换 MQTT_MCP_REST_FORWARD_PROTO_HTTPS)
  • Content-Type: application/json

成功是HTTP 202 Accepted --它证实了 clock-server 接受了命令，不是 设备应用了它。设备确认通过MQTT读取单独公开 下面的路径。非-2xx和传输错误映射到类型 DispatchError；令牌永远不会 记录。

MQTT读取合约（读取路径）

设备事件主题（只读）

主题 有效载荷 保留
clocks/events/{id}/heartbeat {"uptimeSeconds": N, "wifiRssi": N, ...} 没有
clocks/events/{id}/alarm_triggered {"time": N, "label": "..."} 没有
clocks/events/{id}/alarm_acknowledged {"action": "dismiss", ...} 没有
clocks/events/{id}/command_result {"type": "set_alarm", "result": "applied"} 没有
clocks/state/{id}/display {"mode": "clock", "brightness": 50} 是的
clocks/state/{id}/alarm {"armed": true} 是的
clocks/state/{id}/presence "online" / "offline" （LWT） 是

发展

# Install dev dependencies
uv sync --group dev

# Run all quality gates
make ci

# Individual gates
make lint          # ruff check + format
make type-check    # mypy strict
make test          # pytest
make coverage      # pytest with coverage (≥70%)
make audit         # pip-audit (CVE scan)
make build-check   # uv build

添加新工具

  1. 将域模型添加到 src/clock_mcp/domain/models.py
  2. 添加验证 src/clock_mcp/domain/safety.py
  3. 添加一个方法 src/clock_mcp/services/clock_service.py
  4. 将处理程序添加到 src/clock_mcp/tools/commands.py
  5. 注册于 src/clock_mcp/tools/__init__.py::register_all()
  6. 增添 KNOWN_TOOL_NAMES 在 src/clock_mcp/config/models.py
  7. 在中添加测试 tests/unit/tools/

认证

两种身份验证模式：

  • 无：无限制访问（本地/家庭使用）
  • 静态：带有设备范围凭据的承载令牌

设备作用域支持精确匹配(clock-1)，前缀匹配(clock-*)，以及通配符(*).

令牌比较使用恒定时间 hmac.compare_digest 以防止定时攻击。

项目状态

版本：0.1.0——积极发展。尚未发布生产版本。

许可证

麻省理工学院

相关 MCP