Calculate Service Level Percentage from Raw Analytics API Interval Data

What You Will Build

• A script that queries Genesys Cloud Analytics for raw conversation interval data and calculates the Service Level percentage for a specific queue.
• The code uses the Genesys Cloud analytics/conversations/details/query endpoint to retrieve granular metrics.
• The tutorial covers implementation in Python using the genesyscloud SDK and raw HTTP requests via requests.

Prerequisites

• OAuth Client Type: A Confidential Client (Client Credentials Grant) or Public Client (Authorization Code Grant) with appropriate permissions.
• Required OAuth Scopes:
  • analytics:conversation:view
  • analytics:report:view
• SDK Version: genesyscloud-python >= 2.0.0 or direct API access.
• Language/Runtime: Python 3.8+.
• External Dependencies:
  • pip install genesyscloud
  • pip install requests (for raw API examples)

Authentication Setup

Genesys Cloud uses OAuth 2.0. For server-side scripts, the Client Credentials Grant is the standard flow. You must store your Client ID, Client Secret, and Environment URL securely.

Python SDK Authentication

The Genesys Cloud Python SDK handles token caching and refreshing automatically when initialized correctly.

import os
from purecloudplatformclientv2 import ApiClient, Configuration, PureCloudAuthFlow

def get_purecloud_client() -> ApiClient:
    """
    Initializes the Genesys Cloud API Client using Client Credentials Flow.
    """
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    environment = os.getenv("GENESYS_ENVIRONMENT", "https://api.mypurecloud.com")

    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET environment variables are required.")

    config = Configuration(
        host=environment,
        access_token=None,
        client_id=client_id,
        client_secret=client_secret,
        auth_flow=PureCloudAuthFlow.CLIENT_CREDENTIALS
    )

    api_client = ApiClient(configuration=config)
    return api_client

Raw HTTP Authentication

If you are not using the SDK, you must manually manage the OAuth token. The response includes an expires_in field (in seconds). You must cache this token and request a new one before expiration.

import requests
import time
from typing import Optional

class GenesysAuth:
    def __init__(self, client_id: str, client_secret: str, base_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = base_url.rstrip('/')
        self.token: Optional[str] = None
        self.token_expiry: float = 0

    def get_token(self) -> str:
        """
        Retrieves an OAuth access token. Caches it until expiry.
        """
        # Check if current token is still valid (subtract 60s for safety margin)
        if self.token and time.time() < (self.token_expiry - 60):
            return self.token

        url = f"{self.base_url}/oauth/token"
        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }
        data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        response = requests.post(url, headers=headers, data=data)
        response.raise_for_status()
        
        token_data = response.json()
        self.token = token_data["access_token"]
        self.token_expiry = time.time() + token_data["expires_in"]
        
        return self.token

Implementation

Step 1: Constructing the Analytics Query

To calculate Service Level accurately from raw data, you cannot rely on pre-aggregated report metrics if you need granular control over the definition (e.g., specific skill-based routing nuances or custom wait time thresholds). The analytics/conversations/details/query endpoint returns individual conversation records.

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

We need two specific data points from the API response:

1. wrapUpCode or outcome: To filter for answered calls (exclude abandoned).
2. waitTime: To compare against your threshold (e.g., 20 seconds).

Building the Query Body

The query body uses a JSON structure that defines the date range, entity filters, and metric selectors.

def build_analytics_query(queue_id: str, start_time: str, end_time: str) -> dict:
    """
    Builds the JSON body for the analytics/conversations/details/query endpoint.
    
    Args:
        queue_id: The ID of the queue to analyze.
        start_time: ISO 8601 start time (inclusive).
        end_time: ISO 8601 end time (exclusive).
    """
    query = {
        "dateFrom": start_time,
        "dateTo": end_time,
        "groupBy": [],
        "metrics": [
            "wrapUpCode",
            "outcome",
            "waitTime",
            "queueId"
        ],
        "entity": {
            "type": "queue",
            "id": queue_id
        },
        "conversationTypes": [
            "voice"
        ],
        "filters": [
            {
                "dimension": "outcome",
                "operator": "equals",
                "value": "answered"
            }
        ]
    }
    return query

Critical Note on Metrics:

• waitTime is returned in milliseconds.
• outcome can be answered, abandoned, no-answer, etc. We filter for answered in the query to reduce payload size, but you can also fetch all and filter locally for more complex logic (e.g., counting abandoned calls for Abandon Rate).

Step 2: Executing the Query and Handling Pagination

The Analytics API returns paginated results. You must iterate through pages until nextPageToken is null.

Using the Python SDK

from purecloudplatformclientv2 import AnalyticsApi, ConversationDetailsQueryRequest
from purecloudplatformclientv2.rest import ApiException

