MQTT con Flask: flask-mqtt, paho-mqtt y dashboard en vivo

Q: ¿Por qué mi app Flask recibe cada mensaje MQTT dos veces?

Normalmente hay dos clientes MQTT en marcha: el reloader de depuración ejecuta tu código en dos procesos, o Gunicorn arrancó varios workers. Desactiva el reloader (use_reloader=False) o arranca MQTT solo en el proceso que sirve, y da a cada proceso un client ID único.

Q: ¿Uso flask-mqtt o paho-mqtt directamente?

flask-mqtt es un envoltorio fino que lee la configuración de app.config y ofrece decoradores. Es cómodo para apps pequeñas, pero tiene poco mantenimiento, así que comprueba su compatibilidad con paho-mqtt 2.x. Usar paho-mqtt directamente te da control total y la API de callbacks actual.

Q: ¿Puedo usar Flask-SQLAlchemy dentro de un callback MQTT?

Sí, pero los callbacks MQTT se ejecutan en el hilo de red de paho, fuera de cualquier petición, así que no hay contexto de aplicación. Envuelve el código de base de datos en with app.app_context(): y mantén el trabajo breve o pásalo a una cola.

Flask no tiene event loop propio y atiende cada petición en un hilo, así que MQTT en Flask significa ejecutar un cliente MQTT junto al servidor web: un hilo de red en segundo plano recibe los mensajes y los handlers de las peticiones publican a través del mismo cliente. Puedes hacerlo con la extensión flask-mqtt o con paho-mqtt 2.x directamente. Ambas opciones funcionan, y ambas comparten las mismas trampas de hilos y multiproceso, que son el verdadero tema de esta guía.

Configuraremos cada opción, resolveremos el problema de "cada mensaje llega dos veces" y terminaremos con un pequeño dashboard en vivo que envía datos MQTT al navegador con Server-Sent Events.

Opción 1: flask-mqtt

flask-mqtt envuelve paho-mqtt y lee su configuración de app.config:

Instalación

pip install flask flask-mqtt

app_flask_mqtt.py

import os
from flask import Flask
from flask_mqtt import Mqtt

app = Flask(__name__)
app.config["MQTT_BROKER_URL"] = "broker.connectmq.com"
app.config["MQTT_BROKER_PORT"] = 1883
app.config["MQTT_USERNAME"] = os.getenv("MQTT_USER", "")
app.config["MQTT_PASSWORD"] = os.getenv("MQTT_PASS", "")
app.config["MQTT_KEEPALIVE"] = 30
app.config["MQTT_TLS_ENABLED"] = False
mqtt = Mqtt(app)

@mqtt.on_connect()
def handle_connect(client, userdata, flags, rc):
    mqtt.subscribe("testmqtt/flask/#")

@mqtt.on_message()
def handle_message(client, userdata, message):
    print(message.topic, message.payload.decode())

@app.post("/ping")
def ping():
    mqtt.publish("testmqtt/flask/ping", "pong")
    return {"ok": True}

Se configura rápido, pero ten en cuenta dos cosas. La extensión tiene poco mantenimiento, y las versiones antiguas se escribieron para la API de callbacks de paho-mqtt 1.x: si ves errores de firma de callbacks tras instalarse paho-mqtt 2.0, busca una versión más reciente de flask-mqtt o fija paho-mqtt<2. Además, se conecta cuando se ejecuta Mqtt(app), así que cada proceso que importa la app abre una conexión (más sobre esto abajo).

Opción 2: paho-mqtt 2.x directamente

Usar paho directamente requiere unas pocas líneas más y te da la API actual, reconexiones automáticas y control total. Pon el cliente en su propio módulo:

mqtt_service.py

import os
import queue
import threading
import uuid
import paho.mqtt.client as mqtt

TELEMETRY = "testmqtt/flask/+/telemetry"


