Search the docs, videos, blog, apps, teams, skills and release notes.
At a glance
What it is
Guides MQTT message broker architecture including topic design, QoS levels, retained messages, security configuration, and client implementation patterns.
A skill is a written procedure a bot loads when a job calls for it. This one is a single file, SKILL.md, and the whole file is on this page.
The whole file, exactly as it downloads. Read it before you rely on it.
SKILL.md
517 lines · 1,888 words
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.
Use this skill when:
User asks about mqtt specialist techniques or best practices
Guides MQTT message broker architecture including topic design, QoS levels, retained messages, security configuration, and client implementation patterns.
Nothing. It ships with Murage, which is open source. Your bots run on the AI plan you already have.
Yes. Open it in Settings → Skills to read, edit or duplicate it. Skill Guard scans it again after every edit.
Download the app, connect the AI you already pay for, and tell your Chief of Staff what needs doing. Plan on about ten minutes from install to a working team.
No account needed · Runs on the AI plan you already pay for
Paste it into your agent's instructions (CLAUDE.md, AGENTS.md or a custom GPT), or add it to a bot in Murage.
Ask Ember for the bot, then switch the skill on in its settings.
Step 1
Send your Chief of Staff this message. It proposes the bot and its role first, and creates it when you say go.
Send this to Ember
“Create a bot that uses the MQTT Specialist skill, and show me the plan first.”
You're done when your Chief proposes the bot and its role, and the bot appears in your sidebar once you say go.
Step 2
You're done when you see the bot’s skills and the library.
Step 3
You're done when the skill shows as on for that bot.
Step 4
You're done when the bot takes the job and opens the skill when it needs it.
Prefer to do it yourself? Choose + in the sidebar, then New Bot.Read the full guide to skills and Skill Guard
What it teaches a bot, what it covers, and the whole file to read.
The principle
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.
What it covers
When to Use
MQTT Protocol Fundamentals
Topic Design Architecture
Broker Configuration
Client Implementation Patterns
MQTT 5.0 Features
Monitoring and Debugging
Common Pitfalls
Exercises
Process
Output Format
Edge Cases
Example
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
| 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 | 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.
# 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| 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 |
# 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# [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]# [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/##!/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[str, tuple[int, Callable]] = {}
self._setup_callbacks()
self._setup_auth()
def _setup_callbacks(self):
self.client.on_connect = self._on_connect
self.client.on_disconnect = self._on_disconnect
self.client.on_message = self._on_message
self.client.on_subscribe = self._on_subscribe
def _setup_auth(self):
cfg = self.config
if cfg.username:
self.client.username_pw_set(cfg.username, cfg.password)
if cfg.ca_cert:
tls_ctx = ssl.create_default_context(cafile=cfg.ca_cert)
if cfg.client_cert:
tls_ctx.load_cert_chain(cfg.client_cert, cfg.client_key)
self.client.tls_set_context(tls_ctx)
def _on_connect(self, client, userdata, flags, rc):
if rc == 0:
logger.info("Connected to broker")
# Re-subscribe on reconnection
for topic, (qos, _) in self._subscriptions.items():
client.subscribe(topic, qos)
logger.info(f"Re-subscribed: {topic}")
else:
codes = {
1: "Incorrect protocol",
2: "Invalid client ID",
3: "Server unavailable",
4: "Bad credentials",
5: "Not authorized"
}
logger.error(f"Connect failed: {codes.get(rc, f'Unknown ({rc})')}")
def _on_disconnect(self, client, userdata, rc):
if rc != 0:
logger.warning(f"Unexpected disconnect (rc={rc}), reconnecting...")
def _on_message(self, client, userdata, msg):
for pattern, (_, handler) in self._subscriptions.items():
if mqtt.topic_matches_sub(pattern, msg.topic):
try:
handler(msg.topic, msg.payload)
except Exception as e:
logger.error(f"Handler error for {msg.topic}: {e}")
def _on_subscribe(self, client, userdata, mid, granted_qos):
logger.debug(f"Subscribed (mid={mid}, qos={granted_qos})")
def connect(self):
"""Connect with automatic reconnection."""
self.client.reconnect_delay_set(min_delay=1, max_delay=120)
self.client.connect_async(
self.config.broker,
self.config.port,
self.config.keepalive
)
self.client.loop_start()
def subscribe(self, topic: str, qos: int, handler: Callable):
"""Subscribe with handler that persists across reconnections."""
self._subscriptions[topic] = (qos, handler)
self.client.subscribe(topic, qos)
def publish(self, topic: str, payload: dict, qos: int = 1,
retain: bool = False):
"""Publish JSON payload."""
msg = json.dumps(payload)
result = self.client.publish(topic, msg, qos=qos, retain=retain)
if result.rc != mqtt.MQTT_ERR_SUCCESS:
logger.error(f"Publish failed: {mqtt.error_string(result.rc)}")
return result
def disconnect(self):
self.client.loop_stop()
self.client.disconnect()# 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 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)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)# 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# 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 countdef 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()| 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 |
Topic Design: Design a complete topic hierarchy for a 3-floor smart building with HVAC, lighting, and occupancy sensors
QoS Comparison: Publish 1000 messages at each QoS level, measure delivery time, message loss, and broker load
Retained Config: Implement a device configuration system using retained messages where devices read config on boot
LWT Monitor: Build a fleet status monitor using LWT that displays online/offline status for 10 simulated devices
Request/Response: Implement a command-response pattern where a dashboard requests current sensor readings from specific devices
Gather information. Ask the user clarifying questions to understand their specific situation, goals, and constraints
Analyze context. Review the information provided and identify key factors relevant to mqtt specialist
Develop recommendations. Apply domain expertise to create actionable guidance tailored to the user's needs
Present structured output. Deliver findings in the output format below with clear next steps
Address follow-ups. Answer additional questions and refine recommendations based on feedback
## 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]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
Input: "Help me with mqtt specialist for my current situation"
Output:
Based on your situation, here is a structured approach to mqtt specialist:
Assessment: Evaluate your current state and identify key areas for improvement
Strategy: Develop a targeted plan based on best practices
Implementation: Execute the plan with specific, measurable steps
Review: Monitor progress and adjust as needed
SKILL.md
517 lines · 1,888 words
The text of SKILL.md is shown as it downloads, under the Apache-2.0 license, by Ferrox Labs. The license is the one in the file's own frontmatter.
Your new bot
MQTT Specialist bot
Waiting in your sidebar once you say go.
No account needed · Runs on the AI plan you already pay for