Email-Triggered AI Agents: Processing Inbound Emails and Generating Responses

Why Email Remains a Critical Agent Channel

Despite the proliferation of chat tools and ticket systems, email remains the dominant communication channel for business. Over 300 billion emails are sent daily, and most customer inquiries, partner requests, and internal approvals still arrive via email. An AI agent that can process inbound emails, understand intent, and generate contextual responses handles a massive volume of repetitive communication.

The challenge with email agents is complexity. Emails have threading, HTML formatting, attachments, CC lists, and forwarded chains. Building an agent that handles all of this correctly requires careful parsing before the AI reasoning layer even begins.

Two Approaches to Email Ingestion

There are two main ways to feed emails to your agent: webhook-based (services like SendGrid or Mailgun forward parsed emails to your endpoint) and IMAP polling (your agent connects directly to the mailbox).

flowchart LR
    CLIENT(["Client SDK"])
    GW["API Gateway<br/>auth plus rate limit"]
    APP["FastAPI app<br/>handlers and DI"]
    VAL["Pydantic validation"]
    SVC["Service layer<br/>business logic"]
    DB[(Database)]
    QUEUE[(Background queue)]
    OBS[(Tracing)]
    CLIENT --> GW --> APP --> VAL --> SVC
    SVC --> DB
    SVC --> QUEUE
    SVC --> OBS
    SVC --> CLIENT
    style GW fill:#4f46e5,stroke:#4338ca,color:#fff
    style APP fill:#f59e0b,stroke:#d97706,color:#1f2937
    style DB fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b

Webhook-Based Ingestion

from fastapi import FastAPI, Request, BackgroundTasks
from pydantic import BaseModel
from openai import AsyncOpenAI

app = FastAPI()
llm = AsyncOpenAI()

class InboundEmail(BaseModel):
    from_email: str
    from_name: str | None = None
    to: str
    subject: str
    text: str | None = None
    html: str | None = None
    in_reply_to: str | None = None
    message_id: str
    attachments: list[dict] | None = None

@app.post("/email/inbound")
async def receive_email(request: Request, background_tasks: BackgroundTasks):
    form_data = await request.form()
    email = InboundEmail(
        from_email=form_data.get("from", ""),
        from_name=form_data.get("from_name"),
        to=form_data.get("to", ""),
        subject=form_data.get("subject", ""),
        text=form_data.get("text"),
        html=form_data.get("html"),
        in_reply_to=form_data.get("In-Reply-To"),
        message_id=form_data.get("Message-ID", ""),
    )
    background_tasks.add_task(process_inbound_email, email)
    return {"status": "accepted"}

IMAP Polling

import aioimaplib
import email
from email.header import decode_header
import asyncio

async def poll_inbox(interval: int = 30):
    imap = aioimaplib.IMAP4_SSL("imap.gmail.com")
    await imap.wait_hello_from_server()
    await imap.login("agent@example.com", "app-password-here")

    while True:
        await imap.select("INBOX")
        _, message_numbers = await imap.search("UNSEEN")
        nums = message_numbers[0].split()

        for num in nums:
            _, msg_data = await imap.fetch(num, "(RFC822)")
            raw_email = email.message_from_bytes(msg_data[1])
            parsed = parse_raw_email(raw_email)
            await process_inbound_email(parsed)
            await imap.store(num, "+FLAGS", "\\Seen")

        await asyncio.sleep(interval)

Intent Detection

Before generating a response, classify what the sender wants. This determines which workflow the agent triggers.

async def detect_intent(email_obj: InboundEmail) -> dict:
    body = email_obj.text or strip_html(email_obj.html or "")

    prompt = f"""Classify this email's intent. Return a JSON object with:
- intent: one of [support_request, sales_inquiry, meeting_request,
  information_request, complaint, feedback, spam, auto_reply]
- urgency: one of [high, medium, low]
- summary: one sentence summary of what the sender wants
- requires_human: boolean, true if this needs human attention

From: {email_obj.from_email}
Subject: {email_obj.subject}
Body: {body[:2000]}"""

    response = await llm.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
    )
    import json
    return json.loads(response.choices[0].message.content)

Response Generation with Thread Context

For replies, the agent needs the full thread context to avoid repetition and maintain conversation continuity.

async def process_inbound_email(email_obj: InboundEmail):
    if await is_auto_reply(email_obj):
        return

    intent = await detect_intent(email_obj)

    if intent["intent"] == "spam":
        await mark_as_spam(email_obj.message_id)
        return

    if intent["requires_human"]:
        await escalate_to_human(email_obj, intent)
        return

    thread_history = await get_thread_history(email_obj.in_reply_to)
    response_text = await generate_response(email_obj, intent, thread_history)

    await send_reply(
        to=email_obj.from_email,
        subject=f"Re: {email_obj.subject}",
        body=response_text,
        in_reply_to=email_obj.message_id,
        thread_id=email_obj.in_reply_to,
    )
    await store_interaction(email_obj, intent, response_text)

async def generate_response(
    email_obj: InboundEmail,
    intent: dict,
    thread_history: list[dict],
) -> str:
    thread_context = ""
    if thread_history:
        thread_context = "Previous messages in this thread:\n"
        for msg in thread_history[-5:]:
            thread_context += f"- {msg['from']}: {msg['summary']}\n"

    body = email_obj.text or strip_html(email_obj.html or "")

    prompt = f"""Generate a professional email response.

Intent: {intent['intent']}
{thread_context}
Original email from {email_obj.from_name or email_obj.from_email}:
Subject: {email_obj.subject}
Body: {body[:2000]}

Rules:
- Be professional and helpful
- Address the sender's specific question or request
- If you cannot fully resolve the issue, say what you can do and
  set expectations for follow-up
- Keep the response concise (under 200 words)
- Do not make up specific numbers, dates, or policies"""

    response = await llm.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
    )
    return response.choices[0].message.content

Auto-Reply Detection

Prevent infinite email loops by detecting auto-replies and out-of-office messages.

async def is_auto_reply(email_obj: InboundEmail) -> bool:
    auto_headers = ["auto-submitted", "x-auto-response-suppress"]
    subject_patterns = [
        "out of office", "automatic reply",
        "auto-reply", "autoreply", "delivery status",
    ]
    subject_lower = email_obj.subject.lower()
    return any(pattern in subject_lower for pattern in subject_patterns)

FAQ

How do I prevent my email agent from creating infinite reply loops?

Three safeguards: detect auto-reply headers and subjects, maintain a per-address reply counter with a daily limit (e.g., max 3 agent replies per thread), and add a custom header like X-Agent-Generated: true to all outgoing messages so you can filter them on inbound.

Should I use HTML or plain text for agent responses?

Use plain text for initial implementation. HTML emails require careful template rendering and testing across email clients. Once your plain text agent is working reliably, upgrade to HTML templates with a library like mjml or jinja2.

How do I handle email attachments?

Parse attachments separately from the email body. For common file types like PDFs or CSVs, extract text content and include it in the LLM prompt. For images, use a multimodal model. Always validate attachment size and type before processing to prevent abuse.

---

#EmailAutomation #AIAgents #NaturalLanguageProcessing #FastAPI #IMAP #AgenticAI #LearnAI #AIEngineering