# How to Build an Email AI Agent

- Tool: Gmail API
- Difficulty: Intermediate
- Time required: 45-60 min
- Compatibility: Python 3.10+, google-api-python-client, openai-agents, a Google Workspace or Gmail account with API access enabled
- Last updated: November 2026

## TL;DR

An email AI agent polls an inbox, decides what each message needs, and acts through the Gmail API instead of just summarizing. This guide builds a support-inbox triage agent with the OpenAI Agents SDK: it reads unread mail, drafts a reply as a Gmail draft, and labels the thread needs-review so a human sends it. Stack: Python, google-api-python-client, openai-agents. Budget 45-60 minutes plus a one-time OAuth setup.

## What you'll build

You'll build a triage agent for a small support inbox. On each run it lists unread messages that don't already carry an agent-processed label, reads each one, and for anything that looks like a routine question, drafts a reply as a Gmail draft and labels the thread needs-review. It never calls Gmail's send endpoint itself; a human opens the draft, edits if needed, and hits send.

The OpenAI Agents SDK's Agent and Runner give you the loop: define instructions, hand the agent two function_tools (list_unread_support_emails and create_draft_reply), and call Runner.run once per polling cycle with a prompt like 'triage the inbox.' The agent decides how many emails to read and when a draft is warranted; your code only decides what the tools are allowed to do.

Drafting instead of sending is the real safety mechanism here, not a permission callback. It's the right default for most inbox agents: a wrong tool call costs you a draft someone deletes, not an email a customer actually received. If you need true autonomous sending later, add an approval step the way the Slack and Claude Agent SDK guides do before removing this guardrail.

## Before you start

- A Google Cloud project with the Gmail API enabled and OAuth 2.0 credentials (Desktop app type is simplest for testing)
- An OpenAI API key (OPENAI_API_KEY)
- pip install google-api-python-client google-auth-oauthlib openai-agents
- A support inbox you're allowed to read and draft in, or a test Gmail account

## Step-by-step guide

### 1. Enable the Gmail API and get OAuth credentials

In Google Cloud Console, enable the Gmail API for your project, then create an OAuth 2.0 Client ID of type Desktop app under Credentials and download it as credentials.json. The scopes you need are gmail.readonly, gmail.modify (for labels), and gmail.compose (for drafts); request the narrowest set your use case allows rather than the broad gmail scope.

```
pip install google-api-python-client google-auth-oauthlib openai-agents
export OPENAI_API_KEY="sk-..."
# credentials.json downloaded from Google Cloud Console next to this script
```

**Expected result:** A credentials.json file in your project directory and the Gmail API showing as enabled in Cloud Console.

### 2. Write the one-time auth helper

gmail_service() loads a saved token.json if one exists and refreshes it, or runs the installed-app OAuth flow the first time and saves the result. Every tool function below calls this helper to get an authorized service object; it's the only place credentials touch your code.

```
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from googleapiclient.discovery import build
import os

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

def gmail_service():
    creds = None
    if os.path.exists("token.json"):
        creds = Credentials.from_authorized_user_file("token.json", SCOPES)
    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)
        open("token.json", "w").write(creds.to_json())
    return build("gmail", "v1", credentials=creds)
```

**Expected result:** Running this once opens a browser for consent; every call after that works silently from the saved token.json.

### 3. Define the read-only tool: list unread support email

list_unread_support_emails queries Gmail with is:unread -label:agent-processed so it skips anything already handled, pulls each message's snippet, sender, and subject, and returns them as a compact list. Decorate it with @function_tool from the agents package so the Agent can call it; the docstring becomes the description the model reads.

```
from agents import function_tool

@function_tool
def list_unread_support_emails(max_results: int = 10) -> list[dict]:
    """List unread support emails that haven't been processed yet, newest first."""
    service = gmail_service()
    resp = service.users().messages().list(
        userId="me", q="is:unread -label:agent-processed", maxResults=max_results
    ).execute()
    out = []
    for m in resp.get("messages", []):
        msg = service.users().messages().get(userId="me", id=m["id"], format="metadata",
                                              metadataHeaders=["From", "Subject"]).execute()
        headers = {h["name"]: h["value"] for h in msg["payload"]["headers"]}
        out.append({"id": m["id"], "from": headers.get("From"), "subject": headers.get("Subject"),
                    "snippet": msg.get("snippet")})
    return out
```

**Expected result:** A list of dicts, one per unread support email, each with id, from, subject, and snippet.

### 4. Define the write tool: draft a reply and label the thread

create_draft_reply builds a reply as an EmailMessage threaded to the original message, base64-encodes it, calls drafts().create, and then adds the agent-processed and needs-review labels to the original message so it's skipped next run and surfaced for a human. This is the only tool that writes anything, and it never calls messages().send.

