MQTT = Message Queuing Telemetry Transport, Message Bus for Enterprise Orchestration

Flynn · Sep 24, 2025

MQTT = Message Queuing Telemetry Transport, Message Bus for Enterprise Orchestration

Here’s the nutshell up front, then a complete, commented, runnable mini-stack you can drop onto a Raspberry Pi 5 + an NVIDIA Jetson + a laptop/phone (for smart-glasses/Web UI):

MQTT = Message Queuing Telemetry Transport. It’s a super-light pub/sub protocol over TCP (often on port 1883 or 8883/TLS) designed for unreliable links and tiny devices. Core wins: topics, QoS (0/1/2), retained messages, will messages, tiny headers, and easy auth/TLS.
Why it’s perfect here: it’s the “message bus” that lets your Pi sensors, Jetson vision, voice-to-text ML, and smart glasses publish/subscribe to the same live event fabric. Each component stays decoupled, but the enterprise behaves like one coordinated organism.

Architecture (practical & scalable)

Broker (Eclipse Mosquitto) = the hub. Turn on username/password + TLS, define topic ACLs.
Edge (Raspberry Pi 5) = publishes sensor readings (env, RFID/NFC, door states), subscribes to control commands.
Jetson (Orin) = runs vision models, publishes detections (person/plate/tool), subscribes to inference control.
Voice STT/NLU = transcribes mic audio into intents (“open bay one”, “start exhaust fan 60%”), publishes intents.
Orchestrator = subscribes to intents + detections + sensors, emits deterministic actions and notifications.
Smart Glasses / HUD (Web App) = connects via MQTT over WebSockets, shows alerts/tasks, supports push-to-talk.
Logging/BI (optional) = a timeseries sink (e.g., Influx, SQL via worker) that subscribes to everything.

Topic convention (namespaced, role-aware):

site/{siteId}/edge/{deviceId}/sensor/{name}
site/{siteId}/jetson/{nodeId}/vision/{stream}
site/{siteId}/voice/{deviceId}/intent
site/{siteId}/cmd/{target}/{action}
site/{siteId}/notify/{role}/{channel}

Example: site/sd-shop/edge/pi5-A/sensor/temp, site/sd-shop/voice/glasses-12/intent

Security essentials:

TLS on 8883 + per-client creds, ACLs by prefix (least privilege).
Use retained messages for “last known state” sensors.
LWT (Last Will & Testament) on each client to flip health flags if it drops.

Files & Full Code (commented)

docker-compose.yml
mosquitto.conf
.env.example
requirements.txt
lib/schema.py (shared payload helpers)
services/pi_edge/sensor_publisher.py (Pi 5)
services/jetson_vision/vision_publisher.py (Jetson)
services/voice_transcriber/stt_server.py (voice → intent; Whisper/faster-whisper ready)
services/orchestrator/command_router.py (enterprise rules / routing)
services/glasses_client/glasses.html (HUD over WebSockets; works on a phone or smart-glasses browser)

–––

docker-compose.yml

version: "3.9" services: mqtt: image: eclipse-mosquitto:2 container_name: mqtt restart: unless-stopped ports: - "1883:1883" # MQTT TCP - "9001:9001" # MQTT over WebSocket (for glasses/web) volumes: - ./mosquitto.conf:/mosquitto/config/mosquitto.conf:ro - ./data/mosquitto:/mosquitto/data - ./log/mosquitto:/mosquitto/log # Optional: a lightweight MQTT web console could be added later 

mosquitto.conf

persistence true persistence_location /mosquitto/data/ log_dest file /mosquitto/log/mosquitto.log listener 1883 allow_anonymous false password_file /mosquitto/config/passwd # WebSockets for smart glasses / web HUD listener 9001 protocol websockets # (For production add TLS listeners and point to certs) # listener 8883 # cafile /mosquitto/config/ca.crt # certfile /mosquitto/config/server.crt # keyfile /mosquitto/config/server.key 

.env.example

MQTT_HOST=localhost MQTT_PORT=1883 MQTT_WS=ws://localhost:9001/mqtt MQTT_USER=shopuser MQTT_PASS=shoppass SITE_ID=sd-shop PI_DEVICE_ID=pi5-A JETSON_NODE_ID=orin-1 

requirements.txt

