AI Agents · Lesson

Building a Simple Email Assistant Agent

End-to-end: read inbox → summarize → draft reply → await approval.

Building a Simple Email Assistant Agent is a free AI Agents lesson on CoddyKit — lesson 4 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Agents learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

Email Assistant Agent Architecture

An email assistant agent follows a standard pipeline: Fetch → Classify → Decide → Draft → Approve → Send. The agent reads unread emails, uses an LLM to classify and draft responses, then waits for human approval before sending. This human-in-the-loop design prevents costly mistakes from fully autonomous email sending.

# Email Assistant Pipeline:
#
# 1. FETCH: Pull unread emails from Gmail API
# 2. CLASSIFY: LLM labels each email
#    - 'action_needed': requires a reply
#    - 'fyi': informational, no reply needed
#    - 'spam': should be archived
# 3. DRAFT: LLM generates reply for 'action_needed' emails
# 4. APPROVE: Human reviews drafts in Gmail UI
# 5. SEND: Agent sends approved drafts
#
# Tools: Gmail API, Anthropic/OpenAI API, json, base64

print('Email assistant pipeline: fetch -> classify -> draft -> approve -> send')

Defining Agent Tools

Structure the agent with clear tool definitions. Each tool is a Python function with a specific responsibility. This modularity makes the agent testable, debuggable, and easy to extend with new capabilities.

class EmailAssistantTools:
    def __init__(self, gmail_service, llm_client):
        self.gmail = gmail_service
        self.llm = llm_client

    def fetch_unread(self, max_emails=10):
        '''Fetch unread emails from inbox.'''
        pass

    def classify_email(self, subject, body, sender):
        '''Ask LLM to classify: action_needed / fyi / spam.'''
        pass

    def draft_reply(self, subject, body, sender, context):
        '''Ask LLM to draft a reply to an email.'''
        pass

    def create_draft(self, message_id, reply_text):
        '''Save draft reply in Gmail for human review.'''
        pass

    def send_approved_drafts(self):
        '''Send all drafts marked as approved.'''
        pass

print('Tool-based architecture enables testing each step independently')

Step 1: Fetching and Parsing Emails

The first tool fetches unread emails and extracts the key fields the LLM needs: subject, sender, and plain-text body. Keep the body short — trim to the first 2000 characters to stay within LLM context limits and reduce API costs.

import base64

def fetch_emails_for_classification(gmail_service, max_emails=10):
    messages_list = gmail_service.users().messages().list(
        userId='me',
        q='is:unread label:inbox',
        maxResults=max_emails
    ).execute().get('messages', [])

    emails = []
    for ref in messages_list:
        msg = gmail_service.users().messages().get(
            userId='me', id=ref['id'], format='full'
        ).execute()

        headers = {h['name'].lower(): h['value']
                   for h in msg['payload'].get('headers', [])}
        body = extract_plain_text_body(msg)[:2000]  # trim for LLM

        emails.append({
            'id': msg['id'],
            'thread_id': msg['threadId'],
            'from': headers.get('from', ''),
            'subject': headers.get('subject', '(no subject)'),
            'body': body
        })

    print(f'Fetched {len(emails)} unread emails')
    return emails

Step 2: LLM Classification

Send each email to an LLM for classification. Use a structured prompt that asks the model to return a JSON object with the category and reasoning. Request JSON output explicitly — it's more reliable than parsing free text.

import json
import anthropic
import os

client = anthropic.Anthropic(api_key=os.environ['ANTHROPIC_API_KEY'])

def classify_email(subject, sender, body):
    prompt = (
        'Classify this email as exactly one of: action_needed, fyi, spam.\n'
        'Return JSON only: {"category": "...", "reason": "..."}\n\n'
        f'From: {sender}\n'
        f'Subject: {subject}\n\n'
        f'Body:\n{body[:1500]}'
    )

    response = client.messages.create(
        model='claude-opus-4-5',
        max_tokens=200,
        messages=[{'role': 'user', 'content': prompt}]
    )

    try:
        result = json.loads(response.content[0].text)
        return result.get('category', 'fyi'), result.get('reason', '')
    except json.JSONDecodeError:
        return 'fyi', 'Could not parse LLM response'

Step 3: Generating Reply Drafts

For emails classified as action_needed, ask the LLM to generate a draft reply. Provide context about the agent's role and tone. Ask for a professional but concise reply — include a placeholder for anything the human needs to fill in.

def draft_reply(subject, sender, body, agent_context):
    prompt = (
        'You are an email assistant drafting a professional reply.\n'
        'Guidelines:\n'
        '- Be concise and professional\n'
        '- Use [FILL IN] for any info you don\'t know\n'
        '- Start with a greeting, end with a sign-off\n\n'
        f'Context about the recipient\'s work: {agent_context}\n\n'
        f'Original email from {sender}:\n'
        f'Subject: {subject}\n\n'
        f'{body[:1500]}\n\n'
        'Draft a reply:'
    )

    response = client.messages.create(
        model='claude-opus-4-5',
        max_tokens=500,
        messages=[{'role': 'user', 'content': prompt}]
    )

    return response.content[0].text.strip()

