Customer Service Triage Agent

mkdir customer-triage && cd customer-triage
python3 -m venv venv
source venv/bin/activate
pip install anthropic google-auth google-auth-oauthlib \
    google-api-python-client gspread requests python-dotenv

ANTHROPIC_API_KEY=sk-ant-...
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/...
GOOGLE_SHEET_ID=your-triage-log-sheet-id
# Gmail credentials from credentials.json + token.json

import os
import base64
import json
import time
import requests
from datetime import datetime
from dotenv import load_dotenv
import anthropic
import gspread
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build

load_dotenv()

claude = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

# ── Routing configuration — edit to match your team ──────────────────────────

ROUTING_CONFIG = {
    "P1": {
        "team":  "Senior Support",
        "email": "senior-support@yourcompany.com",
        "slack": True,
        "sla":   "1 hour"
    },
    "P2": {
        "team":  "Technical Support",
        "email": "tech-support@yourcompany.com",
        "slack": True,
        "sla":   "4 hours"
    },
    "P3": {
        "team":  "Support Team",
        "email": "support@yourcompany.com",
        "slack": False,
        "sla":   "24 hours"
    },
    "P4": {
        "team":  "Customer Success",
        "email": "success@yourcompany.com",
        "slack": False,
        "sla":   "72 hours"
    },
    "Billing": {
        "team":  "Finance Team",
        "email": "billing@yourcompany.com",
        "slack": False,
        "sla":   "24 hours"
    }
}

# ── Gmail helpers ─────────────────────────────────────────────────────────────

def get_gmail_service():
    creds = Credentials.from_authorized_user_file('token.json')
    return build('gmail', 'v1', credentials=creds)

def get_support_emails(service, support_label="INBOX", max_results=50):
    results = service.users().messages().list(
        userId='me', q='is:unread in:inbox', maxResults=max_results
    ).execute()
    return results.get('messages', [])

def get_email_content(service, msg_id):
    msg     = service.users().messages().get(userId='me', id=msg_id, format='full').execute()
    headers = {h['name']: h['value'] for h in msg['payload']['headers']}
    body    = ''
    for part in msg['payload'].get('parts', [msg['payload']]):
        if part.get('mimeType') == 'text/plain':
            data = part['body'].get('data', '')
            body = base64.urlsafe_b64decode(data).decode('utf-8', errors='ignore')
            break
    return {
        'id':      msg_id,
        'from':    headers.get('From', ''),
        'subject': headers.get('Subject', ''),
        'body':    body[:2000]
    }

def mark_processed(service, msg_id):
    service.users().messages().modify(
        userId='me', id=msg_id, body={'removeLabelIds': ['UNREAD']}
    ).execute()

# ── Triage with Claude ────────────────────────────────────────────────────────

TRIAGE_PROMPT = """You are a customer service triage specialist. Analyse this 
support ticket and classify it.

PRIORITY LEVELS:
- P1 (Critical): System outage, data loss, security breach, account lockout
- P2 (High): Feature broken, billing error, data incorrect, API down
- P3 (Medium): How-to questions, slow performance, minor bugs, feature requests
- P4 (Low): General feedback, praise, newsletter requests, non-urgent enquiries
- Billing: Any issue specifically about invoices, payments, refunds, subscriptions

CUSTOMER SENTIMENT:
- Angry: Threatening to cancel, using caps, exclamation marks, frustrated language
- Frustrated: Repeated issue, second contact, disappointed tone
- Neutral: Matter-of-fact tone
- Positive: Satisfied customer with a question

OUTPUT FORMAT — respond only with valid JSON:
{
  "priority": "",
  "category": "",
  "sentiment": "",
  "issue_summary": "",
  "key_details": "",
  "suggested_response": "",
  "internal_notes": ""
}"""

def triage_ticket(email: dict) -> dict:
    response = claude.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=600,
        messages=[{
            "role": "user",
            "content": f"{TRIAGE_PROMPT}\n\nFROM: {email['from']}\nSUBJECT: {email['subject']}\n\nBODY:\n{email['body']}"
        }]
    )
    try:
        return json.loads(response.content[0].text)
    except json.JSONDecodeError:
        return {
            "priority": "P3",
            "category": "General Enquiry",
            "sentiment": "Neutral",
            "issue_summary": "Could not auto-classify — manual review needed",
            "suggested_response": "",
            "internal_notes": "Auto-triage failed. Review manually."
        }

# ── Slack alerting ────────────────────────────────────────────────────────────

def send_slack_alert(email: dict, triage: dict):
    priority = triage['priority']
    emoji    = {"P1": "🚨", "P2": "🔥"}.get(priority, "📋")
    color    = {"P1": "#e74c3c", "P2": "#e67e22"}.get(priority, "#3498db")

    payload = {
        "attachments": [{
            "color": color,
            "blocks": [
                {"type": "header", "text": {"type": "plain_text", "text": f"{emoji} {priority} Support Ticket — Immediate Attention"}},
                {"type": "section", "fields": [
                    {"type": "mrkdwn", "text": f"*From:*\n{email['from']}"},
                    {"type": "mrkdwn", "text": f"*Priority:*\n{priority}"},
                    {"type": "mrkdwn", "text": f"*Category:*\n{triage['category']}"},
                    {"type": "mrkdwn", "text": f"*Sentiment:*\n{triage['sentiment']}"},
                ]},
                {"type": "section", "text": {"type": "mrkdwn", "text": f"*Issue:* {triage['issue_summary']}"}},
                {"type": "section", "text": {"type": "mrkdwn", "text": f"*Subject:* {email['subject']}"}},
            ]
        }]
    }
    requests.post(os.getenv("SLACK_WEBHOOK_URL"), json=payload)

# ── Logging ───────────────────────────────────────────────────────────────────

def log_to_sheet(email: dict, triage: dict, routing: dict):
    creds  = Credentials.from_authorized_user_file('token.json')
    client = gspread.authorize(creds)
    sheet  = client.open_by_key(os.getenv("GOOGLE_SHEET_ID")).sheet1
    sheet.append_row([
        datetime.now().strftime("%Y-%m-%d %H:%M"),
        email['from'],
        email['subject'],
        triage['priority'],
        triage['category'],
        triage['sentiment'],
        triage['issue_summary'],
        routing['team'],
        triage.get('internal_notes', '')
    ])

# ── Main agent loop ───────────────────────────────────────────────────────────

def run_agent():
    print("Customer Service Triage Agent — Starting...")
    gmail = get_gmail_service()

    while True:
        emails = get_support_emails(gmail)
        print(f"\nFound {len(emails)} unread tickets")

        for msg_ref in emails:
            try:
                email   = get_email_content(gmail, msg_ref['id'])
                triage  = triage_ticket(email)
                routing = ROUTING_CONFIG.get(triage['priority'], ROUTING_CONFIG['P3'])

                print(f"  [{triage['priority']}] {email['subject'][:50]} "
                      f"→ {routing['team']} ({triage['category']})")

                # Alert Slack for P1 and P2
                if routing.get('slack'):
                    send_slack_alert(email, triage)

                # Log everything
                log_to_sheet(email, triage, routing)
                mark_processed(gmail, email['id'])

            except Exception as e:
                print(f"  Error: {e}")

        print("Sleeping 3 minutes...")
        time.sleep(180)

if __name__ == "__main__":
    run_agent()