Alertas por email con MQTT y Python: notificaciones SMTP con límite de envío

Las alertas por email con MQTT y Python son una forma fiable de asegurarte de que la persona adecuada se entera de un problema, incluso cuando nadie está mirando un dashboard. En este tutorial crearás un pequeño servicio que se suscribe a topics de alarmas y telemetría, evalúa umbrales con histéresis y envía correos SMTP con cooldowns y resúmenes, para que un sensor ruidoso nunca inunde una bandeja de entrada. También verás dónde encajan las reglas de alerta integradas en un broker y en qué casos un script propio sigue siendo la mejor herramienta.

Qué necesita un buen servicio de alertas por email

Enviar un email desde on_message cuesta cinco líneas. Enviar correos que la gente no aprenda a ignorar requiere algo más de diseño. Un servicio útil tiene:

Estructura de topics

El servicio escucha dos tipos de topics bajo un prefijo de prueba en el servidor MQTT público gratuito de broker.connectmq.com, pensado solo para pruebas y prototipos:

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

La telemetría se compara con los umbrales definidos en el script. Los mensajes de alarma son eventos que el dispositivo decide por su cuenta y se reenvían tal cual, sujetos al cooldown. El área pública de broker.connectmq.com la puede leer cualquiera, así que elige tu propio prefijo aleatorio y no incluyas direcciones ni datos reales en los payloads de prueba.

Prepara las credenciales SMTP

Necesitas un servidor SMTP que acepte envíos autenticados. Las opciones habituales son el servidor de correo de tu empresa, un proveedor de email transaccional o un buzón personal con contraseña de aplicación. Reúne:

• Host y puerto: normalmente 587 con STARTTLS o 465 con TLS implícito
• Usuario y contraseña (una contraseña específica de aplicación si la cuenta usa autenticación en dos pasos)
• Una dirección de remitente desde la que el servidor te permita enviar

Guárdalos en variables de entorno, nunca en el script.

El servicio de alertas por email MQTT en Python

Instala la única dependencia de terceros:

pip install "paho-mqtt>=2.0"

Guarda lo siguiente 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

# --- Ajustes 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")

# --- Ajustes 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 -> (disparo por encima de, normalización por debajo de)
THRESHOLDS = {
    "temperature": (80.0, 75.0),
    "pressure": (3.0, 2.7),
}

outbox: "queue.Queue[EmailMessage]" = queue.Queue()
active = {}      # (device, metric) -> True mientras está en alarma
last_sent = {}   # clave -> marca de tiempo del último email


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
            # enviar siempre la resolución, incluso dentro de la ventana de cooldown
            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()

Cómo funciona la lógica

Todos los callbacks MQTT se ejecutan en el hilo de red de paho, así que active y last_sent solo se tocan desde un hilo y no necesitan bloqueo. El worker SMTP es dueño de la conexión de red con el servidor de correo y reintenta hasta tres veces con esperas crecientes. Si tu servidor de correo es lento o deja de estar disponible un momento, MQTT sigue fluyendo y los correos simplemente llegan más tarde.

La histéresis es la parte más valiosa. Una temperatura que ronda los 80 generaría, si no, una alerta cada pocos segundos al cruzar la línea en ambos sentidos. Con el disparo en 80 y la normalización en 75, recibes una alerta cuando sube y una resolución cuando se ha recuperado de verdad.

Pruébalo de extremo a extremo

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

Publica un valor por encima del límite, luego un valor recuperado y después una alarma del 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"}'

Deberías recibir tres correos: una alerta, una resolución y la alarma crítica. Publica 84.1 varias veces más y verás que no llega nada más; la máquina de estados sabe que la alarma ya está activa.

Resúmenes para instalaciones ruidosas

Los cooldowns evitan que un dispositivo inunde una bandeja de entrada, pero una instalación con cincuenta dispositivos puede seguir generando cincuenta correos distintos durante una bajada de tensión. Cuando muchas alertas suelen llegar juntas, pasa de los correos inmediatos a un resumen. La idea es sencilla: en lugar de poner cada email directamente en la cola de salida, añade una línea a una lista y, cada pocos minutos, el worker envía un único mensaje que resume todo lo ocurrido, agrupado por dispositivo. Las alarmas críticas pueden saltarse el resumen para seguir llegando en segundos.

