Coreflux MQTT MCP Server

Coreflux MQTT MCP 服务

一款企业级模型上下文协议（MCP）服务，可安全、可扩展地访问 Coreflux MQTT 代理，并为 Claude 及其他兼容 MCP 的 AI 助手提供全面的自动化能力。

🚀 功能

核心功能

• 🔌 MQTT 集成：无缝连接 Coreflux MQTT 代理，并全面支持 TLS
• 🛠️ 完整的 Coreflux API：全面访问模型、操作、规则和路由
• 🤖 AI 代码生成：通过 Coreflux Copilot API 生成 LOT（Language-of-Things）代码
• 🔍 动态发现：自动发现并列出可用操作
• 🏥 健康监控：全面的系统健康检查与监控

企业功能

• 🔒 生产环境安全：全面的日志清理、输入验证和安全功能
• ⚡ 异步处理：非阻塞消息处理，并支持速率限制和队列管理
• � 增强日志记录：结构化日志记录，支持轮换、筛选和安全清理
• ✅ 配置验证：全面的环境与文件验证系统
• 🧪 测试框架：完整的单元测试套件，支持模拟和覆盖率报告

DevOps 与部署

• 🐳 支持容器部署：全面支持 Docker 和 Kubernetes 部署，并提供健康检查
• 🔄 CI/CD 流水线：通过 GitHub Actions 实现自动化测试、安全扫描和质量检查
• 📦 开发工具：Pre-commit 钩子、代码格式化、代码检查和文档生成
• ⚙️ 轻松设置：交互式设置助手，支持验证和测试
• 📚 丰富的文档：API 文档、安全指南和部署说明

快速开始

Docker 部署（推荐）

1. 克隆并配置：

   git clone https://github.com/CorefluxCommunity/Coreflux-MQTT-MCP-Server.git
   cd Coreflux-MQTT-MCP-Server
   cp .env.example .env
   # Edit .env with your configuration
   

2. 使用 Docker 部署：

   docker-compose up -d
   

🚀 快速开始

先决条件

• Python 3.11 或更高版本
• Docker（可选，用于容器化部署）
• 可访问 Coreflux MQTT 代理
• Coreflux Copilot API 密钥（可选，用于 AI 辅助）

选项 1：Docker 部署（推荐）

1. 克隆并配置：

   git clone https://github.com/CorefluxCommunity/Coreflux-MQTT-MCP-Server.git
   cd Coreflux-MQTT-MCP-Server
   cp .env.example .env
   # Edit .env with your configuration
   

2. 使用 Docker 部署：

   docker-compose up -d
   

3. 验证部署：

   docker-compose logs -f coreflux-mcp-server
   

选项 2：开发环境安装

1. 克隆并设置：

   git clone https://github.com/CorefluxCommunity/Coreflux-MQTT-MCP-Server.git
   cd Coreflux-MQTT-MCP-Server
   

2. 安装依赖项：

   pip install -r requirements.txt
   # For development
   pip install -r requirements-dev.txt
   

3. 配置环境：

   python setup_assistant.py  # Interactive configuration
   # OR
   cp .env.example .env && nano .env  # Manual configuration
   

4. 验证并测试：

   make validate  # Validate configuration
   make test      # Run tests
   

5. 启动服务：

   python server.py
   # OR
   make run
   

详细部署说明请参阅 DEPLOYMENT.md。

⚙️ 配置

交互式设置助手

服务包含一款全面的设置助手，可引导你完成配置：

python setup_assistant.py

助手可协助完成：

• 🔧 MQTT 代理连接设置
• 🔐 TLS 证书配置
• 🤖 Coreflux Copilot API 集成
• 📝 日志记录和监控设置
• ✅ 配置验证与测试

以下情况适合使用设置助手：

• 创建初始配置
• 更新现有设置
• 排查连接问题
• 设置 TLS 证书
• 在不同环境之间迁移

环境配置

将 .env.example 复制为 .env 并进行配置：

# MQTT Broker Configuration
MQTT_BROKER=your-broker-host.com
MQTT_PORT=8883
MQTT_USER=your-username
MQTT_PASSWORD=your-password
MQTT_USE_TLS=true

# TLS Configuration (when MQTT_USE_TLS=true)
MQTT_CA_CERT=/path/to/ca.crt
MQTT_CERT_FILE=/path/to/client.crt  
MQTT_KEY_FILE=/path/to/client.key

# Coreflux Copilot API
DO_AGENT_API_KEY=your-api-key-here

# Logging Configuration
LOG_LEVEL=INFO
LOG_FILE=/var/log/coreflux-mcp.log

详细配置选项请参阅 配置指南。

🔌 将 Claude 连接到 MCP 服务

使用 Claude Desktop