```
import base64
from email.message import EmailMessage

@function_tool
def create_draft_reply(message_id: str, to: str, subject: str, body: str) -> str:
    """Create a Gmail draft reply to message_id and label the thread needs-review. Never sends."""
    service = gmail_service()
    msg = EmailMessage()
    msg.set_content(body)
    msg["To"] = to
    msg["Subject"] = f"Re: {subject}"
    raw = base64.urlsafe_b64encode(msg.as_bytes()).decode()
    service.users().drafts().create(
        userId="me", body={"message": {"raw": raw, "threadId": message_id}}
    ).execute()
    label_ids = _ensure_labels(service, ["agent-processed", "needs-review"])
    service.users().messages().modify(
        userId="me", id=message_id, body={"addLabelIds": label_ids}
    ).execute()
    return f"Drafted a reply to {to} and labeled the thread needs-review."
```

**Expected result:** A new draft appears in Gmail threaded under the original message, and the original message gains two labels.

### 5. Build the Agent and run one triage pass

Create an Agent with both tools and instructions that tell it when a draft is warranted versus when to leave a message alone (billing disputes, anything angry, anything you can't answer from the FAQ). Call Runner.run once with a simple instruction; the agent decides how many of the listed emails to read and reply to in that pass.

```
from agents import Agent, Runner
import asyncio

inbox_agent = Agent(
    name="Inbox triage agent",
    instructions=(
        "You triage a support inbox. List unread emails, then for each routine question "
        "(order status, how-to, pricing) draft a short, friendly reply. Skip anything that "
        "sounds like a billing dispute, a complaint, or something you're not confident about; "
        "leave those unread for a human."
    ),
    tools=[list_unread_support_emails, create_draft_reply],
)

async def triage_once():
    result = await Runner.run(inbox_agent, "Triage the inbox now.")
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(triage_once())
```

**Expected result:** Running the script lists the unread count, drafts replies for the routine ones, and leaves tricky messages untouched and still unread.

### 6. Run it on a schedule and harden the dedupe

Wrap triage_once() in a loop with a sleep interval, or call it from a cron job or scheduled Cloud Function for production; polling every 5-10 minutes is plenty for most support inboxes and avoids hitting Gmail's quota. Wrap every Gmail call in try/except googleapiclient.errors.HttpError so a transient API error doesn't crash the whole pass, and double-check the agent-processed label was actually applied before trusting the dedupe.

```
import time
from googleapiclient.errors import HttpError

async def triage_loop(poll_seconds: int = 300):
    while True:
        try:
            await triage_once()
        except HttpError as e:
            print(f"Gmail API error, will retry next cycle: {e}")
        time.sleep(poll_seconds)
```

**Expected result:** A long-running process that triages new mail every 5 minutes and survives a transient Gmail API error without crashing.

## Complete code example

File: `inbox_triage_agent.py`

```python
"""Inbox triage agent: drafts replies to routine support email, never sends.
Run: python inbox_triage_agent.py
(pip install google-api-python-client google-auth-oauthlib openai-agents; set OPENAI_API_KEY)
"""
import asyncio
import base64
import os
import time
from email.message import EmailMessage

from agents import Agent, Runner, function_tool
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError

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


def gmail_service():
    creds = None
    if os.path.exists("token.json"):
        creds = Credentials.from_authorized_user_file("token.json", SCOPES)
    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)
        open("token.json", "w").write(creds.to_json())
    return build("gmail", "v1", credentials=creds)


def _ensure_labels(service, names: list[str]) -> list[str]:
    existing = {l["name"]: l["id"] for l in service.users().labels().list(userId="me").execute().get("labels", [])}
    ids = []
    for name in names:
        if name in existing:
            ids.append(existing[name])
        else:
            created = service.users().labels().create(
                userId="me", body={"name": name, "labelListVisibility": "labelShow"}
            ).execute()
            ids.append(created["id"])
    return ids


@function_tool
def list_unread_support_emails(max_results: int = 10) -> list[dict]:
    """List unread support emails that haven't been processed yet, newest first."""
    service = gmail_service()
    resp = service.users().messages().list(
        userId="me", q="is:unread -label:agent-processed", maxResults=max_results
    ).execute()
    out = []
    for m in resp.get("messages", []):
        msg = service.users().messages().get(
            userId="me", id=m["id"], format="metadata", metadataHeaders=["From", "Subject"]
        ).execute()
        headers = {h["name"]: h["value"] for h in msg["payload"]["headers"]}
        out.append({
            "id": m["id"], "from": headers.get("From"), "subject": headers.get("Subject"),
            "snippet": msg.get("snippet"),
        })
    return out


@function_tool
def create_draft_reply(message_id: str, to: str, subject: str, body: str) -> str:
    """Create a Gmail draft reply to message_id and label the thread needs-review. Never sends."""
    service = gmail_service()
    msg = EmailMessage()
    msg.set_content(body)
    msg["To"] = to
    msg["Subject"] = f"Re: {subject}"
    raw = base64.urlsafe_b64encode(msg.as_bytes()).decode()
    service.users().drafts().create(
        userId="me", body={"message": {"raw": raw, "threadId": message_id}}
    ).execute()
    label_ids = _ensure_labels(service, ["agent-processed", "needs-review"])
    service.users().messages().modify(
        userId="me", id=message_id, body={"addLabelIds": label_ids}
    ).execute()
    return f"Drafted a reply to {to} and labeled the thread needs-review."


inbox_agent = Agent(
    name="Inbox triage agent",
    instructions=(
        "You triage a support inbox. List unread emails, then for each routine question "
        "(order status, how-to, pricing) draft a short, friendly reply. Skip anything that "
        "sounds like a billing dispute, a complaint, or something you're not confident about; "
        "leave those unread for a human."
    ),
    tools=[list_unread_support_emails, create_draft_reply],
)


async def triage_once():
    result = await Runner.run(inbox_agent, "Triage the inbox now.")
    print(result.final_output)


async def triage_loop(poll_seconds: int = 300):
    while True:
        try:
            await triage_once()
        except HttpError as e:
            print(f"Gmail API error, will retry next cycle: {e}")
        time.sleep(poll_seconds)


if __name__ == "__main__":
    asyncio.run(triage_loop())
```

## Common mistakes

- **Calling messages().send directly from a tool to 'save a step'** — undefined Fix: Draft first. The cost of a bad draft is someone deleting it; the cost of a bad send is an email your customer already has.
- **Querying is:unread with no label exclusion** — undefined Fix: Without -label:agent-processed, every poll re-reads and re-drafts for the same messages. Always exclude what you've already handled.
- **Using the broad gmail scope instead of gmail.modify plus gmail.compose** — undefined Fix: The broad scope grants send and full account access this agent doesn't need. Request only the scopes the tools actually call.
- **Skipping the try/except around Gmail API calls** — undefined Fix: A single transient 500 from Gmail's API will otherwise kill a long-running poll loop. Catch HttpError and retry next cycle.
- **Not re-running the OAuth consent flow after changing SCOPES** — undefined Fix: A saved token.json from a narrower scope won't silently gain the new permission. Delete it and re-authenticate after any scope change.

## Best practices

- Default every write action to its safest form (draft, not send) unless you have a specific reason not to.
- Dedupe with a label, not an in-memory set, so restarts don't cause reprocessing.
- Write the instructions with explicit skip conditions (disputes, complaints, low confidence), not just what to do.
- Keep gmail.readonly scope on any tool that only reads, and only add gmail.compose/modify to tools that need them.
- Log which messages were drafted versus skipped on every pass so you can audit why.
- Poll on an interval that respects Gmail API quota instead of reacting to every inbox change in real time.

## Use cases

### undefined



### undefined



### undefined



## Frequently asked questions

### Why draft instead of letting the email AI agent send replies automatically?

A draft gives a human the last look before anything reaches a customer, at the cost of one click. For a support inbox that's almost always worth it; reserve full auto-send for agents you've tested extensively on a narrow, low-risk message type.

### Does this work with Outlook or other email providers?

The agent logic (Agent, Runner, function_tool) is provider-agnostic; swap the Gmail-specific tool implementations for Microsoft Graph API calls (messages, drafts endpoints) and the rest of this guide applies unchanged.

### How do I avoid hitting Gmail API rate limits?

Gmail's per-user quota is generous for a single support inbox polling every few minutes, but batch your messages().get calls with Gmail's batch HTTP request feature if you ever need to read more than a few dozen messages per pass.

### What if the agent drafts a reply that's factually wrong?

That's exactly what the needs-review label and draft-only behavior protect against: nothing goes out until a human reads it. If wrong drafts are common, narrow the instructions' definition of 'routine question' or add a tool that looks up real order/account data instead of guessing.

### Can the same agent also handle outbound cold email instead of inbox triage?

The Agent/Runner/function_tool pattern carries over, but outbound sending is a different risk profile (deliverability, compliance, opt-outs) than replying to someone who already emailed you. Treat it as a separate agent with its own guardrails.

### Can RapidDev build an email agent like this for our support team?

Yes. RapidDev builds inbox and ticketing agents on the OpenAI Agents SDK and Claude Agent SDK, including the draft-first safety pattern and label-based routing this guide uses. A free 30-minute consultation is the fastest way to scope yours.

---

Source: https://www.rapidevelopers.com/build-ai-agent/how-to-build-an-email-ai-agent
© RapidDev — https://www.rapidevelopers.com/build-ai-agent/how-to-build-an-email-ai-agent