De MQTT a API REST con Flask: tutorial en Python

Construir una API REST sobre MQTT con Flask permite que cualquier aplicación que hable HTTP lea datos de dispositivos en vivo y envíe comandos sin tener que aprender MQTT. En este tutorial escribirás una app Flask que se suscribe a topics de dispositivos, mantiene en memoria el último valor de cada dispositivo, lo expone mediante endpoints JSON y publica comandos desde una ruta POST. También conocerás los dos errores de despliegue que rompen en silencio la mayoría de las combinaciones de Flask y MQTT.

Por qué poner una API REST delante de MQTT

MQTT es el protocolo adecuado para los dispositivos: conexiones persistentes, paquetes pequeños y entrega push. Sin embargo, muchos consumidores de esos datos no son dispositivos. El backend de una app móvil, una integración con el ERP, un script de informes o el sistema de un socio suelen esperar petición y respuesta por HTTP. En lugar de enseñar MQTT a cada uno de ellos, un puente ligero hace la traducción:

El consumidor quiere	Llamada REST	Qué hace el puente
Estado actual de todos los dispositivos	`GET /api/devices`	Devuelve el último mensaje cacheado de cada dispositivo
La última lectura de un dispositivo	`GET /api/devices/<id>`	Devuelve una entrada cacheada con su antigüedad
Enviar un comando	`POST /api/devices/<id>/command`	Publica JSON en el topic de comandos del dispositivo
Comprobación de salud	`GET /health`	Indica si MQTT está conectado

El puente es una "caché del último valor". No es un histórico: responde a "¿cuál es el valor ahora?", no a "¿cuál era el martes pasado?". Para el histórico, guarda el flujo en una base de datos junto a la caché.

Qué necesitas

Python 3.9+
pip install "flask>=3.0" "paho-mqtt>=2.0"
Un broker. Para desarrollo, usa el servidor MQTT público gratuito broker.connectmq.com con un prefijo de topic aleatorio; para producción, un broker privado con credenciales.

El puente Flask MQTT

Guarda esto como 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"))  # segundos

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 es imprescindible: el reloader arrancaría un segundo cliente MQTT
    app.run(host="0.0.0.0", port=5000, debug=True, use_reloader=False)

Detalles clave de la implementación

El cliente MQTT arranca al importar el módulo con connect_async y loop_start, así que se ejecuta en su propio hilo tanto si inicias la app con python app.py como con un servidor WSGI. connect_async también hace que la app arranque aunque el broker no esté disponible durante un momento; paho sigue reintentando.
Un lock protege la caché. paho escribe desde su hilo de red mientras Flask lee desde los hilos de las peticiones.
La ruta POST devuelve 202 Accepted, no 200. El broker ha confirmado el mensaje (PUBACK de QoS 1), pero eso no significa que el dispositivo lo haya ejecutado. Si necesitas confirmación, haz que el dispositivo publique un resultado en un topic de respuesta con el mismo request_id, o usa el patrón de response topic de MQTT 5 descrito en las funciones de MQTT 5.
Los IDs de dispositivo se validan para que nadie pueda inyectar +, # o niveles de topic adicionales.

Prueba el puente de MQTT a API REST con Flask

Arranca la app:

export API_KEY=dev-secret
python app.py

Publica un mensaje de estado retenido como lo haría un dispositivo. Los retained messages son útiles aquí: cuando el puente se reinicia, el broker entrega de inmediato el último estado de cada dispositivo, de modo que la caché está llena sin esperar al siguiente informe.

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

Consulta y envía comandos por 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

Observa cómo llega el comando con mosquitto_sub -h broker.connectmq.com -p 1883 -i flask-sub-3391 -v -t "connectmq/demo/flask-3391/+/cmd". Como el área pública la puede leer cualquiera, borra el mensaje retenido de prueba al terminar publicando un payload retenido vacío en el mismo topic.

Errores de despliegue

La mayoría de las configuraciones de Flask y MQTT que fallan lo hacen por uno de estos motivos:

El reloader de depuración ejecuta tu código dos veces. El reloader de Flask lanza un proceso hijo, así que dos clientes MQTT se conectan con el mismo client ID y se desconectan mutuamente en bucle. Mantén use_reloader=False.
Varios workers WSGI. gunicorn -w 4 importa la app cuatro veces: cuatro clientes MQTT y cuatro cachés separadas, de modo que peticiones consecutivas pueden devolver valores distintos. Ejecuta un único worker con hilos: gunicorn -w 1 --threads 8 app:app.
Client IDs fijos en varias instancias. Si escalas horizontalmente detrás de un balanceador de carga, da a cada instancia un client ID único. En un broker donde los client IDs están ligados a las credenciales, emite credenciales distintas por instancia.
Tratar la caché como la verdad. Un dispositivo que se ha desconectado sigue teniendo un valor en caché. Expón age_seconds y stale, como hace el código, para que los consumidores decidan.
Sin autenticación. Cualquiera que pueda hacer POST puede operar el equipo. La cabecera con la API key es lo mínimo; pon la app detrás de HTTPS y de una capa de autenticación adecuada para todo lo que esté expuesto a internet.

