Automate Email with AI: Build a Smart Email Processor Using Python and Claude

The average professional receives 121 emails per day. Of those, maybe 10 require thoughtful responses, 30 need a quick acknowledgment, and the rest are newsletters, notifications, and CC’d threads you never asked to be part of.

What if an AI could handle the triage? Classify every incoming email, draft responses for the straightforward ones, flag the important ones, and quietly archive the noise — all before you open your inbox.

That’s what we’re building. A Python script that connects to your Gmail, uses Claude to understand each email, and takes action based on intelligent classification. No third-party apps. No monthly subscriptions. Just your Gmail API credentials and an Anthropic API key.

---

What We’re Building

Incoming email
  → Fetch from Gmail API
  → Send to Claude for classification
  → Claude returns:
      ├── Category: urgent / needs-reply / informational / spam
      ├── Priority: high / medium / low
      ├── Summary: one-sentence description
      └── Draft reply (if needs-reply)
  → Take action:
      ├── Urgent: Label "URGENT", send Slack notification
      ├── Needs reply: Save draft response in Gmail
      ├── Informational: Label and archive
      └── Spam/noise: Archive immediately

---

Prerequisites

• Python 3.11+
• Google Cloud project with Gmail API enabled
• Anthropic API key
• Gmail account

---

Step 1: Set Up Gmail API

Create Google Cloud Project

1. Go to Google Cloud Console
2. Create a new project (or select existing)
3. Enable the Gmail API:
   • Navigate to APIs & Services > Library
   • Search “Gmail API”
   • Click Enable

Create OAuth Credentials

1. Go to APIs & Services > Credentials
2. Click “Create Credentials” > “OAuth 2.0 Client ID”
3. Application type: “Desktop application”
4. Download the JSON file as credentials.json

Install Dependencies

mkdir email-ai && cd email-ai
python3 -m venv venv
source venv/bin/activate

pip install google-auth-oauthlib google-api-python-client anthropic python-dotenv

---

Step 2: Gmail Authentication

# gmail_auth.py
"""Handle Gmail API authentication."""

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.modify',
    'https://www.googleapis.com/auth/gmail.compose',
]


def get_gmail_service():
    """Authenticate and return Gmail API service."""
    creds = None
    
    # Load saved credentials
    if os.path.exists('token.pickle'):
        with open('token.pickle', 'rb') as token:
            creds = pickle.load(token)
    
    # Refresh or create new credentials
    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)
        
        # Save credentials for future runs
        with open('token.pickle', 'wb') as token:
            pickle.dump(creds, token)
    
    return build('gmail', 'v1', credentials=creds)

---

Step 3: Email Fetcher

# email_fetcher.py
"""Fetch and parse emails from Gmail."""

import base64
from email.mime.text import MIMEText
from typing import Optional


def get_unread_emails(service, max_results: int = 20) -> list[dict]:
    """Fetch unread emails from inbox."""
    results = service.users().messages().list(
        userId='me',
        labelIds=['INBOX', 'UNREAD'],
        maxResults=max_results
    ).execute()
    
    messages = results.get('messages', [])
    emails = []
    
    for msg_ref in messages:
        msg = service.users().messages().get(
            userId='me',
            id=msg_ref['id'],
            format='full'
        ).execute()
        
        email_data = parse_email(msg)
        if email_data:
            emails.append(email_data)
    
    return emails


def parse_email(message: dict) -> Optional[dict]:
    """Parse a Gmail message into a clean dict."""
    headers = message.get('payload', {}).get('headers', [])
    
    def get_header(name: str) -> str:
        for h in headers:
            if h['name'].lower() == name.lower():
                return h['value']
        return ''
    
    # Extract body
    body = extract_body(message.get('payload', {}))
    
    return {
        'id': message['id'],
        'thread_id': message['threadId'],
        'from': get_header('From'),
        'to': get_header('To'),
        'subject': get_header('Subject'),
        'date': get_header('Date'),
        'body': body[:3000],  # Limit body length for API
        'snippet': message.get('snippet', ''),
        'labels': message.get('labelIds', []),
    }


def extract_body(payload: dict) -> str:
    """Extract plain text body from email payload."""
    if payload.get('mimeType') == 'text/plain':
        data = payload.get('body', {}).get('data', '')
        if data:
            return base64.urlsafe_b64decode(data).decode('utf-8', errors='ignore')
    
    # Check parts for multipart messages
    for part in payload.get('parts', []):
        if part.get('mimeType') == 'text/plain':
            data = part.get('body', {}).get('data', '')
            if data:
                return base64.urlsafe_b64decode(data).decode('utf-8', errors='ignore')
        # Recurse into nested parts
        if part.get('parts'):
            result = extract_body(part)
            if result:
                return result
    
    return ''


