AI Email Auto-Responder with Context

AI Email Auto-Responder with Context

Problem Statement

Support teams, solo founders, and on-call engineers routinely drown in repetitive email. A large fraction of incoming messages — "what are your hours?", "how do I reset my password?", "is the API down?" — can be answered instantly with the right context, but manually triaging every message wastes hours per week. Worse, generic canned responses feel robotic and often miss the specific question the sender asked.

This script watches an IMAP mailbox, pulls recent unread messages, and uses the OpenAI API to draft a context-aware reply. It can operate in dry-run mode (log suggested replies without sending) or live mode (send via SMTP). By feeding the model recent thread history and a user-supplied knowledge base, responses stay grounded in facts rather than hallucinated guesses.

How It Works

The script follows a simple poll → filter → enrich → generate → deliver pipeline. On each run it connects to IMAP over SSL, searches for unread messages from the last N hours, and fetches each one's body plus prior messages in the same thread (matched by In-Reply-To/subject). That thread context, plus a plain-text knowledge base loaded from disk, is packed into a chat-style prompt for gpt-4o-mini. The model returns a reply, which is either logged (dry-run) or sent via SMTP with proper In-Reply-To and References headers so it threads correctly in the recipient's client.

Design choices worth noting: idempotency is achieved by marking messages \Seen after processing, so re-running the script won't double-reply. Dry-run is the default to prevent embarrassing accidental sends. Structured logging makes it easy to audit what the bot said. And the whole thing is dependency-light — only openai beyond the standard library — so it drops cleanly into a cron job or systemd timer on any box with Python 3.10+.

Full Script

#!/usr/bin/env python3
"""
AI Email Auto-Responder with Context

Polls an IMAP mailbox for unread messages, generates context-aware replies
using the OpenAI API, and optionally sends them via SMTP.

Usage:
    python3 email_bot.py --imap mail.example.com --dry-run
"""

from __future__ import annotations

import argparse
import email
import imaplib
import logging
import os
import smtplib
import ssl
import sys
from datetime import datetime, timedelta
from email.header import decode_header, make_header
from email.message import EmailMessage
from email.utils import parseaddr, formataddr, make_msgid
from pathlib import Path
from typing import Iterable

from openai import OpenAI

# ---------------------------------------------------------------------------
# Configuration (env vars with sensible defaults)
# ---------------------------------------------------------------------------
IMAP_HOST = os.environ.get("IMAP_HOST", "")
IMAP_PORT = int(os.environ.get("IMAP_PORT", "993"))
IMAP_USER = os.environ.get("IMAP_USER", "")
IMAP_PASSWORD = os.environ.get("IMAP_PASSWORD", "")
IMAP_FOLDER = os.environ.get("IMAP_FOLDER", "INBOX")

SMTP_HOST = os.environ.get("SMTP_HOST", "")
SMTP_PORT = int(os.environ.get("SMTP_PORT", "587"))
SMTP_USER = os.environ.get("SMTP_USER", IMAP_USER)
SMTP_PASSWORD = os.environ.get("SMTP_PASSWORD", IMAP_PASSWORD)

OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "")
OPENAI_MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")

# Only process messages newer than this many hours.
LOOKBACK_HOURS = int(os.environ.get("LOOKBACK_HOURS", "24"))
# Skip messages from these senders (comma-separated).
SKIP_SENDERS = {s.strip().lower() for s in os.environ.get("SKIP_SENDERS", "").split(",") if s.strip()}
# Path to a plain-text knowledge base injected into the system prompt.
KB_PATH = Path(os.environ.get("KB_PATH", "knowledge_base.txt"))
# Signature appended to every reply.
SIGNATURE = os.environ.get("SIGNATURE", "\n\n— Sent by an automated assistant")

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
log = logging.getLogger("email_bot")


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def decode_mime_header(raw: str | None) -> str:
    """Decode RFC 2047 encoded headers into a plain string."""
    if not raw:
        return ""
    try:
        return str(make_header(decode_header(raw)))
    except Exception:
        return raw


