MQTT to REST API Flask Tutorial: Expose Live Device Data over HTTP

Building an MQTT to REST API with Flask lets any application that speaks HTTP read live device data and send commands, without learning MQTT. In this tutorial you will write a Flask app that subscribes to device topics, keeps the latest value for each device in memory, exposes it through JSON endpoints, and publishes commands from a POST route. You will also learn the two deployment mistakes that silently break most Flask and MQTT combinations.

Why Put a REST API in Front of MQTT

MQTT is the right protocol for devices: persistent connections, small packets and push delivery. Many consumers of that data are not devices, though. A mobile app's backend, an ERP integration, a reporting script or a partner system usually expects request and response over HTTP. Rather than teaching every one of them MQTT, a thin bridge translates:

The bridge is a "last value cache". It is not a historian: it answers "what is the value now?", not "what was it last Tuesday?". For history, store the stream in a database alongside the cache.

What You Need

• Python 3.9+
• pip install "flask>=3.0" "paho-mqtt>=2.0"
• A broker. For development, use the free public MQTT broker broker.connectmq.com with a random topic prefix; for production, a private broker with credentials.

The Flask MQTT Bridge

Save this as app.py:

import json
import os
import threading
import time
import uuid
from functools import wraps

import paho.mqtt.client as mqtt
from flask import Flask, abort, jsonify, request

BROKER = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", "1883"))
USERNAME = os.environ.get("MQTT_USERNAME")
PASSWORD = os.environ.get("MQTT_PASSWORD")
CLIENT_ID = os.environ.get("MQTT_CLIENT_ID", f"flask-bridge-{uuid.uuid4().hex[:8]}")
PREFIX = os.environ.get("TOPIC_PREFIX", "connectmq/demo/flask-3391")
API_KEY = os.environ.get("API_KEY", "change-me")
STALE_AFTER = int(os.environ.get("STALE_AFTER", "120"))  # seconds

app = Flask(__name__)
cache = {}                 # device -> {"data": ..., "topic": ..., "received_at": ...}
cache_lock = threading.Lock()
mqtt_connected = threading.Event()


# ---------- MQTT ----------

def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"MQTT connect failed: {reason_code}")
        return
    mqtt_connected.set()
    client.subscribe(f"{PREFIX}/+/state", qos=1)


def on_disconnect(client, userdata, flags, reason_code, properties):
    mqtt_connected.clear()


def on_message(client, userdata, msg):
    device = msg.topic.split("/")[-2]
    try:
        data = json.loads(msg.payload)
    except (json.JSONDecodeError, UnicodeDecodeError):
        data = msg.payload.decode("utf-8", errors="replace")
    with cache_lock:
        cache[device] = {"data": data, "topic": msg.topic, "received_at": time.time()}


mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
if USERNAME:
    mqttc.username_pw_set(USERNAME, PASSWORD)
if PORT == 8883:
    mqttc.tls_set()
mqttc.on_connect = on_connect
mqttc.on_disconnect = on_disconnect
mqttc.on_message = on_message
mqttc.connect_async(BROKER, PORT, keepalive=60)
mqttc.loop_start()


# ---------- HTTP ----------

def require_api_key(view):
    @wraps(view)
    def wrapper(*args, **kwargs):
        if request.headers.get("X-API-Key") != API_KEY:
            abort(401)
        return view(*args, **kwargs)
    return wrapper


def serialize(device, entry):
    age = time.time() - entry["received_at"]
    return {
        "device": device,
        "data": entry["data"],
        "topic": entry["topic"],
        "age_seconds": round(age, 1),
        "stale": age > STALE_AFTER,
    }


@app.get("/health")
def health():
    ok = mqtt_connected.is_set()
    return jsonify({"mqtt_connected": ok}), (200 if ok else 503)


@app.get("/api/devices")
@require_api_key
def list_devices():
    with cache_lock:
        items = [serialize(d, e) for d, e in sorted(cache.items())]
    return jsonify(items)


@app.get("/api/devices/<device>")
@require_api_key
def get_device(device):
    with cache_lock:
        entry = cache.get(device)
        if entry is None:
            abort(404)
        return jsonify(serialize(device, entry))


@app.post("/api/devices/<device>/command")
@require_api_key
def send_command(device):
    if any(ch in device for ch in "/+#"):
        abort(400)
    body = request.get_json(silent=True)
    if not isinstance(body, dict) or "action" not in body:
        return jsonify({"error": "JSON body with an 'action' field is required"}), 400
    if not mqtt_connected.is_set():
        return jsonify({"error": "MQTT broker not connected"}), 503

    body["request_id"] = uuid.uuid4().hex
    info = mqttc.publish(f"{PREFIX}/{device}/cmd", json.dumps(body), qos=1)
    try:
        info.wait_for_publish(timeout=5)
    except (ValueError, RuntimeError):
        pass
    if not info.is_published():
        return jsonify({"error": "publish not acknowledged"}), 504
    return jsonify({"published": True, "request_id": body["request_id"]}), 202


if __name__ == "__main__":
    # use_reloader=False is essential: the reloader would start a second MQTT client
    app.run(host="0.0.0.0", port=5000, debug=True, use_reloader=False)

Key Implementation Details

