Tutoriel alertes e-mail MQTT en Python : notifications SMTP à débit limité

Les alertes e-mail MQTT en Python sont un moyen fiable de s'assurer que la bonne personne est informée d'un problème, même lorsque personne ne regarde un tableau de bord. Dans ce tutoriel, vous allez construire un petit service qui s'abonne aux topics d'alarme et de télémétrie, évalue des seuils avec hystérésis et envoie des e-mails SMTP avec délais de carence et récapitulatifs, pour qu'un capteur bruyant n'inonde jamais une boîte de réception. Vous verrez aussi où se situent les règles d'alerte intégrées d'un broker et dans quels cas un script personnalisé reste le meilleur outil.

Ce dont un bon service d'alertes e-mail a besoin

Envoyer un e-mail depuis on_message tient en cinq lignes. Envoyer des e-mails que les gens n'apprennent pas à ignorer demande un peu plus de conception. Un service utile comporte :

Organisation des topics

Le service écoute deux types de topics sous un préfixe de test sur le serveur MQTT public gratuit :

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

La télémétrie est comparée aux seuils définis dans le script. Les messages d'alarme sont des événements que l'appareil a décidés lui-même : ils sont transmis tels quels, sous réserve du délai de carence. La zone publique de broker.connectmq.com est lisible par n'importe qui et réservée aux tests : choisissez votre propre préfixe aléatoire et gardez les vraies adresses et données hors des payloads de test.

Préparer les identifiants SMTP

Il vous faut un serveur SMTP qui accepte les envois authentifiés. Les options typiques sont le serveur de messagerie de votre entreprise, un fournisseur d'e-mails transactionnels ou une boîte personnelle avec un mot de passe d'application. Rassemblez :

• L'hôte et le port : généralement 587 avec STARTTLS ou 465 avec TLS implicite
• Le nom d'utilisateur et le mot de passe (un mot de passe d'application si le compte utilise l'authentification à deux facteurs)
• Une adresse d'expéditeur que le serveur est autorisé à utiliser

Stockez-les dans des variables d'environnement, jamais dans le script.

Le service d'alertes e-mail MQTT en Python

Installez la seule dépendance tierce :

pip install "paho-mqtt>=2.0"

Enregistrez ce qui suit sous 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

# --- Paramètres 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")

# --- Paramètres 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étrique -> (déclenchement au-dessus de, retour en dessous de)
THRESHOLDS = {
    "temperature": (80.0, 75.0),
    "pressure": (3.0, 2.7),
}

outbox: "queue.Queue[EmailMessage]" = queue.Queue()
active = {}      # (appareil, métrique) -> True tant que l'alarme est active
last_sent = {}   # clé -> horodatage du dernier 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
            # toujours envoyer la résolution, même pendant le délai de carence
            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()

Comment fonctionne la logique

Tous les callbacks MQTT s'exécutent sur le thread réseau de paho : active et last_sent ne sont donc manipulés que depuis un seul thread et n'ont besoin d'aucun verrou. Le worker SMTP possède la connexion réseau au serveur de messagerie et réessaie jusqu'à trois fois avec des délais croissants. Si votre serveur de messagerie est lent ou brièvement injoignable, MQTT continue de circuler et les e-mails arrivent simplement en retard.

L'hystérésis est la partie la plus précieuse. Une température oscillant autour de 80 générerait sinon une alerte toutes les quelques secondes, en franchissant la limite dans les deux sens. Avec un déclenchement à 80 et un retour à 75, vous obtenez une alerte lors de la hausse et une résolution lorsque la situation est réellement rétablie.

Tester de bout en bout

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

Publiez une valeur au-dessus de la limite, puis une valeur rétablie, puis une alarme d'appareil :

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

Vous devriez recevoir trois e-mails : une alerte, une résolution et l'alarme critique. Publiez 84.1 plusieurs fois de plus et constatez que rien d'autre n'arrive ; la machine à états sait que l'alarme est déjà active.

Des récapitulatifs pour les sites bruyants

Les délais de carence empêchent un appareil d'inonder une boîte de réception, mais un site de cinquante appareils peut encore produire cinquante e-mails distincts lors d'une microcoupure de courant. Lorsque de nombreuses alertes arrivent généralement ensemble, passez des e-mails immédiats à un récapitulatif. L'idée est simple : au lieu de placer chaque e-mail directement dans la file d'envoi, ajoutez une ligne à une liste, et toutes les quelques minutes le worker envoie un message unique résumant tout ce qui s'est passé, regroupé par appareil. Les alarmes critiques peuvent contourner le récapitulatif pour continuer d'arriver en quelques secondes.

