MQTT com Flask: flask-mqtt, paho-mqtt e dashboard ao vivo

Q: Por que meu app Flask recebe cada mensagem MQTT duas vezes?

Normalmente há dois clientes MQTT rodando: o reloader de debug executa seu código em dois processos, ou o Gunicorn iniciou vários workers. Desative o reloader (use_reloader=False) ou inicie o MQTT só no processo que atende, e dê a cada processo um client ID único.

Q: Devo usar flask-mqtt ou paho-mqtt diretamente?

O flask-mqtt é um wrapper fino que lê as configurações de app.config e oferece decorators. É prático para apps pequenos, mas tem pouca manutenção, então confira a compatibilidade com o paho-mqtt 2.x. Usar o paho-mqtt diretamente dá controle total e a API de callbacks atual.

Q: Posso usar Flask-SQLAlchemy dentro de um callback MQTT?

Sim, mas os callbacks MQTT rodam na thread de rede do paho, fora de qualquer requisição, então não há contexto de aplicação. Envolva o código de banco em with app.app_context(): e mantenha o trabalho curto ou passe-o para uma fila.

O Flask não tem event loop próprio e atende cada requisição em uma thread, então MQTT no Flask significa rodar um cliente MQTT ao lado do servidor web: uma thread de rede em segundo plano recebe as mensagens e os handlers de requisição publicam pelo mesmo cliente. Você pode fazer isso com a extensão flask-mqtt ou com o paho-mqtt 2.x diretamente. As duas opções funcionam, e as duas compartilham as mesmas armadilhas de threads e múltiplos processos — o verdadeiro assunto deste guia.

Vamos configurar cada opção, resolver o problema de "toda mensagem chega duas vezes" e terminar com um pequeno dashboard ao vivo que envia dados MQTT para o navegador com Server-Sent Events.

Opção 1: flask-mqtt

O flask-mqtt encapsula o paho-mqtt e lê as configurações de app.config:

Instalação

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}

É rápido de configurar, mas lembre-se de duas coisas. A extensão tem pouca manutenção, e versões antigas foram escritas para a API de callbacks do paho-mqtt 1.x — se aparecerem erros de assinatura de callback depois que o paho-mqtt 2.0 foi instalado, procure uma versão mais nova do flask-mqtt ou fixe paho-mqtt<2. Além disso, ela conecta quando Mqtt(app) é executado, então todo processo que importa o app abre uma conexão (mais sobre isso abaixo).

Opção 2: paho-mqtt 2.x diretamente

Usar o paho diretamente exige algumas linhas a mais e dá a API atual, reconexão automática e controle total. Coloque o cliente em um módulo próprio:

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 processo
        )
        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()          # thread de rede, reconecta sozinha

    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)  # roda de novo a cada reconexão

    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: descarta, nunca bloqueia a thread de rede

    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() com loop_start() faz o Flask iniciar mesmo que o broker esteja inacessível, e o paho continua tentando em segundo plano com o backoff definido em reconnect_delay_set(). Assinar dentro de on_connect restaura a assinatura após cada reconexão. Para o básico da API do paho, veja MQTT em Python.

Regras de threads

Nesta montagem, três threads mexem em dados MQTT, então algumas regras mantêm tudo seguro:

Regra	Por quê
Mantenha `on_message` curto	Ele roda na única thread de rede do paho. Um handler lento atrasa todas as mensagens e pode atrasar os pings de keep-alive.
Proteja o estado compartilhado com um lock	As threads de requisição leem `latest` enquanto a thread de rede escreve.
`publish()` a partir dos handlers de requisição é ok	O `publish()` do paho é thread-safe; confira `is_connected()` antes e responda `503` caso contrário.
Não há contexto de app nos callbacks	`current_app`, `g` e sessões do Flask-SQLAlchemy precisam de `with app.app_context():` dentro do callback.
Nunca bloqueie em `wait_for_publish()` sem timeout	Um broker desconectado travaria a thread da requisição para sempre.

Trabalho pesado — escritas em banco, chamadas HTTP para outros serviços — vai para uma fila consumida por uma thread worker ou por um sistema de tarefas como Celery ou RQ, não para on_message.

O problema de "toda mensagem duas vezes"

