Creating an AI Email Assistant Agent: Triage, Draft, and Schedule with Gmail API

The Email Overload Problem

The average professional receives 120+ emails per day and spends 2.5 hours managing their inbox. An AI email assistant agent can reduce this to minutes by automatically triaging incoming mail, drafting responses for routine messages, and scheduling sends at optimal times.

In this tutorial, you will build an email assistant that connects to Gmail via the API, classifies emails by urgency and category, drafts contextually appropriate responses, and schedules sends. The agent handles the mechanical parts of email management while keeping you in control of final decisions.

Architecture

┌─────────────┐     ┌────────────────────┐     ┌────────────┐
│  Gmail API   │────▶│  Email Assistant    │────▶│  Gmail API  │
│  (Inbox)     │     │  Agent             │     │  (Send)     │
└─────────────┘     │                    │     └────────────┘
                     │  Tools:            │
                     │  - read_inbox      │     ┌────────────┐
                     │  - classify_email  │────▶│  Calendar   │
                     │  - draft_response  │     │  (Schedule) │
                     │  - schedule_send   │     └────────────┘
                     │  - search_email    │
                     └────────────────────┘

Prerequisites

• Python 3.11+
• Google Cloud project with Gmail API enabled
• OAuth 2.0 credentials (Desktop app type)
• OpenAI API key

Step 1: Set Up Gmail API Access

First, install the required packages:

flowchart LR
    CALLER(["Student or Parent"])
    subgraph TEL["Telephony"]
        SIP["Twilio SIP and PSTN"]
    end
    subgraph BRAIN["Education AI Agent"]
        STT["Streaming STT<br/>Deepgram or Whisper"]
        NLU{"Intent and<br/>Entity Extraction"}
        TOOLS["Tool Calls"]
        TTS["Streaming TTS<br/>ElevenLabs or Rime"]
    end
    subgraph DATA["Live Data Plane"]
        CRM[("CRM and Notes")]
        CAL[("Calendar and<br/>Schedule")]
        KB[("Knowledge Base<br/>and Policies")]
    end
    subgraph OUT["Outcomes"]
        O1(["Enrollment captured"])
        O2(["Tour scheduled"])
        O3(["Counselor callback"])
    end
    CALLER --> SIP --> STT --> NLU
    NLU -->|Lookup| TOOLS
    TOOLS <--> CRM
    TOOLS <--> CAL
    TOOLS <--> KB
    NLU --> TTS --> SIP --> CALLER
    NLU -->|Resolved| O1
    NLU -->|Schedule| O2
    NLU -->|Escalate| O3
    style CALLER fill:#f1f5f9,stroke:#64748b,color:#0f172a
    style NLU fill:#4f46e5,stroke:#4338ca,color:#fff
    style O1 fill:#059669,stroke:#047857,color:#fff
    style O2 fill:#0ea5e9,stroke:#0369a1,color:#fff
    style O3 fill:#f59e0b,stroke:#d97706,color:#1f2937

pip install openai-agents google-auth-oauthlib google-api-python-client python-dotenv

Set up OAuth credentials. Download your credentials.json from Google Cloud Console and place it in the project root:

# auth/gmail_auth.py
import os
import pickle
from google.auth.transport.requests import Request
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build

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

def get_gmail_service():
    """Authenticate and return a Gmail API service instance."""
    creds = None
    token_path = "token.pickle"

    if os.path.exists(token_path):
        with open(token_path, "rb") as token:
            creds = pickle.load(token)

    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            flow = InstalledAppFlow.from_client_secrets_file(
                "credentials.json", SCOPES
            )
            creds = flow.run_local_server(port=0)
        with open(token_path, "wb") as token:
            pickle.dump(creds, token)

    return build("gmail", "v1", credentials=creds)

Step 2: Build the Inbox Reading Tool

# tools/inbox.py
from agents import function_tool
from auth.gmail_auth import get_gmail_service
import base64
from email.utils import parsedate_to_datetime