Step 4: Saving Drafts for Human Review

Never send AI-generated emails without human review. Save them as Gmail drafts so the human can open Gmail, review, edit if needed, and send. The draft creation API is the same as the send API, just using drafts().create().

import base64
from email.mime.text import MIMEText

def save_draft_reply(gmail_service, original_message, reply_text):
    sender_header = next(
        (h['value'] for h in original_message['payload'].get('headers', [])
         if h['name'].lower() == 'from'), ''
    )
    subject = next(
        (h['value'] for h in original_message['payload'].get('headers', [])
         if h['name'].lower() == 'subject'), ''
    )
    msg_id_header = next(
        (h['value'] for h in original_message['payload'].get('headers', [])
         if h['name'].lower() == 'message-id'), ''
    )

    mime_msg = MIMEText(reply_text, 'plain', 'utf-8')
    mime_msg['To'] = sender_header
    mime_msg['Subject'] = 'Re: ' + subject
    mime_msg['In-Reply-To'] = msg_id_header
    mime_msg['References'] = msg_id_header

    raw = base64.urlsafe_b64encode(mime_msg.as_bytes()).decode()
    draft = gmail_service.users().drafts().create(
        userId='me',
        body={'message': {'raw': raw, 'threadId': original_message['threadId']}}
    ).execute()

    print(f'Draft saved: {draft["id"]}')
    return draft['id']

# --- demo: minimal stand-in for the Gmail API's service object ---
class _Exec:
    def __init__(self, result):
        self._result = result
    def execute(self):
        return self._result

class _FakeDrafts:
    def create(self, userId, body):
        return _Exec({'id': 'r9000abc'})

class _FakeUsers:
    def drafts(self):
        return _FakeDrafts()

class _FakeGmailService:
    def users(self):
        return _FakeUsers()

original_message = {
    'threadId': 'thread_1',
    'payload': {'headers': [
        {'name': 'From', 'value': 'customer@example.com'},
        {'name': 'Subject', 'value': 'Question about my order'},
        {'name': 'Message-ID', 'value': '<abc123@mail.example.com>'}
    ]}
}
save_draft_reply(_FakeGmailService(), original_message, 'Thanks for reaching out, we will look into it.')

Step 5: Marking Processed Emails

After processing an email (classified + draft created or archived), mark it to prevent re-processing. Add a custom label like AgentProcessed and remove the UNREAD label. Create the label once if it doesn't exist.

def get_or_create_label(gmail_service, label_name):
    labels = gmail_service.users().labels().list(userId='me').execute()
    for label in labels.get('labels', []):
        if label['name'] == label_name:
            return label['id']

    # Create the label
    new_label = gmail_service.users().labels().create(
        userId='me',
        body={
            'name': label_name,
            'labelListVisibility': 'labelShow',
            'messageListVisibility': 'show'
        }
    ).execute()
    print(f'Created label: {label_name}')
    return new_label['id']

def mark_processed(gmail_service, message_id, agent_label_id):
    gmail_service.users().messages().modify(
        userId='me',
        id=message_id,
        body={
            'addLabelIds': [agent_label_id],
            'removeLabelIds': ['UNREAD']
        }
    ).execute()

# --- demo: minimal stand-in for the Gmail API's service object ---
class _Exec:
    def __init__(self, result):
        self._result = result
    def execute(self):
        return self._result

class _FakeUsers:
    def __init__(self):
        self._labels = [{'id': 'Label_1', 'name': 'Processed'}]
    def labels(self):
        return self
    def list(self, userId):
        return _Exec({'labels': self._labels})
    def create(self, userId, body):
        print(f'Created label: {body["name"]}')
        return _Exec({'id': 'Label_2', 'name': body['name']})
    def messages(self):
        return self
    def modify(self, userId, id, body):
        print(f'Marked {id} processed with {body}')
        return _Exec({'id': id})

class _FakeGmailService:
    def users(self):
        return _FakeUsers()

gmail_service = _FakeGmailService()
label_id = get_or_create_label(gmail_service, 'Processed')
print(f'Label id: {label_id}')
new_label_id = get_or_create_label(gmail_service, 'AgentHandled')
mark_processed(gmail_service, 'msg_99', new_label_id)

Full Agent Run Loop

Connect all steps into a single run() method. The agent loops through emails, classifies each one, creates drafts for action items, archives spam, and marks all as processed. Log a summary at the end.

