Architecture

Architecture

mcmqtt’s architecture balances simplicity for individual developers with the sophistication needed for enterprise-scale AI coordination. This deep dive explores the technical decisions and patterns that make mcmqtt uniquely powerful.

High-Level Architecture

graph TB
    subgraph "MCP Layer"
        A[Claude Code] --> B[FastMCP Server]
        C[Other MCP Clients] --> B
    end

    subgraph "mcmqtt Core"
        B --> D[MCP Tools Engine]
        D --> E[MQTT Client Manager]
        D --> F[Broker Spawner]
        D --> G[Agent Coordinator]
    end

    subgraph "MQTT Infrastructure"
        E --> H[External MQTT Brokers]
        F --> I[Embedded Brokers]
        G --> I
    end

    subgraph "Agent Swarms"
        G --> J[Container Orchestrator]
        J --> K[Agent Pod 1]
        J --> L[Agent Pod 2]
        J --> M[Agent Pod N]
        K --> I
        L --> I
        M --> I
    end

    style B fill:#3b82f6,color:#fff
    style D fill:#10b981,color:#fff
    style G fill:#f59e0b,color:#fff
    style J fill:#ef4444,color:#fff

Core Components

FastMCP Server

The foundation of mcmqtt is a high-performance FastMCP server that provides the Model Context Protocol interface:

# Simplified FastMCP server structure
class MCMQTTServer(FastMCPServer):
    def __init__(self):
        self.mqtt_manager = MQTTManager()
        self.broker_spawner = BrokerSpawner()
        self.agent_coordinator = AgentCoordinator()

    @tool
    async def mqtt_connect(self, host: str, port: int = 1883):
        """Connect to MQTT broker with connection pooling"""
        return await self.mqtt_manager.connect(host, port)

    @tool
    async def swarm_deploy(self, name: str, count: int):
        """Deploy coordinated agent swarm"""
        return await self.agent_coordinator.deploy_swarm(name, count)

Key Features:

• Async by Design: All operations are non-blocking
• Connection Pooling: Efficient MQTT connection management
• Error Recovery: Automatic reconnection and retry logic
• Type Safety: Full Pydantic validation for all parameters

MQTT Client Manager

Manages persistent MQTT connections with advanced features:

class MQTTManager:
    def __init__(self):
        self.connections: Dict[str, MQTTClient] = {}
        self.message_handlers: Dict[str, List[Callable]] = {}
        self.connection_pool = ConnectionPool(max_size=100)

    async def connect(self, broker_config: BrokerConfig) -> MQTTClient:
        """Establish connection with automatic reconnection"""
        client_id = f"{broker_config.host}:{broker_config.port}"

        if client_id in self.connections:
            return self.connections[client_id]

        client = await self._create_client(broker_config)
        self.connections[client_id] = client

        # Set up automatic reconnection
        client.on_disconnect = self._handle_disconnect

        return client

Advanced Features:

• Automatic Reconnection: Exponential backoff with jitter
• Message Buffering: Queue messages during disconnections
• QoS Management: Intelligent QoS level selection
• Topic Wildcards: Efficient subscription management

Broker Spawner

Creates and manages embedded MQTT brokers on-demand:

class BrokerSpawner:
    def __init__(self):
        self.managed_brokers: Dict[str, BrokerInstance] = {}
        self.port_manager = PortManager()

    async def spawn_broker(self, config: BrokerConfig) -> BrokerInstance:
        """Spawn new MQTT broker with configuration"""
        port = await self.port_manager.allocate_port()

        broker = MQTTBroker(
            host=config.host or "localhost",
            port=port,
            config=config.to_dict()
        )

        await broker.start()
        self.managed_brokers[broker.id] = broker

        return broker

Broker Management:

• Dynamic Port Allocation: Automatic port management
• Configuration Templates: Predefined broker configurations
• Resource Monitoring: CPU and memory usage tracking
• Graceful Shutdown: Clean broker termination

Agent Coordinator

Orchestrates swarms of AI agents with advanced coordination patterns:

