Mqtt Mcp

MQTT MCP 服务

一个轻量级的 Model Context Protocol (MCP) 服务，以安全、标准化的方式将大模型 Agent 连接到 MQTT 设备，实现 AI 驱动的工作流与楼宇自动化 (BAS)、工业控制 (ICS) 和智能家居系统的无缝集成，使 Agent 能够监控实时传感器数据、操控设备并编排复杂的自动化任务。

快速开始

使用 uv 将 MQTT MCP 服务作为依赖项添加到你的项目中并进行管理，或通过 uv pip install 或 pip install 直接安装。完整的安装说明和更多详细信息请参阅文档中的安装部分。

uv add mqtt-mcp

该服务可以嵌入你的应用程序并直接从其中运行。默认情况下，它在 http://127.0.0.1:8000/mcp/ 暴露一个 Streamable HTTP 端点。

# app.py
from mqtt_mcp import MQTTMCP

mcp = MQTTMCP()

if __name__ == "__main__":
    mcp.run(transport="http")

它也可以在不修改源代码的情况下，使用提供的 CLI 从命令行启动。

mqtt-mcp

或者使用 uvx 在临时的隔离环境中运行。更多详细信息请查看使用工具指南。

uvx mqtt-mcp

配置

对于大多数操作都针对特定 MQTT broker 的使用场景，可以在运行时使用环境变量指定其连接设置（host 和 port），这样所有省略显式连接参数的提示词都会被路由到该 broker。

export MQTT_MCP_MQTT__HOST=10.0.0.1
export MQTT_MCP_MQTT__PORT=1883

这些设置也可以在工作目录的 .env 文件中指定。

# .env
mqtt_mcp_mqtt__host=10.0.0.1
mqtt_mcp_mqtt__port=1883

MCP Inspector

要确认服务已启动并运行，并浏览可用的资源和工具，请运行 MCP Inspector 并将其连接到 http://127.0.0.1:8000/mcp/ 上的 MQTT MCP 服务。请确保将传输方式设置为 Streamable HTTP。

npx @modelcontextprotocol/inspector

核心概念

MQTT MCP 服务利用 FastMCP 2.0 的核心构建块——资源模板、工具和提示词——以最少的样板代码和简洁的 Pythonic 接口来简化 MQTT 的接收和发布操作。

接收消息

设备上的每个主题都映射到一个资源（并作为工具暴露），并使用资源模板来指定连接详情（host、port）和接收参数（topic、timeout）。

@mcp.resource("mqtt://{host}:{port}/{topic*}")
@mcp.tool(
    annotations={
        "title": "Receive Message",
        "readOnlyHint": True,
        "openWorldHint": True,
    }
)
async def receive_message(
    topic: str,
    host: str = settings.mqtt.host,
    port: int = settings.mqtt.port,
    timeout: int = 60,
) -> str:
    """Receives a message published to the specified topic, if any."""
    ...

发布消息

发布操作作为工具暴露，接受相同的连接详情（host、port），并允许在单次原子调用中向特定主题发布消息。

@mcp.tool(
    annotations={
        "title": "Publish Message",
        "readOnlyHint": False,
        "openWorldHint": True,
    }
)
async def publish_message(
    topic: str,
    message: str,
    host: str = settings.mqtt.host,
    port: int = settings.mqtt.port,
) -> str:
    """Publishes a message to the specified topic."""
    ...

身份验证

要使用 Streamable HTTP 传输方式的内置 AuthKit 提供程序启用身份验证，请在 .env 文件中提供 AuthKit 域和重定向 URL。更多详细信息请查看 AuthKit Provider 部分。

交互式提示词

结构化响应消息使用提示词实现，帮助引导交互、澄清缺失的参数并优雅地处理错误。

@mcp.prompt(name="mqtt_help", tags={"mqtt", "help"})
def mqtt_help() -> list[Message]:
    """Provides examples of how to use the MQTT MCP server."""
    ...

以下是一些可用于与服务交互的示例文本输入。

Publish {"foo":"bar"} to topic "devices/foo" on 127.0.0.1:1883.
Receive a message from topic "devices/bar", waiting up to 30 seconds.

示例

examples 文件夹包含示例项目，展示如何使用各种客户端 API 与 MQTT MCP 服务集成，从而为大模型提供工具和上下文。

• openai-agents - 展示如何使用 OpenAI Agents SDK 连接到 MQTT MCP 服务。
• openai - 一个最小化应用，利用 OpenAI Python 库中的远程 MCP 服务支持。
• pydantic-ai - 展示如何使用 PydanticAI Agent Framework 连接到 MQTT MCP 服务。

Docker

MQTT MCP 服务可以按如下方式部署为 Docker 容器：

docker run -dit \
  --name mqtt-mcp \
  --restart=always \
  -p 8080:8000 \
  --env-file .env \
  ghcr.io/ezhuk/mqtt-mcp:latest

这会将主机上的端口 8080 映射到容器内 MCP 服务的端口 8000，并在存在 .env 文件时从中加载设置。

许可证

本服务基于 MIT License 授权。