Email Integration¶

Aegis integrates with Google Workspace for email triage, calendar access, and document management.

Configuration¶

Email Account: aegis@richardbankole.com MCP Server: google-workspace

Configuration in ~/.claude.json:

{
  "mcpServers": {
    "google-workspace": {
      "command": "node",
      "args": ["/home/agent/mcp-servers/google-workspace/dist/index.js"],
      "env": {
        "GOOGLE_APPLICATION_CREDENTIALS": "/home/agent/.secure/google-workspace-credentials.json"
      }
    }
  }
}

MCP Tools¶

Gmail Operations¶

# Search emails
results = mcp__google_workspace__gmail_search(
    query="is:unread",
    max_results=50
)

# Get specific email
email = mcp__google_workspace__gmail_get(
    message_id="abc123"
)

# Send email
mcp__google_workspace__gmail_send(
    to="user@example.com",
    subject="Status Update",
    body="System is healthy."
)

# Mark as read
mcp__google_workspace__gmail_modify(
    message_id="abc123",
    add_labels=[],
    remove_labels=["UNREAD"]
)

Email Triage Workflow¶

The /email-triage skill automates email classification.

Classification Categories¶

Category	Description	Auto-action
urgent	Requires immediate attention	Flag + notify
action_required	Needs response within 24h	Add to tasks
informational	FYI, no action needed	Archive
promotional	Marketing, newsletters	Archive
spam	Unwanted content	Mark spam

Database Schema¶

CREATE TABLE email_cache (
    id SERIAL PRIMARY KEY,
    message_id TEXT UNIQUE NOT NULL,
    thread_id TEXT,
    from_email TEXT,
    from_name TEXT,
    subject TEXT,
    snippet TEXT,
    body_text TEXT,
    body_html TEXT,
    received_at TIMESTAMPTZ,
    is_read BOOLEAN DEFAULT false,
    labels TEXT[],
    category TEXT,  -- Classification result
    priority INTEGER,
    metadata JSONB DEFAULT '{}',
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

Usage¶

from aegis.email_triage import EmailTriageService

service = EmailTriageService()

# Triage unread emails
results = await service.triage_unread()
# Returns: {"processed": 25, "urgent": 2, "action_required": 8, ...}

# Classify single email
category = await service.classify_email(
    subject="Meeting tomorrow at 3pm",
    body="Let's discuss the new feature...",
    sender="boss@company.com"
)
# Returns: "action_required"

Search Queries¶

Gmail supports powerful search syntax:

# Unread emails
query = "is:unread"

# From specific sender
query = "from:user@example.com"

# Date range
query = "after:2026/01/20 before:2026/01/25"

# With attachment
query = "has:attachment"

# Combine filters
query = "is:unread from:important@client.com subject:urgent"

# Exclude categories
query = "is:unread -category:promotions"

Daily Email Check¶

Part of the morning routine:

# Check unread count
unread = db.fetch_one("""
    SELECT COUNT(*) as cnt FROM email_cache
    WHERE is_read = false AND received_at > NOW() - INTERVAL '24 hours'
""")

if unread.get('cnt', 0) > 10:
    # Run triage
    await service.triage_unread()

    # Notify about urgent emails
    urgent_emails = db.fetch_all("""
        SELECT from_email, subject FROM email_cache
        WHERE category = 'urgent' AND is_read = false
    """)

    if urgent_emails:
        send_discord_alert(f"🚨 {len(urgent_emails)} urgent emails")

Best Practices¶

Run Triage Regularly: Schedule via cron (e.g., every 2 hours during work hours)
Cache Locally: Store email metadata in PostgreSQL to avoid repeated API calls
Rate Limits: Gmail API has quotas (10,000 quota units per day for free tier)
Privacy: Never log full email bodies in plain text logs
Filters: Use Gmail filters for common patterns before triage

Troubleshooting¶

Authentication Failures¶

Regenerate credentials:

# Check credentials file
cat ~/.secure/google-workspace-credentials.json

# Verify service account has domain-wide delegation
# See: https://developers.google.com/workspace/guides/create-credentials

Rate Limit Errors¶

Reduce frequency:

# Add exponential backoff
import time

retries = 0
while retries < 3:
    try:
        results = mcp__google_workspace__gmail_search(query="is:unread")
        break
    except RateLimitError:
        wait = 2 ** retries
        time.sleep(wait)
        retries += 1

Missing Emails¶

Check label filters:

# Get all labels
labels = mcp__google_workspace__gmail_list_labels()

# Include all mail
results = mcp__google_workspace__gmail_search(
    query="in:anywhere is:unread"
)

Advanced Features¶

Thread Management¶

# Get full thread
thread = mcp__google_workspace__gmail_get_thread(
    thread_id="abc123"
)

# Reply to thread
mcp__google_workspace__gmail_send(
    to="user@example.com",
    subject="Re: Original Subject",
    body="Reply text",
    thread_id="abc123"
)

Label Management¶

# Create custom label
mcp__google_workspace__gmail_create_label(
    name="Aegis/Processed"
)

# Apply label
mcp__google_workspace__gmail_modify(
    message_id="abc123",
    add_labels=["Aegis/Processed"]
)

Batch Operations¶

# Process multiple emails
for email_id in unread_ids:
    # Classify
    category = classify_email(email)

    # Apply label based on category
    if category == "urgent":
        gmail_modify(email_id, add_labels=["STARRED", "IMPORTANT"])
    elif category == "promotional":
        gmail_modify(email_id, add_labels=["TRASH"])

Integration with Other Systems¶

Add to Beads Tasks¶

if category == "action_required":
    import subprocess
    subprocess.run([
        "bd", "create",
        f"Respond to email: {subject}",
        "-p", "2",  # Priority 2
        "-t", "task"
    ])

Post to Discord¶

mcp__discord__discord_send_message(
    channel_id="1455049133189627926",  # #tasks
    content=f"📧 Urgent email from {sender}: {subject}"
)

WhatsApp Notifications¶

from aegis.whatsapp import send_whatsapp_message

await send_whatsapp_message(
    to_number="447490195079",
    text=f"📧 {urgent_count} urgent emails require attention"
)