MQTT vers API REST avec Flask : tutoriel Python

Construire une passerelle MQTT vers API REST avec Flask permet à toute application qui parle HTTP de lire les données des appareils en direct et d'envoyer des commandes, sans avoir à apprendre MQTT. Dans ce tutoriel, vous allez écrire une application Flask qui s'abonne aux topics des appareils, conserve en mémoire la dernière valeur de chacun, l'expose via des endpoints JSON et publie des commandes depuis une route POST. Vous découvrirez aussi les deux erreurs de déploiement qui cassent silencieusement la plupart des combinaisons Flask et MQTT.

Pourquoi placer une API REST devant MQTT

MQTT est le bon protocole pour les appareils : connexions persistantes, paquets légers et livraison en push. Pourtant, beaucoup de consommateurs de ces données ne sont pas des appareils. Le backend d'une application mobile, une intégration ERP, un script de reporting ou le système d'un partenaire attendent généralement un échange requête-réponse en HTTP. Plutôt que d'apprendre MQTT à chacun d'eux, une passerelle légère fait la traduction :

Le consommateur veut	Appel REST	Ce que fait la passerelle
L'état actuel de tous les appareils	`GET /api/devices`	Renvoie le dernier message en cache pour chaque appareil
La dernière mesure d'un appareil	`GET /api/devices/<id>`	Renvoie une entrée du cache avec son âge
Envoyer une commande	`POST /api/devices/<id>/command`	Publie du JSON sur le topic de commande de l'appareil
Vérification de santé	`GET /health`	Indique si MQTT est connecté

La passerelle est un « cache de dernière valeur ». Ce n'est pas un historian : elle répond à « quelle est la valeur maintenant ? », pas à « quelle était-elle mardi dernier ? ». Pour l'historique, stockez le flux dans une base de données en parallèle du cache.

Ce dont vous avez besoin

Python 3.9+
pip install "flask>=3.0" "paho-mqtt>=2.0"
Un broker. Pour le développement, utilisez le serveur MQTT public gratuit broker.connectmq.com avec un préfixe de topic aléatoire ; pour la production, un broker MQTT managé privé avec identifiants.

La passerelle Flask MQTT

Enregistrez ce fichier sous le nom 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"))  # secondes

app = Flask(__name__)
cache = {}                 # appareil -> {"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 est indispensable : le reloader démarrerait un second client MQTT
    app.run(host="0.0.0.0", port=5000, debug=True, use_reloader=False)

Points clés de l'implémentation

Le client MQTT démarre à l'import avec connect_async et loop_start, il tourne donc dans son propre thread, que vous lanciez l'application avec python app.py ou avec un serveur WSGI. connect_async permet aussi à l'application de démarrer même si le broker est brièvement injoignable ; paho continue de réessayer.
Un verrou protège le cache. paho écrit depuis son thread réseau pendant que Flask lit depuis les threads de requête.
La route POST renvoie 202 Accepted, pas 200. Le broker a accusé réception du message (PUBACK en QoS 1), mais cela ne signifie pas que l'appareil l'a exécuté. Si vous avez besoin d'une confirmation, faites publier par l'appareil un résultat sur un topic de réponse avec le même request_id, ou utilisez le pattern de response topic de MQTT 5 décrit dans les fonctionnalités de MQTT 5.
Les identifiants d'appareil sont validés afin qu'un appelant ne puisse pas injecter +, # ou des niveaux de topic supplémentaires.

Tester la passerelle MQTT vers API REST Flask

Démarrez l'application :

export API_KEY=dev-secret
python app.py

Publiez un message d'état retained, comme le ferait un appareil. Les retained messages sont utiles ici : quand la passerelle redémarre, le broker livre immédiatement le dernier état de chaque appareil, et le cache est donc chaud sans attendre le prochain rapport.

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}'

Interrogez et commandez via 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

Observez l'arrivée de la commande avec mosquitto_sub -h broker.connectmq.com -p 1883 -i flask-sub-3391 -v -t "connectmq/demo/flask-3391/+/cmd". Comme l'espace public est lisible par tout le monde, effacez le message de test retained une fois terminé en publiant un payload retained vide sur le même topic.

Pièges de déploiement

La plupart des configurations Flask et MQTT défaillantes échouent pour l'une de ces raisons :

Le reloader de debug exécute votre code deux fois. Le reloader de Flask lance un processus enfant : deux clients MQTT se connectent alors avec le même client ID et se déconnectent mutuellement en boucle. Gardez use_reloader=False.
Plusieurs workers WSGI. gunicorn -w 4 importe l'application quatre fois : quatre clients MQTT et quatre caches distincts, si bien que des requêtes consécutives peuvent renvoyer des valeurs différentes. Utilisez plutôt un seul worker avec des threads : gunicorn -w 1 --threads 8 app:app.
Des client ID fixes entre instances. Si vous faites du scaling horizontal derrière un load balancer, donnez à chaque instance un client ID unique. Sur un broker où les client ID sont liés aux identifiants, délivrez des identifiants distincts par instance.
Considérer le cache comme la vérité. Un appareil passé hors ligne possède toujours une valeur en cache. Exposez age_seconds et stale, comme le fait le code, pour que les consommateurs puissent décider.
Aucune authentification. Tout ce qui peut envoyer un POST peut piloter des équipements. L'en-tête de clé d'API est un minimum ; placez l'application derrière HTTPS et une véritable couche d'authentification dès qu'elle est exposée sur Internet.

