Coreflux MQTT MCP Server

An enterprise-grade Model Context Protocol (MCP) server that provides secure, scalable access to Coreflux MQTT brokers and comprehensive automation capabilities for Claude and other MCP-compatible AI assistants.

🚀 Features

Core Functionality

• 🔌 MQTT Integration: Seamless connection to Coreflux MQTT brokers with full TLS support

• 🛠️ Complete Coreflux API: Full access to models, actions, rules, and routes

• 🤖 AI Code Generation: LOT (Language-of-Things) code generation via Coreflux Copilot API

• 🔍 Dynamic Discovery: Automatic discovery and listing of available actions

• 🏥 Health Monitoring: Comprehensive system health checks and monitoring

Enterprise Features

• 🔒 Production Security: Comprehensive log sanitization, input validation, and security features

• ⚡ Async Processing: Non-blocking message processing with rate limiting and queue management

• � Enhanced Logging: Structured logging with rotation, filtering, and security sanitization

• ✅ Configuration Validation: Comprehensive environment and file validation system

• 🧪 Testing Framework: Complete unit testing suite with mocking and coverage reporting

DevOps & Deployment

• 🐳 Container Ready: Full Docker and Kubernetes deployment support with health checks

• 🔄 CI/CD Pipeline: GitHub Actions with automated testing, security scanning, and quality checks

• 📦 Development Tools: Pre-commit hooks, code formatting, linting, and documentation generation

• ⚙️ Easy Setup: Interactive setup assistant with validation and testing

• 📚 Rich Documentation: API documentation, security guides, and deployment instructions

Quick Start

Docker Deployment (Recommended)

1. Clone and configure:

   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. Deploy with Docker:

   docker-compose up -d

🚀 Quick Start

Prerequisites

• Python 3.11 or higher

• Docker (optional, for containerized deployment)

• Access to a Coreflux MQTT broker

• Coreflux Copilot API key (optional, for AI assistance)

Option 1: Docker Deployment (Recommended)

1. Clone and configure:

   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. Deploy with Docker:

   docker-compose up -d

3. Verify deployment:

   docker-compose logs -f coreflux-mcp-server

Option 2: Development Installation

1. Clone and setup:

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

2. Install dependencies:

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

3. Configure environment:

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

4. Validate and test:

   make validate # Validate configuration make test # Run tests

5. Start the server:

   python server.py # OR make run

For detailed deployment instructions, see DEPLOYMENT.md.

⚙️ Configuration

Interactive Setup Assistant

The server includes a comprehensive setup assistant that guides you through configuration:

The assistant helps with:

• 🔧 MQTT broker connection settings

• 🔐 TLS certificate configuration

• 🤖 Coreflux Copilot API integration

• 📝 Logging and monitoring setup

• ✅ Configuration validation and testing

Use the setup assistant when:

• Creating initial configuration

• Updating existing settings

• Troubleshooting connection issues

• Setting up TLS certificates

• Migrating between environments

Environment Configuration

Copy .env.example to .env and configure:

For detailed configuration options, see the Configuration Guide.

🔌 Connecting Claude to the MCP Server

Using Claude Desktop

1. Locate Claude Desktop config file:

   • macOS/Linux: ~/Library/Application Support/Claude/claude_desktop_config.json

   • Windows: %USERPROFILE%\AppData\Roaming\Claude\claude_desktop_config.json