def create_draft(service, to: str, subject: str, body: str, 
                 thread_id: Optional[str] = None) -> dict:
    """Create a draft reply in Gmail."""
    message = MIMEText(body)
    message['to'] = to
    message['subject'] = subject
    
    raw = base64.urlsafe_b64encode(
        message.as_bytes()
    ).decode('utf-8')
    
    draft_body = {'message': {'raw': raw}}
    if thread_id:
        draft_body['message']['threadId'] = thread_id
    
    return service.users().drafts().create(
        userId='me',
        body=draft_body
    ).execute()


def add_label(service, message_id: str, label_ids: list[str]) -> None:
    """Add labels to a message."""
    service.users().messages().modify(
        userId='me',
        id=message_id,
        body={'addLabelIds': label_ids}
    ).execute()


def archive_message(service, message_id: str) -> None:
    """Remove INBOX label (archive)."""
    service.users().messages().modify(
        userId='me',
        id=message_id,
        body={'removeLabelIds': ['INBOX']}
    ).execute()

---

Step 4: AI Email Classifier

# ai_classifier.py
"""Use Claude to classify and respond to emails."""

import json
import anthropic
from typing import Optional


class EmailClassifier:
    def __init__(self, api_key: str):
        self.client = anthropic.Anthropic(api_key=api_key)
    
    def classify_email(self, email: dict) -> dict:
        """Classify an email and optionally draft a response."""
        
        prompt = f"""Analyze this email and return a JSON response.

From: {email['from']}
Subject: {email['subject']}
Date: {email['date']}

Body:
{email['body']}

Return ONLY valid JSON with these fields:
{{
  "category": "urgent|needs-reply|informational|noise",
  "priority": "high|medium|low",
  "summary": "one sentence summary",
  "reasoning": "brief explanation of classification",
  "needs_response": true/false,
  "suggested_reply": "draft reply text if needs_response is true, null otherwise",
  "action": "draft-reply|label-archive|archive|flag"
}}

Classification rules:
- urgent: Time-sensitive requests, payment issues, system alerts, boss/client requests
- needs-reply: Questions directed at the recipient, meeting requests, action items
- informational: Newsletters, updates, FYI emails, automated reports
- noise: Marketing, promotions, social notifications, CC'd threads with no action needed

For suggested_reply:
- Keep it professional and concise
- Match the tone of the original email
- If unclear what to respond, set needs_response to false"""

        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            system=(
                "You are an email triage assistant. Classify emails accurately "
                "and draft helpful responses. Always return valid JSON."
            ),
            messages=[{"role": "user", "content": prompt}]
        )
        
        result_text = response.content[0].text
        
        # Parse JSON from response (handle markdown code blocks)
        if "```json" in result_text:
            result_text = result_text.split("```json")[1].split("```")[0]
        elif "```" in result_text:
            result_text = result_text.split("```")[1].split("```")[0]
        
        try:
            return json.loads(result_text.strip())
        except json.JSONDecodeError:
            return {
                "category": "informational",
                "priority": "low",
                "summary": "Could not classify",
                "reasoning": "JSON parsing failed",
                "needs_response": False,
                "suggested_reply": None,
                "action": "label-archive"
            }

---

Step 5: Main Processor

# processor.py
"""Main email processing pipeline."""

import os
import json
from datetime import datetime
from dotenv import load_dotenv

from gmail_auth import get_gmail_service
from email_fetcher import (
    get_unread_emails, create_draft, add_label, archive_message
)
from ai_classifier import EmailClassifier

load_dotenv()


def ensure_labels(service) -> dict:
    """Create custom labels if they don't exist."""
    existing = service.users().labels().list(userId='me').execute()
    existing_names = {l['name']: l['id'] for l in existing.get('labels', [])}
    
    needed = ['AI/Urgent', 'AI/NeedsReply', 'AI/Informational', 'AI/Processed']
    label_ids = {}
    
    for name in needed:
        if name in existing_names:
            label_ids[name] = existing_names[name]
        else:
            label = service.users().labels().create(
                userId='me',
                body={
                    'name': name,
                    'labelListVisibility': 'labelShow',
                    'messageListVisibility': 'show',
                }
            ).execute()
            label_ids[name] = label['id']
    
    return label_ids


