AI for Email Automation: Triggers, Parsing & Actions

TL;DR: AI email automation works across three layers: trigger detection (what fires the agent), inbound parsing (extracting structured data from raw MIME), and agent actions (what the agent does next). Getting these layers right requires careful attention to threading, idempotency, and webhook delivery guarantees.

AI for email automation isn't a single capability — it's a pipeline of three distinct engineering layers that must fit together cleanly. The trigger layer detects that something happened. The parsing layer extracts structured signal from raw email. The action layer decides and acts. Collapse these or blur their boundaries and you get agents that hallucinate context, double-send, or miss replies entirely.

This guide covers each layer with concrete mechanisms, code patterns, and the failure modes that bite you in production.

What makes email automation different for AI agents

Traditional email automation is stateless and rule-based: if subject contains "invoice", forward to accounting. AI agent automation is stateful and semantic — the agent maintains conversation context, makes decisions based on message content, and takes actions with real-world consequences like updating a CRM, spawning a sub-agent, or sending a reply that continues a multi-turn thread.

That difference has hard engineering implications:

• Thread identity matters. Agents need to correlate In-Reply-To and References headers to know which conversation a message belongs to. Without this, every inbound email looks orphaned.
• Idempotency is non-negotiable. Webhooks retry. Your agent must not double-process a message or double-send a reply.
• Parsing must be deterministic before reasoning is attempted. LLMs shouldn't be guessing at MIME structure — they should receive clean, pre-extracted text and metadata.

Layer 1: Trigger architecture

Triggers are the events that wake an agent. There are three main patterns.

Inbound webhook triggers

The most common pattern: your email infrastructure parses the inbound MIME message and HTTP-POSTs a structured JSON payload to your webhook endpoint. The agent process handles the webhook synchronously (or enqueues it), then acts.

POST /webhooks/email HTTP/1.1
Content-Type: application/json

{
  "message_id": "<abc123@mail.example.com>",
  "from": "user@example.com",
  "to": ["agent+t-9f3a@yourdomain.ai"],
  "subject": "Re: Invoice #4892",
  "in_reply_to": "<orig-msg-id@yourdomain.ai>",
  "references": ["<orig-msg-id@yourdomain.ai>"],
  "text": "Yes, please go ahead with the approval.",
  "html": "<p>Yes, please go ahead with the approval.</p>",
  "timestamp": "2026-09-05T14:23:00Z"
}

The to address is load-bearing here. If you encode state in the local part (agent+t-9f3a) via a tagged address scheme, you can recover thread context from the address alone — no database lookup required to get started.

Polling triggers (IMAP)

Some setups poll an inbox via IMAP IDLE or periodic FETCH. Latency is lower than most people expect — IMAP IDLE can push within seconds — but it adds operational complexity: you need a persistent connection, reconnection logic, and you're parsing raw MIME yourself.

import imaplib
import email

with imaplib.IMAP4_SSL("imap.yourdomain.ai") as conn:
    conn.login(user, password)
    conn.select("INBOX")
    # Poll unseen messages
    status, data = conn.search(None, "UNSEEN")
    for num in data[0].split():
        _, raw = conn.fetch(num, "(RFC822)")
        msg = email.message_from_bytes(raw[0][1])
        # hand off to parser layer

For production agents, prefer webhooks over polling. Polling introduces artificial latency and the IMAP connection lifecycle is fragile under load.

Schedule + condition triggers

Some agents fire on a schedule (cron) and check conditions — e.g., "every morning, scan the inbox for unresolved support threads older than 24 hours." This is a pull model where the agent queries a mailbox or a database of tracked threads rather than reacting to an event.

This pattern works well for escalation logic but requires careful state tracking to avoid re-processing already-handled threads.

Layer 2: Inbound parsing

Parsing raw MIME correctly is harder than it looks. Email has decades of accumulated complexity: multipart structures, encoded headers, base64 and quoted-printable bodies, nested attachments, inline images, and forwarded messages that look like attachments.

MIME extraction pipeline

A reliable parser needs to handle this in order:

1. Decode headers — Subject, From, To, CC, Date, Message-ID, In-Reply-To, References. All can be encoded (=?UTF-8?B?...?= or =?UTF-8?Q?...?=).
2. Walk the MIME tree — find text/plain first (preferred for LLM input), fall back to text/html if no plain part exists.
3. Strip quoted reply content — agents should reason about the new content, not re-process the entire thread history in every message.
4. Extract attachments — decode base64, identify MIME type, store separately. Don't pass binary blobs to the LLM — parse them first (PDF → text, CSV → rows).
5. Validate SPF/DKIM/DMARC results — check Authentication-Results headers. Never trust the From display name without this.

import email
from email import policy
from email.headerregistry import Address