gmail = get_gmail_service()

@function_tool
def read_inbox(max_results: int = 10, query: str = "is:unread") -> str:
    """Read emails from the inbox. Use Gmail search syntax for the query.
    Examples: 'is:unread', 'from:boss@company.com', 'subject:urgent'.
    Returns sender, subject, date, snippet, and message ID for each email."""
    try:
        results = gmail.users().messages().list(
            userId="me", q=query, maxResults=max_results
        ).execute()

        messages = results.get("messages", [])
        if not messages:
            return "No emails matching the query."

        emails = []
        for msg_ref in messages:
            msg = gmail.users().messages().get(
                userId="me", id=msg_ref["id"], format="metadata",
                metadataHeaders=["From", "Subject", "Date"]
            ).execute()

            headers = {h["name"]: h["value"] for h in msg["payload"]["headers"]}
            emails.append(
                f"ID: {msg['id']}\n"
                f"From: {headers.get('From', 'unknown')}\n"
                f"Subject: {headers.get('Subject', '(no subject)')}\n"
                f"Date: {headers.get('Date', 'unknown')}\n"
                f"Snippet: {msg.get('snippet', '')[:200]}\n"
                f"Labels: {', '.join(msg.get('labelIds', []))}"
            )

        return f"Found {len(emails)} emails:\n\n" + "\n\n---\n\n".join(emails)
    except Exception as e:
        return f"Error reading inbox: {str(e)}"

@function_tool
def read_full_email(message_id: str) -> str:
    """Read the full content of an email by its message ID. Use this when
    you need the complete email body to draft a response."""
    try:
        msg = gmail.users().messages().get(
            userId="me", id=message_id, format="full"
        ).execute()

        headers = {h["name"]: h["value"] for h in msg["payload"]["headers"]}

        # Extract body
        body = ""
        payload = msg["payload"]
        if "parts" in payload:
            for part in payload["parts"]:
                if part["mimeType"] == "text/plain" and "data" in part.get("body", {}):
                    body = base64.urlsafe_b64decode(
                        part["body"]["data"]
                    ).decode("utf-8")
                    break
        elif "body" in payload and "data" in payload["body"]:
            body = base64.urlsafe_b64decode(
                payload["body"]["data"]
            ).decode("utf-8")

        return (
            f"From: {headers.get('From', 'unknown')}\n"
            f"To: {headers.get('To', 'unknown')}\n"
            f"Subject: {headers.get('Subject', '(no subject)')}\n"
            f"Date: {headers.get('Date', 'unknown')}\n\n"
            f"Body:\n{body[:3000]}"
        )
    except Exception as e:
        return f"Error reading email: {str(e)}"

Step 3: Build the Classification Tool

# tools/classifier.py
from agents import function_tool

@function_tool
def classify_email(
    sender: str,
    subject: str,
    snippet: str,
    labels: str = ""
) -> str:
    """Classify an email by urgency and category. Returns a structured
    classification with urgency (critical, high, medium, low),
    category (action_required, informational, meeting, newsletter,
    spam, personal), and a suggested action."""

    # Rule-based pre-classification for known patterns
    sender_lower = sender.lower()
    subject_lower = subject.lower()
    snippet_lower = snippet.lower()

    # Urgency detection
    urgency = "medium"
    if any(w in subject_lower for w in ["urgent", "asap", "critical", "emergency", "blocked"]):
        urgency = "critical"
    elif any(w in subject_lower for w in ["important", "action required", "deadline", "eod"]):
        urgency = "high"
    elif any(w in subject_lower for w in ["fyi", "newsletter", "digest", "weekly"]):
        urgency = "low"

    # Category detection
    category = "informational"
    if any(w in subject_lower for w in ["invite", "meeting", "calendar", "sync", "standup"]):
        category = "meeting"
    elif any(w in subject_lower for w in ["unsubscribe", "newsletter", "digest", "promotion"]):
        category = "newsletter"
    elif any(w in snippet_lower for w in ["please", "could you", "can you", "need you to", "action"]):
        category = "action_required"

    # Suggested action
    actions = {
        ("critical", "action_required"): "Respond immediately",
        ("high", "action_required"): "Respond within 2 hours",
        ("medium", "action_required"): "Respond today",
        ("low", "informational"): "Read when free or archive",
        ("low", "newsletter"): "Archive or batch read later",
    }
    action = actions.get((urgency, category), "Review and respond as appropriate")

    return (
        f"Classification:\n"
        f"  Urgency: {urgency}\n"
        f"  Category: {category}\n"
        f"  Suggested action: {action}\n"
        f"  Sender: {sender}\n"
        f"  Subject: {subject}"
    )

