MQTT para API REST com Flask em Python: Tutorial

Construir uma ponte de MQTT para API REST com Flask permite que qualquer aplicação que fale HTTP leia dados de dispositivos ao vivo e envie comandos, sem precisar aprender MQTT. Neste tutorial você vai escrever um app Flask que assina topics de dispositivos, mantém em memória o último valor de cada dispositivo, o expõe por endpoints JSON e publica comandos a partir de uma rota POST. Você também vai conhecer os dois erros de implantação que quebram silenciosamente a maioria das combinações de Flask com MQTT.

Por que Colocar uma API REST na Frente do MQTT

O MQTT é o protocolo certo para dispositivos: conexões persistentes, pacotes pequenos e entrega por push. Muitos consumidores desses dados, porém, não são dispositivos. O backend de um app móvel, uma integração com ERP, um script de relatórios ou o sistema de um parceiro normalmente esperam requisição e resposta via HTTP. Em vez de ensinar MQTT a cada um deles, uma ponte enxuta faz a tradução:

O consumidor quer	Chamada REST	O que a ponte faz
Estado atual de todos os dispositivos	`GET /api/devices`	Retorna a última mensagem em cache de cada dispositivo
Última leitura de um dispositivo	`GET /api/devices/<id>`	Retorna uma entrada do cache com sua idade
Enviar um comando	`POST /api/devices/<id>/command`	Publica JSON no topic de comando do dispositivo
Verificação de saúde	`GET /health`	Informa se o MQTT está conectado

A ponte é um "cache do último valor". Ela não é um historiador: responde "qual é o valor agora?", e não "qual era na terça passada?". Para histórico, armazene o fluxo em um banco de dados ao lado do cache.

O que Você Precisa

Python 3.9+
pip install "flask>=3.0" "paho-mqtt>=2.0"
Um broker. Para desenvolvimento, use o servidor MQTT público (gratuito) broker.connectmq.com com um prefixo de topic aleatório; para produção, um broker privado com credenciais.

A Ponte Flask MQTT

Salve 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 = {}                 # dispositivo -> {"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 é essencial: o reloader iniciaria um segundo cliente MQTT
    app.run(host="0.0.0.0", port=5000, debug=True, use_reloader=False)

Detalhes Importantes da Implementação

O cliente MQTT é iniciado no momento da importação com connect_async e loop_start, então ele roda em sua própria thread, quer você inicie o app com python app.py ou com um servidor WSGI. connect_async também significa que o app sobe mesmo se o broker estiver brevemente inacessível; o paho continua tentando.
Um lock protege o cache. O paho grava a partir de sua thread de rede enquanto o Flask lê a partir das threads de requisição.
A rota POST retorna 202 Accepted, e não 200. O broker confirmou o recebimento da mensagem (PUBACK de QoS 1), mas isso não significa que o dispositivo a executou. Se precisar de confirmação, faça o dispositivo publicar um resultado em um topic de resposta com o mesmo request_id, ou use o padrão de response topic do MQTT 5 descrito em recursos do MQTT 5.
Os IDs de dispositivo são validados para que quem faz a chamada não consiga injetar +, # ou níveis extras de topic.

Experimente a Ponte de MQTT para API REST com Flask

Inicie o app:

export API_KEY=dev-secret
python app.py

Publique uma mensagem de estado retida, como um dispositivo faria. Retained messages são úteis aqui: quando a ponte reinicia, o broker entrega imediatamente o último estado de cada dispositivo, então o cache já começa preenchido sem esperar o próximo relatório.

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

Consulte e envie comandos 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

Veja o comando chegar com mosquitto_sub -h broker.connectmq.com -p 1883 -i flask-sub-3391 -v -t "connectmq/demo/flask-3391/+/cmd". Como a área pública pode ser lida por todos, limpe a mensagem retida de teste quando terminar, publicando um payload retido vazio no mesmo topic.

Armadilhas de Implantação

A maioria das configurações de Flask com MQTT que não funcionam falha por um destes motivos:

O reloader de debug executa seu código duas vezes. O reloader do Flask cria um processo filho, então dois clientes MQTT se conectam com o mesmo client ID e derrubam um ao outro em loop. Mantenha use_reloader=False.
Vários workers WSGI. gunicorn -w 4 importa o app quatro vezes: quatro clientes MQTT e quatro caches separados, de modo que requisições consecutivas podem retornar valores diferentes. Em vez disso, execute um worker com threads: gunicorn -w 1 --threads 8 app:app.
Client IDs fixos entre instâncias. Se você escalar horizontalmente atrás de um balanceador de carga, dê a cada instância um client ID único. Em um broker onde os client IDs são vinculados às credenciais, emita credenciais separadas para cada instância.
Tratar o cache como verdade absoluta. Um dispositivo que ficou offline ainda tem um valor em cache. Exponha age_seconds e stale, como o código faz, para que os consumidores possam decidir.
Nenhuma autenticação. Qualquer coisa que consiga fazer um POST pode operar equipamentos. O cabeçalho com a chave de API é o mínimo; coloque o app atrás de HTTPS e de uma camada de autenticação adequada para qualquer coisa exposta à internet.