class AgentCoordinator:
    def __init__(self):
        self.swarms: Dict[str, AgentSwarm] = {}
        self.container_runtime = ContainerRuntime()
        self.task_queue = TaskQueue()

    async def deploy_swarm(self, swarm_config: SwarmConfig) -> AgentSwarm:
        """Deploy coordinated agent swarm"""
        swarm = AgentSwarm(
            name=swarm_config.name,
            agent_count=swarm_config.count,
            coordination_pattern=swarm_config.pattern
        )

        # Deploy agents in parallel
        tasks = []
        for i in range(swarm_config.count):
            agent_config = self._create_agent_config(swarm_config, i)
            task = self._deploy_agent(agent_config)
            tasks.append(task)

        agents = await asyncio.gather(*tasks)
        swarm.agents = agents

        # Start coordination layer
        await self._setup_coordination(swarm)

        self.swarms[swarm.name] = swarm
        return swarm

Design Patterns

Fractal Architecture

mcmqtt uses a fractal pattern where coordination patterns repeat at different scales:

graph TD
    A[Global Coordinator] --> B[Region 1 Coordinator]
    A --> C[Region 2 Coordinator]
    A --> D[Region N Coordinator]

    B --> E[Swarm 1]
    B --> F[Swarm 2]
    C --> G[Swarm 3]
    C --> H[Swarm 4]

    E --> I[Agent 1]
    E --> J[Agent 2]
    F --> K[Agent 3]
    F --> L[Agent 4]

    style A fill:#3b82f6,color:#fff
    style B fill:#10b981,color:#fff
    style C fill:#10b981,color:#fff
    style E fill:#f59e0b,color:#fff

This enables:

• Hierarchical Coordination: Parent-child relationships at any scale
• Local Decision Making: Reduce coordination overhead
• Fault Isolation: Problems don’t cascade across the hierarchy
• Emergent Behavior: Complex behaviors from simple rules

Message-Driven Architecture

All coordination happens via MQTT messages with structured topics:

agents/
├── coordination/
│   ├── {swarm_id}/
│   │   ├── tasks/assign      # Task distribution
│   │   ├── tasks/complete    # Task results
│   │   └── health/status     # Health monitoring
├── {agent_id}/
│   ├── commands/            # Direct agent commands
│   ├── status/              # Agent status updates
│   └── results/             # Agent task results
└── global/
    ├── announcements/       # System-wide messages
    └── coordination/        # Cross-swarm coordination

Container Isolation Strategy

Each agent runs in an isolated container with configurable resources:

# Agent container specification
apiVersion: v1
kind: Pod
metadata:
  name: agent-{agent_id}
  labels:
    swarm: {swarm_name}
    role: agent
spec:
  containers:
  - name: agent
    image: mcmqtt/agent:{agent_type}
    resources:
      limits:
        memory: "512Mi"
        cpu: "500m"
        ephemeral-storage: "1Gi"
      requests:
        memory: "256Mi"
        cpu: "250m"
    securityContext:
      readOnlyRootFilesystem: true
      runAsNonRoot: true
      runAsUser: 1000
      capabilities:
        drop: ["ALL"]

Performance Characteristics

Throughput

mcmqtt is designed for high-throughput scenarios:

Metric	Single Instance	Clustered
MQTT Messages/sec	50,000+	500,000+
Concurrent Connections	10,000+	100,000+
Agent Deployment Rate	100/min	1,000/min
Cross-Agent Latency	<10ms	<50ms

Scalability Limits

# Theoretical scaling limits
class ScalingLimits:
    # Per mcmqtt instance
    MAX_MQTT_CONNECTIONS = 10_000
    MAX_CONCURRENT_AGENTS = 1_000
    MAX_BROKER_INSTANCES = 100

    # Per agent swarm
    MAX_AGENTS_PER_SWARM = 10_000
    MAX_SWARMS_PER_COORDINATOR = 100

    # Message throughput
    MAX_MESSAGES_PER_SECOND = 50_000
    MAX_MESSAGE_SIZE = 256 * 1024  # 256KB

Memory Management

Intelligent memory management prevents resource exhaustion:

class MemoryManager:
    def __init__(self):
        self.connection_pool = LRUCache(maxsize=1000)
        self.message_buffer = CircularBuffer(maxsize=10000)
        self.agent_registry = WeakValueDictionary()

    async def monitor_memory(self):
        """Continuous memory monitoring"""
        while True:
            usage = psutil.virtual_memory()

            if usage.percent > 85:
                await self._cleanup_inactive_connections()
                await self._flush_message_buffers()

            if usage.percent > 95:
                await self._emergency_cleanup()

            await asyncio.sleep(30)