2. Add server configuration:

   { "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. Restart Claude Desktop

Security Note: For production deployments, store secrets in secure environment variables or secret management systems rather than the Claude config file.

Using Environment Variables

For better security, use environment variables instead of hardcoding credentials:

Testing the Connection

Once configured, test the connection by asking Claude:

Claude should respond with system status and broker details if the connection is successful.

🛠️ Available Tools

The server provides the following tools to Claude:

Core MQTT Tools

• publish_to_coreflux - Publish messages to MQTT topics with QoS and retention options

• get_broker_info - Get detailed information about the MQTT broker connection

AI Assistance Tools

• copilot_assist - Query the Coreflux Copilot AI for automation assistance and code generation

System Management Tools

• comprehensive_health_check - Perform detailed health checks of all system components

For detailed API documentation, see API_DOCUMENTATION.md.

🧪 Development & Testing

Development Setup

1. Install development dependencies:

   pip install -r requirements-dev.txt

2. Install pre-commit hooks:

   pre-commit install

3. Run the full development setup:

   make dev-setup # Complete development environment setup

Testing

Run the comprehensive test suite:

Code Quality

Maintain code quality with automated tools:

Available development commands:

🔧 System Architecture

Core Components

• server.py - Main MCP server with tool implementations

• config_validator.py - Configuration validation and environment checking

• message_processor.py - Asynchronous MQTT message processing with rate limiting

• enhanced_logging.py - Structured logging with rotation and security filtering

• config_schema.py - Pydantic schemas for type-safe configuration

• parser.py - Sanitization and parsing utilities

Security Features

• Input Sanitization - All inputs are sanitized to prevent injection attacks

• Log Security - Automatic sanitization of sensitive data in logs

• TLS Support - Full TLS encryption for MQTT connections

• Configuration Validation - Comprehensive validation of all configuration parameters

• Secret Management - Secure handling of credentials and API keys

Performance Features

• Async Processing - Non-blocking message processing

• Connection Pooling - Efficient MQTT connection management

• Rate Limiting - Configurable rate limits to prevent abuse

• Health Monitoring - Real-time health checks and system monitoring

📚 Documentation

• API Documentation - Complete API reference

• Deployment Guide - Production deployment instructions

• Secret Management - Security and secret management guide

• Configuration Reference - Complete configuration options

🐳 Docker Deployment

Quick Start with Docker

Production Docker Deployment

See DEPLOYMENT.md for comprehensive production deployment instructions including:

• Multi-stage Docker builds

• Kubernetes deployments

• Health checks and monitoring

• Load balancing and scaling

• Security configurations

🔑 Coreflux Copilot Integration

The server includes powerful AI assistance through the Coreflux Copilot API:

Setup

1. Obtain API Key from the Coreflux Copilot dashboard

2. Configure the key:

   # 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

Features

• LOT Code Generation - Generate Language-of-Things code from natural language

• Automation Assistance - Get help with Coreflux automation tasks

• Best Practices - Receive guidance on optimal implementations

• Troubleshooting - Get assistance with debugging and optimization

Usage Examples

Ask Claude to help with Coreflux automation:

🚀 Advanced Features

Asynchronous Message Processing

The server includes a robust async message processor that:

• Prevents Blocking - Handles messages without blocking the main thread

• Rate Limiting - Configurable limits to prevent system overload

• Queue Management - Intelligent queue handling with backpressure

• Statistics - Real-time processing metrics and monitoring

Enhanced Logging System

Comprehensive logging with enterprise features:

• Structured Logging - JSON formatted logs for easy parsing

• Log Rotation - Automatic log file rotation to manage disk space

• Security Filtering - Automatic sanitization of sensitive information

• Multiple Outputs - Console, file, and syslog support

Configuration Validation

Robust validation system that checks:

• Environment Variables - Validates all required configuration

• File Permissions - Ensures certificate files are accessible

• Network Connectivity - Tests MQTT broker connectivity

• API Availability - Validates Copilot API access

🛡️ Security & Compliance

Security Features

• Input Sanitization - All inputs validated and sanitized

• TLS Encryption - Full TLS support for MQTT connections

• Secret Management - Secure credential handling

• Audit Logging - Comprehensive security event logging

• Non-root Execution - Runs with minimal privileges

Compliance Support

The server supports various compliance requirements:

• SOC 2 - Security controls and monitoring

• GDPR - Data protection and privacy

• HIPAA - Healthcare data protection (when properly configured)

For detailed security information, see SECRET_MANAGEMENT.md.

📊 Monitoring & Health Checks

Health Check Tool

Comprehensive health monitoring with the comprehensive_health_check tool:

Monitoring Metrics

The server provides detailed metrics:

• Connection Status - MQTT broker connectivity

• Message Processing - Queue size and processing rates

• System Resources - Memory and CPU usage

• Error Rates - Failed operations and error statistics

• API Status - Copilot API availability and response times

Alerting

Configure alerts for:

• Connection failures

• High error rates

• Resource exhaustion

• Security events

🤝 Contributing

We welcome contributions! Please see our contribution guidelines:

Development Process

1. Fork the repository

2. Create a feature branch: git checkout -b feature/amazing-feature

3. Install development dependencies: pip install -r requirements-dev.txt

4. Setup pre-commit hooks: pre-commit install

5. Make your changes with tests

6. Run quality checks: make quality-check

7. Commit your changes: git commit -am 'Add amazing feature'

8. Push to the branch: git push origin feature/amazing-feature

9. Create a Pull Request

Code Standards

• Python 3.11+ compatibility

• Type hints for all functions

• Comprehensive tests with >90% coverage

• Security scanning with bandit

• Code formatting with black and isort

• Documentation for all public APIs

📄 License

This project is licensed under the Apache License 2.0 - see the LICENSE file for details.

🆘 Support & Troubleshooting

Common Issues

Connection Refused

• Check broker hostname and port

• Verify network connectivity

• Confirm TLS configuration

Authentication Failed

• Verify username/password

• Check API key validity

• Confirm broker permissions

TLS Handshake Failed

• Verify certificate paths

• Check certificate validity

• Confirm TLS version compatibility

Debug Mode

Enable detailed logging for troubleshooting:

Getting Help

• GitHub Issues: Report bugs and request features

• Discussions: Community support and questions

• Documentation: Complete documentation

• Security Issues: Report to security@coreflux.org

🗺️ Roadmap

Current Status: v1.0.0 ✅

• ✅ Core MQTT functionality

• ✅ Copilot API integration

• ✅ Enterprise security features

• ✅ Comprehensive testing

• ✅ Production deployment support

Upcoming Features

• v1.1.0 - Enhanced monitoring and metrics

• v1.2.0 - Additional Coreflux API endpoints

• v1.3.0 - WebSocket support for real-time data

• v2.0.0 - Multi-broker support and federation

📋 Quick Reference

Essential Commands

Key Files

• server.py - Main MCP server

• .env - Configuration file

• requirements.txt - Python dependencies

• docker-compose.yml - Docker deployment

• Makefile - Development commands

Built with ❤️ by the Coreflux Community

• remove_action: Remove an action event/function

• run_action: Run an action event/function

• remove_all_models: Remove all models

• remove_all_actions: Remove all actions

• remove_all_routes: Remove all routes

• list_discovered_actions: List all discovered Coreflux actions

• request_lot_code: Generate LOT code using Coreflux Copilot API based on natural language prompts

Debugging and Troubleshooting

The MCP server now starts even if the MQTT broker is not available, allowing you to troubleshoot and configure connections through the MCP tools.

Connection Status and Recovery

• The server will start successfully even if the MQTT broker is unreachable

• Use the get_connection_status tool to check connection health and get troubleshooting guidance

• Use the setup_mqtt_connection tool to configure a new broker connection without restarting

• Use the check_broker_health or reconnect_mqtt tools to test and retry connections

Available Tools for Connection Management

• get_connection_status: Get detailed connection status with troubleshooting guidance

• setup_mqtt_connection: Configure a new MQTT broker connection dynamically

• mqtt_connect: Connect to a specific MQTT broker with custom parameters

• check_broker_health: Test broker connectivity and attempt reconnection

• reconnect_mqtt: Force reconnection to the configured broker

Traditional Troubleshooting Steps

If you encounter issues:

1. Verify your MQTT broker credentials in your Claude configuration

2. Ensure the broker is accessible

3. Run the setup assistant to verify or update your configuration:

   python setup_assistant.py

4. Check Claude Desktop logs:

   # 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. Run the server with debug logging:

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

References and Documentation

• DEPLOYMENT.md - Production deployment guide

• SECURITY.md - Security guidelines and best practices

• MCP Documentation - Official MCP documentation

• Coreflux Platform - Coreflux automation platform

Contributing

Contributions are welcome! Please read our contributing guidelines and submit pull requests to the development branch.

License

This project is licensed under the Apache License 2.0 - see the LICENSE file for details.

Support

• 📖 Documentation: Check the README, DEPLOYMENT.md, and SECURITY.md files

• 🐛 Issues: Report bugs and feature requests on GitHub

• 💬 Community: Join the Coreflux community for discussions