Étendre la passerelle

Une fois les endpoints de base fonctionnels, quelques ajouts couvrent la plupart des besoins réels :

Server-Sent Events pour les mises à jour en direct. Interroger GET /api/devices chaque seconde fonctionne, mais un endpoint GET /api/stream qui émet un événement à chaque changement du cache permet aux clients HTTP de recevoir des notifications push sans parler MQTT. Flask prend en charge les réponses en streaming via un générateur ; donnez à chaque client connecté sa propre queue.Queue et alimentez-les toutes depuis on_message.
Filtrage et projection. Des paramètres de requête comme ?stale=false ou ?fields=temperature gardent des réponses compactes pour les clients mobiles.
Historique. Écrivez chaque message dans SQLite ou une base de séries temporelles depuis on_message, et ajoutez GET /api/devices/<id>/history?since=.... Conservez le cache en mémoire pour la dernière valeur, car il est bien plus rapide que d'interroger la base à chaque requête.
Documentation OpenAPI. Si d'autres équipes consomment l'API, décrivez les endpoints dans un fichier OpenAPI pour qu'elles puissent générer des clients et sachent exactement quels champs attendre.

Chaque ajout reste dans le même schéma : un client MQTT par processus, un verrou autour de l'état partagé et des handlers HTTP qui ne bloquent jamais sur le broker.

Passer à un broker privé

En production, connectez la passerelle à un broker privé avec TLS sur le port 8883. Dans le portail ConnectMQ, créez un broker et copiez le nom d'utilisateur, le mot de passe et le client ID générés dans MQTT_USERNAME, MQTT_PASSWORD et MQTT_CLIENT_ID. Un broker MQTT cloud comme celui-ci vous évite de faire tourner Mosquitto à côté de Flask, et l'explorateur de topics en direct du portail aide à vérifier ce que le cache devrait contenir.

Avec l'offre ConnectMQ Serverless (Free), la QoS va jusqu'à 1, ce qui correspond exactement à ce que la passerelle utilise pour les états et les commandes. L'offre Serverless (Paid) ajoute la QoS 2 et les shared subscriptions, ces dernières étant pratiques si vous exécutez plus tard plusieurs instances de la passerelle qui doivent se répartir le trafic entrant.

Intégrations associées

Cette passerelle fait partie d'une famille de petits services abonnés. Envoyer des données MQTT vers Google Sheets couvre le reporting dans un tableur, et les alertes e-mail MQTT avec Python couvre les notifications. Si vos consommateurs tournent dans un navigateur, vous pouvez vous passer complètement de REST et les connecter en WebSocket, comme expliqué dans MQTT over WebSocket. Pour d'autres langages, consultez le guide des bibliothèques clientes MQTT. Flask est lui-même documenté sur flask.palletsprojects.com.

Démarrer avec ConnectMQ

Placez votre passerelle Flask sur un broker privé sécurisé par TLS en quelques minutes. Inscrivez-vous sur app.connectmq.com, choisissez l'offre Serverless gratuite sans carte bancaire, créez un broker et collez les identifiants dans l'environnement de la passerelle. Utilisez le client MQTT intégré au navigateur pour publier des états d'appareils pendant que vous testez les endpoints.

Questions fréquentes

Pourquoi ne pas utiliser Flask-MQTT ?

C'est possible, mais il s'agit d'une surcouche légère qui rencontre les mêmes problèmes de reloader et de workers multiples, et vous devez vérifier quelles versions de paho-mqtt elle prend en charge. Utiliser directement paho-mqtt 2.x vous laisse la maîtrise du threading et du cycle de vie du client.

L'API REST peut-elle renvoyer des données historiques ?

Pas avec un cache en mémoire. Stockez les messages dans une base de données comme SQLite, PostgreSQL ou une base de séries temporelles depuis le même abonné, et ajoutez des endpoints de requête qui la lisent.

Comment savoir qu'une commande a été exécutée ?

La réponse 202 confirme seulement que le broker a accepté le message. Faites publier par l'appareil un accusé de réception contenant le request_id, et exposez-le via un autre endpoint ou un champ de statut dans l'état de l'appareil.

Puis-je utiliser FastAPI au lieu de Flask ?

Oui. La structure est la même : un client MQTT par processus, un cache protégé par un verrou et des routes qui le lisent. Avec FastAPI, vous préférerez peut-être un client asynchrone comme aiomqtt.

Tutoriel MQTT vers API REST avec Flask : exposer les données des appareils en HTTP