Calculate Service Level Percentage Using Raw Genesys Cloud Analytics API Data

Calculate Service Level Percentage Using Raw Genesys Cloud Analytics API Data

What You Will Build

  • A Python script that queries raw interval data from the Genesys Cloud Analytics API and calculates the Service Level (SL) percentage for a specific queue.
  • The logic handles the nuance of “answered within SL” versus “total answered” using waitTime and answerTime fields from the API response.
  • The tutorial uses Python with the official genesys-cloud-purecloud-platform-client SDK and the requests library for direct API comparisons.

Prerequisites

  • OAuth Client: A Genesys Cloud OAuth Client with the scope analytics:query:read.
  • SDK Version: Genesys Cloud Python SDK v2 (latest stable release).
  • Runtime: Python 3.8+ with pip.
  • Dependencies:
    • genesys-cloud-purecloud-platform-client
    • python-dotenv (for secure credential management)
    • pandas (optional, for data manipulation, but we will use standard libraries to keep dependencies low).

Authentication Setup

Genesys Cloud uses OAuth 2.0 for authentication. For server-to-server integrations, the Client Credentials Flow is the standard approach. The SDK handles token acquisition and refresh automatically if configured correctly.

Create a .env file in your project root:

GENESYS_CLIENT_ID=your_client_id
GENESYS_CLIENT_SECRET=your_client_secret
GENESYS_ENVIRONMENT=us-east-1.api.mypurecloud.com

Initialize the SDK client in your script:

import os
from dotenv import load_dotenv
from purecloud_platform_client import PlatformApiClient, Configuration, OAuthClientCredentials

# Load environment variables
load_dotenv()

def get_platform_client():
    """
    Initializes and returns a configured PlatformApiClient.
    """
    config = Configuration(
        host=os.getenv("GENESYS_ENVIRONMENT"),
        client_id=os.getenv("GENESYS_CLIENT_ID"),
        client_secret=os.getenv("GENESYS_CLIENT_SECRET")
    )
    
    # Create the OAuth client credentials object
    oauth = OAuthClientCredentials(config)
    
    # Initialize the API client
    api_client = PlatformApiClient(config, oauth_client=oauth)
    
    # Verify connectivity by forcing a token fetch
    try:
        api_client.get_oauth_token()
        print("Authentication successful.")
    except Exception as e:
        print(f"Authentication failed: {e}")
        raise e
        
    return api_client

Implementation

Step 1: Constructing the Analytics Query

The core of this tutorial is the POST /api/v2/analytics/conversations/details/query endpoint. Unlike summary reports, this endpoint returns raw interaction data. To calculate Service Level accurately, you need the waitTime and answerTime for every conversation.

OAuth Scope Required: analytics:query:read

We must construct a request body that:

  1. Filters by Queue ID.
  2. Filters by Time Interval (e.g., last 24 hours).
  3. Specifies Time Grouping as interval to get data per hour/minute.
  4. Selects specific Metrics and Dimensions.
from purecloud_platform_client.rest import ApiException
from datetime import datetime, timedelta

def build_analytics_query(queue_id: str, start_time: datetime, end_time: datetime):
    """
    Builds the request body for the Analytics Query API.
    """
    # Calculate the interval size (e.g., 1 hour)
    # Note: For precise SL, smaller intervals are better, but 1 hour is a good start.
    interval_size = "hour" 
    
    request_body = {
        "dateFrom": start_time.isoformat() + "Z",
        "dateTo": end_time.isoformat() + "Z",
        "groupBy": ["time"],
        "interval": interval_size,
        "filter": {
            "type": "queue",
            "id": queue_id
        },
        "metrics": [
            "waitTime",
            "answerTime",
            "abandonTime",
            "handleTime"
        ],
        "dimensions": [
            "queue",
            "wrapupcode",
            "skill"
        ],
        # Crucial: We need the raw data points to calculate SL manually
        # The API returns 'details' which contain the individual conversation stats
        "timeGrouping": "interval" 
    }
    
    return request_body

Step 2: Executing the Query and Handling Pagination

The Analytics API returns paginated results. You must handle the nextPageToken to ensure you retrieve all data within the specified time window. If you miss pages, your Service Level calculation will be skewed.