def fetch_conversation_data(api_client: ApiClient, queue_id: str, start_time: str, end_time: str) -> list:
    """
    Fetches all conversation details for a queue within a time range.
    Handles pagination automatically.
    """
    analytics_api = AnalyticsApi(api_client)
    
    # Build the request object
    # Note: The SDK expects specific model objects. 
    # For complex queries, constructing the JSON body directly and passing it 
    # via post_analytics_conversations_details_query is often more flexible.
    
    query_body = build_analytics_query(queue_id, start_time, end_time)
    
    all_conversations = []
    page_token = None
    
    while True:
        try:
            # Post the query
            # The SDK method signature may vary slightly by version, 
            # but generally accepts the query body directly.
            response = analytics_api.post_analytics_conversations_details_query(
                body=query_body,
                page_token=page_token
            )
            
            # Append results
            if response.entities:
                all_conversations.extend(response.entities)
            
            # Check for next page
            if response.next_page_token:
                page_token = response.next_page_token
            else:
                break
                
        except ApiException as e:
            print(f"Exception when calling AnalyticsApi->post_analytics_conversations_details_query: {e}\n")
            raise

    return all_conversations

Using Raw HTTP (Requests)

def fetch_conversation_data_raw(auth: GenesysAuth, queue_id: str, start_time: str, end_time: str) -> list:
    """
    Fetches conversation data using raw HTTP requests.
    """
    base_url = auth.base_url
    url = f"{base_url}/api/v2/analytics/conversations/details/query"
    
    query_body = build_analytics_query(queue_id, start_time, end_time)
    headers = {
        "Authorization": f"Bearer {auth.get_token()}",
        "Content-Type": "application/json"
    }
    
    all_conversations = []
    page_token = None
    
    while True:
        if page_token:
            headers["pageToken"] = page_token
            
        response = requests.post(url, json=query_body, headers=headers)
        response.raise_for_status()
        
        data = response.json()
        
        if "entities" in data and data["entities"]:
            all_conversations.extend(data["entities"])
            
        next_page_token = data.get("nextPageToken")
        if next_page_token:
            page_token = next_page_token
        else:
            break
            
    return all_conversations

Step 3: Calculating Service Level Percentage

Now that you have the raw list of conversation objects, you calculate the Service Level.

Parameters:

• threshold_ms: The target wait time in milliseconds (e.g., 20,000 ms for 20 seconds).
• total_answered: Count of conversations where outcome is answered.
• answered_within_threshold: Count of conversations where waitTime <= threshold_ms.

def calculate_service_level(conversations: list, threshold_seconds: float = 20.0) -> dict:
    """
    Calculates Service Level percentage from raw conversation data.
    
    Args:
        conversations: List of conversation detail objects from the API.
        threshold_seconds: Target wait time in seconds. Default 20s.
        
    Returns:
        Dictionary containing total_answered, answered_within_threshold, and service_level_percentage.
    """
    threshold_ms = threshold_seconds * 1000
    
    total_answered = 0
    answered_within_threshold = 0
    
    for conv in conversations:
        # Ensure the conversation was actually answered
        # The API filter might have already done this, but it is safer to check locally
        outcome = conv.get("outcome")
        if outcome != "answered":
            continue
            
        total_answered += 1
        
        # Get wait time. 
        # Note: waitTime might be null for some edge cases (e.g., instant transfer), 
        # treat null as 0 wait time.
        wait_time_ms = conv.get("waitTime") or 0
        
        if wait_time_ms <= threshold_ms:
            answered_within_threshold += 1
            
    # Calculate percentage
    if total_answered == 0:
        service_level_percentage = 0.0
    else:
        service_level_percentage = (answered_within_threshold / total_answered) * 100
        
    return {
        "total_answered": total_answered,
        "answered_within_threshold": answered_within_threshold,
        "service_level_percentage": round(service_level_percentage, 2),
        "threshold_seconds": threshold_seconds
    }

Complete Working Example

This script combines authentication, data fetching, and calculation into a single runnable module.

import os
import sys
from datetime import datetime, timedelta, timezone
from purecloudplatformclientv2 import ApiClient, Configuration, PureCloudAuthFlow, AnalyticsApi
from purecloudplatformclientv2.rest import ApiException

def get_purecloud_client() -> ApiClient:
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    environment = os.getenv("GENESYS_ENVIRONMENT", "https://api.mypurecloud.com")

    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET environment variables are required.")

    config = Configuration(
        host=environment,
        access_token=None,
        client_id=client_id,
        client_secret=client_secret,
        auth_flow=PureCloudAuthFlow.CLIENT_CREDENTIALS
    )
    return ApiClient(configuration=config)

def build_analytics_query(queue_id: str, start_time: str, end_time: str) -> dict:
    return {
        "dateFrom": start_time,
        "dateTo": end_time,
        "groupBy": [],
        "metrics": [
            "wrapUpCode",
            "outcome",
            "waitTime",
            "queueId"
        ],
        "entity": {
            "type": "queue",
            "id": queue_id
        },
        "conversationTypes": [
            "voice"
        ],
        "filters": [
            {
                "dimension": "outcome",
                "operator": "equals",
                "value": "answered"
            }
        ]
    }

