Query Agent Utilization Metrics by 30-Minute Intervals in Genesys Cloud

What You Will Build

• This tutorial builds a Python script that retrieves agent utilization metrics (tHandle, tAcw, tHold) broken down by 30-minute intervals from Genesys Cloud.
• It uses the Genesys Cloud CX Analytics API (/api/v2/analytics/conversations/details/query) via the official Python SDK.
• The implementation is written in Python 3.9+ using the genesyscloud SDK and requests for fallback authentication if needed.

Prerequisites

• OAuth Client: A Genesys Cloud OAuth client with the following scopes:
  • analytics:report:read
  • analytics:conversation:read (required for detailed conversation data)
• SDK Version: genesyscloud Python SDK version >= 135.0.0 (supports modern async patterns and updated analytics models).
• Language/Runtime: Python 3.9 or higher.
• External Dependencies:
  • genesyscloud
  • python-dotenv (for secure credential management)
  • pandas (for optional data manipulation/display)

Install dependencies via pip:

pip install genesyscloud python-dotenv pandas

Authentication Setup

Genesys Cloud uses OAuth 2.0 for API access. The Python SDK handles token acquisition and refresh automatically when initialized with client credentials.

Create a .env file in your project root with your OAuth client details:

GENESYS_CLOUD_REGION=us-east-1
GENESYS_CLOUD_CLIENT_ID=your_client_id
GENESYS_CLOUD_CLIENT_SECRET=your_client_secret

The authentication code initializes the PlatformClientV2 which manages the OAuth token lifecycle:

import os
from dotenv import load_dotenv
from genesyscloud.platform.client_v2 import PlatformClientV2
from genesyscloud.auth.jwt_authenticator import JwtAuthenticator

load_dotenv()

def get_platform_client() -> PlatformClientV2:
    """
    Initializes and returns a Genesys Cloud PlatformClientV2 instance.
    Handles OAuth token acquisition automatically.
    """
    region = os.getenv("GENESYS_CLOUD_REGION")
    client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")

    if not all([region, client_id, client_secret]):
        raise ValueError("Missing required environment variables: GENESYS_CLOUD_REGION, GENESYS_CLOUD_CLIENT_ID, GENESYS_CLOUD_CLIENT_SECRET")

    # Construct the base URL for the specified region
    base_url = f"https://api.{region}.mygenesys.com"
    
    # Initialize the platform client
    platform_client = PlatformClientV2(base_url=base_url)
    
    # Set up JWT authentication
    # The SDK will handle token refresh automatically
    platform_client.auth.set_jwt_authenticator(
        JwtAuthenticator(
            client_id=client_id,
            client_secret=client_secret
        )
    )
    
    return platform_client

Implementation

Step 1: Configure the Analytics Query

The core of this tutorial relies on the AnalyticsApi within the SDK. To retrieve metrics broken down by 30-minute intervals, you must use the groupBy parameter in the query body. The groupBy value interval(30 minutes) tells the Genesys Cloud analytics engine to aggregate data into 30-minute buckets.

You also need to specify the metrics you want: tHandle (total handle time), tAcw (after-call work time), and tHold (hold time). These are standard conversation metrics.

from datetime import datetime, timedelta
from typing import Dict, Any

def build_analytics_query(start_time: str, end_time: str, user_ids: list[str]) -> Dict[str, Any]:
    """
    Builds the JSON payload for the analytics query.
    
    Args:
        start_time: ISO 8601 formatted start time (UTC)
        end_time: ISO 8601 formatted end time (UTC)
        user_ids: List of agent user IDs to filter by
    
    Returns:
        Dictionary representing the query body
    """
    query_body = {
        "dateFrom": start_time,
        "dateTo": end_time,
        "groupBy": ["interval(30 minutes)"],  # Critical: Defines the 30-minute buckets
        "metrics": [
            "tHandle",
            "tAcw",
            "tHold"
        ],
        "filters": {
            "types": ["voice"],  # Filter for voice conversations only
            "users": user_ids    # Filter for specific agents
        },
        "select": [
            "interval",  # The time bucket identifier
            "user.id",   # Agent ID
            "user.name"  # Agent Name
        ],
        "size": 1000  # Max records per page
    }
    
    return query_body

