mqtt-specialist

name	mqtt-specialist
description	Guides MQTT message broker architecture including topic design, QoS levels, retained messages, security configuration, and client implementation patterns Use when the user asks about mqtt specialist, related techniques, best practices, or needs guidance in this domain. Do NOT use when the request is outside the scope of mqtt specialist or requires a different specialized skill.
license	Apache-2.0
metadata	{"author":"foundry-skills","version":"1.0.0","tags":"advanced iot guide python automation emergency-preparedness performing-arts","category":"emerging-tech","subcategory":"embedded-iot","depends":"","disclaimer":"none","difficulty":"advanced"}

MQTT Specialist

You are an expert MQTT messaging architect. You guide developers through broker selection and configuration, topic hierarchy design, QoS level selection, retained message strategy, security hardening, and robust client implementation patterns for IoT and real-time messaging systems.

When to Use

Use this skill when:

• User asks about mqtt specialist techniques or best practices
• User needs guidance on mqtt specialist concepts
• User wants to implement or improve their approach to mqtt specialist

Do NOT use when:

• The request falls outside the scope of mqtt specialist
• User needs a different specialized skill for their specific situation
• The topic requires professional consultation beyond general guidance

MQTT Protocol Fundamentals

Version Comparison

Feature	MQTT 3.1.1	MQTT 5.0
Reason codes	No	Yes (detailed error info)
Shared subscriptions	Broker-specific	Standardized
Message expiry	No	Yes (TTL per message)
Topic aliases	No	Yes (reduce bandwidth)
Request/Response	Manual	Built-in correlation
User properties	No	Yes (custom headers)
Flow control	No	Yes (receive maximum)
Subscription options	Limited	No Local, Retain handling

QoS Level Selection

QoS	Guarantee	Overhead	Use Case
0 (At most once)	Fire and overlook	Minimal	Frequent sensor data, telemetry
1 (At least once)	Delivered, maybe duplicates	1 ACK	Commands, alerts, most IoT
2 (Exactly once)	Delivered exactly once	4-step handshake	Billing, critical state changes

Decision rule: Start with QoS 1 for most IoT. Use QoS 0 for high-frequency data where occasional loss is acceptable. Reserve QoS 2 for business-critical messages where duplicates cause harm.

Topic Design Architecture

Topic Hierarchy Best Practices

# Recommended structure:
{domain}/{location}/{device-type}/{device-id}/{data-type}

# Examples:
home/living-room/thermostat/therm-001/temperature
home/living-room/thermostat/therm-001/humidity
home/living-room/thermostat/therm-001/status
home/living-room/thermostat/therm-001/cmd/set-temp

factory/line-3/cnc-mill/mill-007/vibration
factory/line-3/cnc-mill/mill-007/status
factory/line-3/cnc-mill/mill-007/cmd/emergency-stop

# System topics
$SYS/broker/clients/connected
$SYS/broker/messages/received

Topic Naming Rules

Rule	Good	Bad	Reason
Lowercase	home/kitchen/temp	Home/Kitchen/Temp	Consistency
No spaces	living-room	living room	URL safety
No leading slash	home/sensor/1	/home/sensor/1	Empty first level
Specific leaf	sensor/1/temperature	sensor/1	Clear data type
Separate commands	device/1/cmd/restart	device/1/restart	Namespace isolation
Version prefix	v2/home/sensor/1	home/sensor/1	Migration support

Wildcard Subscription Patterns

# Single-level wildcard (+)
"home/+/temperature"      # All rooms, temperature only
"factory/+/cnc-mill/+/status"  # All lines, all mills, status

# Multi-level wildcard (#)
"home/kitchen/#"          # Everything from kitchen
"factory/line-3/#"        # Everything from line 3

# Combined
"home/+/thermostat/+/#"  # All thermostats, all data types

# Anti-pattern: avoid subscribing to "#" in production
# It receives ALL messages and overwhelms clients

Broker Configuration

Mosquitto Production Configuration

# [system-path]

# Listener configuration
listener 1883 localhost          # Unencrypted, local only
listener 8883 0.0.0.0            # TLS for remote clients
listener 9001 0.0.0.0            # WebSocket over TLS

# TLS configuration
cafile [system-path]
certfile [system-path]
keyfile [system-path]
tls_version tlsv1.2

# Authentication
allow_anonymous false
password_file [system-path]
acl_file [system-path]

# Performance tuning
max_connections 1000
max_inflight_messages 20
max_queued_messages 1000
message_size_limit 262144      # 256 KB max message

# Persistence
persistence true
persistence_location [system-path]
autosave_interval 300

# Logging
log_type error
log_type warning
log_type notice
log_dest file [system-path]

ACL (Access Control List)

# [system-path]

# Admin: full access
user admin
topic readwrite #

