Building an AI Email Assistant with APIs: Architecture, Code, and Best Practices

+-------------------+        +------------------+        +------------------+
| Email Provider    |  Push  | Ingestion Layer  |  Msg   | Work Queue       |
| (Gmail/Graph/IMAP)| -----> | (Webhook/Watcher)| -----> | (SQS/PubSub/Kafka)|
+-------------------+        +------------------+        +------------------+
                                                        |  Dead-letter Q    |
                                                        +---------+--------+
                                                                  |
                                                                  v
                                                        +------------------+
                                                        | Orchestrator     |
                                                        | (Functions/      |
                                                        |  Workers)        |
                                                        +----+--------+----+
                                                             |        |
                        +---------------------+              |        |
                        | Tools & Integrations| <------------+        |
                        | (Calendar/CRM/DB/   |                       |
                        |  Search/RAG)        |                       |
                        +----------+----------+                       |
                                   |                                  |
                                   v                                  v
                            +--------------+                   +--------------+
                            | LLM Gateway  | <---------------> | Policy/Guard |
                            | (provider(s))|                   | Rails        |
                            +------+-------+                   +------+-------+
                                   |                                  |
                                   v                                  v
                            +--------------+                   +--------------+
                            | Outbox/Approval                  | Observability|
                            | Workflow (HITL)  | ------------> | (logs, traces|
                            +--------------+                   |  metrics)    |
                                                              +--------------+

import base64, json, os, time
from email.message import EmailMessage
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
import requests

GMAIL_SCOPES = [
    "https://www.googleapis.com/auth/gmail.readonly",
    "https://www.googleapis.com/auth/gmail.send",
    "https://www.googleapis.com/auth/gmail.modify",
]

LLM_ENDPOINT = os.environ["LLM_ENDPOINT"]  # e.g., https://api.your-llm.com/v1/chat/completions
LLM_KEY = os.environ["LLM_KEY"]

creds = Credentials.from_authorized_user_file("token.json", GMAIL_SCOPES)
gmail = build("gmail", "v1", credentials=creds)

SCHEMA = {
    "type": "object",
    "properties": {
        "intent": {"type": "string", "enum": [
            "faq", "support", "sales", "scheduling", "spam", "other"
        ]},
        "confidence": {"type": "number"},
        "entities": {"type": "object"},
        "draft": {"type": "string"},
        "auto_send_ok": {"type": "boolean"}
    },
    "required": ["intent", "confidence", "draft", "auto_send_ok"]
}

SYSTEM_PROMPT = (
    "You are an email assistant. Classify intent, extract entities, and draft a concise,"
    " on-brand reply. Obey policy: never invent facts, cite provided context snippets,"
    " and only set auto_send_ok=true for routine FAQs with >=0.85 confidence."
)


def call_llm(thread_context: str) -> dict:
    payload = {
        "model": "your-model-name",
        "messages": [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": thread_context}
        ],
        "temperature": 0.2,
        "response_format": {"type": "json_schema", "json_schema": {"name": "email_decision", "schema": SCHEMA}}
    }
    r = requests.post(LLM_ENDPOINT, headers={"Authorization": f"Bearer {LLM_KEY}"}, json=payload, timeout=30)
    r.raise_for_status()
    return json.loads(r.json()["choices"][0]["message"]["content"])  # provider-dependent


def fetch_unread_messages(user_id="me", max_results=10):
    res = gmail.users().messages().list(userId=user_id, q="is:unread in:inbox", maxResults=max_results).execute()
    return res.get("messages", [])


def get_thread_text(message_id: str) -> str:
    msg = gmail.users().messages().get(userId="me", id=message_id, format="full").execute()
    headers = {h['name'].lower(): h['value'] for h in msg.get('payload', {}).get('headers', [])}
    subject = headers.get('subject', '(no subject)')
    frm = headers.get('from', '')
    snippet = msg.get('snippet', '')
    # Extract plain text body
    def walk(p):
        if p.get('mimeType') == 'text/plain' and 'data' in p.get('body', {}):
            return base64.urlsafe_b64decode(p['body']['data']).decode('utf-8', errors='ignore')
        for part in p.get('parts', []) or []:
            t = walk(part)
            if t:
                return t
        return ''
    body = walk(msg.get('payload', {}))
    return f"From: {frm}\nSubject: {subject}\n\nSnippet: {snippet}\n\nBody:\n{body[:8000]}"


def send_reply(original_id: str, to_addr: str, subject: str, body_text: str):
    message = EmailMessage()
    message['To'] = to_addr
    message['Subject'] = f"Re: {subject}"
    message['In-Reply-To'] = original_id
    message.set_content(body_text)
    raw = base64.urlsafe_b64encode(message.as_bytes()).decode()
    gmail.users().messages().send(userId="me", body={"raw": raw}).execute()


def process():
    for m in fetch_unread_messages():
        mid = m['id']
        context = get_thread_text(mid)
        decision = call_llm(context)
        # Mark as processed to avoid duplicates
        gmail.users().messages().modify(userId="me", id=mid, body={"removeLabelIds": ["UNREAD"], "addLabelIds": ["STARRED"]}).execute()
        # Route
        if decision.get('intent') == 'spam':
            gmail.users().messages().trash(userId="me", id=mid).execute()
            continue
        # HITL gate example
        if decision.get('auto_send_ok') and decision.get('confidence', 0) >= 0.85:
            # Extract reply target and subject from headers again for reliability
            # (left as exercise to parse From/Subject from headers)
            print("Auto-sending reply...")
            # send_reply(mid, to_addr, subject, decision['draft'])
        else:
            # enqueue for approval UI with decision + draft
            print(json.dumps({"message_id": mid, "decision": decision}, indent=2))
        time.sleep(0.25)  # basic pacing; replace with async workers

if __name__ == "__main__":
    process()

import express from "express";
import Ajv from "ajv";
import fetch from "node-fetch";

const app = express();
app.use(express.json({ limit: "1mb" }));

const schema = { type: "object", properties: { intent: { type: "string" }, draft: { type: "string" } }, required: ["intent", "draft"] };
const ajv = new Ajv();
const validate = ajv.compile(schema);

app.post("/classify-draft", async (req, res) => {
  const { threadContext } = req.body;
  const r = await fetch(process.env.LLM_ENDPOINT!, {
    method: "POST",
    headers: { Authorization: `Bearer ${process.env.LLM_KEY}`, "Content-Type": "application/json" },
    body: JSON.stringify({ model: "your-model", messages: [{ role: "system", content: "You are an email assistant." }, { role: "user", content: threadContext }], response_format: { type: "json_schema", json_schema: { name: "out", schema } } })
  });
  const data = await r.json();
  const content = JSON.parse(data.choices[0].message.content);
  if (!validate(content)) return res.status(422).json({ error: validate.errors });
  res.json(content);
});

app.listen(3000, () => console.log("up on :3000"));