# Automate Gmail with AI: A Practical Guide
Most “AI email automation” articles give you high-level concepts and a link to a third-party tool. That’s not helpful if you’re a developer who needs to actually build something.
This guide shows you how to build a real Gmail automation system using Python, Google’s Gmail API, and an LLM of your choice. You’ll see the actual code, the gotchas, and what actually works in production.
## Setting Up Gmail API Access
Before writing code, you need API credentials. Here’s how to get them:
1. Go to the [Google Cloud Console](https://console.cloud.google.com/)
2. Create a new project
3. Enable the Gmail API
4. Go to Credentials → Create Credentials → OAuth Client ID
5. Configure the OAuth consent screen (add your email as a test user)
6. Download the JSON credentials file
The credentials file downloads as something like `credentials.json`. Rename it and keep it safe—this gives your script access to your Gmail.
You’ll also need to enable these Gmail scopes:
– `gmail.readonly` — read emails
– `gmail.modify` — modify emails (mark read, labels)
– `gmail.send` — send replies
## Building the Email Fetching Pipeline
Install the required libraries:
“`bash
pip install google-auth google-auth-oauthlib google-api-python-client
“`
Here’s the core setup to authenticate and fetch emails:
“`python
import os
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.readonly’,
‘https://www.googleapis.com/auth/gmail.modify’,
‘https://www.googleapis.com/auth/gmail.send’]
def get_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:
flow = InstalledAppFlow.from_client_secrets_file(
‘credentials.json’, SCOPES)
creds = flow.run_local_server(port=8080)
with open(‘token.json’, ‘w’) as token:
token.write(creds.to_json())
return build(‘gmail’, ‘v1′, credentials=creds)
def fetch_unread_emails(service, max_results=10):
results = service.users().messages().list(
userId=’me’,
q=’is:unread’,
maxResults=max_results
).execute()
messages = results.get(‘messages’, [])
emails = []
for msg in messages:
email = service.users().messages().get(
userId=’me’,
id=msg[‘id’],
format=’full’
).execute()
emails.append(email)
return emails
“`
This gives you a working service object and a function to pull unread emails. The first run opens a browser window for OAuth authentication. Subsequent runs use the cached `token.json`.
## Parsing Emails and Extracting Content
The Gmail API returns raw messages with headers and payload. You need to parse them to get usable content:
“`python
import base64
import email
from email.parser import Parser
def parse_email(message):
“””Extract subject, sender, and body from Gmail message.”””
payload = message[‘payload’]
headers = {h[‘name’]: h[‘value’] for h in payload[‘headers’]}
subject = headers.get(‘Subject’, ‘(No Subject)’)
sender = headers.get(‘From’, ”)
# Get body – try HTML first, then plain text
body = ”
if ‘parts’ in payload:
for part in payload[‘parts’]:
if part[‘mimeType’] == ‘text/plain’:
body = base64.urlsafe_b64decode(
part[‘body’][‘data’]).decode(‘utf-8’)
break
elif part[‘mimeType’] == ‘text/html’:
body = base64.urlsafe_b64decode(
part[‘body’][‘data’]).decode(‘utf-8’)
return {
‘id’: message[‘id’],
‘subject’: subject,
‘sender’: sender,
‘body’: body,
‘thread_id’: message[‘threadId’]
}
“`
This extracts what you need to feed to an LLM. The parsing handles both plain text and HTML emails—HTML gets priority since it usually contains more information.
## Integrating AI for Classification
Now the interesting part: using an LLM to process emails. I’ll use OpenAI’s API, but you can swap this for Anthropic, local models, or whatever fits your needs:
“`python
from openai import OpenAI
client = OpenAI(api_key=os.environ[‘OPENAI_API_KEY’])
def classify_email(email_data):
“””Use AI to classify email and determine action.”””
prompt = f”””Analyze this email and respond with JSON:
{{
“category”: “urgent|important|newsletter|spam|auto”,
“should_reply”: true/false,
“reply_tone”: “formal|casual|none”,
“summary”: “2 sentence summary”,
“action_items”: [“any specific action needed”]
}}
Email:
From: {email_data[‘sender’]}
Subject: {email_data[‘subject’]}
Body: {email_data[‘body’][:2000]} # Truncate to stay within token limits
“””
response = client.chat.completions.create(
model=”gpt-4o-mini”,
messages=[{“role”: “user”, “content”: prompt}],
response_format={“type”: “json_object”}
)
return json.loads(response.choices[0].message.content)
“`
The prompt asks for structured JSON so you can programmatically act on the results. You’re not just getting a summary—you’re getting actionable metadata.
**Token limits matter here.** I’m truncating the body to 2000 characters. For longer emails, you’d want to implement chunking or use a model with larger context windows.
## Automating Replies and Labels
Once AI classifies an email, you can take action automatically:
“`python
def apply_label(service, message_id, label_id):
“””Add a label to organize processed emails.”””
service.users().messages().modify(
userId=’me’,
id=message_id,
body={‘addLabelIds’: [label_id]}
).execute()
def send_reply(service, thread_id, to, subject, body):
“””Send a reply within the same thread.”””
message = email.message.Message()
message[‘To’] = to
message[‘Subject’] = subject
message.set_payload(body)
encoded_message = base64.urlsafe_b64encode(
message.as_bytes()).decode(‘utf-8′)
service.users().messages().send(
userId=’me’,
threadId=thread_id,
raw=encoded_message
).execute()
def process_email(service, email_data):
“””Main processing logic: classify and act.”””
classification = classify_email(email_data)
# Apply label based on category
label_map = {
‘urgent’: ‘URGENT’,
‘important’: ‘IMPORTANT’,
‘newsletter’: ‘NEWSLETTER’,
‘spam’: ‘SPAM’,
‘auto’: ‘AUTO’
}
label_id = label_map.get(classification[‘category’], ‘UNREAD’)
apply_label(service, email_data[‘id’], label_id)
# Send reply if needed
if classification[‘should_reply’]:
# Extract email from “Sender Name
import re
match = re.search(r’<(.+?)>‘, email_data[‘sender’])
to_email = match.group(1) if match else email_data[‘sender’]
# Generate reply with AI
reply_prompt = f”””Write a {classification[‘reply_tone’]} reply to this email.
Keep it brief and professional.
Original email:
{email_data[‘body’][:1500]}”””
reply_response = client.chat.completions.create(
model=”gpt-4o-mini”,
messages=[{“role”: “user”, “content”: reply_prompt}]
)
reply_text = reply_response.choices[0].message.content
send_reply(
service,
email_data[‘thread_id’],
to_email,
f”Re: {email_data[‘subject’]}”,
reply_text
)
return classification
“`
This is where automation becomes real. The system:
1. Classifies the email
2. Applies a label for organization
3. Generates and sends a reply if warranted
4. Keeps everything in the same email thread
## Handling Rate Limits and Errors
The Gmail API has strict rate limits. Google’s quota page shows 250 quota units per second for most operations, but individual requests are also limited. Here’s how to handle this:
“`python
import time
from googleapiclient.errors import HttpError
def safe_api_call(func, *args, max_retries=3, **kwargs):
“””Wrap API calls with retry logic and rate limit handling.”””
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except HttpError as e:
if e.resp.status == 429:
# Rate limited – wait and retry
wait_time = (attempt + 1) * 5
print(f”Rate limited. Waiting {wait_time}s…”)
time.sleep(wait_time)
elif e.resp.status >= 500:
# Server error – wait and retry
time.sleep(2 ** attempt)
else:
raise e
raise Exception(f”Failed after {max_retries} attempts”)
“`
You’ll also want to add exponential backoff and consider running processing during off-peak hours if you’re dealing with high volumes.
**A real limitation:** Google’s OAuth tokens expire after about an hour. For long-running automation, implement token refresh or use service accounts with domain-wide delegation.
## Putting It All Together
Here’s the main loop that ties everything together:
“`python
def main():
service = get_gmail_service()
emails = fetch_unread_emails(service, max_results=5)
print(f”Processing {len(emails)} emails…”)
for email_msg in emails:
email_data = parse_email(email_msg)
# Skip emails you’ve already processed
if ‘Processed’ in email_data.get(‘labels’, []):
continue
result = process_email(service, email_data)
print(f” {email