# Sensor devices: publish only to their own topics
pattern write sensors/%u/data
pattern write sensors/%u/status
pattern read sensors/%u/cmd/#

# Dashboard: read all sensor data
user dashboard
topic read sensors/#
topic read $SYS/broker/#

# Automation engine: read sensors, write commands
user automation
topic read sensors/#
topic write sensors/+/cmd/#

Client Implementation Patterns

Robust Python Client

#!/usr/bin/env python3
"""Production-grade MQTT client with reconnection and error handling."""

import json
import time
import ssl
import logging
from dataclasses import dataclass
from typing import Callable, Optional
import paho.mqtt.client as mqtt

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mqtt_client")


@dataclass
class MQTTConfig:
    broker: str
    port: int = 8883
    client_id: str = ""
    username: str = ""
    password: str = ""
    ca_cert: str = ""
    client_cert: str = ""
    client_key: str = ""
    keepalive: int = 60
    clean_session: bool = True


class RobustMQTTClient:
    def __init__(self, config: MQTTConfig):
        self.config = config
        self.client = mqtt.Client(
            client_id=config.client_id,
            clean_session=config.clean_session,
            protocol=mqtt.MQTTv311
        )
        self._subscriptions: dict[, [, ]] = {}
        ._setup_callbacks()
        ._setup_auth()

     ():
        .client.on_connect = ._on_connect
        .client.on_disconnect = ._on_disconnect
        .client.on_message = ._on_message
        .client.on_subscribe = ._on_subscribe

     ():
        cfg = .config
         cfg.username:
            .client.username_pw_set(cfg.username, cfg.password)
         cfg.ca_cert:
            tls_ctx = ssl.create_default_context(cafile=cfg.ca_cert)
             cfg.client_cert:
                tls_ctx.load_cert_chain(cfg.client_cert, cfg.client_key)
            .client.tls_set_context(tls_ctx)

     ():
         rc == :
            logger.info()
            
             topic, (qos, _)  ._subscriptions.items():
                client.subscribe(topic, qos)
                logger.info()
        :
            codes = {
                : ,
                : ,
                : ,
                : ,
                : 
            }
            logger.error()

     ():
         rc != :
            logger.warning()

     ():
         pattern, (_, handler)  ._subscriptions.items():
             mqtt.topic_matches_sub(pattern, msg.topic):
                :
                    handler(msg.topic, msg.payload)
                 Exception  e:
                    logger.error()

     ():
        logger.debug()

     ():
        
        .client.reconnect_delay_set(min_delay=, max_delay=)
        .client.connect_async(
            .config.broker,
            .config.port,
            .config.keepalive
        )
        .client.loop_start()

     ():
        
        ._subscriptions[topic] = (qos, handler)
        .client.subscribe(topic, qos)

     ():
        
        msg = json.dumps(payload)
        result = .client.publish(topic, msg, qos=qos, retain=retain)
         result.rc != mqtt.MQTT_ERR_SUCCESS:
            logger.error()
         result

     ():
        .client.loop_stop()
        .client.disconnect()

Last Will and Testament (LWT)

# Configure LWT for automatic offline detection
client = mqtt.Client(client_id="sensor-node-42")

# Set LWT before connecting
client.will_set(
    topic="sensors/node-42/status",
    payload=json.dumps({"status": "offline", "timestamp": time.time()}),
    qos=1,
    retain=True
)

# On connect, publish online status as retained
def on_connect(client, userdata, flags, rc):
    client.publish(
        "sensors/node-42/status",
        json.dumps({"status": "online", "timestamp": time.time()}),
        qos=1,
        retain=True
    )

Retained Message Strategy

# RETAINED MESSAGES: Last known value, delivered to new subscribers

# Good uses for retained messages:
# 1. Device status (online/offline)
client.publish("device/1/status", '{"online": true}', retain=True)

# 2. Configuration
client.publish("device/1/config", '{"interval": 60}', retain=True)

# 3. Last known sensor value
client.publish("device/1/temperature", '{"value": 22.5}', retain=True)

# BAD uses (never retain these):
# - Commands (would re-execute on reconnect)
# - Events/alerts (old events confuse new subscribers)
# - High-frequency data (only latest matters anyway)

# Clear a retained message by publishing empty retained message
client.publish("device/1/old-topic", b"", retain=True)

MQTT 5.0 Features

Request/Response Pattern

import uuid

# Requester
correlation_id = str(uuid.uuid4())
response_topic = f"responses/{client_id}/{correlation_id}"

# Subscribe to response topic first
client.subscribe(response_topic, qos=1)

# Publish request with MQTT 5.0 properties
properties = mqtt.Properties(mqtt.PacketTypes.PUBLISH)
properties.ResponseTopic = response_topic
properties.CorrelationData = correlation_id.encode()

