Alertas por E-mail MQTT com Python: Notificações SMTP com Limite de Envio

Alertas por e-mail MQTT com Python são uma forma confiável de garantir que a pessoa certa fique sabendo de um problema, mesmo quando ninguém está olhando para um dashboard. Neste tutorial você vai construir um pequeno serviço que assina topics de alarme e telemetria, avalia limites com histerese e envia e-mails SMTP com tempos de espera e resumos, para que um sensor ruidoso nunca inunde uma caixa de entrada. Você também verá onde as regras de alerta nativas de um broker se encaixam e onde um script personalizado ainda é a melhor ferramenta.

Do que um Bom Serviço de Alertas por E-mail Precisa

Enviar um e-mail a partir de on_message leva cinco linhas. Enviar e-mails que as pessoas não aprendem a ignorar exige um pouco mais de design. Um serviço útil tem:

Layout dos Topics

O serviço escuta dois tipos de topics sob um prefixo de teste no servidor MQTT público (gratuito):

connectmq/demo/mail-6630/<device>/telemetry   {"temperature": 72.5, "pressure": 2.1}
connectmq/demo/mail-6630/<device>/alarm       {"severity": "critical", "text": "Door open"}

A telemetria é verificada contra os limites definidos no script. As mensagens de alarme são eventos que o próprio dispositivo decidiu gerar, e são encaminhadas como estão, respeitando o tempo de espera. O servidor MQTT público em broker.connectmq.com pode ser lido por qualquer pessoa e serve para testes e protótipos, não para dados de produção, então escolha seu próprio prefixo aleatório e mantenha endereços e dados reais fora dos payloads de teste.

Prepare as Credenciais SMTP

Você precisa de um servidor SMTP que aceite envios autenticados. Opções típicas são o servidor de e-mail da sua empresa, um provedor de e-mail transacional ou uma caixa de e-mail pessoal com senha de app. Reúna:

• Host e porta: normalmente 587 com STARTTLS ou 465 com TLS implícito
• Nome de usuário e senha (uma senha específica de app se a conta usa autenticação de dois fatores)
• Um endereço de remetente que o servidor permita usar

Guarde tudo em variáveis de ambiente, nunca no script.

O Serviço de Alertas por E-mail MQTT em Python

Instale a única dependência de terceiros:

pip install "paho-mqtt>=2.0"

Salve o código a seguir como mqtt_email_alerts.py:

import json
import os
import queue
import smtplib
import ssl
import threading
import time
import uuid
from email.message import EmailMessage

import paho.mqtt.client as mqtt

# --- Configurações MQTT ---
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"mail-alerts-{uuid.uuid4().hex[:8]}")
PREFIX = os.environ.get("TOPIC_PREFIX", "connectmq/demo/mail-6630")

# --- Configurações SMTP ---
SMTP_HOST = os.environ["SMTP_HOST"]
SMTP_PORT = int(os.environ.get("SMTP_PORT", "587"))
SMTP_USER = os.environ["SMTP_USER"]
SMTP_PASS = os.environ["SMTP_PASS"]
MAIL_FROM = os.environ.get("MAIL_FROM", SMTP_USER)
MAIL_TO = [a.strip() for a in os.environ["MAIL_TO"].split(",")]
COOLDOWN = int(os.environ.get("COOLDOWN_MINUTES", "15")) * 60

# métrica -> (dispara acima de, normaliza abaixo de)
THRESHOLDS = {
    "temperature": (80.0, 75.0),
    "pressure": (3.0, 2.7),
}

outbox: "queue.Queue[EmailMessage]" = queue.Queue()
active = {}      # (dispositivo, métrica) -> True enquanto em alarme
last_sent = {}   # chave -> timestamp do último e-mail


def build_email(subject: str, body: str) -> EmailMessage:
    msg = EmailMessage()
    msg["Subject"] = subject
    msg["From"] = MAIL_FROM
    msg["To"] = ", ".join(MAIL_TO)
    msg.set_content(body)
    return msg


def queue_email(key, subject, body, force=False):
    now = time.time()
    if not force and now - last_sent.get(key, 0) < COOLDOWN:
        print(f"Cooldown active for {key}, not emailing")
        return
    last_sent[key] = now
    outbox.put(build_email(subject, body))


