MQTT Email Alerts Python Tutorial: Rate-Limited SMTP Notifications

MQTT email alerts in Python are a dependable way to make sure the right person hears about a problem, even when nobody is looking at a dashboard. In this tutorial you will build a small service that subscribes to alarm and telemetry topics, evaluates thresholds with hysteresis, and sends SMTP emails with cooldowns and digests so a noisy sensor never floods an inbox. You will also see where a broker's built-in alert rules fit and where a custom script is still the better tool.

What a Good Alert Email Service Needs

Sending an email from on_message takes five lines. Sending emails people do not learn to ignore takes a little more design. A useful service has:

Topic Layout

The service listens to two kinds of topics under a test prefix on the public MQTT test broker:

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

Telemetry is checked against thresholds defined in the script. Alarm messages are events the device decided on its own, and they are forwarded as they are, subject to the cooldown. The public area at broker.connectmq.com is readable by anyone, so choose your own random prefix and keep real addresses and data out of test payloads.

Prepare SMTP Credentials

You need an SMTP server that accepts authenticated submissions. Typical options are your company mail server, a transactional email provider, or a personal mailbox with an app password. Gather:

• Host and port: usually 587 with STARTTLS or 465 with implicit TLS
• Username and password (an app-specific password if the account uses two-factor authentication)
• A sender address the server is allowed to send from

Store them in environment variables, never in the script.

The MQTT Email Alerts Python Service

Install the only third-party dependency:

pip install "paho-mqtt>=2.0"

Save the following as 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 settings ---
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 settings ---
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

# metric -> (trigger above, clear below)
THRESHOLDS = {
    "temperature": (80.0, 75.0),
    "pressure": (3.0, 2.7),
}

outbox: "queue.Queue[EmailMessage]" = queue.Queue()
active = {}      # (device, metric) -> True while in alarm
last_sent = {}   # key -> timestamp of last 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
            # always send the resolution, even inside the cooldown window
            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()

How the Logic Works

All MQTT callbacks run on paho's network thread, so active and last_sent are only touched from one thread and need no lock. The SMTP worker owns the network connection to the mail server and retries up to three times with increasing delays. If your mail server is slow or briefly unreachable, MQTT keeps flowing and emails simply arrive late.

Hysteresis is the most valuable part. A temperature hovering around 80 would otherwise generate an alert every few seconds as it crosses the line in both directions. With a trigger at 80 and a clear at 75, you get one alert when it rises and one resolution when it has genuinely recovered.

Test It End to End

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

Publish a value over the limit, then a recovered value, then a device alarm:

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

You should receive three emails: an alert, a resolution, and the critical alarm. Publish 84.1 several more times and notice that nothing else arrives; the state machine knows the alarm is already active.

Digests for Noisy Sites

Cooldowns stop one device from flooding an inbox, but a site with fifty devices can still produce fifty separate emails during a power dip. When many alerts usually arrive together, switch from immediate emails to a digest. The idea is simple: instead of putting each email straight on the outbox, append a line to a list, and every few minutes the worker sends one message summarising everything that happened, grouped by device. Critical alarms can bypass the digest so they still arrive within seconds.

A digest also makes alert fatigue visible. If the same three sensors appear in every summary, the thresholds are wrong or the sensors need maintenance, and that is a conversation to have with the team rather than a filter rule to add in the mail client. Review the thresholds dictionary regularly, and move it to a small YAML or JSON file once non-developers need to adjust limits, so changes do not require editing code.

Built-In Alert Rules vs a Python Script

ConnectMQ includes threshold alert rules and an alerts log in the portal. These are the fastest way to see when a value crossed a limit across your broker, with no code to deploy. A custom script is still worth having when you need behaviour outside the portal:

A common setup uses both: rules in the portal for visibility during commissioning and operations, and a script like this one for the notifications that must reach people's inboxes.

Troubleshooting Checklist

1. SMTPAuthenticationError. The account needs an app password, or SMTP submission is disabled for it.
2. Connection timeouts on port 587 or 465. Some cloud providers and networks block outbound SMTP. Try the other port or a transactional provider.
3. Emails land in spam. The sender domain lacks SPF and DKIM records for your SMTP server. Send from a domain you control and configure them.
4. An alarm is never resolved. The value never drops below the clear level, or the device stopped sending telemetry. Consider a "stale data" alert when no message arrives for N minutes.
5. Duplicate alerts after restarts. State lives in memory. For long-running deployments, persist active to a small JSON file or SQLite.

Production Setup

Run the service against a private broker with credentials and TLS on port 8883, using the username, password and client ID generated in the ConnectMQ portal. On a managed MQTT broker, the broker side is operated for you, so your job shrinks to keeping this one process alive with systemd or a container restart policy. If you prefer chat to email, the MQTT Telegram bot uses the same structure, and if you need an audit trail of every reading, combine this with logging MQTT messages to CSV. For the reconnect behaviour of paho in long-running services, see paho-mqtt reconnect handling, and for library choices beyond Python, the MQTT client libraries guide.

Get Started with ConnectMQ

Set up a private broker for your alert service in minutes. Sign up at app.connectmq.com, choose the free Serverless plan with no credit card, create a broker and use the generated credentials in the environment variables above. Then add a threshold alert rule in the portal and compare it with your Python alerts side by side.

Frequently Asked Questions

Can paho-mqtt send emails directly?

No. paho-mqtt only speaks MQTT. Python's standard smtplib and email modules handle sending, and the script connects the two by queueing an email when a message matches a condition.

How do I avoid getting hundreds of alert emails?

Combine hysteresis (separate trigger and clear levels) with a per-device cooldown. The script sends one alert when a limit is crossed, stays quiet while the condition persists, and sends one resolution when it clears.

Should I use QoS 1 for alarm topics?

Yes. QoS 1 ensures each alarm is delivered at least once to the subscriber while it is connected, and the cooldown absorbs the occasional duplicate.

Does the free ConnectMQ plan work for this?

Yes. The free Serverless plan supports up to 25 connections and QoS up to 1, which covers an alert service and a small set of devices.