Build email automation with Claude and Python. Covers drafting replies, classifying inbound email, extracting action items, bulk processing with Batch API, and Gmail/SMTP integration.
Automate email workflows with Claude — draft replies, classify inbound messages, extract action items, and bulk-process with the Anthropic Batch API at 50% cost.
import anthropic
client = anthropic.Anthropic()
def draft_reply(original_email: str, context: str = "") -> str:
"""Draft a professional reply to an incoming email."""
system = (
"You are an expert email assistant. Draft concise, professional replies. "
"Match the tone of the original. Output only the email body — no subject line, "
"no greeting/sign-off unless explicitly asked."
)
prompt = f"Original email:\n\n{original_email}"
if context:
prompt += f"\n\nAdditional context: {context}"
response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=512,
system=system,
messages=[{"role": "user", "content": prompt}],
)
return response.content[0].text
# Usage
reply = draft_reply(
original_email="Hi, can we move our 3pm meeting to 4pm tomorrow?",
context="I'm free at 4pm. Keep it brief.",
)
print(reply)
import json
CATEGORIES = ["billing", "technical_support", "sales_inquiry", "general", "spam"]
def classify_email(subject: str, body: str) -> dict:
"""Classify email and return category + priority + summary."""
response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=256,
system="You are an email classifier for a SaaS company. Return valid JSON only.",
messages=[{
"role": "user",
"content": (
f"Subject: {subject}\nBody: {body[:2000]}\n\n"
f"Classify into one of {CATEGORIES}. "
"Return JSON: {{\"category\": str, \"priority\": \"high|medium|low\", "
"\"summary\": str (max 20 words), \"auto_reply\": bool}}"
),
}],
)
return json.loads(response.content[0].text)
result = classify_email(
subject="URGENT: Can't log in — production down",
body="Our entire team is locked out. This is blocking us.",
)
# {"category": "technical_support", "priority": "high", "summary": "Team locked out, production down", "auto_reply": false}
def extract_action_items(email_thread: str) -> dict:
"""Extract tasks, deadlines, and owners from an email thread."""
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=512,
system="Extract action items from email threads. Return valid JSON only.",
messages=[{
"role": "user",
"content": (
f"Email thread:\n\n{email_thread}\n\n"
"Return JSON: {{\"action_items\": [{{\"task\": str, \"owner\": str|null, "
"\"deadline\": str|null, \"priority\": \"high|medium|low\"}}]}}"
),
}],
)
return json.loads(response.content[0].text)
thread = """
From: Alice: Can you review the contract by Friday?
From: Bob: Sure, I'll handle it. Can you send the vendor details too?
From: Alice: Yes, I'll send them by EOD Wednesday.
"""
items = extract_action_items(thread)
# action_items: [{task: "Review contract", owner: "Bob", deadline: "Friday", priority: "high"}, ...]
def bulk_classify_emails(emails: list[dict]) -> list[dict]:
"""Classify a list of emails in one Batch API call at 50% cost."""
requests = [
{
"custom_id": f"email-{i}",
"params": {
"model": "claude-haiku-4-5-20251001",
"max_tokens": 128,
"system": "Classify this email. Return JSON: {category, priority, auto_reply}.",
"messages": [{
"role": "user",
"content": f"Subject: {e['subject']}\n{e['body'][:500]}",
}],
},
}
for i, e in enumerate(emails)
]
batch = client.messages.batches.create(message_batch=requests)
print(f"Batch {batch.id} submitted — poll or use webhook for results")
return batch # poll batch.id for results when processing_status == "ended"
pip install google-api-python-client google-auth-httplib2 google-auth-oauthlib anthropic
from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials
import base64, email as email_lib
def fetch_and_reply(creds_path: str, max_emails: int = 5):
"""Fetch unread Gmail messages and draft Claude replies."""
creds = Credentials.from_authorized_user_file(creds_path)
service = build("gmail", "v1", credentials=creds)
results = service.users().messages().list(
userId="me", q="is:unread", maxResults=max_emails
).execute()
for msg_meta in results.get("messages", []):
msg = service.users().messages().get(
userId="me", id=msg_meta["id"], format="full"
).execute()
# Decode body
data = msg["payload"]["body"].get("data", "")
body = base64.urlsafe_b64decode(data + "==").decode("utf-8", errors="ignore")
subject = next(
(h["value"] for h in msg["payload"]["headers"] if h["name"] == "Subject"), ""
)
reply = draft_reply(original_email=f"Subject: {subject}\n\n{body[:1000]}")
print(f"Draft reply for '{subject}':\n{reply}\n")
| Approach | Customizability | Setup | Cost per 1K emails |
|---|---|---|---|
| Claude + Python (this guide) | Full — any task, any rule | ~1 hour | ~$0.08 (Haiku classify) |
| Zapier + ChatGPT action | Low — template flows | 15 min | Zapier plan + OpenAI |
| HubSpot AI | Low — CRM-bound | Minutes | Included in HubSpot plan |
| Fine-tuned BERT classifier | High but static | Days | Hosting cost only |
Key patterns: (1) Use claude-haiku-4-5-20251001 for classification and short drafts — it is 20× cheaper than Sonnet and fast enough for real-time routing. (2) Use the Batch API for bulk processing (nightly digest, backlog clearance) — 50% cost with async delivery. (3) Keep the original email body under 2000 chars for classification prompts; use Sonnet for long-thread extraction. For cost estimates, use the Claude API Cost Calculator. For structured data extraction from email attachments (PDFs, invoices), see the data extraction guide.