def extract_body(msg: email.message.Message) -> str:
    """Return the plain-text body of a message, falling back to HTML-stripped text."""
    if msg.is_multipart():
        for part in msg.walk():
            ctype = part.get_content_type()
            disp = str(part.get("Content-Disposition") or "")
            if ctype == "text/plain" and "attachment" not in disp:
                payload = part.get_payload(decode=True)
                if payload:
                    return payload.decode(part.get_content_charset() or "utf-8", errors="replace")
        # Fallback: first text/html part, crudely stripped
        for part in msg.walk():
            if part.get_content_type() == "text/html":
                payload = part.get_payload(decode=True)
                if payload:
                    import re
                    html = payload.decode(part.get_content_charset() or "utf-8", errors="replace")
                    return re.sub(r"<[^>]+>", "", html)
        return ""
    payload = msg.get_payload(decode=True)
    if payload is None:
        return ""
    return payload.decode(msg.get_content_charset() or "utf-8", errors="replace")


def load_knowledge_base() -> str:
    """Read the optional knowledge base file used to ground the model."""
    if KB_PATH.is_file():
        return KB_PATH.read_text(encoding="utf-8")
    log.warning("Knowledge base %s not found; proceeding without it", KB_PATH)
    return ""


def fetch_thread_context(imap: imaplib.IMAP4_SSL, msg: email.message.Message) -> str:
    """Fetch up to 3 prior messages in the same thread, oldest-first."""
    subject = decode_mime_header(msg.get("Subject"))
    if not subject:
        return ""
    # Strip common reply/forward prefixes before searching.
    base_subject = subject
    for prefix in ("Re:", "RE:", "Fwd:", "FW:"):
        if base_subject.startswith(prefix):
            base_subject = base_subject[len(prefix):].strip()
    try:
        typ, data = imap.search(None, "SUBJECT", f'"{base_subject}"')
    except imaplib.IMAP4.error:
        return ""
    if typ != "OK" or not data or not data[0]:
        return ""
    ids = data[0].split()[-4:-1]  # up to 3 previous messages
    if not ids:
        return ""
    parts: list[str] = []
    for mid in ids:
        typ, fetched = imap.fetch(mid, "(RFC822)")
        if typ != "OK" or not fetched:
            continue
        for item in fetched:
            if isinstance(item, tuple):
                prior = email.message_from_bytes(item[1])
                sender = decode_mime_header(prior.get("From"))
                body = extract_body(prior).strip()[:1500]
                parts.append(f"From: {sender}\n{body}")
    return "\n\n---\n\n".join(parts)


def build_prompt(sender: str, subject: str, body: str, thread: str, kb: str) -> list[dict]:
    """Construct the OpenAI chat messages list."""
    system = (
        "You are a helpful email assistant. Draft a concise, polite reply to the "
        "incoming message. Use ONLY facts from the knowledge base when answering "
        "specific questions; if the answer is unknown, say a human will follow up. "
        "Do not invent policies, prices, or URLs. Reply in plain text, no markdown."
    )
    if kb:
        system += f"\n\nKNOWLEDGE BASE:\n{kb}"
    user = f"New message from {sender}\nSubject: {subject}\n\n{body.strip()[:4000]}"
    if thread:
        user = f"Prior thread context:\n{thread}\n\n{user}"
    return [
        {"role": "system", "content": system},
        {"role": "user", "content": user},
    ]


def generate_reply(client: OpenAI, messages: list[dict]) -> str:
    """Call the OpenAI API and return the assistant's text reply."""
    resp = client.chat.completions.create(
        model=OPENAI_MODEL,
        messages=messages,
        temperature=0.3,
        max_tokens=500,
    )
    return (resp.choices[0].message.content or "").strip()