paho-mqtt==1.6.1 faster-whisper==1.0.3 websockets==12.0 uvicorn==0.30.1 fastapi==0.115.0 python-dotenv==1.0.1 opencv-python==4.10.0.84 numpy==1.26.4 

lib/schema.py

# Shared message helpers to keep payloads consistent. # Use simple dicts; easy for MQTT and logging. from datetime import datetime, timezone from typing import Dict, Any def now_iso(): return datetime.now(timezone.utc).isoformat() def sensor_payload(site_id: str, device_id: str, name: str, value: Any, unit: str = "", meta: Dict[str, Any] = None): return { "ts": now_iso(), "siteId": site_id, "deviceId": device_id, "type": "sensor", "name": name, "value": value, "unit": unit, "meta": meta or {} } def detection_payload(site_id: str, node_id: str, stream: str, label: str, score: float, bbox=None, meta=None): return { "ts": now_iso(), "siteId": site_id, "nodeId": node_id, "type": "detection", "stream": stream, "label": label, "score": score, "bbox": bbox or [], "meta": meta or {} } def intent_payload(site_id: str, device_id: str, intent: str, slots=None, transcript:str=""): return { "ts": now_iso(), "siteId": site_id, "deviceId": device_id, "type": "intent", "intent": intent, # e.g., "OPEN_BAY", "SET_FAN", "WHERE_IS" "slots": slots or {}, # e.g., {"bay":"1"} or {"percent":60} "transcript": transcript } def command_payload(site_id: str, target: str, action: str, args=None, reason:str=""): return { "ts": now_iso(), "siteId": site_id, "target": target, # e.g., "edge/pi5-A/fan" or "edge/*/doors/bay1" "action": action, # e.g., "SET", "OPEN", "CLOSE" "args": args or {}, "reason": reason } def notify_payload(site_id: str, role: str, text: str, severity:str="info", meta=None): return { "ts": now_iso(), "siteId": site_id, "role": role, # "tech", "safety", "admin" "text": text, "severity": severity, "meta": meta or {} } 

services/pi_edge/sensor_publisher.py

# Raspberry Pi 5 edge publisher: # - Publishes temperature/humidity and a fake door sensor. # - Subscribes for control commands (e.g., fan setpoint) to show bidirectional flow. import os, json, random, time from dotenv import load_dotenv import paho.mqtt.client as mqtt from lib.schema import sensor_payload load_dotenv() MQTT_HOST = os.getenv("MQTT_HOST","localhost") MQTT_PORT = int(os.getenv("MQTT_PORT","1883")) MQTT_USER = os.getenv("MQTT_USER","shopuser") MQTT_PASS = os.getenv("MQTT_PASS","shoppass") SITE_ID = os.getenv("SITE_ID","sd-shop") DEVICE_ID = os.getenv("PI_DEVICE_ID","pi5-A") # Topic helpers def t_sensor(name): # publishes return f"site/{SITE_ID}/edge/{DEVICE_ID}/sensor/{name}" CMD_TOPIC = f"site/{SITE_ID}/cmd/edge/{DEVICE_ID}/#" fan_setpoint = 0 # pretend we have a PWM fan def on_connect(client, userdata, flags, rc): print("Pi edge connected rc=", rc) client.subscribe(CMD_TOPIC, qos=1) def on_message(client, userdata, msg): global fan_setpoint try: payload = json.loads(msg.payload.decode("utf-8")) if payload.get("action") == "SET" and "percent" in payload.get("args",{}): fan_setpoint = int(payload["args"]["percent"]) print(f"[EDGE] Fan setpoint updated -> {fan_setpoint}% (reason: {payload.get('reason','')})") else: print("[EDGE] Unknown command:", payload) except Exception as e: print("Command parse error:", e) client = mqtt.Client(client_id=f"pi-edge-{DEVICE_ID}", clean_session=True) client.username_pw_set(MQTT_USER, MQTT_PASS) client.on_connect = on_connect client.on_message = on_message client.will_set(f"site/{SITE_ID}/edge/{DEVICE_ID}/status", "offline", retain=True, qos=1) client.connect(MQTT_HOST, MQTT_PORT, 60) client.loop_start() # Mark this node online client.publish(f"site/{SITE_ID}/edge/{DEVICE_ID}/status", "online", retain=True, qos=1) try: while True: # Replace with real sensor reads (e.g., from I2C, SPI, GPIO) temp_c = round(20 + random.uniform(-1.0, 3.0), 2) hum = round(50 + random.uniform(-3.0, 5.0), 2) door = random.choice(["open","closed"]) if random.random()<0.05 else "closed" client.publish(t_sensor("tempC"), json.dumps(sensor_payload(SITE_ID, DEVICE_ID, "tempC", temp_c, "C")), qos=1, retain=True) client.publish(t_sensor("humidity"), json.dumps(sensor_payload(SITE_ID, DEVICE_ID, "humidity", hum, "%")), qos=1) client.publish(t_sensor("bay1Door"), json.dumps(sensor_payload(SITE_ID, DEVICE_ID, "bay1Door", door)), qos=1) # Report our current fan setpoint as a "sensor" for visibility client.publish(t_sensor("fanSetpoint"), json.dumps(sensor_payload(SITE_ID, DEVICE_ID, "fanSetpoint", fan_setpoint, "%")), qos=1, retain=True) time.sleep(3) except KeyboardInterrupt: pass finally: client.loop_stop() client.disconnect() 

