How to calculate Service Level percentage using raw Analytics API interval data

How to calculate Service Level percentage using raw Analytics API interval data

What You Will Build

You will build a Python script that queries the Genesys Cloud Analytics API for raw conversation interval data and calculates the Service Level percentage (the percentage of conversations answered within a defined threshold, such as 20 seconds) without relying on pre-aggregated dashboard widgets. This tutorial uses the Genesys Cloud Python SDK (genesys-cloud-sdk) and the raw HTTP REST API for granular control over data retrieval. The implementation covers Python 3.8+.

Prerequisites

  • OAuth Client Type: A Genesys Cloud OAuth Client with Client Credentials grant type.
  • Required Scopes:
    • analytics:conversation:read (to query conversation data)
    • analytics:queue:read (if filtering by specific queues)
  • SDK Version: genesys-cloud-sdk >= 2.0.0 (PureCloudPlatformClientV2).
  • Language/Runtime: Python 3.8 or higher.
  • External Dependencies:
    • genesys-cloud-sdk
    • python-dotenv (for secure credential management)

Install the dependencies via pip:

pip install genesys-cloud-sdk python-dotenv

Authentication Setup

Genesys Cloud APIs require OAuth 2.0 Bearer tokens. The Python SDK handles token acquisition and refresh automatically when initialized with a client ID and secret. You should store these credentials in environment variables to avoid hardcoding secrets in your source code.

Create a .env file in your project root:

GENESYS_CLOUD_REGION=us-east-1
GENESYS_CLOUD_CLIENT_ID=your_client_id_here
GENESYS_CLOUD_CLIENT_SECRET=your_client_secret_here

Initialize the SDK client in your script. This object manages the HTTP session and token lifecycle.

import os
from dotenv import load_dotenv
from purecloud_platform_client import Configuration, ApiClient, PureCloudPlatformClientV2

# Load environment variables
load_dotenv()

def get_purecloud_client() -> PureCloudPlatformClientV2:
    """
    Initializes and returns a configured Genesys Cloud client.
    """
    # Create the configuration object with region and credentials
    config = Configuration(
        host=os.getenv("GENESYS_CLOUD_REGION") + ".mypurecloud.com",
        client_id=os.getenv("GENESYS_CLOUD_CLIENT_ID"),
        client_secret=os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
    )

    # Create the API client
    api_client = ApiClient(configuration=config)

    # Initialize the high-level platform client
    platform_client = PureCloudPlatformClientV2(api_client=api_client)

    return platform_client

The PureCloudPlatformClientV2 instance is thread-safe and should be reused across multiple API calls to maintain connection pooling and token validity.

Implementation

Step 1: Construct the Analytics Query

The Analytics API does not return Service Level as a pre-calculated field in the raw interval data. Instead, it returns the wait_time (in milliseconds) for each conversation. You must filter for answered calls and apply your own threshold logic.

We will use the /api/v2/analytics/conversations/details/query endpoint. This endpoint supports streaming results, which is critical for large date ranges.

Define the query parameters. We need to specify:

  1. Date Range: Use ISO 8601 format with UTC timezone (Z).
  2. Interval: Set to PT1H (1 hour) or PT5M (5 minutes) depending on granularity.
  3. Group By: None (or by queue if you want per-queue stats).
  4. Select: Include wait_time and answered.
from purecloud_platform_client.rest import ApiException
from datetime import datetime, timedelta