Step 4: Build the Draft and Send Tools

# tools/compose.py
from agents import function_tool
from auth.gmail_auth import get_gmail_service
import base64
from email.mime.text import MIMEText
from datetime import datetime, timedelta

gmail = get_gmail_service()

@function_tool
def draft_response(
    to: str,
    subject: str,
    body: str,
    reply_to_id: str = ""
) -> str:
    """Create a draft email response. If reply_to_id is provided, the
    draft will be threaded with the original email. The body should be
    plain text. Returns the draft ID for review before sending."""
    try:
        message = MIMEText(body)
        message["to"] = to
        message["subject"] = subject if not subject.startswith("Re:") else subject

        raw = base64.urlsafe_b64encode(message.as_bytes()).decode("utf-8")
        draft_body = {"message": {"raw": raw}}

        if reply_to_id:
            # Get the thread ID for proper threading
            original = gmail.users().messages().get(
                userId="me", id=reply_to_id, format="minimal"
            ).execute()
            draft_body["message"]["threadId"] = original.get("threadId")

        draft = gmail.users().drafts().create(
            userId="me", body=draft_body
        ).execute()

        return (
            f"Draft created successfully.\n"
            f"Draft ID: {draft['id']}\n"
            f"To: {to}\n"
            f"Subject: {subject}\n"
            f"Body preview: {body[:200]}...\n"
            f"Status: Ready for review before sending"
        )
    except Exception as e:
        return f"Draft creation failed: {str(e)}"

@function_tool
def send_draft(draft_id: str) -> str:
    """Send a previously created draft email. Only use this after the
    user has approved the draft content."""
    try:
        result = gmail.users().drafts().send(
            userId="me", body={"id": draft_id}
        ).execute()
        return f"Email sent successfully. Message ID: {result['id']}"
    except Exception as e:
        return f"Send failed: {str(e)}"

@function_tool
def schedule_send(
    to: str,
    subject: str,
    body: str,
    send_at: str
) -> str:
    """Schedule an email to be sent at a specific time. The send_at
    parameter should be in ISO format (e.g., '2026-03-25T09:00:00').
    Creates a draft and returns scheduling confirmation."""
    try:
        # Create the draft
        message = MIMEText(body)
        message["to"] = to
        message["subject"] = subject
        raw = base64.urlsafe_b64encode(message.as_bytes()).decode("utf-8")

        draft = gmail.users().drafts().create(
            userId="me", body={"message": {"raw": raw}}
        ).execute()

        # Parse the scheduled time
        scheduled_time = datetime.fromisoformat(send_at)
        now = datetime.now()

        if scheduled_time <= now:
            return "Cannot schedule in the past. Please provide a future time."

        delay = scheduled_time - now

        return (
            f"Email scheduled successfully.\n"
            f"Draft ID: {draft['id']}\n"
            f"To: {to}\n"
            f"Subject: {subject}\n"
            f"Scheduled for: {send_at}\n"
            f"Time until send: {delay}\n"
            f"Note: A background worker will send this draft at the scheduled time."
        )
    except Exception as e:
        return f"Scheduling failed: {str(e)}"