services/jetson_vision/vision_publisher.py

# Jetson Orin vision publisher: # - Simulates detections from a camera stream. # - Replace simulate_detections() with your actual CV/DeepStream/YOLO code. import os, json, time, random from dotenv import load_dotenv import paho.mqtt.client as mqtt from lib.schema import detection_payload load_dotenv() MQTT_HOST = os.getenv("MQTT_HOST","localhost") MQTT_PORT = int(os.getenv("MQTT_PORT","1883")) MQTT_USER = os.getenv("MQTT_USER","shopuser") MQTT_PASS = os.getenv("MQTT_PASS","shoppass") SITE_ID = os.getenv("SITE_ID","sd-shop") NODE_ID = os.getenv("JETSON_NODE_ID","orin-1") STREAM = "bay1-cam" def topic(): return f"site/{SITE_ID}/jetson/{NODE_ID}/vision/{STREAM}" client = mqtt.Client(client_id=f"jetson-{NODE_ID}", clean_session=True) client.username_pw_set(MQTT_USER, MQTT_PASS) client.will_set(f"site/{SITE_ID}/jetson/{NODE_ID}/status", "offline", retain=True, qos=1) client.connect(MQTT_HOST, MQTT_PORT, 60) client.loop_start() client.publish(f"site/{SITE_ID}/jetson/{NODE_ID}/status", "online", retain=True, qos=1) labels = ["person", "forklift", "no-hardhat", "spill", "tool"] try: while True: label = random.choice(labels) score = round(random.uniform(0.55, 0.99), 2) det = detection_payload(SITE_ID, NODE_ID, STREAM, label, score, bbox=[100,120,200,250]) client.publish(topic(), json.dumps(det), qos=1) time.sleep(5) except KeyboardInterrupt: pass finally: client.loop_stop() client.disconnect() 

services/voice_transcriber/stt_server.py

# Voice → Intent service: # - Accepts short WAV/MP3 uploads via FastAPI endpoint /api/transcribe # - Transcribes with faster-whisper (local, Jetson-friendly if CUDA available) # - Very simple intent parser turns transcript into {intent, slots} # - Publishes to MQTT for the orchestrator. import os, io, json from dotenv import load_dotenv from fastapi import FastAPI, UploadFile, File import uvicorn import paho.mqtt.client as mqtt from faster_whisper import WhisperModel from lib.schema import intent_payload load_dotenv() MQTT_HOST = os.getenv("MQTT_HOST","localhost") MQTT_PORT = int(os.getenv("MQTT_PORT","1883")) MQTT_USER = os.getenv("MQTT_USER","shopuser") MQTT_PASS = os.getenv("MQTT_PASS","shoppass") SITE_ID = os.getenv("SITE_ID","sd-shop") DEVICE_ID = "stt-1" # could be the ID of the glasses or mic device # Load lightweight model first; swap to medium/large as hardware allows model = WhisperModel("small", device="auto", compute_type="auto") client = mqtt.Client(client_id="stt-server", clean_session=True) client.username_pw_set(MQTT_USER, MQTT_PASS) client.connect(MQTT_HOST, MQTT_PORT, 60) client.loop_start() app = FastAPI() def topic_intent(dev_id): return f"site/{SITE_ID}/voice/{dev_id}/intent" def parse_intent(text: str): t = text.lower() # naive parser to prove the pipeline; replace with Rasa/LUIS/Transformers later if "open bay" in t: return "OPEN_BAY", {"bay": "".join([c for c in t if c.isdigit()]) or "1"} if "close bay" in t: return "CLOSE_BAY", {"bay": "".join([c for c in t if c.isdigit()]) or "1"} if "set fan" in t or "fan" in t: # extract percent import re m = re.search(r"(\d+)\s*%", t) pct = int(m.group(1)) if m else 50 return "SET_FAN", {"percent": pct} if "where is" in t: # e.g., "where is torque wrench" item = t.split("where is")[-1].strip() return "WHERE_IS", {"item": item} return "UNKNOWN", {} @app.post("/api/transcribe") async def transcribe(file: UploadFile = File(...), deviceId: str = "glasses-00"): audio_bytes = await file.read() segments, info = model.transcribe(io.BytesIO(audio_bytes), beam_size=1) transcript = " ".join([seg.text for seg in segments]).strip() intent, slots = parse_intent(transcript) payload = intent_payload(SITE_ID, deviceId, intent, slots, transcript) client.publish(topic_intent(deviceId), json.dumps(payload), qos=1) return {"intent": intent, "slots": slots, "transcript": transcript} if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8088) 