def fetch_conversation_data(api_client: ApiClient, queue_id: str, start_time: str, end_time: str) -> list:
    analytics_api = AnalyticsApi(api_client)
    query_body = build_analytics_query(queue_id, start_time, end_time)
    
    all_conversations = []
    page_token = None
    
    while True:
        try:
            response = analytics_api.post_analytics_conversations_details_query(
                body=query_body,
                page_token=page_token
            )
            
            if response.entities:
                all_conversations.extend(response.entities)
            
            if response.next_page_token:
                page_token = response.next_page_token
            else:
                break
                
        except ApiException as e:
            print(f"API Error: {e}")
            raise

    return all_conversations

def calculate_service_level(conversations: list, threshold_seconds: float = 20.0) -> dict:
    threshold_ms = threshold_seconds * 1000
    total_answered = 0
    answered_within_threshold = 0
    
    for conv in conversations:
        outcome = conv.get("outcome")
        if outcome != "answered":
            continue
            
        total_answered += 1
        wait_time_ms = conv.get("waitTime") or 0
        
        if wait_time_ms <= threshold_ms:
            answered_within_threshold += 1
            
    if total_answered == 0:
        service_level_percentage = 0.0
    else:
        service_level_percentage = (answered_within_threshold / total_answered) * 100
        
    return {
        "total_answered": total_answered,
        "answered_within_threshold": answered_within_threshold,
        "service_level_percentage": round(service_level_percentage, 2),
        "threshold_seconds": threshold_seconds
    }

def main():
    # Configuration
    QUEUE_ID = os.getenv("GENESYS_QUEUE_ID")
    if not QUEUE_ID:
        raise ValueError("GENESYS_QUEUE_ID environment variable is required.")
    
    # Time Range: Last 24 Hours
    end_time = datetime.now(timezone.utc)
    start_time = end_time - timedelta(days=1)
    
    start_iso = start_time.isoformat()
    end_iso = end_time.isoformat()
    
    print(f"Fetching data for Queue {QUEUE_ID} from {start_iso} to {end_iso}...")
    
    try:
        client = get_purecloud_client()
        conversations = fetch_conversation_data(client, QUEUE_ID, start_iso, end_iso)
        
        print(f"Fetched {len(conversations)} conversations.")
        
        result = calculate_service_level(conversations, threshold_seconds=20.0)
        
        print("\n--- Service Level Report ---")
        print(f"Total Answered: {result['total_answered']}")
        print(f"Answered within {result['threshold_seconds']}s: {result['answered_within_threshold']}")
        print(f"Service Level: {result['service_level_percentage']}%")
        
    except Exception as e:
        print(f"Failed: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

Cause: The OAuth token lacks the required scope analytics:conversation:view.
Fix: Regenerate your OAuth token ensuring the scope is included. If using the SDK, verify the client credentials in the Genesys Cloud Admin Console under Security > OAuth 2.0 Clients.

# Check scopes in your token response if using raw HTTP
token_data = response.json()
scopes = token_data.get("scope", "").split(" ")
if "analytics:conversation:view" not in scopes:
    raise ValueError("Token missing required scope: analytics:conversation:view")

Error: 429 Too Many Requests

Cause: You have exceeded the rate limit for the Analytics API. Analytics endpoints are heavier than standard CRM endpoints.
Fix: Implement exponential backoff. The Genesys Cloud SDK does not handle retries automatically for Analytics queries due to the large payload sizes.

import time
import random

def make_request_with_retry(url, headers, payload, max_retries=3):
    for attempt in range(max_retries):
        response = requests.post(url, json=payload, headers=headers)
        
        if response.status_code == 429:
            wait_time = (2 ** attempt) + random.uniform(0, 1)
            print(f"Rate limited. Retrying in {wait_time:.2f} seconds...")
            time.sleep(wait_time)
            continue
            
        return response
        
    raise Exception("Max retries exceeded due to 429 Too Many Requests")

Error: Empty Results (entities is empty)

Cause:

1. The date range is in the future.
2. The queue ID is invalid or does not exist.
3. No conversations occurred in that queue during the time range.
4. The conversationTypes filter excludes the data (e.g., querying for voice but the queue only has webchat).

Fix:

• Verify the dateFrom and dateTo are in ISO 8601 format with timezone offset (e.g., 2023-10-27T10:00:00Z).
• Check the queue ID in the Genesys Cloud Admin Console.
• Remove the conversationTypes filter temporarily to see if any data returns.

Error: waitTime is Null

Cause: Some conversations, such as those transferred immediately or answered via callback, may not have a recorded wait time in the queue.
Fix: The code above handles this by treating None as 0. If your business logic requires excluding these calls from the Service Level calculation, add a filter:

wait_time_ms = conv.get("waitTime")
if wait_time_ms is None:
    continue  # Skip conversations without wait time

Official References

• Analytics Conversations Details Query API Documentation
• Genesys Cloud Python SDK Reference
• OAuth 2.0 Client Credentials Flow