1. 找到 Claude Desktop 配置文件：

   • macOS/Linux：~/Library/Application Support/Claude/claude_desktop_config.json
   • Windows：%USERPROFILE%\AppData\Roaming\Claude\claude_desktop_config.json

2. 添加服务配置：

   {
     "mcpServers": {
       "coreflux": {
         "command": "python",
         "args": ["/path/to/your/server.py"],
         "env": {
           "MQTT_BROKER": "your-broker-host.com",
           "MQTT_PORT": "8883",
           "MQTT_USER": "your-username", 
           "MQTT_PASSWORD": "your-password",
           "MQTT_USE_TLS": "true",
           "DO_AGENT_API_KEY": "your-copilot-api-key"
         }
       }
     }
   }
   

3. 重启 Claude Desktop

安全提示：对于生产部署，请将密钥存储在安全的环境变量或密钥管理系统中，而不是 Claude 配置文件中。

使用环境变量

为了提高安全性，请使用环境变量，而不是将凭据硬编码：

{
  "mcpServers": {
    "coreflux": {
      "command": "python",
      "args": ["/path/to/your/server.py"],
      "env": {
        "MQTT_BROKER": "${COREFLUX_MQTT_BROKER}",
        "MQTT_PORT": "${COREFLUX_MQTT_PORT}",
        "MQTT_USER": "${COREFLUX_MQTT_USER}",
        "MQTT_PASSWORD": "${COREFLUX_MQTT_PASSWORD}",
        "DO_AGENT_API_KEY": "${COREFLUX_API_KEY}"
      }
    }
  }
}

测试连接

配置完成后，可以向 Claude 提问来测试连接：

Can you check the health of the Coreflux MCP server and show me the broker information?

如果连接成功，Claude 应会回复系统状态和代理详细信息。

🛠️ 可用工具

该服务向 Claude 提供以下工具：

核心 MQTT 工具

• publish_to_coreflux - 向 MQTT 主题发布消息，并设置 QoS 和保留选项
• get_broker_info - 获取 MQTT 代理连接的详细信息

AI 助手工具

• copilot_assist - 向 Coreflux Copilot AI 请求自动化协助和代码生成

系统管理工具

• comprehensive_health_check - 对所有系统组件执行详细的运行状况检查

详细的 API 文档请参阅 API_DOCUMENTATION.md。

🧪 开发与测试

开发环境设置

1. 安装开发依赖项：

   pip install -r requirements-dev.txt
   

2. 安装 pre-commit 挂钩：

   pre-commit install
   

3. 运行完整的开发环境设置：

   make dev-setup  # Complete development environment setup
   

测试

运行全面的测试套件：

# Run all tests
make test

# Run tests with coverage
make test-coverage

# Run specific test categories
make test-unit        # Unit tests only
make test-integration # Integration tests only

代码质量

使用自动化工具维护代码质量：

# Format code
make format

# Run linters
make lint

# Security scanning
make security-check

# Type checking
make type-check

# Run all quality checks
make quality-check

可用的开发命令：

# Development workflow
make dev-setup     # Set up complete development environment
make validate      # Validate configuration and environment  
make run           # Start the server with validation
make run-debug     # Start server in debug mode

# Testing and validation
make test          # Run all tests
make test-coverage # Run tests with coverage report
make test-unit     # Run unit tests only
make validate-config # Validate configuration files

# Code quality
make format        # Format code with black and isort
make lint          # Run all linters (flake8, bandit, mypy)
make security-check # Run security scanning
make type-check    # Run type checking with mypy

# Docker operations  
make docker-build  # Build Docker image
make docker-run    # Run in Docker container
make docker-test   # Run tests in Docker

# Documentation
make docs          # Generate documentation
make docs-serve    # Serve documentation locally

🔧 系统架构

核心组件

• server.py - 包含工具实现的主 MCP 服务
• config_validator.py - 配置验证和环境检查
• message_processor.py - 具有速率限制功能的异步 MQTT 消息处理
• enhanced_logging.py - 带轮换和安全过滤的结构化日志记录
• config_schema.py - 用于类型安全配置的 Pydantic 架构
• parser.py - 清理和解析工具

安全功能

• 输入清理 - 清理所有输入以防止注入攻击
• 日志安全 - 自动清理日志中的敏感数据
• TLS 支持 - 为 MQTT 连接提供完整的 TLS 加密
• 配置验证 - 全面验证所有配置参数
• 密钥管理 - 安全处理凭据和 API 密钥

性能功能

• 异步处理 - 非阻塞消息处理
• 连接池 - 高效管理 MQTT 连接
• 速率限制 - 可配置速率限制，以防止滥用
• 运行状况监控 - 实时运行状况检查和系统监控

📚 文档