services/orchestrator/command_router.py

# Orchestrator: # - Subscribes to intents, detections, and key sensors. # - Applies rules → emits commands & human notifications. # - This is where "enterprise behavior" lives; expand with a rules engine later. import os, json from dotenv import load_dotenv import paho.mqtt.client as mqtt from lib.schema import command_payload, notify_payload load_dotenv() MQTT_HOST = os.getenv("MQTT_HOST","localhost") MQTT_PORT = int(os.getenv("MQTT_PORT","1883")) MQTT_USER = os.getenv("MQTT_USER","shopuser") MQTT_PASS = os.getenv("MQTT_PASS","shoppass") SITE_ID = os.getenv("SITE_ID","sd-shop") INTENTS = f"site/{SITE_ID}/voice/+/intent" DETECT = f"site/{SITE_ID}/jetson/+/vision/+" SENSORS = f"site/{SITE_ID}/edge/+/sensor/+" client = mqtt.Client(client_id="orchestrator", clean_session=True) client.username_pw_set(MQTT_USER, MQTT_PASS) def publish_cmd(target: str, action: str, args=None, reason=""): topic = f"site/{SITE_ID}/cmd/{target}" payload = command_payload(SITE_ID, target, action, args or {}, reason) client.publish(topic, json.dumps(payload), qos=1) def notify(role: str, text: str, severity="info", meta=None): topic = f"site/{SITE_ID}/notify/{role}/general" payload = notify_payload(SITE_ID, role, text, severity, meta) client.publish(topic, json.dumps(payload), qos=0) def on_connect(c,u,f,rc): print("Orchestrator connected rc=", rc) c.subscribe(INTENTS, qos=1) c.subscribe(DETECT, qos=1) c.subscribe(SENSORS, qos=0) def handle_intent(payload): intent = payload.get("intent") slots = payload.get("slots", {}) dev = payload.get("deviceId","unknown") if intent == "OPEN_BAY": bay = slots.get("bay","1") publish_cmd(f"edge/+/doors/bay{bay}", "OPEN", reason=f"Voice by {dev}") notify("tech", f"Opening Bay {bay} (via {dev})") elif intent == "CLOSE_BAY": bay = slots.get("bay","1") publish_cmd(f"edge/+/doors/bay{bay}", "CLOSE", reason=f"Voice by {dev}") notify("tech", f"Closing Bay {bay} (via {dev})") elif intent == "SET_FAN": pct = int(slots.get("percent",50)) publish_cmd("edge/pi5-A/fan", "SET", args={"percent": pct}, reason=f"Voice by {dev}") notify("tech", f"Setting exhaust fan to {pct}% (via {dev})") elif intent == "WHERE_IS": item = slots.get("item","unknown tool") # In a real system you'd query inventory/RTLS. Here we notify. notify("tech", f"Last seen location for {item}: Tool Cage A") else: notify("tech", f"Unrecognized voice command from {dev}: {payload.get('transcript','')}", "warning") def handle_detection(payload): label = payload.get("label") score = payload.get("score",0) if label == "no-hardhat" and score > 0.7: notify("safety", "Hardhat missing detected at Bay 1", "warning", {"score": score}) if label == "spill" and score > 0.6: publish_cmd("edge/+/beacon/floor", "SET", args={"color":"yellow","blink":True}) notify("safety", "Spill detected near Bay 1. Beacon activated.", "critical") def handle_sensor(topic, payload): # Example: auto-open fans if temp too high if topic.endswith("/sensor/tempC"): val = payload.get("value", 0) if val > 28: publish_cmd("edge/pi5-A/fan", "SET", args={"percent": 70}, reason="Temp > 28C") notify("tech", "Auto cooling: fan set to 70% (temp high)") # Door status could drive geofencing, etc. def on_message(c,u,msg): try: payload = json.loads(msg.payload.decode("utf-8")) except: print("Bad JSON on", msg.topic); return topic = msg.topic if "/voice/" in topic and topic.endswith("/intent"): handle_intent(payload) elif "/vision/" in topic: handle_detection(payload) elif "/sensor/" in topic: handle_sensor(topic, payload) client.on_connect = on_connect client.on_message = on_message client.connect(MQTT_HOST, MQTT_PORT, 60) client.loop_forever() 

