email-agent-builder

Email Agent Builder

Critères de décision — Quelle source connecter ?

Cas	Solution
Microsoft 365 / Exchange Online	Microsoft Graph API + OAuth2 (Delegated ou App-only)
Gmail / Google Workspace	Gmail API + OAuth2 (Service Account pour full-auto)
IMAP générique (hébergement, Outlook on-premise)	IMAP4 + STARTTLS, polling toutes les N secondes
Temps réel critique (SLA < 30 s)	Graph webhooks (changeNotifications) ou Gmail push (Pub/Sub)
Volume > 10 000 emails/jour	Kafka topic + consumer group pour paralléliser

---

Workflow en étapes

1. Connexion et authentification

# Microsoft Graph — App-only (sans interaction utilisateur)
from msal import ConfidentialClientApplication

app = ConfidentialClientApplication(
    client_id=CLIENT_ID,
    client_credential=CLIENT_SECRET,
    authority=f"https://login.microsoftonline.com/{TENANT_ID}"
)
token = app.acquire_token_for_client(scopes=["https://graph.microsoft.com/.default"])
# Stocker token["access_token"] dans un vault (Azure Key Vault, HashiCorp Vault)
# Ne jamais logger ce token, ne jamais le committer

# Gmail — Service Account
from google.oauth2 import service_account
from googleapiclient.discovery import build

creds = service_account.Credentials.from_service_account_file(
    "sa.json",
    scopes=["https://www.googleapis.com/auth/gmail.modify"]
).with_subject("[email protected]")
service = build("gmail", "v1", credentials=creds)

Checklist connexion :

• Refresh token stocké en vault, jamais en .env committé
• Scopes minimaux (lecture seule si l'agent ne répond pas)
• Webhook/subscription renouvelé avant expiration (Graph : 60 min max)

---

2. Parser et normaliser les emails

import email
from email import policy

def parse_raw(raw_bytes: bytes) -> dict:
    msg = email.message_from_bytes(raw_bytes, policy=policy.default)
    body_plain = ""
    body_html  = ""
    attachments = []

    for part in msg.walk():
        ct = part.get_content_type()
        if ct == "text/plain" and not body_plain:
            body_plain = part.get_content()
        elif ct == "text/html" and not body_html:
            body_html = part.get_content()
        elif part.get_filename():
            attachments.append({
                "filename": part.get_filename(),
                "content_type": ct,
                "size": len(part.get_payload(decode=True) or b""),
            })

    return {
        "message_id": msg["Message-ID"],
        "from": msg["From"],
        "to": msg.get_all("To", []),
        "subject": msg["Subject"],
        "date": msg["Date"],
        "body_plain": strip_signature(body_plain),
        "body_html": body_html,
        "attachments": attachments,
    }

def strip_signature(text: str) -> str:
    """Coupe aux marqueurs communs de signature."""
    markers = ["-- \n", "Cordialement,", "Best regards,", "Sent from my"]
    for m in markers:
        if m in text:
            text = text[:text.index(m)]
    return text.strip()