• API 文档 - 完整的 API 参考
• 部署指南 - 生产部署说明
• 密钥管理 - 安全和密钥管理指南
• 配置参考 - 完整的配置选项

🐳 Docker 部署

使用 Docker 快速开始

# Clone and configure
git clone https://github.com/CorefluxCommunity/Coreflux-MQTT-MCP-Server.git
cd Coreflux-MQTT-MCP-Server

# Copy and edit environment file
cp .env.example .env
nano .env  # Configure your settings

# Start with Docker Compose
docker-compose up -d

# Check logs
docker-compose logs -f coreflux-mcp-server

# Health check
docker-compose exec coreflux-mcp-server python -c "
import os
os.system('python server.py --health-check')
"

Docker 生产部署

请参阅 DEPLOYMENT.md，了解全面的生产部署说明，包括：

• 多阶段 Docker 构建
• Kubernetes 部署
• 运行状况检查和监控
• 负载均衡和扩展
• 安全配置

🔑 Coreflux Copilot 集成

该服务通过 Coreflux Copilot API 提供强大的 AI 协助功能：

设置

1. 从 Coreflux Copilot 控制面板获取 API 密钥
2. 配置密钥：

   # Option 1: Environment file
   echo "DO_AGENT_API_KEY=your_api_key_here" >> .env
   
   # Option 2: Environment variable
   export DO_AGENT_API_KEY=your_api_key_here
   

功能

• LOT 代码生成 - 根据自然语言生成 Language-of-Things 代码
• 自动化协助 - 获取 Coreflux 自动化任务方面的帮助
• 最佳实践 - 获取最佳实现方案的指导
• 故障排查 - 获取调试和优化方面的帮助

使用示例

让 Claude 帮助完成 Coreflux 自动化任务：

Generate LOT code for a temperature monitoring system that triggers an alert when the temperature exceeds 75°F

Help me create a rule that processes sensor data and stores it in a database

🚀 高级功能

异步消息处理

该服务包含一个可靠的异步消息处理器，可实现：

• 防止阻塞 - 处理消息时不会阻塞主线程
• 速率限制 - 可配置限制，防止系统过载
• 队列管理 - 通过背压机制实现智能队列处理
• 统计信息 - 实时处理指标与监控

增强型日志系统

提供具备企业级功能的全面日志记录：

• 结构化日志 - 采用 JSON 格式的日志，便于解析
• 日志轮转 - 自动轮转日志文件，以管理磁盘空间
• 安全过滤 - 自动清理敏感信息
• 多种输出方式 - 支持控制台、文件和 syslog

配置验证

可靠的验证系统，可检查：

• 环境变量 - 验证所有必需的配置
• 文件权限 - 确保可以访问证书文件
• 网络连接 - 测试 MQTT 代理连接
• API 可用性 - 验证 Copilot API 的访问权限

🛡️ 安全与合规

安全功能

• 输入清理 - 验证并清理所有输入
• TLS 加密 - 完全支持 MQTT 连接的 TLS
• 密钥管理 - 安全处理凭据
• 审计日志 - 全面记录安全事件
• 非 root 执行 - 以最小权限运行

合规支持

该服务支持多种合规要求：

• SOC 2 - 安全控制与监控
• GDPR - 数据保护与隐私
• HIPAA - 医疗保健数据保护（需正确配置）

如需了解详细的安全信息，请参阅 SECRET_MANAGEMENT.md。

📊 监控与运行状况检查

运行状况检查工具

通过 comprehensive_health_check 工具进行全面的运行状况监控：

# Manual health check
python server.py --health-check

# Or ask Claude:
# "Please run a comprehensive health check on the Coreflux MCP server"

监控指标

该服务提供详细的指标：

• 连接状态 - MQTT 代理连接状态
• 消息处理 - 队列大小和处理速率
• 系统资源 - 内存和 CPU 使用情况
• 错误率 - 操作失败情况和错误统计
• API 状态 - Copilot API 的可用性和响应时间

告警

可针对以下情况配置告警：

• 连接失败
• 错误率过高
• 资源耗尽
• 安全事件

🤝 参与贡献

欢迎参与贡献！请参阅我们的贡献指南：

开发流程

1. Fork 此仓库
2. 创建 功能分支：git checkout -b feature/amazing-feature
3. 安装 开发依赖：pip install -r requirements-dev.txt
4. 设置 pre-commit 钩子：pre-commit install
5. 进行 修改并添加测试
6. 运行 质量检查：make quality-check
7. 提交 修改：git commit -am 'Add amazing feature'
8. 推送 到分支：git push origin feature/amazing-feature
9. 创建 Pull Request

代码规范

• 兼容 Python 3.11+
• 所有函数都应包含类型提示
• 编写全面的测试，覆盖率超过 90%
• 使用 bandit 进行安全扫描
• 使用 black 和 isort 进行代码格式化
• 为所有公共 API 编写文档