Ampliar el puente

Cuando los endpoints básicos funcionen, unas pocas mejoras cubren la mayoría de las peticiones reales:

Server-Sent Events para actualizaciones en vivo. Consultar GET /api/devices cada segundo funciona, pero un endpoint GET /api/stream que emita un evento cada vez que cambia la caché permite a los clientes HTTP recibir datos push sin hablar MQTT. Flask admite respuestas en streaming mediante un generador; da a cada cliente conectado su propia queue.Queue y mete las actualizaciones en todas ellas desde on_message.
Filtrado y proyección. Parámetros de consulta como ?stale=false o ?fields=temperature mantienen las respuestas pequeñas para los clientes móviles.
Histórico. Escribe cada mensaje en SQLite o en una base de datos de series temporales dentro de on_message y añade GET /api/devices/<id>/history?since=.... Mantén la caché en memoria para el último valor, porque es mucho más rápida que consultar la base de datos en cada petición.
Documentación OpenAPI. Si otros equipos consumen la API, describe los endpoints en un archivo OpenAPI para que puedan generar clientes y sepan exactamente qué campos esperar.

Cada mejora se mantiene dentro del mismo patrón: un cliente MQTT por proceso, un lock alrededor del estado compartido y handlers HTTP que nunca se bloquean esperando al broker.

Pasar a un broker privado

En producción, conecta el puente a un broker privado con TLS en el puerto 8883. En el portal de ConnectMQ, crea un broker y copia el usuario, la contraseña y el client ID generados en MQTT_USERNAME, MQTT_PASSWORD y MQTT_CLIENT_ID. Un broker MQTT en la nube como este elimina la necesidad de ejecutar Mosquitto junto a Flask, y el explorador de topics en vivo del portal ayuda a verificar qué debería contener la caché.

En el plan Serverless (Free) de ConnectMQ, QoS llega hasta 1, que es justo lo que usa el puente para estados y comandos. El plan Serverless de pago añade QoS 2 y suscripciones compartidas, estas últimas muy prácticas si más adelante ejecutas varias instancias del puente que deban repartirse el tráfico entrante.

Integraciones relacionadas

Este puente pertenece a una familia de pequeños servicios suscriptores. Enviar datos MQTT a Google Sheets cubre los informes en hojas de cálculo, y alertas por email con MQTT y Python cubre las notificaciones. Si tus consumidores se ejecutan en un navegador, puedes prescindir de REST por completo y conectarlos por WebSocket, como se explica en MQTT sobre WebSocket. Para otros lenguajes, consulta la guía de librerías cliente MQTT. Flask está documentado en flask.palletsprojects.com.

Empieza con ConnectMQ

Pon tu puente Flask en un broker MQTT gestionado, privado y protegido con TLS, en minutos. Regístrate en app.connectmq.com, elige el plan Serverless gratuito sin necesidad de tarjeta de crédito, crea un broker y pega las credenciales en el entorno del puente. Usa el cliente MQTT del navegador para publicar estados de dispositivos mientras pruebas los endpoints.

Preguntas frecuentes

¿Por qué no usar Flask-MQTT?

Puedes, pero es un envoltorio fino que sigue enfrentándose a los mismos problemas del reloader y de los múltiples workers, y tienes que comprobar qué versiones de paho-mqtt admite. Usar paho-mqtt 2.x directamente te da el control de los hilos y del ciclo de vida del cliente.

¿Puede la API REST devolver datos históricos?

No con una caché en memoria. Guarda los mensajes en una base de datos como SQLite, PostgreSQL o un almacén de series temporales desde el mismo suscriptor y añade endpoints de consulta que lean de ella.

¿Cómo sé que un comando se ha ejecutado?

La respuesta 202 solo confirma que el broker ha aceptado el mensaje. Haz que el dispositivo publique un acuse de recibo que contenga el request_id y expónlo mediante otro endpoint o un campo de estado en el estado del dispositivo.

¿Puedo usar FastAPI en lugar de Flask?

Sí. La estructura es la misma: un cliente MQTT por proceso, una caché con lock y rutas que la leen. Con FastAPI quizá prefieras un cliente asíncrono como aiomqtt.

De MQTT a API REST con Flask: expón datos de dispositivos en vivo por HTTP