Security Architecture

Multi-Layer Security

graph TB
    A[TLS/mTLS Transport] --> B[Authentication Layer]
    B --> C[Authorization Engine]
    C --> D[Container Isolation]
    D --> E[Resource Limits]
    E --> F[Audit Logging]

    style A fill:#ef4444,color:#fff
    style D fill:#ef4444,color:#fff

Container Security

Each agent container runs with minimal privileges:

# Agent container security hardening
FROM python:3.11-slim

# Create non-root user
RUN groupadd -r agent && useradd -r -g agent agent

# Install dependencies
COPY requirements.txt /tmp/
RUN pip install -r /tmp/requirements.txt && \
    rm -rf /root/.cache /tmp/requirements.txt

# Copy application
COPY --chown=agent:agent . /app
WORKDIR /app

# Security settings
USER agent
EXPOSE 8080

# Read-only filesystem
VOLUME ["/tmp", "/app/logs"]

ENTRYPOINT ["python", "-m", "mcmqtt.agent"]

Network Security

class NetworkSecurity:
    def __init__(self):
        self.allowed_hosts = set()
        self.rate_limiter = RateLimiter()
        self.tls_config = TLSConfig()

    async def validate_connection(self, host: str, port: int):
        """Validate connection against security policies"""
        if not self._is_host_allowed(host):
            raise SecurityError(f"Host {host} not in allowlist")

        if not await self.rate_limiter.check_rate(host):
            raise RateLimitError(f"Rate limit exceeded for {host}")

        return True

Monitoring and Observability

Structured Logging

All components use structured logging for observability:

import structlog

logger = structlog.get_logger(__name__)

async def deploy_agent(agent_config):
    logger.info(
        "agent_deployment_started",
        agent_id=agent_config.id,
        swarm_name=agent_config.swarm,
        agent_type=agent_config.type,
        resources=agent_config.resources.dict()
    )

    try:
        agent = await create_agent(agent_config)
        logger.info(
            "agent_deployment_completed",
            agent_id=agent.id,
            startup_time=agent.startup_time,
            health_status="healthy"
        )
        return agent
    except Exception as e:
        logger.error(
            "agent_deployment_failed",
            agent_id=agent_config.id,
            error=str(e),
            error_type=type(e).__name__
        )
        raise

Metrics Collection

Built-in metrics for performance monitoring:

class MetricsCollector:
    def __init__(self):
        self.counters = defaultdict(int)
        self.gauges = defaultdict(float)
        self.histograms = defaultdict(list)

    def increment(self, name: str, value: int = 1, tags: Dict = None):
        """Increment counter metric"""
        key = self._make_key(name, tags)
        self.counters[key] += value

    def gauge(self, name: str, value: float, tags: Dict = None):
        """Set gauge metric"""
        key = self._make_key(name, tags)
        self.gauges[key] = value

    def histogram(self, name: str, value: float, tags: Dict = None):
        """Record histogram value"""
        key = self._make_key(name, tags)
        self.histograms[key].append(value)

Extension Points

Custom Agent Types

Implement custom agent behavior:

class CustomAgent(BaseAgent):
    async def process_task(self, task: Task) -> TaskResult:
        """Custom task processing logic"""
        # Your implementation here
        pass

    async def health_check(self) -> HealthStatus:
        """Custom health check logic"""
        # Your implementation here
        pass

Custom Coordination Patterns

Implement new coordination strategies:

class CustomCoordinationPattern(BaseCoordinationPattern):
    async def coordinate_agents(self, agents: List[Agent]) -> None:
        """Custom coordination logic"""
        # Your implementation here
        pass

Plugin System

Extend mcmqtt with plugins:

@plugin_manager.register("custom_broker")
class CustomBrokerPlugin(BrokerPlugin):
    async def create_broker(self, config: BrokerConfig) -> BrokerInstance:
        """Custom broker creation logic"""
        # Your implementation here
        pass

---

mcmqtt’s architecture enables unprecedented AI coordination capabilities while maintaining the simplicity that developers love. Every design decision balances power with usability, performance with safety. 🏗️