Email & Calendar AI Agent
What is an Email and Calendar Agent?
Email and calendar agents automate routine productivity tasks â reading, classifying, and responding to emails, scheduling meetings, extracting action items, and providing daily briefings. They transform an overwhelmed inbox into a managed workflow.
Why this matters: The average professional spends 28% of their workweek on email (McKinsey, 2023). That's 11+ hours per week. An email agent can reduce this by 60% by handling classification, drafting, and scheduling automatically.
Common Misconception
"Email agents just send automated replies."
Production email agents do much more: they classify urgency, extract tasks with deadlines, draft contextually appropriate responses, schedule meetings based on calendar availability, and provide daily briefings that combine email urgency with calendar context.
Real-World Analogy
Think of it as an extremely efficient executive assistant. They read all your emails, prioritize by urgency, draft responses in your voice, extract action items, check your calendar for availability, schedule meetings, and give you a morning briefing â all before your first coffee.
Project Overview
We will build a productivity agent that:
- Reads and classifies incoming emails (urgent, FYI, spam)
- Extracts action items and deadlines from email content
- Drafts contextually appropriate email responses
- Schedules meetings based on email requests and calendar availability
- Provides daily briefings combining email urgency with calendar context
Expected outcome: An agent that manages your inbox and calendar automatically.
Difficulty: Advanced (requires understanding of Google APIs, OAuth2, and NLP)
Tools & Setup
| Tool | Version | Purpose |
|---|---|---|
| Python | 3.11+ | Core language |
| google-api-python-client | 2.0+ | Gmail/Calendar APIs |
| google-auth-oauthlib | 1.0+ | OAuth2 authentication |
| openai | 1.0+ | LLM backbone |
| pydantic | 2.0+ | Data models |
Step 1: Environment Setup
python -m venv venv
source venv/bin/activate
pip install google-api-python-client google-auth-oauthlib openai pydantic
Step 2: Gmail Client
# gmail/client.py
"""Gmail API client with OAuth2 authentication and email operations."""
import base64
import logging
from typing import List, Dict
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
logger = logging.getLogger(__name__)
class GmailClient:
"""Gmail API client with automatic OAuth2 token refresh.
Args:
credentials_path: Path to OAuth2 credentials JSON file.
"""
SCOPES = ["https://mail.google.com/"]
def __init__(self, credentials_path: str = "credentials.json"):
self.service = None
self.credentials_path = credentials_path
def authenticate(self, token_path: str = "token.json") -> None:
"""Authenticate with OAuth2 and build Gmail service."""
creds = None
try:
creds = Credentials.from_authorized_user_file(token_path, self.SCOPES)
except (FileNotFoundError, ValueError):
pass
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(google.auth.transport.requests.Request())
else:
flow = InstalledAppFlow.from_client_secrets_file(
self.credentials_path, self.SCOPES
)
creds = flow.run_local_server(port=0)
with open(token_path, "w") as f:
f.write(creds.to_json())
self.service = build("gmail", "v1", credentials=creds)
logger.info("Gmail authenticated successfully")
def get_recent_emails(self, max_results: int = 10, query: str = "") -> List[Dict]:
"""Fetch recent emails with headers and body.
Args:
max_results: Maximum number of emails to fetch.
query: Gmail search query (e.g., "is:unread").
Returns:
List of email dicts with id, subject, from, body, labels.
"""
results = self.service.users().messages().list(
userId="me", maxResults=max_results, q=query
).execute()
messages = results.get("messages", [])
emails = []
for msg in messages:
full = self.service.users().messages().get(
userId="me", id=msg["id"]
).execute()
headers = {
h["name"]: h["value"]
for h in full["payload"].get("headers", [])
}
body = self._extract_body(full["payload"])
emails.append({
"id": msg["id"],
"subject": headers.get("Subject", ""),
"from": headers.get("From", ""),
"to": headers.get("To", ""),
"date": headers.get("Date", ""),
"snippet": full.get("snippet", ""),
"body": body[:5000],
"labels": full.get("labelIds", []),
})
logger.info(f"Fetched {len(emails)} emails")
return emails
def _extract_body(self, payload: dict) -> str:
"""Extract plain text body from Gmail's MIME structure."""
if "body" in payload and payload["body"].get("data"):
return base64.urlsafe_b64decode(
payload["body"]["data"]
).decode("utf-8", errors="ignore")
if "parts" in payload:
for part in payload["parts"]:
if part["mimeType"] == "text/plain":
data = part.get("body", {}).get("data")
if data:
return base64.urlsafe_b64decode(
data
).decode("utf-8", errors="ignore")
return ""
def send_email(self, to: str, subject: str, body: str) -> Dict:
"""Send an email via Gmail API.
Args:
to: Recipient email address.
subject: Email subject line.
body: Email body text.
Returns:
Gmail API response with sent message ID.
"""
message = {
"raw": base64.urlsafe_b64encode(
f"To: {to}\nSubject: {subject}\n\n{body}".encode()
).decode()
}
return self.service.users().messages().send(
userId="me", body=message
).execute()
Step 3: Calendar Client
# calendar/client.py
"""Google Calendar API client with free slot detection and event creation."""
import logging
from datetime import datetime, timedelta
from typing import List, Dict
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
logger = logging.getLogger(__name__)
class CalendarClient:
"""Google Calendar client with scheduling intelligence.
Args:
credentials_path: Path to OAuth2 credentials JSON file.
"""
SCOPES = ["https://www.googleapis.com/auth/calendar"]
def __init__(self, credentials_path: str = "credentials.json"):
self.service = None
self.credentials_path = credentials_path
def authenticate(self, token_path: str = "calendar_token.json") -> None:
"""Authenticate and build Calendar service."""
creds = None
try:
creds = Credentials.from_authorized_user_file(token_path, self.SCOPES)
except (FileNotFoundError, ValueError):
pass
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(google.auth.transport.requests.Request())
else:
flow = InstalledAppFlow.from_client_secrets_file(
self.credentials_path, self.SCOPES
)
creds = flow.run_local_server(port=0)
with open(token_path, "w") as f:
f.write(creds.to_json())
self.service = build("calendar", "v3", credentials=creds)
logger.info("Calendar authenticated successfully")
def get_events(self, days_ahead: int = 7) -> List[Dict]:
"""Get upcoming events for the next N days."""
now = datetime.utcnow().isoformat() + "Z"
end = (datetime.utcnow() + timedelta(days=days_ahead)).isoformat() + "Z"
events = self.service.events().list(
calendarId="primary",
timeMin=now, timeMax=end,
singleEvents=True, orderBy="startTime",
).execute()
return [
{
"id": e["id"],
"summary": e.get("summary", "No title"),
"start": e["start"].get("dateTime", e["start"].get("date")),
"end": e["end"].get("dateTime", e["end"].get("date")),
"description": e.get("description", ""),
"attendees": [a["email"] for a in e.get("attendees", [])],
}
for e in events.get("items", [])
]
def get_free_slots(self, date: datetime, duration_minutes: int = 60) -> List[Dict]:
"""Find free time slots for a given date.
Args:
date: Date to check availability.
duration_minutes: Required slot duration.
Returns:
List of free time slots with start/end times.
"""
day_start = date.replace(hour=9, minute=0, second=0).isoformat() + "Z"
day_end = date.replace(hour=17, minute=0, second=0).isoformat() + "Z"
events = self.service.events().list(
calendarId="primary",
timeMin=day_start, timeMax=day_end,
singleEvents=True,
).execute()
busy_slots = [
(e["start"].get("dateTime", e["start"].get("date")),
e["end"].get("dateTime", e["end"].get("date")))
for e in events.get("items", [])
]
free_slots = []
current = datetime.fromisoformat(day_start.replace("Z", "+00:00"))
end = datetime.fromisoformat(day_end.replace("Z", "+00:00"))
while current + timedelta(minutes=duration_minutes) <= end:
slot_end = current + timedelta(minutes=duration_minutes)
is_free = all(
slot_end.isoformat() <= busy[0] or current.isoformat() >= busy[1]
for busy in busy_slots
)
if is_free:
free_slots.append({
"start": current.isoformat(),
"end": slot_end.isoformat(),
})
current += timedelta(minutes=30)
return free_slots
def create_event(self, summary: str, start: str, end: str,
description: str = "", attendees: List[str] = None) -> Dict:
"""Create a calendar event with optional attendees."""
event = {
"summary": summary,
"start": {"dateTime": start, "timeZone": "UTC"},
"end": {"dateTime": end, "timeZone": "UTC"},
"description": description,
}
if attendees:
event["attendees"] = [{"email": a} for a in attendees]
return self.service.events().insert(
calendarId="primary", body=event, sendNotifications=True
).execute()
Step 4: Processing Modules
# processing/classifier.py
"""Email classification by urgency, intent, and required action."""
import json
import logging
from openai import OpenAI
logger = logging.getLogger(__name__)
class EmailClassifier:
"""Classify emails by urgency and intent using LLM.
Args:
model: OpenAI model for classification.
"""
def __init__(self, model: str = "gpt-4o"):
self.client = OpenAI()
self.model = model
def classify(self, email: dict) -> dict:
"""Classify an email by urgency, intent, and required action.
Args:
email: Email dict with subject, from, body.
Returns:
Dict with urgency, intent, requires_response, action_needed.
"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": """Classify email by urgency and intent.
Return JSON:
{
"urgency": "urgent|high|medium|low",
"intent": "request|information|meeting|deadline|spam|newsletter|other",
"requires_response": true/false,
"action_needed": "brief description"
}"""},
{"role": "user", "content": f"Subject: {email['subject']}\nFrom: {email['from']}\nBody: {email['body'][:1000]}"},
],
temperature=0.0,
)
try:
return json.loads(response.choices[0].message.content)
except (json.JSONDecodeError, IndexError):
return {"urgency": "medium", "intent": "other", "requires_response": False}
# processing/task_extractor.py
"""Extract action items and deadlines from email content."""
import json
import logging
from openai import OpenAI
logger = logging.getLogger(__name__)
class TaskExtractor:
"""Extract tasks with deadlines from email content.
Args:
model: OpenAI model for extraction.
"""
def __init__(self, model: str = "gpt-4o"):
self.client = OpenAI()
self.model = model
def extract(self, email: dict) -> list:
"""Extract action items from email.
Returns:
List of task dicts with task, deadline, priority.
"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": """Extract action items from email.
Return JSON array:
[{"task": "description", "deadline": "date or null", "priority": "high|medium|low"}]"""},
{"role": "user", "content": f"Email:\n{email['body'][:2000]}"},
],
temperature=0.0,
)
try:
return json.loads(response.choices[0].message.content)
except (json.JSONDecodeError, IndexError):
return []
# processing/response_drafter.py
"""Draft contextually appropriate email responses."""
import logging
from openai import OpenAI
logger = logging.getLogger(__name__)
class ResponseDrafter:
"""Draft email responses in specified tone.
Args:
model: OpenAI model for drafting.
"""
def __init__(self, model: str = "gpt-4o"):
self.client = OpenAI()
self.model = model
def draft(self, email: dict, context: str = "", tone: str = "professional") -> str:
"""Draft a response to an email.
Args:
email: Original email dict.
context: Additional context for the response.
tone: Response tone (professional, friendly, formal).
Returns:
Draft response text.
"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": f"""Draft email response in {tone} tone.
Be concise, clear, and action-oriented.
Never make promises about refunds or credits without approval."""},
{"role": "user", "content": f"Original email:\nSubject: {email['subject']}\nFrom: {email['from']}\n{email['body'][:2000]}\n\n{context}\n\nDraft response:"},
],
temperature=0.7,
)
return response.choices[0].message.content
Step 5: Complete Agent
# agent.py
"""Complete productivity agent orchestrating email, calendar, and processing modules."""
import logging
from datetime import datetime
from typing import Dict, List
from gmail.client import GmailClient
from calendar.client import CalendarClient
from processing.classifier import EmailClassifier
from processing.task_extractor import TaskExtractor
from processing.response_drafter import ResponseDrafter
logger = logging.getLogger(__name__)
class ProductivityAgent:
"""Email and calendar productivity agent.
Args:
credentials_path: Path to Google OAuth2 credentials JSON.
"""
def __init__(self, credentials_path: str = "credentials.json"):
self.gmail = GmailClient(credentials_path)
self.calendar = CalendarClient(credentials_path)
self.classifier = EmailClassifier()
self.task_extractor = TaskExtractor()
self.drafter = ResponseDrafter()
def authenticate(self) -> None:
"""Authenticate with Gmail and Calendar APIs."""
self.gmail.authenticate()
self.calendar.authenticate()
def get_daily_briefing(self) -> Dict:
"""Generate a daily briefing combining emails and calendar.
Returns:
Dict with urgent emails, tasks, upcoming events, and summary.
"""
emails = self.gmail.get_recent_emails(max_results=20)
classified = []
for email in emails:
classification = self.classifier.classify(email)
tasks = self.task_extractor.extract(email)
classified.append({
**email,
"classification": classification,
"tasks": tasks,
})
urgent = [e for e in classified if e["classification"]["urgency"] == "urgent"]
needs_response = [e for e in classified if e["classification"]["requires_response"]]
all_tasks = [t for e in classified for t in e["tasks"]]
events = self.calendar.get_events(days_ahead=1)
return {
"date": datetime.now().isoformat(),
"total_emails": len(classified),
"urgent_emails": urgent,
"needs_response": needs_response,
"upcoming_events": events,
"all_tasks": all_tasks,
"summary": self._generate_summary(urgent, needs_response, events, all_tasks),
}
def _generate_summary(self, urgent, needs_response, events, tasks) -> str:
"""Generate a natural language daily summary."""
parts = []
if urgent:
parts.append(f"{len(urgent)} urgent emails require immediate attention")
if needs_response:
parts.append(f"{len(needs_response)} emails need responses")
if events:
parts.append(f"{len(events)} meetings scheduled today")
if tasks:
parts.append(f"{len(tasks)} action items extracted")
return "; ".join(parts) if parts else "No urgent items today"
def draft_reply(self, email_id: str) -> str:
"""Draft a reply to a specific email by ID."""
emails = self.gmail.get_recent_emails(max_results=50)
email = next((e for e in emails if e["id"] == email_id), None)
if not email:
return "Email not found"
return self.drafter.draft(email)
def schedule_meeting(self, summary: str, date: str, duration: int = 60) -> Dict:
"""Schedule a meeting by finding the first free slot.
Args:
summary: Meeting title.
date: Date string (ISO format).
duration: Duration in minutes.
Returns:
Dict with created event or error.
"""
dt = datetime.fromisoformat(date)
free_slots = self.calendar.get_free_slots(dt, duration)
if not free_slots:
return {"error": "No free slots available"}
slot = free_slots[0]
return self.calendar.create_event(
summary=summary,
start=slot["start"],
end=slot["end"],
)
Mathematical Foundation
Email Priority Score:
Where:
- â urgency classification (0-1)
- â time sensitivity (deadline proximity)
- â sender importance score
Intuition: Balances urgency, time pressure, and sender importance for prioritization. For example: an email from your CEO (S=1.0) with a deadline tomorrow (T=0.9) marked urgent (U=1.0) scores 0.97.
Daily Briefing Priority Ranking:
Where are configurable weights. Typical: .
Performance Considerations
| Metric | Value | Cost Impact |
|---|---|---|
| Email Classification Latency | 50ms | LLM + rule-based hybrid |
| Draft Response Time | 2s | GPT-4o + context |
| Classification Accuracy | 85% | Urgency + Intent |
| Gmail API Quota | 250 units/sec | 5 units per message.list |
| Cost per Daily Briefing | $0.04 | ~4K tokens for 20 emails |
| Memory per Session | 50MB | With conversation history |
Security Notes
- OAuth2 token security â Never hardcode credentials; use environment variables
- Least-privilege scopes â Only request what you need (read-only vs full access)
- Token encryption â Encrypt token.json at rest
- Audit logging â Log all send actions for compliance
- PII handling â Strip PII before sending to LLM APIs
- Draft mode â Implement draft_mode flag to prevent accidental sends
Interview Questions
1. How does the agent handle OAuth2 token refresh automatically?
Answer: The agent stores the token in token.json and checks creds.valid before each API call. When expired, it uses creds.refresh(google.auth.transport.requests.Request()) to get a new access token without user interaction. The refresh token (obtained during initial OAuth consent) is long-lived and enables silent renewal. Handle RefreshError by re-authenticating.
2. What is the trade-off between Gmail Push Notifications vs polling?
Answer: Push notifications (via Pub/Sub) provide real-time updates but require Google Cloud infrastructure and webhook endpoints. Polling with users().messages().list() is simpler but introduces latency proportional to the polling interval. For production, push notifications reduce latency and API quota usage; for prototyping, polling at 5-minute intervals is sufficient.
3. How do you prevent the agent from sending drafts prematurely?
Answer: Implement a two-phase workflow: (1) the agent generates drafts and stores them in a local draft queue, (2) a human review step confirms before calling send_email(). In code, add a draft_mode: bool = True flag to the agent and only call send_email when explicitly confirmed. Never auto-send without human approval.
4. How does the classifier handle ambiguous emails?
Answer: The LLM classifier uses temperature=0.0 for deterministic output and returns structured JSON with confidence indicators. For ambiguous cases, the urgency defaults to "medium" and requires_response is False. A confidence threshold (e.g., >0.8) can be added to route uncertain emails to human review.
5. What rate limits exist for Gmail/Calendar APIs and how do you handle them?
Answer: Gmail API quota is 250 quota units/second/user. Each messages.list costs 5 units, messages.get costs 5 units. Calendar API has similar limits. Implement exponential backoff, batch API calls where possible, and cache recent results. The googleapiclient library handles 429 responses with automatic retry.
6. How do you handle email threading and context preservation?
Answer: Use the Gmail threadId field to group related messages. Store conversation history in a local database (SQLite) keyed by thread_id. When drafting responses, retrieve the full thread context to maintain continuity. The response drafter receives the last 3 messages in the thread as context.
7. How would you extend this to handle multiple calendars for a team?
Answer: Use Google Calendar's calendarList API to access shared calendars. For each team member, query their free/busy information using freebusy.query(). The scheduler then finds overlapping free slots across all attendees' calendars before creating the event.
8. What security considerations are important for this agent?
Answer: Never store OAuth credentials in source code â use environment variables or a secrets manager. Implement least-privilege OAuth scopes (only request what you need). Encrypt token.json at rest. Log all send actions for audit. Never include email body content in LLM training data. Use service accounts for enterprise deployments. Implement rate limiting to prevent abuse.
Common Pitfalls & Solutions
| Pitfall | Solution |
|---|---|
| OAuth token expiry errors | Implement automatic refresh with creds.refresh() and handle RefreshError by re-authenticating |
| API rate limits (429) | Use exponential backoff, batch requests, cache results, and respect Retry-After headers |
| Incorrect email classification | Fine-tune with domain-specific examples; use few-shot prompting with known categories |
| Calendar timezone confusion | Always pass timeZone parameter; use pytz for timezone conversions |
| Large email bodies slow processing | Truncate to 5000 chars; focus on subject + first paragraph for classification |
| Draft emails sent accidentally | Implement draft_mode flag; require human confirmation before sending |
| Recurring events not handled | Use singleEvents=True to expand; handle recurrence rules separately |
| Privacy concerns with email content | Process sensitive content locally; strip PII before sending to LLM APIs |
Summary with Key Takeaways
- Email classification enables automatic prioritization and routing based on urgency and intent
- Task extraction transforms unstructured emails into actionable items with deadlines
- Response drafting saves time while maintaining personalization and tone consistency
- Calendar integration enables automated scheduling with conflict-free slot detection
- Daily briefings provide actionable summaries combining email urgency with calendar context
- OAuth2 security is critical â never hardcode credentials, always implement token refresh
- Rate limit handling with exponential backoff ensures production reliability
- Always implement draft mode to prevent accidental email sends
KnowledgeCheck
-
What OAuth2 flow does Gmail API use for desktop applications?
- a) Authorization Code flow with refresh token
- b) Client Credentials flow
- c) Implicit flow
- d) PKCE flow only
-
What is the primary advantage of using
singleEvents=Truein Calendar API queries?- a) Faster response times
- b) Expands recurring events into individual instances
- c) Reduces API quota usage
- d) Includes cancelled events
-
Which LLM temperature setting is recommended for email classification?
- a) 0.7 (creative)
- b) 1.0 (maximum randomness)
- c) 0.0 (deterministic)
- d) 0.5 (balanced)
-
How does the agent extract plain text from Gmail's nested MIME structure?
- a) Always reads the top-level body
- b) Recursively searches
partsfortext/plainmimeType - c) Converts HTML to text
- d) Uses a third-party email parser
-
What happens when the agent encounters a Gmail API rate limit (429 error)?
- a) The request fails permanently
- b) The agent crashes
- c) Exponential backoff with automatic retry
- d) It switches to the Calendar API
-
What is the recommended approach for preventing accidental email sends?
- a) Don't implement email sending
- b) Implement draft_mode flag requiring human confirmation
- c) Use a weak LLM model
- d) Don't authenticate with Gmail
Answers: 1-a, 2-b, 3-c, 4-b, 5-c, 6-b