Important Note on groupBy:
The groupBy parameter determines how data is aggregated. Using interval(30 minutes) creates rows for each 30-minute segment within the query window. If you also include user.id in the select array but not in groupBy, the API may return nulls or aggregate across all users for that interval. To get per-agent breakdowns per interval, you typically include user in the groupBy as well, or rely on the select fields to distinguish users within the interval bucket if the API supports that granularity for the specific metric type. For this tutorial, we will group by both interval and user to ensure accurate per-agent, per-interval data.

Updated query_body for accurate per-agent breakdown:

def build_analytics_query_detailed(start_time: str, end_time: str, user_ids: list[str]) -> Dict[str, Any]:
    """
    Builds a more detailed query grouping by both interval and user.
    """
    query_body = {
        "dateFrom": start_time,
        "dateTo": end_time,
        "groupBy": [
            "interval(30 minutes)", 
            "user"  # Groups by agent ID within each interval
        ],
        "metrics": [
            "tHandle",
            "tAcw",
            "tHold"
        ],
        "filters": {
            "types": ["voice"],
            "users": user_ids
        },
        "select": [
            "interval",
            "user.id",
            "user.name"
        ],
        "size": 1000
    }
    
    return query_body

Step 2: Execute the Query and Handle Pagination

The Genesys Cloud Analytics API returns paginated results. You must handle the nextPage token to retrieve all data if it exceeds the page size. The SDK provides get_analytics_conversations_details_query which returns a response object containing the data and pagination links.

from genesyscloud.platform.client_v2.api.analytics_api import AnalyticsApi
from genesyscloud.platform.client_v2.model.conversation_details_query import ConversationDetailsQuery
from genesyscloud.platform.client_v2.model.conversation_details_response import ConversationDetailsResponse

def fetch_utilization_metrics(
    platform_client: PlatformClientV2, 
    query_body: Dict[str, Any]
) -> list[Dict[str, Any]]:
    """
    Executes the analytics query and handles pagination.
    
    Args:
        platform_client: The initialized PlatformClientV2 instance
        query_body: The query configuration dictionary
    
    Returns:
        List of dictionaries containing the metric data
    """
    analytics_api = AnalyticsApi(platform_client)
    all_results = []
    
    # Convert dict to SDK model if necessary, or use raw dict if SDK supports it
    # The SDK typically accepts dicts for complex bodies in recent versions
    # However, for strict typing, we can use the model class
    
    try:
        # Initial request
        response = analytics_api.post_analytics_conversations_details_query(body=query_body)
        
        # Check if response is valid
        if response is None:
            raise ValueError("Received None response from API")
            
        # Accumulate data
        if response.data:
            all_results.extend(response.data)
            
        # Handle pagination
        while response.next_page:
            # Extract the next page token
            next_page_token = response.next_page
            
            # Construct the query for the next page
            # Note: The SDK might handle this via a specific method or by passing the token
            # In Genesys Cloud Python SDK, pagination is often handled by re-querying with the token
            
            # Re-construct the body with the nextPage token
            # This approach depends on the specific SDK version implementation
            # A more robust way is to use the SDK's built-in pagination if available
            # For manual control:
            
            query_body["nextPage"] = next_page_token
            
            response = analytics_api.post_analytics_conversations_details_query(body=query_body)
            
            if response.data:
                all_results.extend(response.data)
                
    except Exception as e:
        print(f"Error fetching metrics: {e}")
        raise
    
    return all_results