def send_reply(original: email.message.Message, reply_text: str) -> None:
    """Send the reply via SMTP, preserving thread headers."""
    to_addr = decode_mime_header(original.get("Reply-To") or original.get("From"))
    from_addr = SMTP_USER
    subject = decode_mime_header(original.get("Subject")) or "(no subject)"
    if not subject.lower().startswith("re:"):
        subject = f"Re: {subject}"

    outgoing = EmailMessage()
    outgoing["From"] = formataddr(("AI Assistant", from_addr))
    outgoing["To"] = to_addr
    outgoing["Subject"] = subject
    outgoing["Message-ID"] = make_msgid()
    if original.get("Message-ID"):
        outgoing["In-Reply-To"] = original["Message-ID"]
        refs = original.get("References", "")
        outgoing["References"] = f"{refs} {original['Message-ID']}".strip()
    outgoing.set_content(reply_text + SIGNATURE)

    context = ssl.create_default_context()
    with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=30) as smtp:
        smtp.starttls(context=context)
        smtp.login(SMTP_USER, SMTP_PASSWORD)
        smtp.send_message(outgoing)


# ---------------------------------------------------------------------------
# Main pipeline
# ---------------------------------------------------------------------------
def process_mailbox(dry_run: bool) -> int:
    """Fetch unread messages, generate replies, optionally send them."""
    if not IMAP_HOST or not IMAP_USER or not IMAP_PASSWORD:
        raise SystemExit("Missing IMAP_HOST / IMAP_USER / IMAP_PASSWORD env vars")
    if not OPENAI_API_KEY:
        raise SystemExit("Missing OPENAI_API_KEY env var")
    if not dry_run and (not SMTP_HOST or not SMTP_USER or not SMTP_PASSWORD):
        raise SystemExit("Live mode requires SMTP_HOST / SMTP_USER / SMTP_PASSWORD")

    client = OpenAI(api_key=OPENAI_API_KEY)
    kb = load_knowledge_base()
    since = (datetime.utcnow() - timedelta(hours=LOOKBACK_HOURS)).strftime("%d-%b-%Y")
    handled = 0

    log.info("Connecting to IMAP %s:%d folder=%s", IMAP_HOST, IMAP_PORT, IMAP_FOLDER)
    with imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT) as imap:
        imap.login(IMAP_USER, IMAP_PASSWORD)
        imap.select(IMAP_FOLDER)
        typ, data = imap.search(None, f'(UNSEEN SINCE "{since}")')
        if typ != "OK":
            log.error("IMAP search failed: %s", data)
            return 0
        ids = data[0].split()
        log.info("Found %d unread message(s)", len(ids))

        for mid in ids:
            typ, fetched = imap.fetch(mid, "(RFC822)")
            if typ != "OK" or not fetched:
                log.warning("Failed to fetch message %s", mid)
                continue
            raw = next((item[1] for item in fetched if isinstance(item, tuple)), None)
            if raw is None:
                continue
            msg = email.message_from_bytes(raw)
            sender = decode_mime_header(msg.get("From"))
            _, sender_email = parseaddr(sender)
            if sender_email.lower() in SKIP_SENDERS:
                log.info("Skipping sender on blocklist: %s", sender_email)
                imap.store(mid, "+FLAGS", "\\Seen")
                continue

            subject = decode_mime_header(msg.get("Subject"))
            body = extract_body(msg)
            if not body.strip():
                log.info("Empty body from %s; skipping", sender_email)
                imap.store(mid, "+FLAGS", "\\Seen")
                continue

            thread = fetch_thread_context(imap, msg)
            messages = build_prompt(sender, subject, body, thread, kb)

            try:
                reply = generate_reply(client, messages)
            except Exception as exc:
                log.exception("OpenAI call failed for %s: %s", sender_email, exc)
                continue

            log.info("--- Draft reply for %s ---\n%s\n---", sender_email, reply)

            if dry_run:
                log.info("[dry-run] Would send reply to %s", sender_email)
            else:
                try:
                    send_reply(msg, reply)
                    log.info("Sent reply to %s", sender_email)
                except Exception as exc:
                    log.exception("SMTP send failed for %s: %s", sender_email, exc)
                    continue

            # Mark as seen so we don't loop on it next run.
            imap.store(mid, "+FLAGS", "\\Seen")
            handled += 1

    return handled


