khalilbenaz/claude-skills-collection/agent-email-agent-builder

# 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("inbox@company.com")
service = build("gmail", "v1", credentials=creds)

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

import anthropic, json
SYSTEM = """Tu es un classificateur d'emails B2B. Réponds UNIQUEMENT en JSON.
Schéma : {"category": "support|commercial|rh|facturation|autre",
           "urgency": "critique|haute|normale|basse",
           "intent": "demande|reclamation|information|confirmation|spam",
           "confidence": 0.0-1.0}"""
def classify(email_data: dict) -> dict:
    client = anthropic.Anthropic()
    prompt = f"Sujet: {email_data['subject']}\n\nCorps:\n{email_data['body_plain'][:1500]}"
    resp = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=200,
        system=SYSTEM,
        messages=[{"role": "user", "content": prompt}]
    )
    return json.loads(resp.content[0].text)

EXTRACT_SYSTEM = """Extrais les entités de l'email en JSON strict.
Schéma : {"references": [], "amounts": [], "dates": [], "persons": [],
           "companies": [], "action_required": bool, "deadline": null|"ISO8601"}"""
def extract_entities(email_data: dict) -> dict:
    client = anthropic.Anthropic()
    resp = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=400,
        system=EXTRACT_SYSTEM,
        messages=[{"role": "user", "content": email_data["body_plain"][:2000]}]
    )
    return json.loads(resp.content[0].text)

RESPONSE_SYSTEM = """Tu es l'assistant email de {company}. Rédige une réponse professionnelle
en {lang} sur la base du contexte fourni. Sois concis (< 150 mots). Ne promets pas
de délais sans les avoir vérifiés. Ne divulgue pas d'informations internes."""
def draft_response(email_data: dict, classification: dict, context: str) -> str:
    client = anthropic.Anthropic()
    resp = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=500,
        system=RESPONSE_SYSTEM.format(company="Acme", lang="français"),
        messages=[{
            "role": "user",
            "content": f"Email reçu:\n{email_data['body_plain'][:1000]}\n\nContexte CRM:\n{context}"
        }]
    )
    return resp.content[0].text

def route(classification: dict, entities: dict) -> str:
    if classification["intent"] == "spam":
        return "archive"
    if classification["urgency"] == "critique":
        return "escalate_human"  # alerte Slack/PagerDuty immédiate
    if classification["category"] == "facturation" and entities.get("amounts"):
        return "queue:finance"
    if classification["confidence"] < 0.85:
        return "queue:review"
    return "queue:auto_reply"

AUTO_REPLY_HEADERS = {"X-Auto-Reply": "true", "Auto-Submitted": "auto-replied"}
def is_auto_reply(headers: dict) -> bool:
    """Détecte les emails déjà automatiques pour éviter les boucles infinies."""
    return any([
        headers.get("X-Auto-Reply"),
        headers.get("Auto-Submitted", "").startswith("auto"),
        "MAILER-DAEMON" in headers.get("From", "").upper(),
        headers.get("Precedence") in ("bulk", "list", "junk"),
    ])

key = f"autoreply:{sender_email}"
if redis.incr(key) == 1:
    redis.expire(key, 3600)
elif redis.get(key) > 1:
    raise AutoReplyThrottled(sender_email)