Note on Pagination:
The nextPage token is a string returned in the response header or body. The Genesys Cloud Python SDK often simplifies this by allowing you to pass the nextPage value back into the same endpoint. Ensure your SDK version supports this pattern. If using an older SDK, you may need to use the get_analytics_conversations_details_query_by_id endpoint with the queryId returned from the initial post, but the direct nextPage approach is standard for the details/query endpoint.

Step 3: Process and Format the Results

The raw API response contains nested objects. You need to flatten this data into a usable format, such as a list of dictionaries or a Pandas DataFrame. The interval field is a string representing the start of the 30-minute bucket in ISO 8601 format.

import pandas as pd
from datetime import datetime

def process_metrics_data(raw_data: list[Dict[str, Any]]) -> pd.DataFrame:
    """
    Transforms raw API response into a structured DataFrame.
    
    Args:
        raw_data: List of metric objects from the API response
    
    Returns:
        Pandas DataFrame with flattened metric data
    """
    records = []
    
    for item in raw_data:
        # Extract interval start time
        interval_start = item.get("interval", {}).get("start", "")
        
        # Extract user details
        user_id = item.get("user", {}).get("id", "")
        user_name = item.get("user", {}).get("name", "")
        
        # Extract metrics
        # Metrics are typically in a 'metrics' object within the item
        # The structure depends on the specific API response format
        # For conversation details, metrics might be flat or nested
        
        t_handle = item.get("tHandle", 0)
        t_acw = item.get("tAcw", 0)
        t_hold = item.get("tHold", 0)
        
        # If metrics are nested under a 'metrics' key, adjust accordingly
        # Example: item['metrics']['tHandle']
        
        records.append({
            "interval_start": interval_start,
            "user_id": user_id,
            "user_name": user_name,
            "tHandle_seconds": t_handle,
            "tAcw_seconds": t_acw,
            "tHold_seconds": t_hold
        })
    
    if not records:
        return pd.DataFrame()
        
    df = pd.DataFrame(records)
    
    # Convert interval_start to datetime for better analysis
    if not df.empty:
        df["interval_start"] = pd.to_datetime(df["interval_start"], utc=True)
        
    return df

Complete Working Example

The following script combines all steps into a runnable module. It retrieves agent utilization metrics for the last 24 hours for a specified list of agents.

import os
import sys
import pandas as pd
from datetime import datetime, timedelta
from dotenv import load_dotenv
from genesyscloud.platform.client_v2 import PlatformClientV2
from genesyscloud.auth.jwt_authenticator import JwtAuthenticator
from genesyscloud.platform.client_v2.api.analytics_api import AnalyticsApi
from typing import Dict, Any, List

# Load environment variables
load_dotenv()

def get_platform_client() -> PlatformClientV2:
    region = os.getenv("GENESYS_CLOUD_REGION")
    client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")

    if not all([region, client_id, client_secret]):
        raise ValueError("Missing required environment variables.")

    base_url = f"https://api.{region}.mygenesys.com"
    platform_client = PlatformClientV2(base_url=base_url)
    platform_client.auth.set_jwt_authenticator(
        JwtAuthenticator(client_id=client_id, client_secret=client_secret)
    )
    return platform_client

def build_query(start_time: str, end_time: str, user_ids: List[str]) -> Dict[str, Any]:
    return {
        "dateFrom": start_time,
        "dateTo": end_time,
        "groupBy": ["interval(30 minutes)", "user"],
        "metrics": ["tHandle", "tAcw", "tHold"],
        "filters": {
            "types": ["voice"],
            "users": user_ids
        },
        "select": ["interval", "user.id", "user.name"],
        "size": 1000
    }

def fetch_data(platform_client: PlatformClientV2, query_body: Dict[str, Any]) -> List[Dict[str, Any]]:
    analytics_api = AnalyticsApi(platform_client)
    all_results = []
    
    try:
        response = analytics_api.post_analytics_conversations_details_query(body=query_body)
        
        if response and response.data:
            all_results.extend(response.data)
            
        while response and response.next_page:
            query_body["nextPage"] = response.next_page
            response = analytics_api.post_analytics_conversations_details_query(body=query_body)
            
            if response and response.data:
                all_results.extend(response.data)
                
    except Exception as e:
        print(f"Error: {e}")
        raise
        
    return all_results

