AI Agents · Lektion

Einen einfachen E-Mail-Assistenten als Agent erstellen

Ende zu Ende: Posteingang lesen → zusammenfassen → Antwortentwurf erstellen → Genehmigung abwarten.

Einen einfachen E-Mail-Assistenten als Agent erstellen ist eine kostenlose AI Agents-Lektion auf CoddyKit. Dies ist Lektion 4 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Agents-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.

Architektur eines E-Mail-Assistenten-Agenten

Ein E-Mail-Assistenten-Agent folgt einer standardisierten Pipeline: Abrufen → Klassifizieren → Entscheiden → Entwurf erstellen → Genehmigen → Senden. Der Agent liest ungelesene E-Mails, verwendet ein LLM zum Klassifizieren und Erstellen von Antwortentwürfen und wartet dann vor dem Senden auf die Genehmigung durch eine Person. Dieses Human-in-the-Loop-Design verhindert kostspielige Fehler durch vollständig autonomes Versenden von E-Mails.

# 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')

Agenten-Tools definieren

Strukturieren Sie den Agenten mit klar definierten Tools. Jedes Tool ist eine Python-Funktion mit einer bestimmten Aufgabe. Diese Modularität macht den Agenten testbar, leichter zu debuggen und einfach um neue Funktionen erweiterbar.

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')

Schritt 1: E-Mails abrufen und parsen

Das erste Tool ruft ungelesene E-Mails ab und extrahiert die wichtigsten Felder, die das LLM benötigt: Betreff, Absender und Textkörper im Nur-Text-Format. Halten Sie den Textkörper kurz – kürzen Sie ihn auf die ersten 2000 Zeichen, um die Kontextgrenzen des LLM einzuhalten und API-Kosten zu reduzieren.

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

Schritt 2: Klassifizierung durch das LLM

Senden Sie jede E-Mail zur Klassifizierung an ein LLM. Verwenden Sie einen strukturierten Prompt, der das Modell auffordert, ein JSON-Objekt mit Kategorie und Begründung zurückzugeben. Fordern Sie die JSON-Ausgabe ausdrücklich an – sie ist zuverlässiger als das Parsen von Freitext.

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'

Schritt 3: Antwortentwürfe erstellen

Bitten Sie das LLM für E-Mails, die als action_needed klassifiziert wurden, einen Antwortentwurf zu erstellen. Geben Sie den Aufgabenbereich und den gewünschten Ton des Agenten als Kontext an. Fordern Sie eine professionelle, aber prägnante Antwort an und fügen Sie für alle Angaben, die eine Person ergänzen muss, einen Platzhalter ein.

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()

Schritt 4: Entwürfe zur Prüfung speichern

Senden Sie niemals von einer KI erstellte E-Mails ohne Prüfung durch eine Person. Speichern Sie sie als Gmail-Entwürfe, damit die Person Gmail öffnen, sie prüfen, bei Bedarf bearbeiten und anschließend senden kann. Die API zum Erstellen von Entwürfen ist dieselbe wie die Sende-API, verwendet jedoch 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.')

Schritt 5: Verarbeitete E-Mails markieren

Markieren Sie eine E-Mail nach der Verarbeitung (klassifiziert und Entwurf erstellt oder archiviert), damit sie nicht erneut verarbeitet wird. Fügen Sie ein benutzerdefiniertes Label wie AgentProcessed hinzu und entfernen Sie das Label UNREAD. Erstellen Sie das Label einmalig, falls es noch nicht vorhanden ist.

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)

Vollständige Agenten-Schleife

Verbinden Sie alle Schritte in einer einzigen run()-Methode. Der Agent durchläuft die E-Mails, klassifiziert jede einzelne, erstellt Entwürfe für Vorgänge mit Handlungsbedarf, archiviert Spam und markiert alle E-Mails als verarbeitet. Protokollieren Sie am Ende eine Zusammenfassung.

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

Fehlerbehandlung und Robustheit

Umschließen Sie jeden Verarbeitungsschritt für E-Mails mit try/except. Ein Fehler bei einer E-Mail darf den Agenten niemals daran hindern, die übrigen E-Mails zu verarbeiten. Protokollieren Sie Fehler zusammen mit der E-Mail-ID und dem Betreff, damit Sie sie später prüfen können.

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'

Agenten regelmäßig ausführen

Führen Sie den E-Mail-Agenten nach einem Zeitplan aus, indem Sie zwischen den Durchläufen in einer einfachen Schleife time.sleep() verwenden. Verwenden Sie in der Produktion einen Aufgabenplaner wie cron, APScheduler oder eine Cloud-Funktion. Fügen Sie immer eine Prüfung hinzu, die eine Ausführung außerhalb der Arbeitszeiten verhindert und so die Zeitzonen der Empfänger berücksichtigt.

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)

Protokollierung und Prüfspur

Führen Sie ein JSON-Protokoll über jede E-Mail, die der Agent verarbeitet. Dadurch entsteht eine Prüfspur, anhand derer Sie das Verhalten des Agenten überprüfen, Fehlklassifizierungen diagnostizieren und die Prompts im Laufe der Zeit verbessern können.

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'))

Kurzer Test: Human-in-the-Loop

Testen Sie Ihr Verständnis des Designs für den E-Mail-Assistenten.

Zusammenfassung des E-Mail-Assistenten-Agenten

Sie haben eine vollständige Pipeline für einen E-Mail-Assistenten-Agenten erstellt:

Abrufen: ungelesene E-Mails mit Gmail-API-Suchanfragen abrufen
Klassifizieren: Das LLM kategorisiert jede E-Mail mit JSON-Ausgabe als action_needed/fyi/spam
Entwurf erstellen: Das LLM erstellt eine Antwort in professionellem Ton mit Platzhaltern wie [FILL IN]
Entwurf speichern: zur Prüfung durch eine Person in Gmail speichern – niemals automatisch senden
Als verarbeitet markieren: ein benutzerdefiniertes Label hinzufügen und UNREAD entfernen, um eine erneute Verarbeitung zu verhindern
Fehler isolieren: jede E-Mail mit try/except umschließen, damit ein Fehler den Stapel nicht anhält
Prüfprotokoll: JSONL-Datei, die jede Aktion zur Prüfung und Verbesserung aufzeichnet

Häufig gestellte Fragen

Ist die Lektion „Einen einfachen E-Mail-Assistenten als Agent erstellen“ kostenlos?

Ja — der vollständige Text von „Einen einfachen E-Mail-Assistenten als Agent erstellen“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Agents-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Einen einfachen E-Mail-Assistenten als Agent erstellen“?

Ende zu Ende: Posteingang lesen → zusammenfassen → Antwortentwurf erstellen → Genehmigung abwarten. Du übst AI Agents mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um AI Agents zu starten?

Keine Vorkenntnisse erforderlich. AI Agents auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 4 von 4.

Wie lange dauert die Lektion „Einen einfachen E-Mail-Assistenten als Agent erstellen“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser AI Agents-Lektion Code schreiben und ausführen?

Ja. Jede AI Agents-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

Über die API mit Gmail verbinden
E-Mails programmgesteuert lesen und senden
Kalenderereignisse erstellen und abfragen
Einen einfachen E-Mail-Assistenten als Agent erstellen

← Zurück zu AI Agents