MQTT-E-Mail-Benachrichtigung mit Python: SMTP-Alarme mit Ratenbegrenzung

Eine MQTT-E-Mail-Benachrichtigung mit Python ist ein zuverlässiger Weg, damit die richtige Person von einem Problem erfährt, auch wenn niemand auf ein Dashboard schaut, etwa die Rufbereitschaft der Instandhaltung am Wochenende, wenn ein Härteofen überhitzt oder der Druck in einer Druckluftleitung abfällt. In diesem Tutorial bauen Sie einen kleinen Dienst, der Alarm- und Telemetrie-Topics abonniert, Schwellwerte mit Hysterese auswertet und SMTP-E-Mails mit Abklingzeiten und Sammelberichten versendet, damit ein lauter Sensor nie ein Postfach überflutet. Außerdem sehen Sie, wo die eingebauten Alarmregeln eines Brokers passen und wo ein eigenes Skript weiterhin das bessere Werkzeug ist.

Was ein guter E-Mail-Alarmdienst braucht

Eine E-Mail aus on_message zu senden, braucht fünf Zeilen. E-Mails zu senden, die Menschen nicht zu ignorieren lernen, braucht etwas mehr Design. Ein nützlicher Dienst hat:

Topic-Struktur

Der Dienst lauscht auf zwei Arten von Topics unter einem Testpräfix auf dem öffentlichen MQTT-Broker:

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

Telemetrie wird gegen die im Skript definierten Schwellwerte geprüft. Alarmnachrichten sind Ereignisse, die das Gerät selbst festgestellt hat, und werden unter Beachtung der Abklingzeit unverändert weitergeleitet. Der öffentliche MQTT-Broker unter broker.connectmq.com ist für Tests und Prototypen gedacht und für jeden lesbar, wählen Sie also ein eigenes zufälliges Präfix und halten Sie echte Adressen und Daten aus Test-Payloads heraus.

SMTP-Zugangsdaten vorbereiten

Sie benötigen einen SMTP-Server, der authentifizierte Einlieferungen akzeptiert. Typische Optionen sind der Mailserver Ihres Unternehmens, ein Anbieter für Transaktions-E-Mails oder ein persönliches Postfach mit App-Passwort. Sammeln Sie:

• Host und Port: meist 587 mit STARTTLS oder 465 mit implizitem TLS
• Benutzername und Passwort (ein App-spezifisches Passwort, wenn das Konto Zwei-Faktor-Authentifizierung nutzt)
• Eine Absenderadresse, von der der Server senden darf

Speichern Sie diese in Umgebungsvariablen, niemals im Skript.

Der Python-Dienst für MQTT-E-Mail-Benachrichtigungen

Installieren Sie die einzige Fremdabhängigkeit:

pip install "paho-mqtt>=2.0"

Speichern Sie Folgendes als 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

# --- MQTT-Einstellungen ---
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")

# --- SMTP-Einstellungen ---
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

# Metrik -> (auslösen oberhalb, zurücksetzen unterhalb)
THRESHOLDS = {
    "temperature": (80.0, 75.0),
    "pressure": (3.0, 2.7),
}

outbox: "queue.Queue[EmailMessage]" = queue.Queue()
active = {}      # (Gerät, Metrik) -> True, solange im Alarm
last_sent = {}   # Schlüssel -> Zeitstempel der letzten 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
            # die Entwarnung immer senden, auch innerhalb der Abklingzeit
            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()

Wie die Logik funktioniert

Alle MQTT-Callbacks laufen auf dem Netzwerk-Thread von paho, daher werden active und last_sent nur von einem Thread berührt und brauchen keinen Lock. Der SMTP-Worker besitzt die Netzwerkverbindung zum Mailserver und versucht es bis zu dreimal mit wachsenden Pausen. Ist Ihr Mailserver langsam oder kurz nicht erreichbar, fließt MQTT weiter, und E-Mails kommen einfach später an.

Die Hysterese ist der wertvollste Teil. Eine Temperatur, die um 80 schwankt, würde sonst alle paar Sekunden einen Alarm erzeugen, weil sie die Grenze in beide Richtungen überschreitet. Mit einer Auslöseschwelle bei 80 und einer Rücksetzschwelle bei 75 erhalten Sie einen Alarm beim Anstieg und eine Entwarnung, wenn sich der Wert wirklich erholt hat.

Durchgängig testen

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

Publizieren Sie einen Wert über dem Grenzwert, dann einen erholten Wert und dann einen Gerätealarm:

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

Sie sollten drei E-Mails erhalten: einen Alarm, eine Entwarnung und den kritischen Alarm. Publizieren Sie 84.1 noch mehrmals und stellen Sie fest, dass nichts weiter ankommt; der Zustandsautomat weiß, dass der Alarm bereits aktiv ist.

Sammelberichte für laute Standorte

Abklingzeiten verhindern, dass ein Gerät ein Postfach überflutet, doch ein Standort mit fünfzig Geräten kann bei einem Spannungseinbruch immer noch fünfzig einzelne E-Mails erzeugen. Kommen viele Alarme meist gleichzeitig, wechseln Sie von sofortigen E-Mails zu einem Sammelbericht. Die Idee ist einfach: Statt jede E-Mail direkt in den Ausgang zu legen, hängen Sie eine Zeile an eine Liste an, und alle paar Minuten sendet der Worker eine Nachricht, die alles Geschehene nach Gerät gruppiert zusammenfasst. Kritische Alarme können den Sammelbericht umgehen, damit sie weiterhin innerhalb von Sekunden ankommen.

