How to Integrate AI Agents with Gmail

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

import os
import base64
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

SCOPES = ['https://www.googleapis.com/auth/gmail.modify']

def get_gmail_service():
    """Authenticate and return Gmail API 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)
        with open('token.json', 'w') as token:
            token.write(creds.to_json())

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

service = get_gmail_service()

from langchain.tools import tool

service = get_gmail_service()


def decode_message_body(payload) -> str:
    """Extract plain text from Gmail message payload."""
    body = ""
    if payload.get("mimeType") == "text/plain":
        data = payload.get("body", {}).get("data", "")
        body = base64.urlsafe_b64decode(data + "==").decode("utf-8", errors="ignore")
    elif payload.get("parts"):
        for part in payload["parts"]:
            if part.get("mimeType") == "text/plain":
                data = part.get("body", {}).get("data", "")
                body = base64.urlsafe_b64decode(data + "==").decode("utf-8", errors="ignore")
                break
    return body[:3000]


@tool
def list_unread_emails(max_results: int = 10) -> str:
    """List unread emails from the inbox with sender, subject, and snippet."""
    messages = service.users().messages().list(
        userId="me",
        q="is:unread in:inbox",
        maxResults=max_results
    ).execute().get("messages", [])

    if not messages:
        return "No unread emails in inbox"

    result = [f"Unread emails ({len(messages)} shown):"]
    for msg_ref in messages[:10]:
        msg = service.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.get("payload", {}).get("headers", [])}
        snippet = msg.get("snippet", "")[:150]
        result.append(f"\nID: {msg_ref['id']}\nFrom: {headers.get('From', 'Unknown')}\nSubject: {headers.get('Subject', 'No subject')}\nPreview: {snippet}")
    return "\n".join(result)


@tool
def get_email_thread(thread_id: str) -> str:
    """Get full conversation thread including all messages by thread ID."""
    thread = service.users().threads().get(userId="me", id=thread_id).execute()
    messages = thread.get("messages", [])

    thread_text = []
    for msg in messages[-5:]:  # Last 5 messages
        headers = {h["name"]: h["value"] for h in msg.get("payload", {}).get("headers", [])}
        body = decode_message_body(msg.get("payload", {}))
        thread_text.append(f"From: {headers.get('From', 'Unknown')}\nDate: {headers.get('Date', '')}\n{body}")

    return "\n\n---\n\n".join(thread_text)


@tool
def create_draft_reply(thread_id: str, to_email: str, subject: str, body: str) -> str:
    """Create a draft email reply in a thread. The draft appears in Drafts for human review."""
    import email.mime.text
    message = email.mime.text.MIMEText(body)
    message["to"] = to_email
    message["subject"] = f"Re: {subject}" if not subject.startswith("Re:") else subject

    raw = base64.urlsafe_b64encode(message.as_bytes()).decode("utf-8")
    draft = service.users().drafts().create(
        userId="me",
        body={"message": {"raw": raw, "threadId": thread_id}}
    ).execute()
    return f"Draft reply created (Draft ID: {draft.get('id')}) — review in Gmail Drafts before sending"


@tool
def apply_label(message_id: str, label_name: str) -> str:
    """Apply a label to an email. Creates the label if it doesn't exist."""
    # Get existing labels
    labels = service.users().labels().list(userId="me").execute().get("labels", [])
    label_map = {l["name"]: l["id"] for l in labels}

    if label_name not in label_map:
        new_label = service.users().labels().create(
            userId="me", body={"name": label_name}
        ).execute()
        label_id = new_label["id"]
    else:
        label_id = label_map[label_name]

    service.users().messages().modify(
        userId="me",
        id=message_id,
        body={"addLabelIds": [label_id]}
    ).execute()
    return f"Label '{label_name}' applied to message {message_id}"


@tool
def archive_email(message_id: str) -> str:
    """Archive an email by removing it from the inbox (removes INBOX label)."""
    service.users().messages().modify(
        userId="me",
        id=message_id,
        body={"removeLabelIds": ["INBOX"]}
    ).execute()
    return f"Email {message_id} archived"

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
tools = [list_unread_emails, get_email_thread, create_draft_reply,
         apply_label, archive_email]

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are an executive email assistant with access to Gmail.

When processing emails:
1. Check unread emails and classify each by urgency and category
2. For urgent/important emails, draft a professional reply and create it as a draft
3. Apply appropriate labels (urgent, follow-up, customer, invoice, newsletter)
4. Archive low-priority newsletters and notifications after labeling
5. Never send emails autonomously — always create drafts for human review

Draft reply guidelines:
- Be professional, concise, and action-oriented
- Match the formality level of the incoming email
- Include specific next steps or clear answers to questions
- Never over-promise timelines or commitments on behalf of the human"""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=8)

# Process inbox
result = executor.invoke({
    "input": "Check my unread emails. For each one: classify urgency, apply an appropriate label, draft replies for customer/urgent emails, and archive newsletters."
})
print(result["output"])