Estendendo a Ponte

Quando os endpoints básicos estiverem funcionando, algumas adições cobrem a maioria das demandas do mundo real:

Server-Sent Events para atualizações ao vivo. Consultar GET /api/devices a cada segundo funciona, mas um endpoint GET /api/stream que emite um evento sempre que o cache muda permite que clientes HTTP recebam pushes sem falar MQTT. O Flask suporta respostas em streaming por meio de um gerador; dê a cada cliente conectado sua própria queue.Queue e coloque as atualizações em todas elas a partir de on_message.
Filtragem e projeção. Parâmetros de consulta como ?stale=false ou ?fields=temperature mantêm as respostas pequenas para clientes móveis.
Histórico. Grave cada mensagem no SQLite ou em um banco de dados de séries temporais dentro de on_message e adicione GET /api/devices/<id>/history?since=.... Mantenha o cache em memória para o último valor, porque ele é muito mais rápido do que consultar o banco a cada requisição.
Documentação OpenAPI. Se outras equipes consomem a API, descreva os endpoints em um arquivo OpenAPI para que possam gerar clientes e saber exatamente quais campos esperar.

Cada adição segue o mesmo padrão: um cliente MQTT por processo, um lock em volta do estado compartilhado e handlers HTTP que nunca ficam bloqueados esperando o broker.

Migrando para um Broker Privado

Em produção, conecte a ponte a um broker privado com TLS na porta 8883. No portal do ConnectMQ, crie um broker e copie o nome de usuário, a senha e o client ID gerados para MQTT_USERNAME, MQTT_PASSWORD e MQTT_CLIENT_ID. Um broker MQTT na nuvem como esse, operado como broker MQTT gerenciado, elimina a necessidade de rodar um Mosquitto ao lado do Flask, e o explorador de topics ao vivo do portal ajuda a verificar o que o cache deveria conter.

No plano Serverless (Free) do ConnectMQ, o QoS vai até 1, que é exatamente o que a ponte usa para estados e comandos. O plano Serverless pago adiciona QoS 2 e shared subscriptions, e estas últimas são úteis se você depois executar várias instâncias da ponte que devam dividir o tráfego de entrada.

Integrações Relacionadas

Esta ponte pertence a uma família de pequenos serviços subscriber. Enviar dados MQTT para o Google Sheets cobre relatórios em planilhas, e alertas por e-mail MQTT com Python cobre notificações. Se seus consumidores rodam em um navegador, você pode dispensar o REST e conectá-los via WebSocket, como explicado em MQTT sobre WebSocket. Para outras linguagens, veja o guia de bibliotecas cliente MQTT. O próprio Flask está documentado em flask.palletsprojects.com.

Comece a Usar o ConnectMQ

Coloque sua ponte Flask em um broker privado e protegido por TLS em minutos. Cadastre-se em app.connectmq.com, escolha o plano Serverless gratuito, sem cartão de crédito, crie um broker e cole as credenciais no ambiente da ponte. Use o cliente MQTT no navegador para publicar estados de dispositivos enquanto testa os endpoints.

Perguntas frequentes

Por que não usar o Flask-MQTT?

Você pode, mas ele é um wrapper fino que enfrenta os mesmos problemas de reloader e de múltiplos workers, e você precisa verificar quais versões do paho-mqtt ele suporta. Usar o paho-mqtt 2.x diretamente mantém você no controle das threads e do ciclo de vida do cliente.

A API REST pode retornar dados históricos?

Não com um cache em memória. Armazene as mensagens em um banco de dados como SQLite, PostgreSQL ou um armazenamento de séries temporais a partir do mesmo subscriber e adicione endpoints de consulta que leiam dele.

Como sei que um comando foi executado?

A resposta 202 só confirma que o broker aceitou a mensagem. Faça o dispositivo publicar uma confirmação contendo o request_id e exponha-a por outro endpoint ou por um campo de status no estado do dispositivo.

Posso usar FastAPI em vez de Flask?

Sim. A estrutura é a mesma: um cliente MQTT por processo, um cache protegido por lock e rotas que o leem. Com FastAPI, talvez você prefira um cliente assíncrono como o aiomqtt.

MQTT para API REST com Flask: Exponha Dados de Dispositivos ao Vivo via HTTP