📄 许可证

本项目采用 Apache License 2.0 许可证授权，详情请参阅 LICENSE 文件。

🆘 支持与故障排查

常见问题

连接被拒绝

Error: MQTT connection failed

• 检查代理主机名和端口
• 验证网络连接
• 确认 TLS 配置

身份验证失败

Error: Authentication failed

• 验证用户名/密码
• 检查 API 密钥是否有效
• 确认代理权限

TLS 握手失败

Error: TLS handshake failed

• 验证证书路径
• 检查证书是否有效
• 确认 TLS 版本兼容性

调试模式

启用详细日志以进行故障排查：

export LOG_LEVEL=DEBUG
python server.py

获取帮助

• GitHub Issues：报告错误并请求新功能
• 讨论区：社区支持与问题交流
• 文档：完整文档
• 安全问题：发送邮件至 [email protected] 报告

🗺️ 路线图

当前状态：v1.0.0 ✅

• ✅ MQTT 核心功能
• ✅ Copilot API 集成
• ✅ 企业级安全功能
• ✅ 全面测试
• ✅ 支持生产环境部署

即将推出的功能

• v1.1.0 - 增强监控和指标
• v1.2.0 - 更多 Coreflux API 端点
• v1.3.0 - 支持 WebSocket 实时数据
• v2.0.0 - 支持多代理和联邦

---

📋 快速参考

常用命令

# Setup and configuration
python setup_assistant.py    # Interactive setup
make validate                 # Validate configuration

# Development
make dev-setup               # Complete dev environment
make test                    # Run all tests
make quality-check           # Run all quality checks

# Deployment
docker-compose up -d         # Docker deployment
make docker-build           # Build Docker image

# Monitoring
make health-check           # System health check
docker-compose logs -f      # View logs

关键文件

• server.py - MCP 主服务
• .env - 配置文件
• requirements.txt - Python 依赖项
• docker-compose.yml - Docker 部署
• Makefile - 开发命令

---

由 Coreflux 社区倾情打造 ❤️

• remove_action：移除一个操作事件/函数
• run_action：运行一个操作事件/函数
• remove_all_models：移除所有模型
• remove_all_actions：移除所有操作
• remove_all_routes：移除所有路由
• list_discovered_actions：列出所有已发现的 Coreflux 操作
• request_lot_code：根据自然语言提示词，使用 Coreflux Copilot API 生成 LOT 代码

调试与故障排查

即使 MQTT 代理不可用，MCP 服务现在也会启动，因此你可以通过 MCP 工具排查问题并配置连接。

连接状态与恢复

• 即使无法访问 MQTT 代理，服务也会成功启动
• 使用 get_connection_status 工具检查连接状况并获取故障排查指引
• 使用 setup_mqtt_connection 工具配置新的代理连接，无需重启
• 使用 check_broker_health 或 reconnect_mqtt 工具测试连接并重试

可用的连接管理工具

• get_connection_status：获取详细连接状态和故障排查指引
• setup_mqtt_connection：动态配置新的 MQTT 代理连接
• mqtt_connect：使用自定义参数连接到指定的 MQTT 代理
• check_broker_health：测试代理连接并尝试重新连接
• reconnect_mqtt：强制重新连接到已配置的代理

传统故障排查步骤

如果遇到问题：

1. 在 Claude 配置中验证 MQTT 代理凭据
2. 确保代理可访问
3. 运行设置助手以验证或更新配置：

   python setup_assistant.py
   

4. 检查 Claude Desktop 日志：

   # Check Claude's logs for errors (macOS/Linux)
   tail -n 20 -f ~/Library/Logs/Claude/mcp*.log
   # Windows PowerShell
   Get-Content -Path "$env:USERPROFILE\AppData\Roaming\Claude\Logs\mcp*.log" -Tail 20 -Wait
   

5. 使用调试日志运行服务：

   # Direct execution with debug logging
   python server.py --mqtt-host localhost --mqtt-port 1883 --log-level DEBUG
   

参考资料与文档

• DEPLOYMENT.md - 生产环境部署指南
• SECURITY.md - 安全指南与最佳实践
• MCP 文档 - MCP 官方文档
• Coreflux 平台 - Coreflux 自动化平台

参与贡献

欢迎贡献！请阅读我们的贡献指南，并向 development 分支提交拉取请求。

许可证

本项目采用 Apache License 2.0 许可，详情请参阅 LICENSE 文件。

支持

• 📖 文档：查看 README、DEPLOYMENT.md 和 SECURITY.md 文件
• 🐛 问题：在 GitHub 上报告错误和功能请求
• 💬 社区：加入 Coreflux 社区参与讨论