def fetch_all_conversation_details(api_client, queue_id: str, start_time: datetime, end_time: datetime):
    """
    Fetches all conversation detail records for a queue within a time range.
    Handles pagination automatically.
    """
    from purecloud_platform_client import AnalyticsApi
    
    analytics_api = AnalyticsApi(api_client)
    request_body = build_analytics_query(queue_id, start_time, end_time)
    
    all_details = []
    next_page_token = None
    
    try:
        while True:
            # Execute the query
            response = analytics_api.post_analytics_conversations_details_query(
                body=request_body,
                page_token=next_page_token
            )
            
            # Append details to our list
            if response.details:
                all_details.extend(response.details)
            
            # Check for next page
            if response.next_page_token:
                next_page_token = response.next_page_token
            else:
                break
                
    except ApiException as e:
        if e.status == 429:
            print("Rate limit hit. Implement exponential backoff in production.")
        elif e.status == 400:
            print(f"Bad Request. Check queue ID and date range. Body: {e.body}")
        else:
            print(f"API Error: {e.status} - {e.reason}")
        raise e
    except Exception as e:
        print(f"Unexpected error: {e}")
        raise e
        
    return all_details

Step 3: Calculating Service Level from Raw Data

Service Level is defined as:
$$ \text{Service Level %} = \left( \frac{\text{Calls Answered Within SL Threshold}}{\text{Total Calls Answered}} \right) \times 100 $$

Critical Logic Note:
The waitTime in the API response represents the time the customer spent in queue. The answerTime represents the time the agent spent handling the call.

  • If answerTime is null or 0, the call was not answered (abandoned or missed).
  • We only count calls where answerTime is greater than 0.
  • We compare waitTime against the SL threshold (e.g., 20 seconds).
def calculate_service_level(details: list, sl_threshold_seconds: int = 20):
    """
    Calculates Service Level percentage from raw conversation details.
    
    Args:
        details: List of ConversationDetail objects from the API.
        sl_threshold_seconds: The SL threshold in seconds (e.g., 20 for 20s).
        
    Returns:
        dict: Contains 'total_answered', 'answered_within_sl', and 'service_level_percent'.
    """
    total_answered = 0
    answered_within_sl = 0
    
    for detail in details:
        # Ensure we have valid timing data
        if not detail.total_wait_time and not detail.total_answer_time:
            continue
            
        # Parse durations (API returns ISO 8601 duration format, e.g., "PT20S")
        wait_seconds = parse_duration(detail.total_wait_time)
        answer_seconds = parse_duration(detail.total_answer_time)
        
        # Only count calls that were actually answered
        if answer_seconds > 0:
            total_answered += 1
            
            # Check if wait time was within the SL threshold
            if wait_seconds <= sl_threshold_seconds:
                answered_within_sl += 1
                
    if total_answered == 0:
        return {
            "total_answered": 0,
            "answered_within_sl": 0,
            "service_level_percent": 0.0,
            "message": "No answered calls found in this interval."
        }
        
    sl_percent = (answered_within_sl / total_answered) * 100
    
    return {
        "total_answered": total_answered,
        "answered_within_sl": answered_within_sl,
        "service_level_percent": round(sl_percent, 2)
    }

def parse_duration(iso_duration: str) -> float:
    """
    Converts ISO 8601 duration string (e.g., 'PT20S', 'PT1M30S') to seconds.
    """
    if not iso_duration:
        return 0.0
        
    try:
        from datetime import timedelta
        # Simple regex extraction for standard ISO 8601 durations
        import re
        # Pattern for PT[H]H[M]M[S]S
        match = re.match(r'PT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?', iso_duration)
        if match:
            hours = int(match.group(1) or 0)
            minutes = int(match.group(2) or 0)
            seconds = int(match.group(3) or 0)
            return hours * 3600 + minutes * 60 + seconds
        return 0.0
    except Exception:
        return 0.0

Complete Working Example

This script ties everything together. It authenticates, fetches data for the last 24 hours, and prints the Service Level.

import os
import sys
from datetime import datetime, timedelta
from dotenv import load_dotenv
from purecloud_platform_client import PlatformApiClient, Configuration, OAuthClientCredentials, AnalyticsApi
from purecloud_platform_client.rest import ApiException

# --- Helper Functions from Previous Steps ---

def get_platform_client():
    load_dotenv()
    config = Configuration(
        host=os.getenv("GENESYS_ENVIRONMENT"),
        client_id=os.getenv("GENESYS_CLIENT_ID"),
        client_secret=os.getenv("GENESYS_CLIENT_SECRET")
    )
    oauth = OAuthClientCredentials(config)
    api_client = PlatformApiClient(config, oauth_client=oauth)
    try:
        api_client.get_oauth_token()
    except Exception as e:
        raise Exception(f"Auth failed: {e}")
    return api_client

def build_analytics_query(queue_id: str, start_time: datetime, end_time: datetime):
    return {
        "dateFrom": start_time.isoformat() + "Z",
        "dateTo": end_time.isoformat() + "Z",
        "groupBy": ["time"],
        "interval": "hour",
        "filter": {
            "type": "queue",
            "id": queue_id
        },
        "metrics": ["waitTime", "answerTime"],
        "dimensions": ["queue"],
        "timeGrouping": "interval"
    }