def run_email_agent(gmail_service, agent_context, max_emails=10):
    processed_label = get_or_create_label(gmail_service, 'AgentProcessed')
    emails = fetch_emails_for_classification(gmail_service, max_emails)

    summary = {'action_needed': 0, 'fyi': 0, 'spam': 0, 'drafts_created': 0}

    for email in emails:
        category, reason = classify_email(
            email['subject'], email['from'], email['body']
        )
        summary[category] += 1
        print(f'[{category}] {email["subject"][:60]} - {reason[:50]}')

        if category == 'action_needed':
            # Fetch full message for reply context
            full_msg = gmail_service.users().messages().get(
                userId='me', id=email['id'], format='full'
            ).execute()
            reply = draft_reply(
                email['subject'], email['from'],
                email['body'], agent_context
            )
            save_draft_reply(gmail_service, full_msg, reply)
            summary['drafts_created'] += 1

        mark_processed(gmail_service, email['id'], processed_label)

    print('\nAgent run complete:')
    for k, v in summary.items():
        print(f'  {k}: {v}')
    return summary

Error Handling and Resilience

Wrap each email processing step in try/except. A failure on one email should never stop the agent from processing the rest. Log errors with the email ID and subject so you can review them later.

from googleapiclient.errors import HttpError

def process_email_safely(gmail_service, email, agent_context, label_id):
    try:
        category, reason = classify_email(
            email['subject'], email['from'], email['body']
        )

        if category == 'action_needed':
            full_msg = gmail_service.users().messages().get(
                userId='me', id=email['id'], format='full'
            ).execute()
            reply = draft_reply(
                email['subject'], email['from'],
                email['body'], agent_context
            )
            save_draft_reply(gmail_service, full_msg, reply)

        mark_processed(gmail_service, email['id'], label_id)
        return category

    except HttpError as e:
        print(f'Gmail API error on {email["id"]}: {e.resp.status}')
        return 'error'
    except Exception as e:
        print(f'Unexpected error on {email["subject"][:50]}: {e}')
        return 'error'

Scheduling the Agent to Run Periodically

Run the email agent on a schedule using a simple loop with a time.sleep() between runs. For production, use a task scheduler like cron, APScheduler, or a cloud function. Always add a check to avoid running during off-hours to respect recipients' time zones.

import time
import datetime

def is_business_hours():
    now = datetime.datetime.now()
    # Mon-Fri, 9am-6pm local time
    return (now.weekday() < 5 and 9 <= now.hour < 18)

def run_scheduler(gmail_service, agent_context,
                  interval_minutes=30, max_emails=20):
    print(f'Email agent started. Checking every {interval_minutes} min.')

    while True:
        if is_business_hours():
            print(f'\n[{datetime.datetime.now().strftime("%H:%M")}] Running agent...')
            try:
                run_email_agent(gmail_service, agent_context, max_emails)
            except Exception as e:
                print(f'Agent run failed: {e}')
        else:
            print('Outside business hours, skipping run')

        time.sleep(interval_minutes * 60)

Logging and Audit Trail

Maintain a JSON log of every email the agent processes. This creates an audit trail you can review to verify the agent's behavior, diagnose misclassifications, and improve the prompts over time.

import json
import datetime
from pathlib import Path

LOG_FILE = Path('agent_audit.jsonl')

def log_action(email_id, subject, sender, category, action, draft_id=None):
    entry = {
        'timestamp': datetime.datetime.now().isoformat(),
        'email_id': email_id,
        'subject': subject[:100],
        'from': sender,
        'category': category,
        'action': action,
        'draft_id': draft_id
    }
    with open(LOG_FILE, 'a', encoding='utf-8') as f:
        f.write(json.dumps(entry) + '\n')

# Usage in the main loop:
log_action(
    email_id='18abc123',
    subject='Invoice #1234',
    sender='billing@vendor.com',
    category='action_needed',
    action='draft_created',
    draft_id='r9000abc'
)

# --- demo ---
print('Audit log contents:')
print(LOG_FILE.read_text(encoding='utf-8'))

Quick Check: Human-in-the-Loop

Test your understanding of the email assistant design.

Email Assistant Agent Recap

You've built a complete email assistant agent pipeline:

Fetch: pull unread emails with Gmail API search queries
Classify: LLM categorizes each email as action_needed/fyi/spam with JSON output
Draft: LLM generates reply with professional tone and [FILL IN] placeholders
Save draft: store in Gmail for human review — never auto-send
Mark processed: add custom label + remove UNREAD to prevent re-processing
Error isolation: wrap each email in try/except to prevent one failure from stopping the batch
Audit log: JSONL file recording every action for review and improvement

Frequently asked questions

Is the “Building a Simple Email Assistant Agent” lesson free?

Yes — the full text of “Building a Simple Email Assistant Agent” is free to read here on the web, and the AI Agents course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Agents course, upgrade to CoddyKit PRO.

What will I learn in “Building a Simple Email Assistant Agent”?

End-to-end: read inbox → summarize → draft reply → await approval. You practise AI Agents with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start AI Agents?

No prior experience is required. AI Agents on CoddyKit is structured for beginners through advanced learners; this is — lesson 4 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Building a Simple Email Assistant Agent” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this AI Agents lesson?

Yes. Every AI Agents lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

Connecting to Gmail via the API
Reading and Sending Emails Programmatically
Calendar Event Creation and Querying
Building a Simple Email Assistant Agent

← Back to AI Agents