• The MQTT client starts at import time with connect_async and loop_start, so it runs in its own thread whether you start the app with python app.py or a WSGI server. connect_async also means the app boots even if the broker is briefly unreachable; paho keeps retrying.
• A lock protects the cache. paho writes from its network thread while Flask reads from request threads.
• The POST route returns 202 Accepted, not 200. The broker acknowledged the message (QoS 1 PUBACK), but that does not mean the device executed it. If you need confirmation, have the device publish a result on a reply topic with the same request_id, or use the MQTT 5 response topic pattern described in MQTT 5 features.
• Device IDs are validated so a caller cannot inject +, # or extra topic levels.

Try the MQTT to REST API Flask Bridge

Start the app:

export API_KEY=dev-secret
python app.py

Publish a retained state message as a device would. Retained messages are useful here: when the bridge restarts, the broker immediately delivers the last state of every device, so the cache is warm without waiting for the next report.

mosquitto_pub -h broker.connectmq.com -p 1883 -i flask-dev-3391 -r -q 1 \
  -t connectmq/demo/flask-3391/valve7/state -m '{"open": true, "flow": 12.4}'

Query and command through HTTP:

curl -s -H "X-API-Key: dev-secret" http://localhost:5000/api/devices
curl -s -H "X-API-Key: dev-secret" http://localhost:5000/api/devices/valve7
curl -s -X POST -H "X-API-Key: dev-secret" -H "Content-Type: application/json" \
  -d '{"action": "close"}' http://localhost:5000/api/devices/valve7/command

Watch the command arrive with mosquitto_sub -h broker.connectmq.com -p 1883 -i flask-sub-3391 -v -t "connectmq/demo/flask-3391/+/cmd". Because the public area is readable by everyone, clear the retained test message when you are done by publishing an empty retained payload to the same topic.

Deployment Pitfalls

Most broken Flask and MQTT setups fail for one of these reasons:

1. The debug reloader runs your code twice. Flask's reloader spawns a child process, so two MQTT clients connect with the same client ID and disconnect each other in a loop. Keep use_reloader=False.
2. Multiple WSGI workers. gunicorn -w 4 imports the app four times: four MQTT clients and four separate caches, so consecutive requests may return different values. Run one worker with threads instead: gunicorn -w 1 --threads 8 app:app.
3. Fixed client IDs across instances. If you scale horizontally behind a load balancer, give each instance a unique client ID. On a broker where client IDs are tied to credentials, issue separate credentials per instance.
4. Treating the cache as truth. A device that went offline still has a cached value. Expose age_seconds and stale, as the code does, so consumers can decide.
5. No authentication. Anything that can POST can operate equipment. The API key header is a minimum; put the app behind HTTPS and a proper auth layer for anything internet-facing.

Extending the Bridge

Once the basic endpoints work, a few additions cover most real-world requests:

• Server-Sent Events for live updates. Polling GET /api/devices every second works, but a GET /api/stream endpoint that yields an event whenever the cache changes lets HTTP clients receive pushes without speaking MQTT. Flask supports streaming responses through a generator; give each connected client its own queue.Queue and put updates into all of them from on_message.
• Filtering and projection. Query parameters like ?stale=false or ?fields=temperature keep responses small for mobile clients.
• History. Write each message to SQLite or a time-series database inside on_message, and add GET /api/devices/<id>/history?since=.... Keep the in-memory cache for the latest value, because it is far faster than querying the database for every request.
• OpenAPI documentation. If other teams consume the API, describe the endpoints in an OpenAPI file so they can generate clients and know exactly which fields to expect.

Each addition stays inside the same pattern: one MQTT client per process, a lock around shared state, and HTTP handlers that never block on the broker.

Moving to a Private Broker

For production, connect the bridge to a private broker with TLS on port 8883. In the ConnectMQ portal, create a broker and copy the generated username, password and client ID into MQTT_USERNAME, MQTT_PASSWORD and MQTT_CLIENT_ID. A cloud MQTT broker like this removes the need to run Mosquitto next to Flask, and the portal's live topic explorer helps verify what the cache should contain.

On the ConnectMQ Serverless (Free) plan, QoS goes up to 1, which is exactly what the bridge uses for state and commands. The Paid Serverless plan adds QoS 2 and shared subscriptions, the latter being handy if you later run several bridge instances that should split incoming traffic.

Related Integrations

This bridge belongs to a family of small subscriber services. Send MQTT data to Google Sheets covers spreadsheet reporting, and MQTT email alerts with Python covers notifications. If your consumers run in a browser, you can skip REST entirely and connect them over WebSocket, as explained in MQTT over WebSocket. For other languages, see the guide to MQTT client libraries. Flask itself is documented at flask.palletsprojects.com.

Get Started with ConnectMQ

Put your Flask bridge on a private, TLS-secured broker in minutes. Sign up at app.connectmq.com, choose the free Serverless plan with no credit card required, create a broker and paste the credentials into the bridge's environment. Use the in-browser MQTT client to publish device states while you test the endpoints.

Frequently Asked Questions

Why not use Flask-MQTT?

You can, but it is a thin wrapper that still faces the same reloader and multi-worker issues, and you must check which paho-mqtt versions it supports. Using paho-mqtt 2.x directly keeps you in control of threading and the client lifecycle.

Can the REST API return historical data?

Not with an in-memory cache. Store messages in a database such as SQLite, PostgreSQL or a time-series store from the same subscriber, and add query endpoints that read from it.

How do I know a command was executed?

The 202 response only confirms the broker accepted the message. Have the device publish an acknowledgement containing the request_id, and expose it through another endpoint or a status field in the device's state.

Can I use FastAPI instead of Flask?

Yes. The structure is the same: one MQTT client per process, a locked cache and routes that read it. With FastAPI you may prefer an async client such as aiomqtt.