mqtt MCP

MQTT MCP服务器

轻量级 模型上下文协议（MCP） 将LLM代理连接到的服务器 消息队列遥测传输 以安全、标准化的方式使用设备，实现人工智能驱动的工作流程与楼宇自动化（BAS）、工业控制（ICS）和智能家居系统的无缝集成，使代理能够监控实时传感器数据、启动设备和协调复杂的自动化任务。

  ](https://pypi.org/p/mqtt-mcp)

入门指南

uv add mqtt-mcp

服务器可以嵌入到您的应用程序中并直接从中运行。默认情况下，它公开一个 Streamable HTTP 端点位于 http://127.0.0.1:8000/mcp/.

# app.py
from mqtt_mcp import MQTTMCP

mcp = MQTTMCP()

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

它也可以使用提供的从命令行启动 CLI 而无需修改源代码。

mqtt-mcp

或者在短暂、孤立的环境中使用 uvx.看看 使用工具 更多详细信息，请参阅指南。

uvx mqtt-mcp

配置

对于大多数操作都针对特定MQTT代理的用例，其连接设置(host 和 port)可以在运行时使用环境变量指定，以便将所有省略显式连接参数的提示路由到此代理。

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检查员

要确认服务器已启动并正在运行，并探索可用的资源和工具，请运行 MCP检查员 并将其连接到MQTT MCP服务器 http://127.0.0.1:8000/mcp/.确保将运输设置为 Streamable HTTP.

npx @modelcontextprotocol/inspector

核心概念

MQTT MCP服务器利用FastMCP 2.0的核心构建块——资源模板、工具和提示——以最少的样板和干净的Pythonic界面简化MQTT接收和发布操作。

接收消息

设备上的每个主题都映射到一个资源（并作为工具公开） 资源模板 用于指定连接详细信息（主机、端口）和接收参数（主题、超时）。

@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."""
    ...

发布消息

发布操作公开为 工具，接受相同的连接细节（主机、端口），并允许在单个原子调用中将消息发布到特定主题。

@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."""
    ...

认证

使用内置的启用身份验证 AuthKit 提供者为 Streamable HTTP 传输，在中提供AuthKit域和重定向URL .env 文件。看看 AuthKit提供程序 有关更多详细信息，请参阅第节。

互动提示

结构化响应消息是通过以下方式实现的 鼓励 这有助于指导交互，澄清缺失的参数，并优雅地处理错误。

@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服务器集成，为LLM提供工具和上下文。

码头工人

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许可证.