class MqttService:
    def __init__(self, host="broker.connectmq.com", port=1883):
        self.host, self.port = host, port
        self.latest: dict[str, str] = {}
        self.listeners: list[queue.Queue] = []
        self.lock = threading.Lock()
        self.client = mqtt.Client(
            mqtt.CallbackAPIVersion.VERSION2,
            client_id=f"testmqtt-flask-{uuid.uuid4().hex[:8]}",  # único por proceso
        )
        if os.getenv("MQTT_USER"):
            self.client.username_pw_set(os.environ["MQTT_USER"], os.getenv("MQTT_PASS"))
        self.client.reconnect_delay_set(min_delay=1, max_delay=30)
        self.client.on_connect = self._on_connect
        self.client.on_message = self._on_message

    def start(self):
        self.client.connect_async(self.host, self.port, keepalive=30)
        self.client.loop_start()          # hilo de red, se reconecta solo

    def _on_connect(self, client, userdata, flags, reason_code, properties):
        if reason_code.is_failure:
            print("MQTT connect failed:", reason_code)
            return
        client.subscribe(TELEMETRY, qos=1)  # se repite tras cada reconexión

    def _on_message(self, client, userdata, msg):
        item = {"topic": msg.topic, "payload": msg.payload.decode(errors="replace")}
        with self.lock:
            self.latest[msg.topic] = item["payload"]
            listeners = list(self.listeners)
        for q in listeners:
            try:
                q.put_nowait(item)
            except queue.Full:
                pass                        # navegador lento: descartar, nunca bloquear el hilo de red

    def join(self) -> queue.Queue:
        q: queue.Queue = queue.Queue(maxsize=100)
        with self.lock:
            self.listeners.append(q)
        return q

    def leave(self, q: queue.Queue):
        with self.lock:
            self.listeners.remove(q)

    def publish(self, topic: str, payload: str, qos: int = 1) -> bool:
        if not self.client.is_connected():
            return False
        return self.client.publish(topic, payload, qos=qos).rc == mqtt.MQTT_ERR_SUCCESS

connect_async() más loop_start() hace que Flask arranque aunque el broker no esté accesible, y paho sigue reintentando en segundo plano con el backoff definido en reconnect_delay_set(). Suscribirse dentro de on_connect restaura la suscripción tras cada reconexión. Para lo básico de la API de paho, consulta MQTT en Python.

Reglas de hilos

En este montaje tres hilos tocan datos MQTT, así que unas pocas reglas lo mantienen seguro:

Regla	Por qué
Mantén `on_message` breve	Se ejecuta en el único hilo de red de paho. Un handler lento retrasa todos los mensajes y puede retrasar los pings de keep-alive.
Protege el estado compartido con un lock	Los hilos de peticiones leen `latest` mientras el hilo de red lo escribe.
`publish()` desde los handlers de peticiones está bien	`publish()` de paho es thread-safe; comprueba antes `is_connected()` y responde `503` si no.
No hay contexto de app en los callbacks	`current_app`, `g` y las sesiones de Flask-SQLAlchemy necesitan `with app.app_context():` dentro del callback.
Nunca bloquees en `wait_for_publish()` sin timeout	Un broker desconectado colgaría el hilo de la petición para siempre.

El trabajo pesado — escrituras en base de datos, llamadas HTTP a otros servicios — va a una cola consumida por un hilo worker o a un sistema de tareas como Celery o RQ, no a on_message.

El problema de "cada mensaje dos veces"

El bug más común de MQTT en Flask: los mensajes se procesan dos veces, o el log muestra el cliente conectándose y desconectándose en bucle. Causas y soluciones:

Reloader de depuración. app.run(debug=True) ejecuta tu módulo en un proceso vigilante y en un proceso que sirve, así que tienes dos clientes MQTT. Usa use_reloader=False, o arranca MQTT solo en el proceso que sirve: if os.environ.get("WERKZEUG_RUN_MAIN") == "true" or not app.debug:.
Varios workers de Gunicorn. -w 4 significa cuatro clientes. Para un dashboard está bien (cada worker necesita todos los mensajes para sus navegadores), pero todo lo que deba ocurrir una sola vez — un insert en base de datos, una alerta — debería usar una shared subscription para que el broker entregue cada mensaje a un solo worker. Consulta shared subscriptions.
Client ID fijo. Si todos esos procesos usan el mismo client ID, el broker desconecta la conexión anterior cada vez que llega una nueva, y se expulsan unos a otros sin fin. Los sufijos aleatorios lo evitan — consulta client ID duplicado.
gunicorn --preload. La app se importa en el master antes del fork, pero el hilo de paho no sobrevive a fork(), así que los workers tienen un objeto cliente sin loop de red. No uses preload, o arranca el cliente en un hook post_fork.

Un dashboard en vivo sencillo

Ahora conecta el servicio con Flask: un snapshot JSON, un endpoint de comandos, un stream SSE y una pequeña página HTML.

app.py

import json
import queue
import re
from flask import Flask, Response, abort, jsonify, request
from mqtt_service import MqttService

app = Flask(__name__)
mqtt_service = MqttService()
mqtt_service.start()
DEVICE_ID = re.compile(r"^[A-Za-z0-9_-]{1,64}$")