def smtp_worker():
    context = ssl.create_default_context()
    while True:
        msg = outbox.get()
        for attempt in range(1, 4):
            try:
                if SMTP_PORT == 465:
                    server = smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, context=context, timeout=20)
                else:
                    server = smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=20)
                    server.starttls(context=context)
                with server:
                    server.login(SMTP_USER, SMTP_PASS)
                    server.send_message(msg)
                print(f"Sent: {msg['Subject']}")
                break
            except (smtplib.SMTPException, OSError) as exc:
                print(f"SMTP attempt {attempt} failed: {exc}")
                time.sleep(10 * attempt)


def check_thresholds(device: str, data: dict):
    for metric, (trigger, clear) in THRESHOLDS.items():
        value = data.get(metric)
        if not isinstance(value, (int, float)):
            continue
        key = (device, metric)
        if not active.get(key) and value > trigger:
            active[key] = True
            queue_email(key, f"[ALERT] {device} {metric} = {value}",
                        f"{device} reported {metric} = {value}, above the limit of {trigger}.\n"
                        f"Topic: {PREFIX}/{device}/telemetry\nTime: {time.strftime('%Y-%m-%d %H:%M:%S %Z')}")
        elif active.get(key) and value < clear:
            active[key] = False
            # sempre envia a normalização, mesmo dentro da janela de espera
            queue_email(key, f"[RESOLVED] {device} {metric} = {value}",
                        f"{device} {metric} is back below {clear} (now {value}).", force=True)


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"MQTT connect failed: {reason_code}")
        return
    client.subscribe(f"{PREFIX}/+/telemetry", qos=0)
    client.subscribe(f"{PREFIX}/+/alarm", qos=1)
    print("Subscribed to telemetry and alarm topics")


def on_message(client, userdata, msg):
    device, kind = msg.topic.split("/")[-2:]
    try:
        data = json.loads(msg.payload)
    except (json.JSONDecodeError, UnicodeDecodeError):
        print(f"Ignoring non-JSON payload on {msg.topic}")
        return
    if not isinstance(data, dict):
        return
    if kind == "telemetry":
        check_thresholds(device, data)
    elif kind == "alarm" and not msg.retain:
        severity = str(data.get("severity", "alarm")).upper()
        queue_email((device, "alarm"), f"[{severity}] {device}: {data.get('text', 'alarm')}",
                    json.dumps(data, indent=2))


def main():
    threading.Thread(target=smtp_worker, daemon=True).start()
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        client.username_pw_set(USERNAME, PASSWORD)
    if PORT == 8883:
        client.tls_set()
    client.on_connect = on_connect
    client.on_message = on_message
    client.reconnect_delay_set(min_delay=1, max_delay=60)
    client.connect(BROKER, PORT, keepalive=60)
    client.loop_forever(retry_first_connection=True)


if __name__ == "__main__":
    main()

Como a Lógica Funciona

Todos os callbacks MQTT rodam na thread de rede do paho, então active e last_sent só são acessados a partir de uma thread e não precisam de lock. A thread de trabalho SMTP é dona da conexão com o servidor de e-mail e tenta até três vezes, com intervalos crescentes. Se o seu servidor de e-mail estiver lento ou temporariamente inacessível, o MQTT continua fluindo e os e-mails simplesmente chegam mais tarde.

A histerese é a parte mais valiosa. Uma temperatura oscilando em torno de 80 geraria, de outra forma, um alerta a cada poucos segundos, cruzando o limite nas duas direções. Com disparo em 80 e normalização em 75, você recebe um alerta quando ela sobe e uma notificação de resolução quando ela realmente se recupera.

Teste de Ponta a Ponta

export SMTP_HOST=smtp.example.com SMTP_PORT=587
export SMTP_USER=alerts@example.com SMTP_PASS='app-password'
export MAIL_TO=you@example.com
python mqtt_email_alerts.py

Publique um valor acima do limite, depois um valor já recuperado e, por fim, um alarme do dispositivo:

mosquitto_pub -h broker.connectmq.com -p 1883 -i mail-pub-6630 \
  -t connectmq/demo/mail-6630/oven2/telemetry -m '{"temperature": 84.1}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i mail-pub-6630 \
  -t connectmq/demo/mail-6630/oven2/telemetry -m '{"temperature": 72.0}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i mail-pub-6630 \
  -t connectmq/demo/mail-6630/oven2/alarm -m '{"severity": "critical", "text": "Door open"}'

Você deve receber três e-mails: um alerta, uma resolução e o alarme crítico. Publique 84.1 mais algumas vezes e repare que nada mais chega; a máquina de estados sabe que o alarme já está ativo.

Resumos para Instalações Ruidosas

