mqtt消息中心
项目介绍
MQTT MCP 服务器
一个轻量级的 模型上下文协议（MCP） 服务器，它以安全、标准化的方式连接大型语言模型（LLM）代理与 MQTT 设备，使基于AI的工作流程能够无缝集成到楼宇自动化（BAS）、工业控制（ICS）和智能家居系统中，允许代理监控实时传感器数据、操作设备并协调复杂的自动化任务。
快速开始
使用 uv 将 MQTT MCP 服务器添加并管理为项目中的依赖项，或通过 uv pip install 或 pip install 直接安装。请参阅文档中的 安装 部分获取完整的安装说明和其他详细信息。
uv add mqtt-mcp
该服务器可以嵌入并在应用程序中直接运行。默认情况下，它在 http://127.0.0.1:8000/mcp/ 暴露一个 流式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 代理的情况，可以在运行时使用环境变量指定其连接设置（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 检查器，并将其连接到位于 http://127.0.0.1:8000/mcp/ 的 MQTT MCP 服务器。确保将传输设置为 流式HTTP。
npx @modelcontextprotocol/inspector
核心概念
MQTT MCP 服务器利用 FastMCP 2.0 的核心构建块——资源模板、工具和提示——来简化 MQTT 接收和发布操作，减少样板代码，并提供清晰的 Python 接口。
接收消息
设备上的每个主题都被映射到一个资源（并作为工具暴露出来），并且使用 资源模板 来指定连接细节（主机、端口）和接收参数（主题、超时时间）。
@mcp.resource("mqtt://{host}:{port}/{topic*}")
@mcp.tool(
annotations={
"title": "接收消息",
"readOnlyHint": True,
"openWorldHint": True,
}
)
async def receive_message(
topic: str,
host: str = settings.mqtt.host,
port: int = settings.mqtt.port,
timeout: int = 60,
) -> str:
"""接收发布到指定主题的消息，如果有。"""
...
发布消息
发布操作作为 工具 暴露，接受相同的连接细节（主机、端口），并允许在一个原子调用中向特定主题发布消息。
@mcp.tool(
annotations={
"title": "发布消息",
"readOnlyHint": False,
"openWorldHint": True,
}
)
async def publish_message(
topic: str,
message: str,
host: str = settings.mqtt.host,
port: int = settings.mqtt.port,
) -> str:
"""向指定主题发布消息。"""
...
认证
要启用使用内置的 AuthKit 提供者的 流式HTTP 传输的认证，请在 .env 文件中提供 AuthKit 域名和重定向URL。更多详情请参阅 AuthKit 提供者 部分。
交互式提示
结构化响应消息通过 提示 实现，帮助引导交互、澄清缺失的参数并优雅地处理错误。
@mcp.prompt(name="mqtt_help", tags={"mqtt", "help"})
def mqtt_help() -> list[Message]:
"""提供如何使用 MQTT MCP 服务器的示例。"""
...
这里是一些可以用来与服务器交互的示例文本输入。
将 {"foo":"bar"} 发布到 127.0.0.1:1883 上的主题 "devices/foo"。
从主题 "devices/bar" 接收消息，等待最多 30 秒。
示例
examples 文件夹包含了一些示例项目，展示了如何使用各种客户端API与 MQTT MCP 服务器集成，为LLMs提供工具和上下文。
- 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 许可证。