Un resumen también hace visible la fatiga de alertas. Si los mismos tres sensores aparecen en todos los resúmenes, los umbrales están mal o los sensores necesitan mantenimiento, y eso es una conversación que hay que tener con el equipo, no una regla de filtrado que añadir en el cliente de correo. Revisa el diccionario de umbrales con regularidad y muévelo a un pequeño archivo YAML o JSON cuando personas que no programan necesiten ajustar los límites, para que los cambios no obliguen a editar código.

Reglas de alerta integradas frente a un script en Python

ConnectMQ incluye reglas de alerta por umbral y un registro de alertas en el portal. Son la forma más rápida de ver cuándo un valor ha superado un límite en todo tu broker, sin código que desplegar. Aun así, merece la pena tener un script propio cuando necesitas un comportamiento que el portal no cubre:

Una configuración habitual usa ambos: reglas en el portal para tener visibilidad durante la puesta en marcha y la operación, y un script como este para las notificaciones que deben llegar a la bandeja de entrada de las personas.

Lista de comprobación para resolver problemas

1. SMTPAuthenticationError. La cuenta necesita una contraseña de aplicación o tiene desactivado el envío por SMTP.
2. Timeouts de conexión en el puerto 587 o 465. Algunos proveedores cloud y redes bloquean el SMTP saliente. Prueba el otro puerto o un proveedor transaccional.
3. Los correos acaban en spam. El dominio remitente no tiene registros SPF y DKIM para tu servidor SMTP. Envía desde un dominio que controles y configúralos.
4. Una alarma nunca se resuelve. El valor nunca baja del nivel de normalización o el dispositivo dejó de enviar telemetría. Plantéate una alerta de "datos obsoletos" cuando no llegue ningún mensaje durante N minutos.
5. Alertas duplicadas tras reinicios. El estado vive en memoria. Para despliegues de larga duración, persiste active en un pequeño archivo JSON o en SQLite.

Configuración para producción

Ejecuta el servicio contra un broker privado con credenciales y TLS en el puerto 8883, usando el usuario, la contraseña y el client ID generados en el portal de ConnectMQ. Con un broker MQTT gestionado, la parte del broker la operan por ti, así que tu trabajo se reduce a mantener vivo este único proceso con systemd o con una política de reinicio de contenedores. Si prefieres el chat al email, el bot de Telegram para MQTT usa la misma estructura, y si necesitas un registro de auditoría de cada lectura, combina esto con guardar mensajes MQTT en CSV. Para el comportamiento de reconexión de paho en servicios de larga duración, consulta la gestión de reconexiones en paho-mqtt, y para elegir librerías más allá de Python, la guía de librerías cliente MQTT.

Empieza con ConnectMQ

Configura un broker privado para tu servicio de alertas en minutos. Regístrate en app.connectmq.com, elige el plan Serverless gratuito sin tarjeta de crédito, crea un broker y usa las credenciales generadas en las variables de entorno anteriores. Después añade una regla de alerta por umbral en el portal y compárala con tus alertas en Python, una al lado de la otra.

Preguntas frecuentes

¿Puede paho-mqtt enviar correos directamente?

No. paho-mqtt solo habla MQTT. Los módulos estándar smtplib y email de Python se encargan del envío, y el script une ambos poniendo un email en cola cuando un mensaje cumple una condición.

¿Cómo evito recibir cientos de correos de alerta?

Combina la histéresis (niveles separados de disparo y de normalización) con un cooldown por dispositivo. El script envía una alerta cuando se supera un límite, permanece en silencio mientras la condición persiste y envía una resolución cuando se normaliza.

¿Debo usar QoS 1 en los topics de alarma?

Sí. QoS 1 garantiza que cada alarma se entregue al menos una vez al suscriptor mientras está conectado, y el cooldown absorbe algún duplicado ocasional.

¿Sirve el plan gratuito de ConnectMQ para esto?

Sí. El plan Serverless gratuito admite hasta 25 conexiones y QoS hasta 1, lo que cubre un servicio de alertas y un pequeño conjunto de dispositivos.