A personal AI assistant email triage system needs to do more than filter spam. It must classify sender intent, summarize threads, and propose concrete actions with enough precision that you trust it to modify your inbox. The following steps build a minimal but production-shaped pipeline for personal AI assistant email triage using standard IMAP and an OpenAI-compatible LLM endpoint.
Step 1: Establish authenticated email access
Use IMAP with an app-specific password or OAuth2. For a personal assistant, IMAP over SSL with a dedicated password is simplest and avoids Google’s OAuth dance for a local script.
import imaplib
IMAP_HOST = "imap.gmail.com"
EMAIL = "you@example.com"
APP_PASSWORD = "xxxx xxxx xxxx xxxx" # 16-char Gmail app password
def connect_imap():
conn = imaplib.IMAP4_SSL(IMAP_HOST)
conn.login(EMAIL, APP_PASSWORD)
conn.select("INBOX")
return conn
if __name__ == "__main__":
imap = connect_imap()
print(imap.search(None, "UNSEEN")[1])If you run this and see a byte string of message IDs, authentication works. Never hardcode credentials in source; load from environment variables.
Step 2: Fetch and normalize the unread batch
Pull the last 20 unseen messages, parse headers and plain text, and flatten into a dict. The LLM should receive clean text, not raw MIME.
import email
from email.header import decode_header
from typing import Dict, List
def fetch_unseen(imap: imaplib.IMAP4_SSL, limit: int = 20) -> List[Dict]:
_, msg_ids = imap.search(None, "UNSEEN")
ids = msg_ids[0].split()[-limit:]
messages = []
for mid in ids:
_, data = imap.fetch(mid, "(RFC822)")
raw = data[0][1]
msg = email.message_from_bytes(raw)
subject = decode_header(msg["Subject"])[0][0]
if isinstance(subject, bytes):
subject = subject.decode()
body = ""
if msg.is_multipart():
for part in msg.walk():
if part.get_content_type() == "text/plain":
body += part.get_payload(decode=True).decode(errors="ignore")
else:
body = msg.get_payload(decode=True).decode(errors="ignore")
messages.append({
"id": mid.decode(),
"from": msg["From"],
"subject": subject,
"body": body[:2000] # truncate to keep token count sane
})
return messagesTruncating at 2,000 characters covers most triage needs. Long threads can be summarized later by the model if needed.
Step 3: Define a strict triage schema
The model must return structured output. Define a JSON schema and enforce it with response_format or post-parse validation. A good schema for personal AI assistant email triage looks like this:
{
"type": "object",
"properties": {
"category": {
"type": "string",
"enum": ["personal", "work", "promo", "finance", "spam", "urgent"]
},
"priority": { "type": "integer", "minimum": 1, "maximum": 3 },
"summary": { "type": "string", "maxLength": 120 },
"action": {
"type": "string",
"enum": ["archive", "label_work", "label_finance", "draft_reply", "none"]
},
"draft": { "type": "string", "description": "Proposed reply if action is draft_reply" }
},
"required": ["category", "priority", "summary", "action"]
}Keep enums tight. Free-form labels cause messy mailbox state.
Step 4: Call the model with a constrained prompt
Use the OpenAI Python client pointed at any OpenAI-compatible endpoint. If you do not want to juggle multiple provider keys, point base_url at n4n.ai—it fronts 240+ models, fails over automatically when a provider is rate-limited, and meters per token. The code below uses gpt-4o-mini as a stand-in; swap the model name for whatever your gateway exposes.
from openai import OpenAI
import os, json
client = OpenAI(
base_url=os.environ.get("LLM_BASE_URL", "https://api.n4n.ai/v1"),
api_key=os.environ["LLM_API_KEY"]
)
SYSTEM_PROMPT = """You are an email triage agent. Given an email, classify it,
summarize in <=120 chars, assign priority 1-3, and pick one action.
If action is draft_reply, write a concise professional reply."""
def triage(messages: List[Dict]) -> List[Dict]:
results = []
for m in messages:
user_content = f"From: {m['from']}\nSubject: {m['subject']}\n\n{m['body']}"
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_content}
],
response_format={"type": "json_object"}
)
parsed = json.loads(resp.choices[0].message.content)
parsed["id"] = m["id"]
results.append(parsed)
return resultsSet LLM_BASE_URL and LLM_API_KEY in your environment. The response_format flag works on most modern endpoints; if yours lacks it, validate with jsonschema after the call.
Step 5: Execute mailbox actions
Map the model’s action field to IMAP commands. Gmail labels are implemented as IMAP folders under [Gmail]/. Create them once if missing.
def apply_actions(imap: imaplib.IMAP4_SSL, triaged: List[Dict]):
for item in triaged:
mid = item["id"]
if item["action"] == "archive":
imap.store(mid, "+FLAGS", "\\Seen")
imap.copy(mid, "[Gmail]/All Mail")
imap.store(mid, "+FLAGS", "\\Deleted")
elif item["action"].startswith("label_"):
label = item["action"].split("_")[1].capitalize()
imap.copy(mid, f"[Gmail]/{label}")
imap.store(mid, "+FLAGS", "\\Seen")
elif item["action"] == "draft_reply" and item.get("draft"):
# Draft creation requires MIME append to [Gmail]/Drafts
import time
from email.mime.text import MIMEText
draft = MIMEText(item["draft"])
draft["Subject"] = "Re: " + item.get("subject", "")
draft["From"] = EMAIL
draft["To"] = item["from"]
imap.append("[Gmail]/Drafts", "\\Draft", time.time(), draft.as_bytes())
imap.expunge()For a real personal AI assistant email triage loop, you would not delete without a dry-run flag. Add DRY_RUN to log intended actions instead of mutating the mailbox.
Step 6: Schedule and verify success
Wrap the pipeline in a function and run it on a cron trigger. A five-minute interval is aggressive for personal use; 15 minutes is fine.
def run_pipeline(dry_run: bool = False):
imap = connect_imap()
msgs = fetch_unseen(imap, limit=20)
if not msgs:
return "No unseen mail"
triaged = triage(msgs)
if dry_run:
for t in triaged:
print(t["id"], t["category"], t["action"], t["summary"])
else:
apply_actions(imap, triaged)
imap.logout()
return f"Processed {len(triaged)} messages"
if __name__ == "__main__":
print(run_pipeline(dry_run=True))How to verify success
- Run with
dry_run=True. Confirm the console prints plausible categories and summaries for your recent mail. - Check token usage from your gateway dashboard (or the
usagefield in the response) to validate cost per run. - Disable dry-run for a single low-risk label (e.g.,
label_finance) and confirm the message appears in that Gmail folder. - Inspect the Drafts folder after a
draft_replyaction to ensure the proposed text is sane.
If the model misclassifies more than ~10% of a labeled test set, tighten the system prompt or add few-shot examples inline. Personal AI assistant email triage is only useful when you stop checking the inbox yourself.
Edge cases that will bite you
HTML-only mail breaks the plain-text extractor. Add a fallback using html2text or strip tags. Threaded replies duplicate content; dedupe by Message-ID before sending to the model. Rate limits on the LLM side should trigger a retry with exponential backoff—the gateway’s automatic fallback handles provider outages, but your client should still catch 429s.
Closing the loop
Once stable, extend the schema with calendar_event detection or receipt_amount extraction. The same IMAP scaffold supports pushing structured data to your task manager. Keep the action set small, log every mutation, and review the Drafts folder weekly. That discipline is what separates a toy script from a personal AI assistant email triage system you actually trust.