Monitor CSAT scores and alert Slack on detractors using an agent skill

mkdir -p .claude/skills/csat-alert/scripts

---
name: csat-alert
description: Checks recent Gorgias satisfaction survey scores and alerts Slack when a customer gives a low rating, so detractors get immediate follow-up.
disable-model-invocation: true
allowed-tools: Bash(python *)
---
 
Check for low CSAT scores and alert Slack:
 
1. Run: `python $SKILL_DIR/scripts/check_csat.py`
2. Review the output — it shows each low score and whether a Slack alert was sent
3. Low scores are tagged in Gorgias for tracking

#!/usr/bin/env python3
"""
CSAT Survey Monitor
Fetches recent satisfaction survey responses from Gorgias,
identifies low scores, and alerts Slack.
"""
import os
import json
from datetime import datetime, timedelta, timezone
from pathlib import Path
 
try:
    import requests
except ImportError:
    os.system("pip install requests -q")
    import requests
 
GORGIAS_EMAIL  = os.environ["GORGIAS_EMAIL"]
GORGIAS_KEY    = os.environ["GORGIAS_API_KEY"]
GORGIAS_DOMAIN = os.environ["GORGIAS_DOMAIN"]
SLACK_WEBHOOK  = os.environ["SLACK_WEBHOOK_URL"]
BASE_URL       = f"https://{GORGIAS_DOMAIN}.gorgias.com/api"
AUTH           = (GORGIAS_EMAIL, GORGIAS_KEY)
 
LOW_SCORE_THRESHOLD = 2  # Alert on scores 1-2 out of 5
LOOKBACK_HOURS      = 2  # Check surveys from the last N hours
 
# Track alerted surveys to avoid duplicates
SKILL_DIR    = Path(__file__).parent.parent
STATE_FILE   = SKILL_DIR / ".alerted_surveys.json"
 
 
def load_alerted_ids() -> set:
    if STATE_FILE.exists():
        data = json.loads(STATE_FILE.read_text())
        return set(data.get("ids", []))
    return set()
 
 
def save_alerted_ids(ids: set) -> None:
    STATE_FILE.write_text(json.dumps({"ids": list(ids)[-500:]}))
 
 
def get_recent_surveys() -> list:
    """Fetch satisfaction surveys from recent tickets."""
    since = datetime.now(timezone.utc) - timedelta(hours=LOOKBACK_HOURS)
 
    resp = requests.get(
        f"{BASE_URL}/satisfaction-surveys",
        auth=AUTH,
        params={
            "created_datetime__gte": since.isoformat(),
            "limit": 50,
            "order_by": "created_datetime:desc",
        },
    )
    resp.raise_for_status()
    return resp.json().get("data", [])
 
 
def get_ticket(ticket_id: int) -> dict:
    resp = requests.get(f"{BASE_URL}/tickets/{ticket_id}", auth=AUTH)
    resp.raise_for_status()
    return resp.json()
 
 
def send_slack_alert(survey: dict, ticket: dict) -> None:
    score = survey.get("score", "?")
    customer = ticket.get("requester", {})
    customer_name = customer.get("name", "Unknown")
    customer_email = customer.get("email", "")
    subject = ticket.get("subject", "No subject")
    agent = ticket.get("assignee_user", {})
    agent_name = agent.get("name", "Unassigned") if agent else "Unassigned"
    ticket_url = f"https://{GORGIAS_DOMAIN}.gorgias.com/app/ticket/{ticket['id']}"
 
    payload = {
        "blocks": [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": f"Low CSAT Score: {score}/5",
                },
            },
            {
                "type": "section",
                "fields": [
                    {"type": "mrkdwn", "text": f"*Customer:* {customer_name}"},
                    {"type": "mrkdwn", "text": f"*Email:* {customer_email}"},
                    {"type": "mrkdwn", "text": f"*Subject:* {subject}"},
                    {"type": "mrkdwn", "text": f"*Agent:* {agent_name}"},
                ],
            },
            {
                "type": "actions",
                "elements": [
                    {
                        "type": "button",
                        "text": {"type": "plain_text", "text": "Open Ticket"},
                        "url": ticket_url,
                    }
                ],
            },
        ]
    }
 
    resp = requests.post(SLACK_WEBHOOK, json=payload)
    resp.raise_for_status()
 
 
def tag_ticket(ticket_id: int, tag: str) -> None:
    resp = requests.put(
        f"{BASE_URL}/tickets/{ticket_id}",
        auth=AUTH,
        json={"tags": [{"name": tag}]},
    )
    resp.raise_for_status()
 
 
def main() -> None:
    print(f"Checking CSAT surveys from the last {LOOKBACK_HOURS} hours...")
    surveys = get_recent_surveys()
    print(f"Found {len(surveys)} recent survey response(s)\n")
 
    alerted = load_alerted_ids()
    new_alerts = 0
 
    for survey in surveys:
        survey_id = survey.get("id")
        score = survey.get("score")
        ticket_id = survey.get("ticket_id")
 
        if survey_id in alerted:
            continue
 
        if score is not None and score <= LOW_SCORE_THRESHOLD:
            ticket = get_ticket(ticket_id)
            send_slack_alert(survey, ticket)
            tag_ticket(ticket_id, "csat-low")
            alerted.add(survey_id)
            new_alerts += 1
 
            customer = ticket.get("requester", {}).get("name", "Unknown")
            print(f"  #{ticket_id}  Score: {score}/5  Customer: {customer}")
            print(f"    Subject: {ticket.get('subject', '')[:60]}")
            print(f"    -> Slack alert sent, tagged csat-low\n")
        else:
            alerted.add(survey_id)
            print(f"  #{ticket_id}  Score: {score}/5  (above threshold, skipped)")
 
    save_alerted_ids(alerted)
 
    if new_alerts == 0:
        print("\nNo new low scores found.")
    else:
        print(f"\nDone. Sent {new_alerts} Slack alert(s).")
 
 
if __name__ == "__main__":
    main()

# Via Claude Code
/csat-alert
 
# Or run directly
python .claude/skills/csat-alert/scripts/check_csat.py

Checking CSAT surveys from the last 2 hours...
Found 8 recent survey response(s)
 
  #15201  Score: 1/5  Customer: Maria S.
    Subject: Damaged item received
    -> Slack alert sent, tagged csat-low
 
  #15198  Score: 5/5  (above threshold, skipped)
  #15195  Score: 2/5  Customer: James T.
    Subject: Late delivery
    -> Slack alert sent, tagged csat-low
 
Done. Sent 2 Slack alert(s).

# crontab -e — every 30 minutes during business hours
*/30 8-18 * * 1-5 cd /path/to/project && python .claude/skills/csat-alert/scripts/check_csat.py

0 * * * * cd /path/to/project && python .claude/skills/csat-alert/scripts/check_csat.py