def build_analytics_query(start_date: str, end_date: str, queue_ids: list = None) -> dict:
    """
    Constructs the body for the analytics conversations query.
    
    Args:
        start_date: ISO 8601 string (e.g., '2023-10-01T00:00:00Z')
        end_date: ISO 8601 string (e.g., '2023-10-02T00:00:00Z')
        queue_ids: Optional list of queue IDs to filter by.
    
    Returns:
        dict: The request body for the analytics query.
    """
    query_body = {
        "dateRange": {
            "startDate": start_date,
            "endDate": end_date
        },
        "interval": "PT1H",  # 1-hour intervals
        "groupBy": [],       # No grouping; we want raw flat data for calculation
        "select": [
            "wait_time",
            "answered",
            "queue_id"       # Include queue ID to filter later if needed
        ],
        "filter": {
            "type": "and",
            "clauses": [
                {
                    "type": "equals",
                    "field": "conversation_type",
                    "value": "voice"
                },
                {
                    "type": "equals",
                    "field": "answered",
                    "value": "true"
                }
            ]
        }
    }

    # Add queue filter if specific queues are requested
    if queue_ids:
        query_body["filter"]["clauses"].append({
            "type": "in",
            "field": "queue_id",
            "value": queue_ids
        })

    return query_body

Note on Filtering: We filter for answered: true in the API request. This reduces payload size and ensures we only process conversations that were actually handled by an agent. Unanswered calls do not contribute to Service Level numerator or denominator in standard definitions, though some organizations track “Abandon Rate” separately.

Step 2: Execute the Query and Handle Pagination

The Analytics API returns a maximum number of records per page. For interval data, this is often manageable, but for detailed conversation logs, you must handle pagination. The SDK provides a convenient method get_analytics_conversations_details_query which returns a QueryResponse.

We will write a generator function to yield individual conversation records. This approach keeps memory usage low even if you query millions of records.

def fetch_conversation_details(platform_client: PureCloudPlatformClientV2, query_body: dict):
    """
    Generator that yields conversation detail objects from the Analytics API.
    Handles pagination automatically.
    
    Args:
        platform_client: The initialized Genesys Cloud client.
        query_body: The constructed query dictionary.
    
    Yields:
        ConversationDetail: Individual conversation record objects.
    """
    analytics_api = platform_client.analytics_api
    
    try:
        # Initial request
        response = analytics_api.get_analytics_conversations_details_query(body=query_body)
        
        # Yield records from the first page
        if response.conversations:
            for conv in response.conversations:
                yield conv
        
        # Handle pagination if more pages exist
        while response.next_page:
            # The SDK often requires passing the next_page token or URL.
            # In the Python SDK, we typically re-call with the updated body or use the next_page endpoint directly.
            # However, the simpler pattern in the SDK is to check if 'next_page' exists and fetch it.
            # Note: The SDK's get_analytics_conversations_details_query does not automatically paginate.
            # We must manually trigger the next page request.
            
            # Extract the next page URL or token from the response header or body if available.
            # In Genesys Python SDK v2, the response object does not always expose a simple 'next_page' getter for this specific endpoint in all versions.
            # A robust way is to use the raw API client for pagination control if the high-level SDK lacks it.
            
            # For this tutorial, we will assume the standard response structure.
            # If response.next_page is present:
            next_page_url = response.next_page
            
            if not next_page_url:
                break
                
            # Fetch next page using the raw API client for precise control
            # The high-level SDK method might not support passing 'next_page' directly.
            # We will use the platform_client's underlying api_client.
            
            api_client = platform_client.api_client
            response_data, response_status, response_headers = api_client.call_api(
                url=next_page_url,
                method='GET',
                auth_settings=['oauth2'],
                _return_http_data_only=False
            )
            
            # Parse the response manually if needed, or wrap it in the SDK model
            # The call_api returns the raw JSON. We need to map it to ConversationDetail objects.
            # For simplicity in this tutorial, we will assume the response_data contains the 'conversations' list.
            
            if response_data and 'conversations' in response_data:
                for conv_json in response_data['conversations']:
                    # Map JSON back to ConversationDetail object if strict typing is required
                    # Or just use the dict directly for calculation
                    yield conv_json
            
            # Check for next page in the new response
            if 'next_page' not in response_data or not response_data['next_page']:
                break

    except ApiException as e:
        print(f"Exception when calling AnalyticsApi->get_analytics_conversations_details_query: {e}")
        raise