services/glasses_client/glasses.html

 <!DOCTYPE html> <html> <head> <meta charset="utf-8" /> <title>Empire HUD</title> <meta name="viewport" content="width=device-width, initial-scale=1" /> <style> body { font-family: system-ui, sans-serif; background:#0a0f18; color:#eaf6ff; padding:16px; } .card { background:#0d1624; border:1px solid #113a59; border-radius:16px; padding:16px; margin-bottom:12px; box-shadow:0 0 20px rgba(0,204,255,0.1); } input, button { font-size:16px; padding:10px; border-radius:10px; border:1px solid #113a59; background:#09101a; color:#eaf6ff; } button { cursor: pointer; } .notif { border-left:4px solid #3fb; padding-left:12px; } </style> </head> <body> <div class="card"> <h2>Empire HUD</h2> <div>Status: disconnected</div> <div>MQTT over WS: <code id="ws"></code></div> </div> <div class="card"> <h3>Quick Command (demo)</h3> Examples: "open bay 1", "set fan 60%", "where is torque wrench" <input id="cmd" placeholder="type a command…" /> <button onclick="sendIntent()">Send</button> </div> <div class="card"> <h3>Notifications</h3> <div id="feed"></div> </div>  <script src="https://unpkg.com/[email protected]/paho-mqtt-min.js"></script> <script> const SITE_ID = 'sd-shop'; const DEVICE_ID = 'glasses-12'; const WS_URL = (new URLSearchParams(location.search)).get('ws') || 'ws://localhost:9001/mqtt'; document.getElementById('ws').textContent = WS_URL; const client = new Paho.MQTT.Client(WS_URL, "hud-" + Math.random().toString(16).slice(2)); client.onConnectionLost = (resp) => { document.getElementById('status').textContent = "lost"; }; client.onMessageArrived = (msg) => { const el = document.createElement('div'); el.className = 'notif'; try { const p = JSON.parse(msg.payloadString); el.textContent = `[${p.severity}] ${p.text}`; } catch { el.textContent = msg.payloadString; } document.getElementById('feed').prepend(el); }; client.connect({ timeout: 5, onSuccess: () => { document.getElementById('status').textContent = "connected"; client.subscribe(`site/${SITE_ID}/notify/+/+`, {qos:0}); }, onFailure: () => document.getElementById('status').textContent = "failed", useSSL: false, // set true when you enable TLS for WS userName: "shopuser", // match mosquitto credentials password: "shoppass" }); function sendIntent(){ const text = document.getElementById('cmd').value || ''; if(!text) return; const payload = { ts: new Date().toISOString(), siteId: SITE_ID, deviceId: DEVICE_ID, type: "intent", intent: "FREE_TEXT", slots: {}, transcript: text }; // Let the orchestrator handle unknowns or add a tiny local parser here const msg = new Paho.MQTT.Message(JSON.stringify

MQTT = Message Queuing Telemetry Transport, Message Bus for Enterprise Orchestration

Architecture (practical & scalable)​

Files & Full Code (commented)​

Similar threads

Architecture (practical & scalable)

Files & Full Code (commented)