Build an AI Escalation Manager with Zendesk APIs

1. 1
   Set Up Zendesk API Access

   First, enable API access and generate a token. In Zendesk, go to Admin Center > Apps and integrations > APIs > Zendesk API. Enable 'Token access' and create a new token. Store this token securely, as it grants access to your Zendesk data.
2. 2
   Define Escalation Triggers and Targets

   Configure the rules for your Escalation Manager Agent. Define a list of keywords that signal urgency and specify the Group ID for your Tier 2 or specialist team. You can find the Group ID in the URL when viewing a group in Zendesk.
3. 3
   Monitor the Ticket Queue

   Write a function to fetch tickets from a specific Zendesk View. This allows the agent to focus only on new, unassigned, or at-risk tickets, making the monitoring process efficient.
4. 4
   Analyze Ticket Content for Escalation Signals

   The Escalation Manager Agent's core logic lives here. For each ticket, check its subject and description against your keyword list. This simple check is highly effective for catching explicit customer frustration.
5. 5
   Route Ticket and Add Context

   Once a ticket is identified, perform the escalation. Use the API to re-assign the ticket to your specialist group and add a private note explaining why the agent escalated it. This ensures a smooth handoff with full context.

import os
import time
import requests

# --- Configuration ---
# In a real app, use a config file or environment variables
ESCALATION_CONFIG = {
    "keywords": [
        "angry", "frustrated", "unacceptable", "cancel my account",
        "refund", "legal", "complaint", "manager"
    ],
    "escalation_group_id": 360008311012,  # IMPORTANT: Replace with your Tier 2 Group ID
    "monitored_view_id": 360029809631,    # IMPORTANT: Replace with the View ID you want to monitor
    "poll_interval_seconds": 60
}

# --- Get Credentials ---
# Before running: export ZENDESK_SUBDOMAIN='your_subdomain' etc.
SUBDOMAIN = os.getenv('ZENDESK_SUBDOMAIN')
EMAIL = os.getenv('ZENDESK_EMAIL')
TOKEN = os.getenv('ZENDESK_API_TOKEN')

if not all([SUBDOMAIN, EMAIL, TOKEN]):
    raise ValueError("Zendesk credentials not found in environment variables.")

AUTH = (f'{EMAIL}/token', TOKEN)

# --- Agent Functions ---
def get_tickets_from_view(view_id):
    """Fetches a list of tickets from a specified view."""
    url = f'https://{SUBDOMAIN}.zendesk.com/api/v2/views/{view_id}/tickets.json'
    try:
        response = requests.get(url, auth=AUTH)
        response.raise_for_status()
        return response.json().get('tickets', [])
    except requests.exceptions.RequestException as e:
        print(f"Error fetching tickets: {e}")
        return []

def escalate_ticket(ticket_id, group_id):
    """Assigns a ticket to the escalation group and adds a private note."""
    url = f'https://{SUBDOMAIN}.zendesk.com/api/v2/tickets/{ticket_id}.json'
    payload = {
        'ticket': {
            'group_id': group_id,
            'comment': {
                'body': 'Auto-escalated by Escalation Manager Agent due to keyword match.',
                'public': False
            }
        }
    }
    try:
        response = requests.put(url, json=payload, auth=AUTH)
        response.raise_for_status()
        print(f"Successfully escalated ticket #{ticket_id} to group {group_id}")
    except requests.exceptions.RequestException as e:
        print(f"Error escalating ticket #{ticket_id}: {e}")

# --- Main Agent Loop ---
def run_escalation_agent():
    """The main loop for the Escalation Manager Agent."""
    print("Starting Escalation Manager Agent...")
    while True:
        print(f"Checking view {ESCALATION_CONFIG['monitored_view_id']} for tickets to analyze...")
        tickets = get_tickets_from_view(ESCALATION_CONFIG['monitored_view_id'])
        
        if not tickets:
            print("No new tickets found.")
        else:
            for ticket in tickets:
                content = (ticket.get('subject', '') + ' ' + ticket.get('description', '')).lower()
                if any(keyword in content for keyword in ESCALATION_CONFIG['keywords']):
                    print(f"Escalation signal found in ticket #{ticket['id']}.")
                    escalate_ticket(ticket['id'], ESCALATION_CONFIG['escalation_group_id'])
        
        print(f"Sleeping for {ESCALATION_CONFIG['poll_interval_seconds']} seconds...")
        time.sleep(ESCALATION_CONFIG['poll_interval_seconds'])

if __name__ == '__main__':
    # To run this script:
    # 1. pip install requests
    # 2. Set environment variables for ZENDESK_SUBDOMAIN, ZENDESK_EMAIL, ZENDESK_API_TOKEN
    # 3. Update ESCALATION_CONFIG with your specific Group and View IDs
    # 4. python your_script_name.py
    run_escalation_agent()