Un récapitulatif rend aussi visible la lassitude face aux alertes. Si les trois mêmes capteurs apparaissent dans chaque résumé, les seuils sont mal réglés ou les capteurs ont besoin de maintenance, et c'est une discussion à avoir avec l'équipe plutôt qu'une règle de filtrage à ajouter dans le client de messagerie. Revoyez régulièrement le dictionnaire des seuils, et déplacez-le dans un petit fichier YAML ou JSON dès que des non-développeurs doivent ajuster les limites, pour que les changements ne nécessitent pas de modifier le code.

Règles d'alerte intégrées ou script Python

ConnectMQ inclut des règles d'alerte à seuil et un journal des alertes dans le portail. C'est le moyen le plus rapide de voir quand une valeur a franchi une limite sur l'ensemble de votre broker, sans code à déployer. Un script personnalisé reste utile lorsque vous avez besoin d'un comportement hors du périmètre du portail :

Une configuration courante utilise les deux : les règles du portail pour la visibilité pendant la mise en service et l'exploitation, et un script comme celui-ci pour les notifications qui doivent atteindre les boîtes de réception.

Liste de dépannage

1. SMTPAuthenticationError. Le compte nécessite un mot de passe d'application, ou l'envoi SMTP est désactivé pour lui.
2. Délais d'expiration de connexion sur le port 587 ou 465. Certains fournisseurs cloud et réseaux bloquent le SMTP sortant. Essayez l'autre port ou un fournisseur transactionnel.
3. Les e-mails arrivent dans les spams. Le domaine expéditeur n'a pas d'enregistrements SPF et DKIM pour votre serveur SMTP. Envoyez depuis un domaine que vous contrôlez et configurez-les.
4. Une alarme n'est jamais résolue. La valeur ne descend jamais sous le niveau de retour, ou l'appareil a cessé d'envoyer de la télémétrie. Envisagez une alerte « données périmées » lorsqu'aucun message n'arrive pendant N minutes.
5. Alertes en double après les redémarrages. L'état vit en mémoire. Pour les déploiements de longue durée, persistez active dans un petit fichier JSON ou dans SQLite.

Configuration de production

Exécutez le service sur un broker privé avec identifiants et TLS sur le port 8883, en utilisant le username, le mot de passe et le client ID générés dans le portail ConnectMQ. Avec un broker MQTT managé, la partie broker est exploitée pour vous : votre travail se réduit à maintenir ce processus en vie avec systemd ou une politique de redémarrage de conteneur. Si vous préférez le chat à l'e-mail, le bot Telegram MQTT utilise la même structure, et si vous avez besoin d'une trace d'audit de chaque relevé, combinez ce service avec l'enregistrement des messages MQTT en CSV. Pour le comportement de reconnexion de paho dans les services de longue durée, consultez la gestion des reconnexions paho-mqtt, et pour le choix de bibliothèques au-delà de Python, le guide des bibliothèques clientes MQTT.

Démarrer avec ConnectMQ

Mettez en place un broker MQTT managé privé pour votre service d'alertes en quelques minutes. Inscrivez-vous sur app.connectmq.com, choisissez le plan Serverless gratuit sans carte bancaire, créez un broker et utilisez les identifiants générés dans les variables d'environnement ci-dessus. Ajoutez ensuite une règle d'alerte à seuil dans le portail et comparez-la côte à côte avec vos alertes Python.

Questions fréquentes

paho-mqtt peut-il envoyer des e-mails directement ?

Non. paho-mqtt ne parle que MQTT. Les modules standard smtplib et email de Python gèrent l'envoi, et le script relie les deux en mettant un e-mail en file lorsqu'un message remplit une condition.

Comment éviter de recevoir des centaines d'e-mails d'alerte ?

Combinez l'hystérésis (niveaux de déclenchement et de retour distincts) avec un délai de carence par appareil. Le script envoie une alerte lorsqu'une limite est franchie, reste silencieux tant que la condition persiste et envoie une résolution lorsqu'elle disparaît.

Faut-il utiliser le QoS 1 pour les topics d'alarme ?

Oui. Le QoS 1 garantit que chaque alarme est livrée au moins une fois au subscriber tant qu'il est connecté, et le délai de carence absorbe les doublons occasionnels.

Le plan ConnectMQ gratuit convient-il pour cela ?

Oui. Le plan Serverless gratuit prend en charge jusqu'à 25 connexions et le QoS jusqu'à 1, ce qui couvre un service d'alertes et un petit ensemble d'appareils.