How to Build an Email AI Agent

1"""Inbox triage agent: drafts replies to routine support email, never sends.
2Run: python inbox_triage_agent.py
3(pip install google-api-python-client google-auth-oauthlib openai-agents; set OPENAI_API_KEY)
4"""
5import asyncio
6import base64
7import os
8import time
9from email.message import EmailMessage
10
11from agents import Agent, Runner, function_tool
12from google.auth.transport.requests import Request
13from google.oauth2.credentials import Credentials
14from google_auth_oauthlib.flow import InstalledAppFlow
15from googleapiclient.discovery import build
16from googleapiclient.errors import HttpError
17
18SCOPES = [
19    "https://www.googleapis.com/auth/gmail.modify",
20    "https://www.googleapis.com/auth/gmail.compose",
21]
22
23
24def gmail_service():
25    creds = None
26    if os.path.exists("token.json"):
27        creds = Credentials.from_authorized_user_file("token.json", SCOPES)
28    if not creds or not creds.valid:
29        if creds and creds.expired and creds.refresh_token:
30            creds.refresh(Request())
31        else:
32            flow = InstalledAppFlow.from_client_secrets_file("credentials.json", SCOPES)
33            creds = flow.run_local_server(port=0)
34        open("token.json", "w").write(creds.to_json())
35    return build("gmail", "v1", credentials=creds)
36
37
38def _ensure_labels(service, names: list[str]) -> list[str]:
39    existing = {l["name"]: l["id"] for l in service.users().labels().list(userId="me").execute().get("labels", [])}
40    ids = []
41    for name in names:
42        if name in existing:
43            ids.append(existing[name])
44        else:
45            created = service.users().labels().create(
46                userId="me", body={"name": name, "labelListVisibility": "labelShow"}
47            ).execute()
48            ids.append(created["id"])
49    return ids
50
51
52@function_tool
53def list_unread_support_emails(max_results: int = 10) -> list[dict]:
54    """List unread support emails that haven't been processed yet, newest first."""
55    service = gmail_service()
56    resp = service.users().messages().list(
57        userId="me", q="is:unread -label:agent-processed", maxResults=max_results
58    ).execute()
59    out = []
60    for m in resp.get("messages", []):
61        msg = service.users().messages().get(
62            userId="me", id=m["id"], format="metadata", metadataHeaders=["From", "Subject"]
63        ).execute()
64        headers = {h["name"]: h["value"] for h in msg["payload"]["headers"]}
65        out.append({
66            "id": m["id"], "from": headers.get("From"), "subject": headers.get("Subject"),
67            "snippet": msg.get("snippet"),
68        })
69    return out
70
71
72@function_tool
73def create_draft_reply(message_id: str, to: str, subject: str, body: str) -> str:
74    """Create a Gmail draft reply to message_id and label the thread needs-review. Never sends."""
75    service = gmail_service()
76    msg = EmailMessage()
77    msg.set_content(body)
78    msg["To"] = to
79    msg["Subject"] = f"Re: {subject}"
80    raw = base64.urlsafe_b64encode(msg.as_bytes()).decode()
81    service.users().drafts().create(
82        userId="me", body={"message": {"raw": raw, "threadId": message_id}}
83    ).execute()
84    label_ids = _ensure_labels(service, ["agent-processed", "needs-review"])
85    service.users().messages().modify(
86        userId="me", id=message_id, body={"addLabelIds": label_ids}
87    ).execute()
88    return f"Drafted a reply to {to} and labeled the thread needs-review."
89
90
91inbox_agent = Agent(
92    name="Inbox triage agent",
93    instructions=(
94        "You triage a support inbox. List unread emails, then for each routine question "
95        "(order status, how-to, pricing) draft a short, friendly reply. Skip anything that "
96        "sounds like a billing dispute, a complaint, or something you're not confident about; "
97        "leave those unread for a human."
98    ),
99    tools=[list_unread_support_emails, create_draft_reply],
100)
101
102
103async def triage_once():
104    result = await Runner.run(inbox_agent, "Triage the inbox now.")
105    print(result.final_output)
106
107
108async def triage_loop(poll_seconds: int = 300):
109    while True:
110        try:
111            await triage_once()
112        except HttpError as e:
113            print(f"Gmail API error, will retry next cycle: {e}")
114        time.sleep(poll_seconds)
115
116
117if __name__ == "__main__":
118    asyncio.run(triage_loop())
119