def process_data(raw_data: List[Dict[str, Any]]) -> pd.DataFrame:
    records = []
    for item in raw_data:
        interval_start = item.get("interval", {}).get("start", "")
        user_id = item.get("user", {}).get("id", "")
        user_name = item.get("user", {}).get("name", "")
        
        # Accessing metrics directly from the item root or nested structure
        # Note: The exact structure may vary slightly based on SDK version
        # Assuming flat structure for tHandle, tAcw, tHold as per typical analytics response
        t_handle = item.get("tHandle", 0)
        t_acw = item.get("tAcw", 0)
        t_hold = item.get("tHold", 0)
        
        records.append({
            "interval_start": interval_start,
            "user_id": user_id,
            "user_name": user_name,
            "tHandle_seconds": t_handle,
            "tAcw_seconds": t_acw,
            "tHold_seconds": t_hold
        })
        
    df = pd.DataFrame(records)
    if not df.empty:
        df["interval_start"] = pd.to_datetime(df["interval_start"], utc=True)
    return df

def main():
    try:
        # 1. Initialize Client
        platform_client = get_platform_client()
        
        # 2. Define Time Range (Last 24 Hours)
        end_time = datetime.utcnow()
        start_time = end_time - timedelta(hours=24)
        
        start_str = start_time.strftime("%Y-%m-%dT%H:%M:%S.000Z")
        end_str = end_time.strftime("%Y-%m-%dT%H:%M:%S.000Z")
        
        # 3. Define Agent IDs (Replace with actual IDs)
        user_ids = ["agent_id_1", "agent_id_2"] 
        
        # 4. Build Query
        query_body = build_query(start_str, end_str, user_ids)
        
        # 5. Fetch Data
        print("Fetching metrics...")
        raw_data = fetch_data(platform_client, query_body)
        print(f"Fetched {len(raw_data)} records.")
        
        # 6. Process Data
        df = process_data(raw_data)
        
        if not df.empty:
            # Display results
            print(df.head())
            
            # Optional: Save to CSV
            df.to_csv("agent_utilization_metrics.csv", index=False)
            print("Results saved to agent_utilization_metrics.csv")
        else:
            print("No data returned.")
            
    except Exception as e:
        print(f"Fatal error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: Invalid OAuth client credentials or expired token.
• Fix: Verify GENESYS_CLOUD_CLIENT_ID and GENESYS_CLOUD_CLIENT_SECRET in your .env file. Ensure the client has the analytics:report:read scope. The SDK handles refresh, but if the initial token fails, check your credentials.

Error: 403 Forbidden

• Cause: The OAuth client lacks the required scopes.
• Fix: Ensure the client has analytics:report:read and analytics:conversation:read. Go to Genesys Cloud Admin > Platform > OAuth clients, edit your client, and add these scopes.

Error: 422 Unprocessable Entity

• Cause: Invalid query parameters, such as malformed groupBy syntax or invalid date ranges.
• Fix: Ensure groupBy uses exact syntax: interval(30 minutes). Verify that dateFrom is before dateTo. Check that user_ids are valid UUIDs.

Error: Empty Results

• Cause: No conversations match the filters within the time range.
• Fix: Verify that the specified agents had voice conversations during the query period. Check the filters object to ensure types matches the conversation type (e.g., voice, chat, email).

Error: Metric Values Are Null

• Cause: The metric is not available for the selected conversation type or interval.
• Fix: tHandle, tAcw, and tHold are primarily voice metrics. Ensure filters.types includes voice. If querying other channels, use appropriate metrics like tQueue or tWait.

Official References

• Genesys Cloud Analytics API Documentation
• Conversation Details Query Endpoint
• Genesys Cloud Python SDK Reference
• OAuth Scopes Reference