Os tempos de espera impedem que um único dispositivo inunde uma caixa de entrada, mas uma instalação com cinquenta dispositivos, como um frigorífico ou uma estação de saneamento, ainda pode gerar cinquenta e-mails separados durante uma queda de energia. Quando muitos alertas costumam chegar juntos, troque os e-mails imediatos por um resumo. A ideia é simples: em vez de colocar cada e-mail direto na fila de saída, adicione uma linha a uma lista e, a cada poucos minutos, a thread de trabalho envia uma mensagem resumindo tudo o que aconteceu, agrupado por dispositivo. Alarmes críticos podem ignorar o resumo para continuar chegando em segundos.

Um resumo também torna visível a fadiga de alertas. Se os mesmos três sensores aparecem em todo resumo, os limites estão errados ou os sensores precisam de manutenção, e essa é uma conversa a ter com a equipe, não uma regra de filtro a criar no cliente de e-mail. Revise o dicionário de limites regularmente e mova-o para um pequeno arquivo YAML ou JSON quando pessoas que não são desenvolvedoras precisarem ajustar os limites, para que as mudanças não exijam editar código.

Regras de Alerta Nativas vs um Script Python

O ConnectMQ inclui regras de alerta por limite e um log de alertas no portal. Elas são a forma mais rápida de ver quando um valor ultrapassou um limite em todo o seu broker, sem nenhum código para implantar. Um script personalizado ainda vale a pena quando você precisa de comportamentos fora do portal:

Uma configuração comum usa os dois: regras no portal para visibilidade durante o comissionamento e a operação, e um script como este para as notificações que precisam chegar à caixa de entrada das pessoas.

Checklist de Solução de Problemas

1. SMTPAuthenticationError. A conta precisa de uma senha de app, ou o envio SMTP está desativado para ela.
2. Timeouts de conexão nas portas 587 ou 465. Alguns provedores de nuvem e redes bloqueiam SMTP de saída. Tente a outra porta ou um provedor transacional.
3. Os e-mails caem no spam. O domínio do remetente não tem registros SPF e DKIM para o seu servidor SMTP. Envie a partir de um domínio que você controla e configure-os.
4. Um alarme nunca é resolvido. O valor nunca cai abaixo do nível de normalização, ou o dispositivo parou de enviar telemetria. Considere um alerta de "dados desatualizados" quando nenhuma mensagem chegar por N minutos.
5. Alertas duplicados após reinícios. O estado fica em memória. Para implantações de longa duração, persista active em um pequeno arquivo JSON ou no SQLite.

Configuração para Produção

Execute o serviço em um broker privado com credenciais e TLS na porta 8883, usando o nome de usuário, a senha e o client ID gerados no portal do ConnectMQ. Em um broker MQTT gerenciado, o lado do broker é operado para você, então seu trabalho se resume a manter esse único processo ativo com systemd ou com uma política de reinício de contêiner. Se você prefere chat a e-mail, o bot do Telegram com MQTT usa a mesma estrutura, e se você precisa de uma trilha de auditoria de cada leitura, combine este serviço com salvar mensagens MQTT em CSV. Para o comportamento de reconexão do paho em serviços de longa duração, veja tratamento de reconexão no paho-mqtt e, para opções de bibliotecas além do Python, o guia de bibliotecas cliente MQTT.

Comece a Usar o ConnectMQ

Configure um broker MQTT gerenciado e privado para o seu serviço de alertas em minutos. Cadastre-se em app.connectmq.com, escolha o plano Serverless gratuito, sem cartão de crédito, crie um broker e use as credenciais geradas nas variáveis de ambiente acima. Depois, adicione uma regra de alerta por limite no portal e compare-a lado a lado com seus alertas em Python.

Perguntas frequentes

O paho-mqtt consegue enviar e-mails diretamente?

Não. O paho-mqtt só fala MQTT. Os módulos padrão smtplib e email do Python cuidam do envio, e o script liga os dois colocando um e-mail na fila quando uma mensagem atende a uma condição.

Como evitar receber centenas de e-mails de alerta?

Combine histerese (níveis separados de disparo e de normalização) com um tempo de espera por dispositivo. O script envia um alerta quando um limite é ultrapassado, fica em silêncio enquanto a condição persiste e envia uma resolução quando ela é normalizada.

Devo usar QoS 1 nos topics de alarme?

Sim. QoS 1 garante que cada alarme seja entregue pelo menos uma vez ao subscriber enquanto ele está conectado, e o tempo de espera absorve uma duplicata ocasional.

O plano gratuito do ConnectMQ funciona para isso?

Sim. O plano Serverless gratuito suporta até 25 conexões e QoS até 1, o que cobre um serviço de alertas e um pequeno conjunto de dispositivos.