def extract_message(raw_bytes: bytes) -> dict:
    msg = email.message_from_bytes(raw_bytes, policy=policy.default)
    
    text_body = None
    attachments = []
    
    for part in msg.walk():
        ct = part.get_content_type()
        cd = part.get_content_disposition()
        
        if ct == "text/plain" and cd != "attachment" and text_body is None:
            text_body = part.get_content()
        elif cd == "attachment":
            attachments.append({
                "filename": part.get_filename(),
                "mime_type": ct,
                "data": part.get_payload(decode=True)
            })
    
    return {
        "message_id": msg["message-id"],
        "in_reply_to": msg["in-reply-to"],
        "references": msg["references"],
        "from": str(msg["from"]),
        "subject": str(msg["subject"]),
        "body": text_body,
        "attachments": attachments
    }

Stripping quote threads

Most email clients append the previous message(s) below a -- separator or after a > quoting pattern. Strip this before passing to the LLM or your token count bloats and the model fixates on stale context.

A simple heuristic for plain text:

import re

def strip_quoted_reply(text: str) -> str:
    # Remove lines starting with > (standard quoting)
    lines = text.split("\n")
    clean = []
    for line in lines:
        if line.startswith(">"):
            break # stop at first quoted block
        clean.append(line)
    # Also strip common "On [date], [person] wrote:" patterns
    result = "\n".join(clean)
    result = re.sub(
        r"On .+?wrote:\s*$", "", result, 
        flags=re.DOTALL | re.MULTILINE
    )
    return result.strip()

For HTML emails, use a proper HTML parser (BeautifulSoup or similar) to remove blockquote elements before converting to plain text.

Threading and context recovery

The References header contains the full chain of Message-ID values for a thread, oldest first. Your agent's conversation store should index by Message-ID so any inbound message can recover its thread context in O(1):

def resolve_thread(parsed: dict, store) -> dict:
    refs = parsed.get("references", "").split()
    in_reply_to = parsed.get("in_reply_to", "")
    
    # Try in_reply_to first (direct parent)
    thread = store.get_by_message_id(in_reply_to.strip())
    if not thread:
        # Walk references chain
        for ref in reversed(refs):
            thread = store.get_by_message_id(ref.strip())
            if thread:
                break
    return thread

Layer 3: Classification and agent actions

Email classification sits between parsing and action. Before the agent reasons, classify the message into a category that determines which action handler fires. This keeps your agent logic modular and prevents a single large prompt from handling every possible input type.

Classification schema

A minimal classification schema for a support or workflow agent:

Detecting auto-replies and OOO messages from headers is more reliable than NLP alone:

def is_auto_reply(headers: dict) -> bool:
    return any([
        headers.get("auto-submitted", "").lower() not in ("", "no"),
        headers.get("x-auto-response-suppress") is not None,
        headers.get("precedence", "").lower() in ("bulk", "auto_reply"),
    ])

Agent action handlers

Once classified, route to a handler. Each handler should be a pure function that takes parsed message + thread context and returns an action descriptor, not execute the action directly. This keeps your pipeline testable.

from dataclasses import dataclass
from typing import Literal

@dataclass
class SendReply:
    type: Literal["send_reply"]
    to: str
    subject: str
    body: str
    in_reply_to: str
    references: list[str]

@dataclass 
class UpdateRecord:
    type: Literal["update_record"]
    record_id: str
    fields: dict

@dataclass
class EscalateToHuman:
    type: Literal["escalate"]
    reason: str
    thread_id: str

def handle_approval_response(parsed: dict, thread: dict, llm) -> list:
    intent = llm.classify_intent(
        prompt=f"Did the user approve, reject, or ask for more info?\n\n{parsed['body']}",
        options=["approve", "reject", "clarify"]
    )
    
    if intent == "approve":
        return [
            UpdateRecord(type="update_record", record_id=thread["workflow_id"], fields={"status": "approved"}),
            SendReply(type="send_reply", to=parsed["from"], subject=parsed["subject"],
                      body="Approval recorded. Processing will begin shortly.",
                      in_reply_to=parsed["message_id"], references=thread["message_ids"])
        ]
    elif intent == "reject":
        return [UpdateRecord(type="update_record", record_id=thread["workflow_id"], fields={"status": "rejected"})]
    else:
        return [EscalateToHuman(type="escalate", reason="Ambiguous approval response", thread_id=thread["id"])]

Idempotency at the action layer

Webhooks retry on delivery failure. Your action handlers must be idempotent. The standard approach: store the Message-ID of every processed inbound message in a deduplication table with a TTL of at least 7 days. Check before processing:

import redis

r = redis.Redis()

def process_inbound(payload: dict):
    msg_id = payload["message_id"]
    key = f"processed:{msg_id}"
    
    if r.set(key, "1", nx=True, ex=604800): # 7 day TTL
        # First time — process
        run_agent_pipeline(payload)
    # else: duplicate delivery, silently discard

Putting it together: full pipeline flow

sequenceDiagram
    participant Sender
    participant MX as MX Server
    participant Parser as Parse Layer
    participant Classifier as Classifier
    participant Agent as Agent Handler
    participant Action as Action Executor
    Sender->>MX: SMTP inbound
    MX->>Parser: Webhook POST raw payload
    Parser->>Parser: Extract headers body attachments
    Parser->>Classifier: Structured message object
    Classifier->>Agent: Class plus thread context
    Agent->>Agent: LLM reasoning
    Agent->>Action: Action descriptors
    Action->>Sender: SMTP reply or API call