O bug mais comum de MQTT no Flask: mensagens processadas duas vezes, ou o log mostrando o cliente conectando e desconectando em loop. Causas e correções:

Reloader de debug. app.run(debug=True) executa seu módulo em um processo observador e em um processo que atende, então você tem dois clientes MQTT. Use use_reloader=False, ou inicie o MQTT só no processo que atende: if os.environ.get("WERKZEUG_RUN_MAIN") == "true" or not app.debug:.
Vários workers do Gunicorn. -w 4 significa quatro clientes. Para um dashboard tudo bem (cada worker precisa de todas as mensagens para os seus navegadores), mas tudo o que deve acontecer uma vez — um insert no banco, um alerta — deveria usar uma shared subscription para que o broker entregue cada mensagem a um único worker. Veja shared subscriptions.
Client ID fixo. Se todos esses processos usam o mesmo client ID, o broker derruba a conexão antiga sempre que uma nova chega, e eles se expulsam para sempre. Sufixos aleatórios evitam isso — veja client ID duplicado.
gunicorn --preload. O app é importado no master antes do fork, mas a thread do paho não sobrevive ao fork(), então os workers ficam com um objeto cliente sem loop de rede. Não use preload, ou inicie o cliente em um hook post_fork.

Um dashboard ao vivo simples

Agora ligue o serviço ao Flask: um snapshot JSON, um endpoint de comandos, um stream SSE e uma pequena 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>Telemetria ao 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 coloque entrada sem validação em um 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"  # impede que proxies fechem o 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 aberto mantém uma conexão SSE e, com o servidor em threads, isso é uma thread por navegador — suficiente para um dashboard interno com poucos espectadores. Em produção rode gunicorn -w 1 --threads 16 app:app (ou mais workers seguindo as regras acima). Para muitos espectadores simultâneos, um framework assíncrono encaixa melhor; veja MQTT com FastAPI, ou deixe os navegadores se conectarem direto ao broker como no tutorial de dashboard ao vivo.

Teste

Rode python app.py e abra http://localhost:5000.
No cliente MQTT online, conecte em broker.connectmq.com e publique {"temp": 23.1} em testmqtt/flask/greenhouse/telemetry. Uma linha deve aparecer no dashboard na hora.
Assine testmqtt/flask/# no cliente online e envie um comando:

HTTP → MQTT

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

Confira se curl localhost:5000/api/latest retorna o último valor por topic.
Pare o broker ou derrube a rede: o endpoint de comandos deve responder 503, e o dashboard deve voltar sozinho quando o paho reconectar.

Envie telemetria de teste ao seu dashboard Flask

Se o seu projeto usa Django em vez de Flask, os mesmos padrões se aplicam com um management command como subscriber de longa duração — veja MQTT com Django.

MQTT no Flask funciona bem quando você o trata como o que ele é: um cliente de longa duração em uma thread de fundo, um por processo. Inicie-o uma vez, mantenha os callbacks curtos e thread-safe, dê a cada processo um client ID único e teste o ciclo completo — do dispositivo ao navegador e do navegador ao dispositivo — antes de adicionar mais workers.

Perguntas frequentes

Por que meu app Flask recebe cada mensagem MQTT duas vezes?

Normalmente há dois clientes MQTT rodando: o reloader de debug executa seu código em dois processos, ou o Gunicorn iniciou vários workers. Desative o reloader (use_reloader=False) ou inicie o MQTT só no processo que atende, e dê a cada processo um client ID único.

Devo usar flask-mqtt ou paho-mqtt diretamente?

O flask-mqtt é um wrapper fino que lê as configurações de app.config e oferece decorators. É prático para apps pequenos, mas tem pouca manutenção, então confira a compatibilidade com o paho-mqtt 2.x. Usar o paho-mqtt diretamente dá controle total e a API de callbacks atual.

Posso usar Flask-SQLAlchemy dentro de um callback MQTT?

Sim, mas os callbacks MQTT rodam na thread de rede do paho, fora de qualquer requisição, então não há contexto de aplicação. Envolva o código de banco em with app.app_context(): e mantenha o trabalho curto ou passe-o para uma fila.

MQTT com Flask (flask-mqtt e paho)