PAGE = """<!doctype html><title>MQTT dashboard</title>
<h1>Telemetría en vivo</h1><table id="t"></table>
<script>
  const rows = {};
  new EventSource("/stream").onmessage = (e) => {
    const { topic, payload } = JSON.parse(e.data);
    let row = rows[topic];
    if (!row) {
      row = rows[topic] = document.getElementById("t").insertRow();
      row.insertCell().textContent = topic;
      row.insertCell();
    }
    row.cells[1].textContent = payload;
  };
</script>"""


@app.get("/")
def index():
    return PAGE


@app.get("/api/latest")
def latest():
    with mqtt_service.lock:
        return jsonify(mqtt_service.latest)


@app.post("/api/devices/<device_id>/command")
def command(device_id):
    if not DEVICE_ID.match(device_id):
        abort(400)                          # nunca metas entrada sin validar en un topic
    body = json.dumps(request.get_json(force=True))
    if not mqtt_service.publish(f"testmqtt/flask/{device_id}/command", body):
        abort(503)
    return jsonify(queued=True), 202


@app.get("/stream")
def stream():
    q = mqtt_service.join()

    def events():
        try:
            while True:
                try:
                    yield f"data: {json.dumps(q.get(timeout=15))}\n\n"
                except queue.Empty:
                    yield ": keep-alive\n\n"  # evita que los proxies cierren el stream
        finally:
            mqtt_service.leave(q)

    return Response(events(), mimetype="text/event-stream", headers={"Cache-Control": "no-cache"})


if __name__ == "__main__":
    app.run(debug=True, use_reloader=False, threaded=True)

Cada dashboard abierto mantiene una conexión SSE y, con el servidor con hilos, eso es un hilo por navegador: suficiente para un dashboard interno con pocos espectadores. En producción ejecuta gunicorn -w 1 --threads 16 app:app (o más workers siguiendo las reglas de arriba). Para muchos espectadores simultáneos encaja mejor un framework asíncrono; consulta MQTT con FastAPI, o deja que los navegadores se conecten directamente al broker como en el tutorial de dashboard en vivo.

Pruébalo

Ejecuta python app.py y abre http://localhost:5000.
En el cliente MQTT online, conéctate a broker.connectmq.com y publica {"temp": 23.1} en testmqtt/flask/greenhouse/telemetry. Debería aparecer una fila en el dashboard al instante.
Suscríbete a testmqtt/flask/# en el cliente online y envía un comando:

HTTP → MQTT

curl -X POST localhost:5000/api/devices/greenhouse/command \
  -H 'content-type: application/json' -d '{"fan": "on"}'

Comprueba que curl localhost:5000/api/latest devuelve el último valor por topic.
Detén el broker o corta la red: el endpoint de comandos debería responder 503, y el dashboard debería reanudarse solo cuando paho se reconecte.

Envía telemetría de prueba a tu dashboard Flask

Si tu proyecto usa Django en lugar de Flask, se aplican los mismos patrones con un management command como suscriptor de larga duración — consulta MQTT con Django.

MQTT en Flask funciona bien en cuanto lo tratas como lo que es: un cliente de larga duración en un hilo en segundo plano, uno por proceso. Arráncalo una vez, mantén los callbacks breves y thread-safe, da a cada proceso un client ID único y prueba el circuito completo — del dispositivo al navegador y del navegador al dispositivo — antes de añadir más workers.

Preguntas frecuentes

¿Por qué mi app Flask recibe cada mensaje MQTT dos veces?

Normalmente hay dos clientes MQTT en marcha: el reloader de depuración ejecuta tu código en dos procesos, o Gunicorn arrancó varios workers. Desactiva el reloader (use_reloader=False) o arranca MQTT solo en el proceso que sirve, y da a cada proceso un client ID único.

¿Uso flask-mqtt o paho-mqtt directamente?

flask-mqtt es un envoltorio fino que lee la configuración de app.config y ofrece decoradores. Es cómodo para apps pequeñas, pero tiene poco mantenimiento, así que comprueba su compatibilidad con paho-mqtt 2.x. Usar paho-mqtt directamente te da control total y la API de callbacks actual.

¿Puedo usar Flask-SQLAlchemy dentro de un callback MQTT?

Sí, pero los callbacks MQTT se ejecutan en el hilo de red de paho, fuera de cualquier petición, así que no hay contexto de aplicación. Envuelve el código de base de datos en with app.app_context(): y mantén el trabajo breve o pásalo a una cola.

MQTT con Flask (flask-mqtt y paho)