Delivering agent replies that thread correctly

When your agent sends a reply, it must set In-Reply-To and References correctly or email clients will display it as a new thread. That breaks the user experience and makes conversation history impossible to follow.

For inbound email parsing platforms that give you these headers in the webhook payload, pass them straight through to your outbound send:

import httpx

def send_agent_reply(parsed: dict, reply_body: str, api_key: str):
    # Build References: append our new message ID to the chain
    existing_refs = parsed.get("references", "")
    in_reply_to = parsed["message_id"]
    
    httpx.post(
        "https://api.mails.ai/v1/send",
        headers={"Authorization": f"Bearer {api_key}"},
        json={
            "from": "agent@yourdomain.ai",
            "to": parsed["from"],
            "subject": parsed["subject"] if parsed["subject"].startswith("Re:") else f"Re: {parsed['subject']}",
            "text": reply_body,
            "headers": {
                "In-Reply-To": in_reply_to,
                "References": f"{existing_refs} {in_reply_to}".strip()
            }
        }
    )

Mails.ai's email API accepts custom headers directly in the send payload, which makes threading from agents straightforward without additional SMTP wrangling.

Deliverability for automated senders

Agent-generated replies carry a different risk profile than human email. They send at unusual hours, use templated prose, and can have reply-to addresses that don't match the sending domain. Inbox providers notice all of it.

Foundational requirements:

• SPF: your sending IP must be in the SPF record for the From domain.
• DKIM: sign with a 2048-bit RSA key (or Ed25519). Align the d= domain with the From header domain for DMARC.
• DMARC: publish at minimum p=none with a rua= reporting address so you can monitor alignment failures before tightening policy.
• List hygiene: never reply to addresses that have hard-bounced. Track bounce DSNs and suppress immediately.
• Volume ramping: if your agent sends from a new IP or subdomain, ramp sending volume over 2-4 weeks. Sudden spikes from cold IPs trigger throttling at major providers.

For agents sending at scale, a dedicated IP with a managed warm-up schedule is worth the operational overhead — your agent's reputation won't be polluted by other senders on a shared pool.

Frequently Asked Questions

What's the right way to detect whether an inbound email is an auto-reply?

Check headers first, content second. The Auto-Submitted header (RFC 3834) is the authoritative signal — any value other than no or absent means it's automated. Also check X-Auto-Response-Suppress, Precedence: bulk, and Precedence: auto_reply. Subject-line detection ("Out of Office", "Automatic reply") is a fallback for senders who don't set proper headers, but it generates false positives. Never reply to auto-replies — you'll create a mail loop.

How should agents handle attachments in inbound email?

Never pass raw binary data to an LLM. Extract the attachment, identify its MIME type, and run type-specific parsing before the agent sees it: PDF → text extraction (pdfminer, pymupdf), XLSX/CSV → structured row data, images → OCR or vision model caption. Store the original binary separately and pass only the extracted text or structured data into the agent context. Log the filename and MIME type so the agent can reference them accurately.

What's the minimum DMARC setup for an agent sending domain?

Publish a v=DMARC1; p=none; rua=mailto:dmarc-reports@yourdomain.com record immediately — this starts collecting alignment reports without risking mail rejection. After reviewing reports for 2-4 weeks and confirming SPF and DKIM are aligning correctly on all send paths, move to p=quarantine, then p=reject. Skipping straight to p=reject without reviewing reports first causes legitimate mail loss if any send path lacks proper DKIM signing.

How do I prevent my agent from re-processing the same email twice?

Deduplication on Message-ID is the standard approach. Store each processed Message-ID in Redis or a database with a TTL of at least 7 days (covering typical webhook retry windows). Before running any agent logic, check if the Message-ID exists. If it does, return a 200 to the webhook caller and exit. If it doesn't, write it atomically (using Redis SET NX or a database unique constraint) before processing. This handles both webhook retries and any infrastructure-level duplicate delivery.

Should agents reply from the same address they received on, or a central address?

Reply from the same address that received the message, or at minimum the same domain. If the agent received email at agent+t-9f3a@yourdomain.ai, it should reply from that address or agent@yourdomain.ai — not a completely different domain. Mismatched From and Reply-To domains damage DMARC alignment and look suspicious to both inbox providers and recipients. If you use tagged addresses for thread routing, preserve the tag or store the thread mapping in your database.

What's the difference between classification at the infrastructure layer vs. the LLM layer?

Infrastructure-layer classification uses deterministic rules: header inspection (auto-reply detection, bounce DSN parsing), address pattern matching (tagged address routing), and structured field extraction. Fast, free, and reliable. LLM-layer classification handles semantic content: intent detection, sentiment analysis, entity extraction from free text. Use infrastructure classification first to filter out mechanical cases (auto-replies, bounces, OOO), then apply LLM classification only to messages that require semantic understanding. This cuts LLM token spend and keeps latency low for the easy cases.

Start sending with Mails.ai