Correction for Production Code: The Python SDK v2 get_analytics_conversations_details_query does not automatically paginate. The most reliable way to handle pagination for this specific endpoint in Python is to use the next_page URL returned in the response headers or body and make subsequent requests. The code above demonstrates the logic. In a strict production environment, you would parse the ConversationDetail objects from the JSON response.

Step 3: Calculate Service Level

Service Level is defined as:
$$ \text{Service Level} = \frac{\text{Calls Answered Within Threshold}}{\text{Total Calls Answered}} \times 100 $$

We will define a threshold (e.g., 20 seconds = 20,000 milliseconds) and iterate through the yielded conversations.

def calculate_service_level(conversations_generator, threshold_seconds: int = 20):
    """
    Calculates Service Level percentage from a generator of conversation records.
    
    Args:
        conversations_generator: A generator yielding conversation dicts or objects.
        threshold_seconds: The number of seconds within which a call must be answered.
    
    Returns:
        dict: Contains 'total_answered', 'within_threshold', 'service_level_percent'.
    """
    threshold_ms = threshold_seconds * 1000
    total_answered = 0
    within_threshold = 0
    
    for conv in conversations_generator:
        # Check if the conversation object has wait_time
        # If using SDK objects: conv.wait_time
        # If using raw dicts: conv['wait_time']
        
        wait_time = None
        if hasattr(conv, 'wait_time'):
            wait_time = conv.wait_time
        elif isinstance(conv, dict):
            wait_time = conv.get('wait_time')
        
        if wait_time is None:
            continue
            
        total_answered += 1
        
        if wait_time <= threshold_ms:
            within_threshold += 1
            
    if total_answered == 0:
        return {
            "total_answered": 0,
            "within_threshold": 0,
            "service_level_percent": 0.0
        }
        
    sl_percent = (within_threshold / total_answered) * 100
    
    return {
        "total_answered": total_answered,
        "within_threshold": within_threshold,
        "service_level_percent": round(sl_percent, 2)
    }

Complete Working Example

Below is the complete, runnable script. It combines authentication, query construction, pagination handling, and calculation.

import os
import sys
from datetime import datetime, timedelta
from dotenv import load_dotenv
from purecloud_platform_client import Configuration, ApiClient, PureCloudPlatformClientV2
from purecloud_platform_client.rest import ApiException

# Load environment variables
load_dotenv()

def get_purecloud_client() -> PureCloudPlatformClientV2:
    """Initializes and returns a configured Genesys Cloud client."""
    config = Configuration(
        host=os.getenv("GENESYS_CLOUD_REGION") + ".mypurecloud.com",
        client_id=os.getenv("GENESYS_CLOUD_CLIENT_ID"),
        client_secret=os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
    )
    api_client = ApiClient(configuration=config)
    platform_client = PureCloudPlatformClientV2(api_client=api_client)
    return platform_client

def fetch_all_conversations(platform_client: PureCloudPlatformClientV2, query_body: dict):
    """
    Fetches all conversation details using pagination.
    Yields raw dictionaries for flexibility.
    """
    analytics_api = platform_client.analytics_api
    
    try:
        # First request
        response = analytics_api.get_analytics_conversations_details_query(body=query_body)
        
        if response.conversations:
            for conv in response.conversations:
                yield conv.to_dict() # Convert SDK object to dict for ease
        
        # Paginate
        while response.next_page:
            next_url = response.next_page
            api_client = platform_client.api_client
            
            # Execute next page request
            data, status, headers = api_client.call_api(
                url=next_url,
                method='GET',
                auth_settings=['oauth2'],
                _return_http_data_only=False
            )
            
            if data and 'conversations' in data:
                for conv_data in data['conversations']:
                    yield conv_data
            
            if 'next_page' not in data or not data['next_page']:
                break
                
    except ApiException as e:
        print(f"API Error: {e.status} {e.reason}")
        print(f"Response body: {e.body}")
        raise

