Build an Email Monitoring Bot in Python

A monitoring bot watches an inbox and does something when a particular kind of email lands: page someone when Stripe says a payment failed, post to a channel when a customer replies, flag a security alert from a provider. The matching logic is ten lines. The work is in not missing messages and not acting on the same one twice, across restarts, network errors and deploys.

This post builds the bot in Python with httpx and the standard library, in two modes. The first long-polls and needs no public URL, so it runs on a laptop or behind a firewall. The second receives webhooks and uses the first mode's catch-up function as its safety net.

Rules and state

Rules match on sender, subject and the category Lumbox assigns on arrival. State lives in SQLite: one row per email the bot has seen, which gives deduplication and a restart checkpoint from the same table.

import os
import sqlite3
import time
from datetime import datetime, timedelta, timezone

import httpx

API = "https://api.lumbox.co/v1"
INBOX_ID = os.environ["LUMBOX_INBOX_ID"]
SKIP = {"outbound", "welcome"}

client = httpx.Client(
    headers={"X-API-Key": os.environ["LUMBOX_API_KEY"]},
    timeout=httpx.Timeout(10.0, read=90.0),
)

db = sqlite3.connect("monitor.db", check_same_thread=False)
db.execute("CREATE TABLE IF NOT EXISTS seen (email_id TEXT PRIMARY KEY, received_at TEXT)")

RULES = [
    {"name": "payment-failed", "sender": "stripe.com", "subject": "payment failed"},
    {"name": "security-alert", "category": "security"},
    {"name": "customer-reply", "category": "conversation"},
]


def iso(dt):
    return dt.isoformat(timespec="milliseconds").replace("+00:00", "Z")


def first_time(email_id, received_at=None):
    with db:
        cur = db.execute("INSERT OR IGNORE INTO seen VALUES (?, ?)", (email_id, received_at))
    return cur.rowcount == 1


def normalize(e):
    if "parsed" in e:
        return {"id": e["id"], "sender": e["from"]["address"] or "", "subject": e["subject"] or "",
                "category": e["parsed"]["category"], "is_spam": e["is_spam"], "received_at": e["received_at"]}
    return {"id": e["email_id"], "sender": e["from"] or "", "subject": e["subject"] or "",
            "category": e["category"], "is_spam": e.get("is_spam", False), "received_at": None}


def matches(rule, e):
    return (
        rule.get("sender", "") in e["sender"]
        and rule.get("subject", "") in e["subject"].lower()
        and rule.get("category") in (None, e["category"])
    )


def handle(e):
    if e["is_spam"] or e["category"] in SKIP:
        return
    for rule in RULES:
        if matches(rule, e):
            print(f"[{rule['name']}] {e['sender']}: {e['subject']}")

normalize exists because the two sources disagree on shape. A full email from the list or wait endpoints has from as {name, address} and the category under parsed, while a webhook payload has from as a bare address and category at the top level. SKIP drops copies of mail the inbox sent (category outbound) and the message Lumbox seeds into an organisation's first inbox (category welcome). Replace the print with whatever the alert should do.

Trusting the sender

A sender rule is a substring match on an address that anyone can type into a From header, and stripe.com also matches stripe.com.example.net. For rules that page a human, match the full domain after the @, and look at spam_flags on the email, where Lumbox records the SPF result for the envelope sender when a check ran, for example {"spf": "pass"}. The envelope sender can differ from the From header, so a pass proves little, while a fail on a message claiming to come from your payment provider is a good reason to hold the alert for review.

Mode 1: long-poll loop

def catch_up(since):
    r = client.get(f"{API}/inboxes/{INBOX_ID}/emails",
                   params={"since": since, "limit": 100, "include_spam": "true"})
    r.raise_for_status()
    page = r.json()
    if page["has_more"]:
        print(f"warning: over 100 emails since {since}; older ones were not fetched")
    for raw in reversed(page["data"]):
        e = normalize(raw)
        if first_time(e["id"], e["received_at"]):
            handle(e)


def checkpoint():
    row = db.execute("SELECT max(received_at) FROM seen").fetchone()
    return row[0] or iso(datetime.now(timezone.utc))