def main(argv: Iterable[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="AI email auto-responder")
    parser.add_argument("--imap", dest="imap_host", help="Override IMAP_HOST env var")
    parser.add_argument("--dry-run", action="store_true",
                        help="Generate replies but do not send them (default)")
    parser.add_argument("--send", action="store_true",
                        help="Actually send replies via SMTP")
    args = parser.parse_args(argv)

    if args.imap_host:
        global IMAP_HOST
        IMAP_HOST = args.imap_host

    # Default to dry-run unless --send is explicitly passed.
    dry_run = not args.send
    try:
        handled = process_mailbox(dry_run=dry_run)
    except imaplib.IMAP4.error as exc:
        log.error("IMAP error: %s", exc)
        return 2
    except smtplib.SMTPException as exc:
        log.error("SMTP error: %s", exc)
        return 3
    log.info("Done. Handled %d message(s).", handled)
    return 0


if __name__ == "__main__":
    sys.exit(main())

Configuration

All configuration is via environment variables so the script slots cleanly into systemd, Docker, or cron:

The knowledge base file is optional but strongly recommended — without it, the model will only be able to answer generic questions and will defer everything else to a human.

Usage Examples

1. Dry-run against your inbox (safe default):

$ export OPENAI_API_KEY=sk-...
$ export IMAP_HOST=imap.gmail.com IMAP_USER=me@example.com IMAP_PASSWORD=app-token
$ python3 email_bot.py --imap imap.gmail.com
2025-01-15 10:12:04 INFO Connecting to IMAP imap.gmail.com:993 folder=INBOX
2025-01-15 10:12:05 INFO Found 3 unread message(s)
2025-01-15 10:12:09 INFO --- Draft reply for alice@customer.com ---
Hi Alice, our support hours are Mon–Fri 9am–6pm ET...
---
2025-01-15 10:12:09 INFO [dry-run] Would send reply to alice@customer.com
2025-01-15 10:12:11 INFO Done. Handled 3 message(s).

2. Live send with a knowledge base:

$ KB_PATH=/etc/email_bot/kb.txt SIGNATURE="— The Support Team" \
  python3 email_bot.py --send
2025-01-15 10:20:00 INFO Sent reply to alice@customer.com
2025-01-15 10:20:02 INFO Done. Handled 1 message(s).

3. Cron entry running every 15 minutes:

*/15 * * * * /usr/bin/python3 /opt/email_bot/email_bot.py --send >> /var/log/email_bot.log 2>&1

Error Handling

• Missing credentials — the script exits early with a clear SystemExit message rather than failing mid-run.
• IMAP search/fetch failures — logged as warnings; the script continues to the next message so one bad email doesn't abort the batch.
• OpenAI API errors (rate limits, timeouts, invalid key) — caught per-message, logged with traceback, and the message is left unread so it will be retried next run.
• SMTP failures — also caught per-message; the message stays unread for retry rather than being silently dropped.
• Idempotency — successfully processed messages are flagged \Seen so a re-run won't duplicate replies. Messages that fail generation or sending are not marked seen.
• Blocked senders — messages from SKIP_SENDERS are marked seen without a reply to avoid loops with auto-responders.

Extending Ideas

• Per-sender policies — replace SKIP_SENDERS with a small YAML config mapping senders/domains to templates, languages, or "always escalate to human."
• Attachment awareness — extract text from PDFs or CSVs in incoming mail and include it in the prompt so the bot can answer questions about invoices or logs.
• Human-in-the-loop queue — instead of sending, write drafts to a Drafts IMAP folder so a human can review and hit send.
• Metrics & alerting — emit Prometheus counters (messages_processed, replies_sent, openai_errors) and alert if the error rate spikes, which usually signals an expired API key or a mailbox outage.