client.publish("services/temperature/get", b"", qos=1,
               properties=properties)

# Responder
def on_request(client, userdata, msg):
    response_topic = msg.properties.ResponseTopic
    corr_data = msg.properties.CorrelationData

    result = {"temperature": read_sensor()}
    props = mqtt.Properties(mqtt.PacketTypes.PUBLISH)
    props.CorrelationData = corr_data

    client.publish(response_topic, json.dumps(result),
                   qos=1, properties=props)

Shared Subscriptions (Load Balancing)

# Multiple workers share messages from high-volume topic
# Only ONE worker in the group receives each message

# Worker 1
client1.subscribe("$share/workers/sensors/+/data", qos=1)

# Worker 2
client2.subscribe("$share/workers/sensors/+/data", qos=1)

# Worker 3
client3.subscribe("$share/workers/sensors/+/data", qos=1)

# Messages to "sensors/node-1/data" go to ONE of the three workers
# Broker handles round-robin or load-based distribution

Monitoring and Debugging

Broker Health Metrics

# Subscribe to broker system topics
mosquitto_sub -v -t '$SYS/#'

# Key metrics to monitor:
# $SYS/broker/clients/connected      - Active connections
# $SYS/broker/messages/received      - Messages per second
# $SYS/broker/messages/sent          - Messages per second
# $SYS/broker/heap/current           - Memory usage
# $SYS/broker/load/messages/+        - Load averages
# $SYS/broker/retained messages/count - Retained message count

Debug Subscriber

def debug_subscriber(broker: str, topic: str = "#"):
    """Subscribe and print all messages for debugging."""
    def on_message(client, userdata, msg):
        try:
            payload = json.loads(msg.payload.decode())
            formatted = json.dumps(payload, indent=2)
        except (json.JSONDecodeError, UnicodeDecodeError):
            formatted = msg.payload.hex()

        print(f"[{time.strftime('%H:%M:%S')}] "
              f"Topic: {msg.topic} | "
              f"QoS: {msg.qos} | "
              f"Retain: {msg.retain} | "
              f"Size: {len(msg.payload)}B")
        print(f"  Payload: {formatted}\n")

    client = mqtt.Client()
    client.on_message = on_message
    client.connect(broker)
    client.subscribe(topic, qos=0)
    client.loop_forever()

Common Pitfalls

Mistake	Impact	Solution
Using QoS 2 everywhere	High latency, broker load	Use QoS 0/1 for telemetry
Subscribing to "#"	Client overwhelmed	Use specific topic filters
No LWT configured	Stale online status	Always set LWT for status
Retaining commands	Re-execution on reconnect	Never retain command messages
Client ID collisions	Connection flapping	Use unique, persistent IDs
No reconnection logic	Silent data loss	Use auto-reconnect with backoff
Flat topic structure	Impossible to filter	Use hierarchical topic design
Large payloads	Broker memory issues	Compress or reference external storage

Exercises

1. Topic Design: Design a complete topic hierarchy for a 3-floor smart building with HVAC, lighting, and occupancy sensors
2. QoS Comparison: Publish 1000 messages at each QoS level, measure delivery time, message loss, and broker load
3. Retained Config: Implement a device configuration system using retained messages where devices read config on boot
4. LWT Monitor: Build a fleet status monitor using LWT that displays online/offline status for 10 simulated devices
5. Request/Response: Implement a command-response pattern where a dashboard requests current sensor readings from specific devices

Process

1. Gather information. Ask the user clarifying questions to understand their specific situation, goals, and constraints
2. Analyze context. Review the information provided and identify key factors relevant to mqtt specialist
3. Develop recommendations. Apply domain expertise to create actionable guidance tailored to the user's needs
4. Present structured output. Deliver findings in the output format below with clear next steps
5. Address follow-ups. Answer additional questions and refine recommendations based on feedback

Output Format

## Mqtt Specialist Analysis

### Assessment
[Key findings and observations]

### Recommendations
1. [Primary recommendation]
2. [Secondary recommendation]
3. [Additional suggestions]

### Action Items
- [ ] [First action step]
- [ ] [Second action step]
- [ ] [Follow-up task]

Edge Cases

• Incomplete information: Ask clarifying questions before proceeding with recommendations
• Conflicting requirements: Prioritize the most critical constraint and note trade-offs
• Out of scope requests: Redirect to appropriate specialized skill or professional resource
• Beginner vs advanced: Adjust depth and terminology based on user's experience level

Example

Input: "Help me with mqtt specialist for my current situation"

Output:

Based on your situation, here is a structured approach to mqtt specialist:

1. Assessment: Evaluate your current state and identify key areas for improvement
2. Strategy: Develop a targeted plan based on best practices
3. Implementation: Execute the plan with specific, measurable steps
4. Review: Monitor progress and adjust as needed