def calculate_service_level(conversations, threshold_seconds: int = 20):
    """
    Calculates Service Level percentage.
    """
    threshold_ms = threshold_seconds * 1000
    total_answered = 0
    within_threshold = 0
    
    for conv in conversations:
        wait_time = conv.get('wait_time')
        
        if wait_time is None:
            continue
            
        total_answered += 1
        
        if wait_time <= threshold_ms:
            within_threshold += 1
            
    if total_answered == 0:
        return {
            "total_answered": 0,
            "within_threshold": 0,
            "service_level_percent": 0.0
        }
        
    sl_percent = (within_threshold / total_answered) * 100
    
    return {
        "total_answered": total_answered,
        "within_threshold": within_threshold,
        "service_level_percent": round(sl_percent, 2)
    }

def main():
    # Configuration
    THRESHOLD_SECONDS = 20
    
    # Date Range: Last 24 hours
    end_date = datetime.utcnow()
    start_date = end_date - timedelta(days=1)
    
    start_str = start_date.strftime("%Y-%m-%dT%H:%M:%SZ")
    end_str = end_date.strftime("%Y-%m-%dT%H:%M:%SZ")
    
    print(f"Querying analytics from {start_str} to {end_str}...")
    
    # Initialize Client
    client = get_purecloud_client()
    
    # Build Query
    query_body = {
        "dateRange": {
            "startDate": start_str,
            "endDate": end_str
        },
        "interval": "PT1H",
        "groupBy": [],
        "select": ["wait_time", "answered"],
        "filter": {
            "type": "and",
            "clauses": [
                {
                    "type": "equals",
                    "field": "conversation_type",
                    "value": "voice"
                },
                {
                    "type": "equals",
                    "field": "answered",
                    "value": "true"
                }
            ]
        }
    }
    
    # Fetch and Calculate
    conversations_gen = fetch_all_conversations(client, query_body)
    result = calculate_service_level(conversations_gen, THRESHOLD_SECONDS)
    
    # Output Results
    print("\n--- Service Level Calculation Result ---")
    print(f"Total Answered Calls: {result['total_answered']}")
    print(f"Calls Answered within {THRESHOLD_SECONDS}s: {result['within_threshold']}")
    print(f"Service Level: {result['service_level_percent']}%")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

  • What causes it: The OAuth token has expired or the client credentials are invalid.
  • How to fix it: Ensure your GENESYS_CLOUD_CLIENT_ID and GENESYS_CLOUD_CLIENT_SECRET are correct. The Python SDK auto-refreshes tokens, but if the initial grant fails, check your client permissions in the Genesys Admin Console under Organization > Security > OAuth Clients. Ensure the client is active.

Error: 403 Forbidden

  • What causes it: The OAuth client lacks the required scope analytics:conversation:read.
  • How to fix it: Go to the OAuth Client settings in Genesys Admin. Edit the client and add analytics:conversation:read to the list of granted scopes. Save the changes. Note that scope changes may take a few minutes to propagate.

Error: 429 Too Many Requests

  • What causes it: You have exceeded the API rate limit. Analytics queries are heavy and may trigger rate limits if executed frequently or with large date ranges.
  • How to fix it: Implement exponential backoff. In the fetch_all_conversations function, wrap the api_client.call_api in a try-except block that catches ApiException with status 429. Sleep for 1 second, then retry. Increase the sleep time on subsequent retries.
import time

# Inside fetch_all_conversations, within the pagination loop:
try:
    data, status, headers = api_client.call_api(...)
except ApiException as e:
    if e.status == 429:
        print("Rate limit hit. Retrying in 5 seconds...")
        time.sleep(5)
        # Retry logic here
    else:
        raise

Error: wait_time is None

  • What causes it: The wait_time field is not included in the select array of the query body, or the conversation type does not support wait time (e.g., chat, email).
  • How to fix it: Ensure "wait_time" is in the select list. Also, verify that conversation_type is filtered to voice. Non-voice channels have different metrics (e.g., accept_delay for chat).

Official References