def wait_for_next(since):
    after = datetime.fromisoformat(since.replace("Z", "+00:00")) + timedelta(milliseconds=1)
    r = client.get(f"{API}/inboxes/{INBOX_ID}/wait", params={"since": iso(after), "timeout": 60})
    if r.status_code == 429:
        time.sleep(r.json().get("retry_after_seconds", 10))
    elif r.status_code not in (200, 408):
        r.raise_for_status()


def run():
    while True:
        try:
            catch_up(checkpoint())
            wait_for_next(checkpoint())
        except httpx.HTTPError as exc:
            print(f"lumbox error: {exc}; retrying in 10s")
            time.sleep(10)


if __name__ == "__main__":
    run()

The loop uses /wait as a wake-up signal, not as the source of emails, and that split is what keeps it from dropping messages. /wait returns only the newest matching email. If two arrive in the same second, a bot that handled only what /wait returned would drop one. Here the list call does the handling: it asks for everything since the checkpoint, oldest first, and first_time filters what was already done.

The checkpoint arithmetic matters too. since is inclusive, so passing the last processed timestamp to /wait would return that same email immediately, forever. Adding one millisecond makes /wait block until something newer exists, and it still returns at once if mail arrived while the bot was busy. A restart resumes from the newest received_at in SQLite.

Three smaller details shape the code. The list endpoint hides spam by default, so the catch-up passes include_spam=true and handle skips flagged mail itself; /wait has no spam filter and would otherwise wake the bot for messages the list never shows. The read timeout (90 seconds) is longer than the 60-second server wait, since httpx's default of 5 seconds would cut every wait short. /wait also marks the email it returns as read, which matters if people use the read flag in the dashboard.

Idle, the bot makes about two requests a minute: one list and one wait that times out after 60 seconds. That is far inside the 120-requests-per-minute limit on an API key. A page of 100 is plenty for a bot that wakes on every arrival; the warning only fires after a long outage.

To try it, add a temporary rule such as {"name": "test", "subject": "monitor test"}, start the bot, and send the inbox an email with that subject from your own account. The alert line should print within a second or two of the message arriving. Stop the bot, send a second one, start it again, and the catch-up prints it on startup.

Mode 2: webhooks with a sweep

Register a webhook with POST /v1/webhooks and store the whsec_ secret from the response. The receiver verifies the HMAC over the raw body, answers straight away, and handles the email in a background task:

import hashlib
import hmac
import json
import os

from fastapi import BackgroundTasks, FastAPI, HTTPException, Request

from monitor import first_time, handle, normalize

app = FastAPI()
SECRET = os.environ["LUMBOX_WEBHOOK_SECRET"].encode()


def process(data):
    e = normalize(data)
    if first_time(e["id"]):
        handle(e)


@app.post("/hooks/lumbox")
async def lumbox_hook(request: Request, background_tasks: BackgroundTasks):
    body = await request.body()
    expected = hmac.new(SECRET, body, hashlib.sha256).hexdigest()
    received = request.headers.get("x-lumbox-signature", "")
    if not hmac.compare_digest(expected.encode(), received.encode()):
        raise HTTPException(status_code=401, detail="bad signature")
    event = json.loads(body)
    if event["event"] == "email.received":
        background_tasks.add_task(process, event["data"])
    return {"ok": True}

hmac.compare_digest runs in constant time, and comparing bytes keeps it from raising on an unexpected header value. The HMAC is computed over request.body(), the bytes exactly as sent, never over re-serialised JSON. Lumbox waits 10 seconds for the response, which is why the work happens after it.

Webhook deliveries are attempted once and not retried, so a deploy or a crash can lose a notification. The email is still stored in the inbox, and the sweep finds it. Run catch_up(iso(datetime.now(timezone.utc) - timedelta(minutes=15))) every five minutes from cron or a background thread. first_time is shared by both paths, so an email the webhook already handled is skipped by the sweep, and one it missed is handled late instead of never.

Webhooks are registered per organisation and fire for every inbox in it. If the org has more inboxes than the one being monitored, add data["inbox_id"] == INBOX_ID to the check in process.

Webhooks vs polling covers when to pick each mode, and the triage post explains how category is assigned and where it misfires, which is worth reading before a rule depends on it. The endpoint parameters used here are in the emails docs.