Ein Sammelbericht macht außerdem Alarmmüdigkeit sichtbar. Tauchen in jeder Zusammenfassung dieselben drei Sensoren auf, sind die Schwellwerte falsch oder die Sensoren brauchen Wartung, und darüber sollte das Team sprechen, statt im Mailprogramm eine Filterregel anzulegen. Überprüfen Sie das Dictionary der Schwellwerte regelmäßig und verschieben Sie es in eine kleine YAML- oder JSON-Datei, sobald auch Nicht-Entwickler Grenzwerte anpassen müssen, damit Änderungen keine Codebearbeitung erfordern.

Eingebaute Alarmregeln im Vergleich zu einem Python-Skript

ConnectMQ enthält im Portal Schwellwert-Alarmregeln und ein Alarmprotokoll. Sie sind der schnellste Weg, um über Ihren gesamten Broker hinweg zu sehen, wann ein Wert einen Grenzwert überschritten hat, ganz ohne Code-Deployment. Ein eigenes Skript lohnt sich trotzdem, wenn Sie Verhalten außerhalb des Portals brauchen:

Ein gängiges Setup nutzt beides: Regeln im Portal für den Überblick während Inbetriebnahme und Betrieb und ein Skript wie dieses für Benachrichtigungen, die die Postfächer der Menschen erreichen müssen.

Checkliste zur Fehlerbehebung

1. SMTPAuthenticationError. Das Konto benötigt ein App-Passwort, oder die SMTP-Einlieferung ist dafür deaktiviert.
2. Verbindungs-Timeouts auf Port 587 oder 465. Manche Cloud-Anbieter und Netzwerke blockieren ausgehendes SMTP. Versuchen Sie den anderen Port oder einen Anbieter für Transaktions-E-Mails.
3. E-Mails landen im Spam. Für die Absenderdomain fehlen SPF- und DKIM-Einträge für Ihren SMTP-Server. Senden Sie von einer Domain, die Sie kontrollieren, und konfigurieren Sie diese Einträge.
4. Ein Alarm wird nie entwarnt. Der Wert fällt nie unter die Rücksetzschwelle, oder das Gerät sendet keine Telemetrie mehr. Erwägen Sie einen Alarm für "veraltete Daten", wenn N Minuten lang keine Nachricht ankommt.
5. Doppelte Alarme nach Neustarts. Der Zustand liegt im Arbeitsspeicher. Für lang laufende Deployments persistieren Sie active in einer kleinen JSON-Datei oder in SQLite.

Setup für die Produktion

Betreiben Sie den Dienst gegen einen privaten Broker mit Zugangsdaten und TLS auf Port 8883 und verwenden Sie dabei den im ConnectMQ-Portal generierten Benutzernamen, das Passwort und die Client ID. Bei einem verwalteten MQTT-Broker wird die Broker-Seite für Sie betrieben, sodass sich Ihre Aufgabe darauf beschränkt, diesen einen Prozess mit systemd oder einer Container-Restart-Policy am Leben zu halten. Bevorzugen Sie Chat statt E-Mail, nutzt der MQTT Telegram Bot dieselbe Struktur, und brauchen Sie einen Prüfpfad jedes Messwerts, kombinieren Sie dies mit dem Protokollieren von MQTT-Nachrichten als CSV. Zum Reconnect-Verhalten von paho in lang laufenden Diensten lesen Sie Reconnect-Handling mit paho-mqtt, und zur Bibliothekswahl jenseits von Python den Leitfaden zu MQTT-Client-Bibliotheken.

Starten Sie mit ConnectMQ

Richten Sie in wenigen Minuten einen privaten Broker für Ihren Alarmdienst ein. Registrieren Sie sich unter app.connectmq.com, wählen Sie den kostenlosen Serverless-Plan ohne Kreditkarte, erstellen Sie einen Broker und verwenden Sie die generierten Zugangsdaten in den Umgebungsvariablen oben. Legen Sie dann im Portal eine Schwellwert-Alarmregel an und vergleichen Sie sie direkt mit Ihren Python-Alarmen.

Häufig gestellte Fragen

Kann paho-mqtt direkt E-Mails senden?

Nein. paho-mqtt spricht nur MQTT. Die Standardmodule smtplib und email von Python übernehmen den Versand, und das Skript verbindet beides, indem es eine E-Mail in die Queue legt, wenn eine Nachricht eine Bedingung erfüllt.

Wie vermeide ich Hunderte Alarm-E-Mails?

Kombinieren Sie Hysterese (getrennte Auslöse- und Rücksetzwerte) mit einer Abklingzeit pro Gerät. Das Skript sendet einen Alarm, wenn ein Grenzwert überschritten wird, bleibt still, solange der Zustand anhält, und sendet eine Entwarnung, wenn er endet.

Sollte ich für Alarm-Topics QoS 1 verwenden?

Ja. QoS 1 stellt sicher, dass jeder Alarm dem Subscriber mindestens einmal zugestellt wird, solange er verbunden ist, und die Abklingzeit fängt gelegentliche Duplikate ab.

Funktioniert der kostenlose ConnectMQ-Plan dafür?

Ja. Der kostenlose Serverless-Plan unterstützt bis zu 25 Verbindungen und QoS bis 1, was für einen Alarmdienst und eine kleine Zahl von Geräten ausreicht.