def fetch_all_conversation_details(api_client, queue_id: str, start_time: datetime, end_time: datetime):
    analytics_api = AnalyticsApi(api_client)
    request_body = build_analytics_query(queue_id, start_time, end_time)
    all_details = []
    next_page_token = None
    
    try:
        while True:
            response = analytics_api.post_analytics_conversations_details_query(
                body=request_body,
                page_token=next_page_token
            )
            if response.details:
                all_details.extend(response.details)
            
            if response.next_page_token:
                next_page_token = response.next_page_token
            else:
                break
    except ApiException as e:
        print(f"API Error: {e.status} - {e.reason}")
        raise e
        
    return all_details

def parse_duration(iso_duration: str) -> float:
    if not iso_duration:
        return 0.0
    import re
    try:
        match = re.match(r'PT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?', iso_duration)
        if match:
            hours = int(match.group(1) or 0)
            minutes = int(match.group(2) or 0)
            seconds = int(match.group(3) or 0)
            return hours * 3600 + minutes * 60 + seconds
        return 0.0
    except Exception:
        return 0.0

def calculate_service_level(details: list, sl_threshold_seconds: int = 20):
    total_answered = 0
    answered_within_sl = 0
    
    for detail in details:
        if not detail.total_wait_time and not detail.total_answer_time:
            continue
            
        wait_seconds = parse_duration(detail.total_wait_time)
        answer_seconds = parse_duration(detail.total_answer_time)
        
        if answer_seconds > 0:
            total_answered += 1
            if wait_seconds <= sl_threshold_seconds:
                answered_within_sl += 1
                
    if total_answered == 0:
        return {"total_answered": 0, "answered_within_sl": 0, "service_level_percent": 0.0}
        
    sl_percent = (answered_within_sl / total_answered) * 100
    return {
        "total_answered": total_answered,
        "answered_within_sl": answered_within_sl,
        "service_level_percent": round(sl_percent, 2)
    }

# --- Main Execution ---

if __name__ == "__main__":
    # Configuration
    QUEUE_ID = "your_queue_id_here" # Replace with actual Queue ID
    SL_THRESHOLD = 20 # Seconds
    
    # Time Window: Last 24 Hours
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(hours=24)
    
    print(f"Calculating SL for Queue: {QUEUE_ID}")
    print(f"Window: {start_time.isoformat()}Z to {end_time.isoformat()}Z")
    
    try:
        # 1. Authenticate
        api_client = get_platform_client()
        
        # 2. Fetch Data
        print("Fetching conversation details...")
        details = fetch_all_conversation_details(api_client, QUEUE_ID, start_time, end_time)
        print(f"Fetched {len(details)} conversation records.")
        
        # 3. Calculate SL
        result = calculate_service_level(details, SL_THRESHOLD)
        
        # 4. Output Results
        print("\n--- Service Level Report ---")
        print(f"Total Answered: {result['total_answered']}")
        print(f"Answered within {SL_THRESHOLD}s: {result['answered_within_sl']}")
        print(f"Service Level: {result['service_level_percent']}%")
        
    except Exception as e:
        print(f"Execution failed: {e}")
        sys.exit(1)

Common Errors & Debugging

Error: 400 Bad Request - “Invalid filter”

  • Cause: The queue_id provided does not exist, or the user associated with the OAuth client does not have permission to view analytics for that queue.
  • Fix: Verify the Queue ID in the Genesys Cloud Admin console. Ensure the OAuth client has the analytics:query:read scope and is assigned to a user with “View Analytics” permissions.

Error: 429 Too Many Requests

  • Cause: The Analytics API has strict rate limits. Fetching large volumes of raw data can trigger this quickly.
  • Fix: Implement exponential backoff in the fetch_all_conversation_details loop.
    import time
    
    # Inside the except block for 429
    if e.status == 429:
        wait_time = 2 ** retry_count
        print(f"Rate limited. Waiting {wait_time} seconds...")
        time.sleep(wait_time)
        retry_count += 1
    

Error: Service Level is 0% despite high call volume

  • Cause: The answerTime is null for all records, or the waitTime parsing is failing.
  • Fix:
    1. Check if the queue actually has answered calls in the selected time window.
    2. Print the raw total_wait_time and total_answer_time values for the first 5 records to verify the ISO 8601 format.
    3. Ensure you are not filtering out answered calls accidentally. In the calculate_service_level function, verify answer_seconds > 0.

Error: “Page token expired”

  • Cause: The Analytics API page tokens are short-lived. If your script takes too long between requests (e.g., due to slow network or processing), the token expires.
  • Fix: Process data in smaller batches or reduce the time window of the query. If processing large datasets, consider using the Analytics Export API (/api/v2/analytics/conversations/details/export) instead, which handles large data extraction more efficiently.

Official References