def process_emails():
    """Main processing function."""
    print(f"\n{'='*60}")
    print(f"Email AI Processor - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"{'='*60}\n")
    
    # Initialize services
    service = get_gmail_service()
    classifier = EmailClassifier(api_key=os.getenv('ANTHROPIC_API_KEY'))
    labels = ensure_labels(service)
    
    # Fetch unread emails
    emails = get_unread_emails(service, max_results=10)
    print(f"Found {len(emails)} unread emails\n")
    
    results = []
    
    for email in emails:
        print(f"Processing: {email['subject'][:60]}...")
        
        # Classify with AI
        classification = classifier.classify_email(email)
        
        # Take action based on classification
        action = classification.get('action', 'label-archive')
        category = classification.get('category', 'informational')
        
        if category == 'urgent':
            add_label(service, email['id'], [labels['AI/Urgent']])
            print(f"  → URGENT: {classification['summary']}")
            
        elif category == 'needs-reply' and classification.get('suggested_reply'):
            # Create draft reply
            sender = email['from']
            subject = email['subject']
            if not subject.startswith('Re:'):
                subject = f"Re: {subject}"
            
            create_draft(
                service,
                to=sender,
                subject=subject,
                body=classification['suggested_reply'],
                thread_id=email['thread_id']
            )
            add_label(service, email['id'], [labels['AI/NeedsReply']])
            print(f"  → DRAFT CREATED: {classification['summary']}")
            
        elif category == 'informational':
            add_label(service, email['id'], [labels['AI/Informational']])
            archive_message(service, email['id'])
            print(f"  → ARCHIVED: {classification['summary']}")
            
        elif category == 'noise':
            archive_message(service, email['id'])
            print(f"  → NOISE (archived): {classification['summary']}")
        
        # Mark as processed
        add_label(service, email['id'], [labels['AI/Processed']])
        
        results.append({
            'subject': email['subject'],
            'from': email['from'],
            **classification
        })
    
    # Print summary
    print(f"\n{'='*60}")
    print("SUMMARY")
    print(f"{'='*60}")
    for r in results:
        icon = {'urgent': '🔴', 'needs-reply': '🟡', 
                'informational': '🔵', 'noise': '⚪'}.get(r['category'], '⚪')
        print(f"{icon} [{r['priority'].upper()}] {r['summary']}")
    
    # Save log
    log_file = f"logs/email_log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
    os.makedirs('logs', exist_ok=True)
    with open(log_file, 'w') as f:
        json.dump(results, f, indent=2)
    print(f"\nLog saved to {log_file}")


if __name__ == '__main__':
    process_emails()

---

Step 6: Schedule Automated Runs

Using cron (Linux/macOS)

# Run every 15 minutes
crontab -e

# Add this line:
*/15 * * * * cd /path/to/email-ai && /path/to/email-ai/venv/bin/python processor.py >> /path/to/email-ai/logs/cron.log 2>&1

Using systemd timer (Linux)

# Create service file
sudo tee /etc/systemd/system/email-ai.service << 'EOF'
[Unit]
Description=AI Email Processor

[Service]
Type=oneshot
User=your-username
WorkingDirectory=/path/to/email-ai
ExecStart=/path/to/email-ai/venv/bin/python processor.py
EOF

# Create timer file
sudo tee /etc/systemd/system/email-ai.timer << 'EOF'
[Unit]
Description=Run AI Email Processor every 15 minutes

[Timer]
OnBootSec=5min
OnUnitActiveSec=15min

[Install]
WantedBy=timers.target
EOF

sudo systemctl enable email-ai.timer
sudo systemctl start email-ai.timer

---

Cost Analysis

Per email processed:
- Input: ~800 tokens (email content + prompt)
- Output: ~200 tokens (classification + draft)
- Cost per email (Claude Sonnet): ~$0.005

Daily (50 emails processed):
- $0.25/day
- $7.50/month

Compare to commercial alternatives:
- SaneBox: $7/month (no AI drafting)
- Superhuman: $30/month
- Your solution: $7.50/month + full control

---

Safety Considerations

Never auto-send replies. Always create drafts that you review before sending. AI can misinterpret tone, miss context from previous threads, or draft responses that are technically correct but socially wrong.

Exclude sensitive senders. Add a blocklist for emails from your boss, legal department, HR, or clients where AI processing is inappropriate:

SKIP_SENDERS = [
    'ceo@company.com',
    'legal@company.com',
    'hr@company.com',
]

# In processor.py, before classification:
sender_email = email['from'].lower()
if any(skip in sender_email for skip in SKIP_SENDERS):
    print(f"  → SKIPPED (sensitive sender)")
    continue

Log everything. Keep records of what the AI classified and what actions it took. If something goes wrong, you need an audit trail.

This system handles the email triage that would otherwise eat 30-60 minutes of your morning. It doesn’t replace your judgment — it does the sorting so you can focus your judgment on the emails that actually need your brain.