Step 5: Assemble the Email Assistant Agent

# agent.py
from agents import Agent
from tools.inbox import read_inbox, read_full_email
from tools.classifier import classify_email
from tools.compose import draft_response, send_draft, schedule_send

email_agent = Agent(
    name="Email Assistant",
    instructions="""You are an intelligent email assistant. You help manage
    the user's inbox efficiently.

    WORKFLOW:
    1. When asked to check email: read the inbox, classify each email by
       urgency and category, and present a prioritized summary.
    2. When asked to respond to an email: read the full email first, then
       draft a response that matches the tone and context. Always create
       a draft for review — never send without confirmation.
    3. When asked to schedule: use schedule_send with the specified time.

    RESPONSE DRAFTING RULES:
    - Match the formality of the original email
    - Be concise but thorough
    - Include specific references to the content of the original email
    - For meeting requests: check conflicts before accepting
    - For action items: acknowledge and provide a timeline
    - Never fabricate information not in the original email

    SAFETY RULES:
    - Never send emails without explicit user approval
    - Always show draft content before sending
    - Flag suspicious or phishing emails clearly
    - Do not open attachments or click links""",
    tools=[read_inbox, read_full_email, classify_email, draft_response,
           send_draft, schedule_send],
    model="gpt-4o",
)

Step 6: Build the Interactive Runner

# run_assistant.py
import asyncio
from agents import Runner
from agent import email_agent
from dotenv import load_dotenv

load_dotenv()

async def main():
    print("Email Assistant ready. Commands:")
    print("  'check'      - Check and triage inbox")
    print("  'respond X'  - Draft a response to email X")
    print("  'schedule'   - Schedule an email")
    print("  'exit'       - Quit")
    print()

    while True:
        user_input = input("You: ").strip()
        if user_input.lower() == "exit":
            break

        result = await Runner.run(email_agent, user_input)
        print(f"\nAssistant: {result.final_output}\n")

if __name__ == "__main__":
    asyncio.run(main())

Extending the Assistant

Here are natural extensions to make the assistant more powerful:

• Contact context — Add a tool that looks up the sender in your CRM or contacts database, giving the agent context about your relationship
• Calendar integration — Connect Google Calendar to check for conflicts before accepting meeting invites
• Template library — Provide response templates for common email types (invoices, meeting requests, follow-ups)
• Analytics — Track response times, email volume, and categories over time to identify workflow improvements
• Multi-account — Support multiple Gmail accounts with per-account OAuth tokens

Security Best Practices

Email access is sensitive. Follow these practices:

1. Least privilege scopes — Only request the Gmail scopes you actually need
2. Token storage — Encrypt the OAuth token at rest, never commit it to version control
3. Audit logging — Log every email read, draft created, and email sent
4. Rate limiting — Implement rate limits on send operations to prevent runaway agents from spamming
5. Human in the loop — Always require explicit approval before sending

FAQ

How do I handle emails with attachments?

The Gmail API provides attachment data in the message payload's parts array. Add a download_attachment tool that extracts attachments by part ID and saves them to disk. For security, scan downloaded files before processing and never execute attachments.

Can the agent learn my writing style over time?

Yes. Store your sent emails in a vector database and use them as few-shot examples when drafting responses. The agent can retrieve your most similar past responses and use them as style references. This significantly improves the naturalness of drafted responses after collecting 50-100 examples.

How do I prevent the agent from reading sensitive emails?

Add a label-based filter. Create a Gmail label called "AI-Excluded" and modify the read_inbox tool to exclude emails with that label: query = "is:unread -label:AI-Excluded". You can also filter by sender domain to exclude specific contacts.

What is the latency for processing an inbox of 50 emails?

Reading 50 email headers takes approximately 3-5 seconds via the Gmail API. Classification of all 50 emails through the agent loop takes about 10-15 seconds. The total end-to-end time for triaging 50 emails is typically under 30 seconds, compared to 15-20 minutes manually.