Building a custom interval report using the Analytics Conversations Aggregates query

What You Will Build

• This tutorial builds a script that retrieves aggregated conversation metrics (handle time, wait time, resolution) for a specific queue over a 24-hour period, segmented by 15-minute intervals.
• This uses the Genesys Cloud CX Analytics Conversations Aggregates API (POST /api/v2/analytics/conversations/aggregates/query).
• The implementation is provided in Python using the official genesys-cloud-purecloud-platform-client SDK.

Prerequisites

• OAuth Client Type: A Genesys Cloud API Client with Client Credentials flow enabled.
• Required Scopes:
  • analytics:conversation:view (Required to read conversation analytics data).
  • queue:view (Optional, if you need to programmatically fetch queue IDs, though hardcoding the ID is sufficient for this tutorial).
• SDK Version: genesys-cloud-purecloud-platform-client version 140.0.0 or later.
• Language/Runtime: Python 3.8+.
• External Dependencies:
  • genesys-cloud-purecloud-platform-client
  • pandas (for optional data manipulation, though not strictly required for the API call).

pip install genesys-cloud-purecloud-platform-client pandas

Authentication Setup

Genesys Cloud uses OAuth 2.0 for authentication. For server-to-server integrations like reporting scripts, the Client Credentials grant type is the standard approach. This flow exchanges a client ID and secret for an access token without user interaction.

The SDK handles token caching and automatic refresh if the token is still valid within the cache window. You must configure the Configuration object with your client ID and secret before initializing the API client.

from platformclientv2 import Configuration, AnalyticsApi
from platformclientv2.rest import ApiException
import os

def get_analytics_api_client():
    """
    Initializes and returns an authenticated Analytics API client.
    """
    # Load credentials from environment variables for security
    client_id = os.getenv('GENESYS_CLIENT_ID')
    client_secret = os.getenv('GENESYS_CLIENT_SECRET')

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

    # Initialize the configuration with OAuth credentials
    configuration = Configuration()
    configuration.client_id = client_id
    configuration.client_secret = client_secret

    # The SDK automatically handles token acquisition and caching.
    # It will fetch a new token if the current one is expired or missing.
    analytics_api = AnalyticsApi(configuration)
    
    return analytics_api

Note on Token Lifecycle: The access token typically lasts for 1 hour. The SDK maintains an internal cache. If you make multiple API calls within that hour, the SDK reuses the token. If the token expires, the SDK automatically triggers a refresh in the background before the next request fails. You do not need to implement manual refresh logic.

Implementation

Step 1: Constructing the Aggregate Query Request Body

The core of this tutorial is the POST /api/v2/analytics/conversations/aggregates/query endpoint. Unlike simple GET requests, this endpoint accepts a complex JSON body that defines what data you want, where it comes from, and how it should be grouped.

The request body consists of three main sections:

1. dateRange: Defines the start and end time for the report.
2. filterBy: Defines the scope of conversations (e.g., specific queues, users, or media types).
3. groupBy and metrics: Defines the segmentation (intervals, users, queues) and the specific KPIs to calculate.

To build a “custom interval report,” we must specify groupBy as interval and define the intervalSize (e.g., 15m for 15 minutes).

from platformclientv2 import ConversationAggregatesQuery, ConversationAggregatesFilterBy, ConversationAggregatesGroupBy, ConversationAggregatesMetric

def build_aggregate_query(queue_id: str, start_time: str, end_time: str, interval_size: str = "15m"):
    """
    Constructs the ConversationAggregatesQuery object for the API call.
    
    Args:
        queue_id (str): The ID of the queue to report on.
        start_time (str): ISO 8601 start time (e.g., '2023-10-25T00:00:00Z').
        end_time (str): ISO 8601 end time (e.g., '2023-10-25T23:59:59Z').
        interval_size (str): Interval granularity (e.g., '15m', '1h', '1d').
        
    Returns:
        ConversationAggregatesQuery: The configured query object.
    """
    
    # 1. Define the Date Range
    # The API expects ISO 8601 format. Ensure you use UTC times.
    date_range = {
        "startDate": start_time,
        "endDate": end_time
    }

    # 2. Define the Filter
    # We filter by a specific queue. You can also filter by 'user', 'group', or 'mediaType'.
    # 'mediaType' is often set to ['voice', 'chat'] to exclude SMS or webchat if desired.
    filter_by = ConversationAggregatesFilterBy()
    filter_by.queues = [queue_id]
    filter_by.media_types = ['voice'] # Restrict to voice conversations for this example

    # 3. Define the Grouping
    # To get interval reports, we must group by 'interval'.
    # We can also group by 'user' or 'queue' simultaneously for multi-dimensional reports.
    group_by = ConversationAggregatesGroupBy()
    group_by.interval_size = interval_size
    group_by.interval_type = "calendar" # 'calendar' aligns with clock times, 'rolling' aligns from start time
    # Optional: group_by.users = True # Uncomment to break down intervals by individual agent

    # 4. Define the Metrics
    # These are the actual KPIs you want to retrieve.
    # Common metrics: 'handleTime', 'waitTime', 'resolution', 'abandoned', 'offerCount'
    metrics = [
        "handleTime",
        "waitTime",
        "resolution",
        "abandoned",
        "offerCount",
        "answered",
        "missed"
    ]

    # Construct the final query object
    query = ConversationAggregatesQuery(
        date_range=date_range,
        filter_by=filter_by,
        group_by=group_by,
        metrics=metrics
    )

    return query

Why use the SDK Object over Raw JSON?
While you could send a raw JSON payload, the SDK provides type safety and documentation hints. For example, interval_type accepts calendar or rolling. If you use calendar, the intervals align with the wall clock (e.g., 10:00-10:15, 10:15-10:30). If you use rolling, the intervals start from your startDate (e.g., if start is 10:07, the first interval is 10:07-10:22). For most business reporting, calendar is preferred for consistency with daily schedules.

Step 2: Executing the Query and Handling Pagination

The Analytics Aggregates API does not return data in a simple list. It returns a structured object containing entities (the data points) and nextUri (for pagination). If your report spans a long period with a small interval (e.g., 30 days with 15-minute intervals), the result set will be large. You must implement pagination to retrieve all data.

The SDK method post_analytics_conversations_aggregates_query returns a ConversationAggregatesResponse object.

def fetch_all_aggregate_data(api_client: AnalyticsApi, query: ConversationAggregatesQuery):
    """
    Fetches all pages of aggregate data using the nextUri pattern.
    
    Args:
        api_client (AnalyticsApi): The authenticated API client.
        query (ConversationAggregatesQuery): The query object defined in Step 1.
        
    Returns:
        list: A list of all conversation aggregate entities.
    """
    all_entities = []
    next_uri = None
    
    # Initial request
    try:
        # The SDK method name corresponds to the endpoint:
        # POST /api/v2/analytics/conversations/aggregates/query
        response = api_client.post_analytics_conversations_aggregates_query(
            body=query
        )
    except ApiException as e:
        print(f"Exception when calling AnalyticsApi->post_analytics_conversations_aggregates_query: {e}\n")
        raise

    if response.entities:
        all_entities.extend(response.entities)

    # Handle Pagination
    # The response object contains a 'next_uri' attribute if more data is available.
    while response.next_uri:
        next_uri = response.next_uri
        try:
            # For subsequent requests, we pass the next_uri instead of the body
            # Note: The SDK method signature allows passing next_uri directly
            response = api_client.post_analytics_conversations_aggregates_query(
                next_uri=next_uri
            )
            
            if response.entities:
                all_entities.extend(response.entities)
        except ApiException as e:
            print(f"Pagination error: {e}\n")
            break

    return all_entities

Understanding the Response Structure:
Each entity in response.entities is a ConversationAggregateEntity object. It contains:

• from: The start timestamp of the interval.
• to: The end timestamp of the interval.
• metrics: A dictionary or object containing the calculated values for the requested metrics.

Step 3: Processing and Formatting the Results

Raw API data is often nested. To make this useful for a developer or a downstream system (like a database or Excel export), we need to flatten the data. We will iterate through the entities and extract the timestamp and metric values into a list of dictionaries.

import pandas as pd
from datetime import datetime

def format_aggregate_data(entities):
    """
    Converts the list of ConversationAggregateEntity objects into a flat list of dictionaries.
    
    Args:
        entities (list): List of ConversationAggregateEntity objects.
        
    Returns:
        list: A list of dictionaries suitable for DataFrame creation or JSON export.
    """
    formatted_data = []
    
    for entity in entities:
        # Extract the interval time range
        interval_start = entity.from_ts if hasattr(entity, 'from_ts') else entity.from_ # SDK variation check
        interval_end = entity.to_ts if hasattr(entity, 'to_ts') else entity.to_
        
        # Extract metrics
        # The metrics attribute is typically a dict or an object with key-value pairs
        metrics_dict = {}
        if entity.metrics:
            # Depending on SDK version, metrics might be a dict or an object
            if isinstance(entity.metrics, dict):
                metrics_dict = entity.metrics
            else:
                # If it is an object, iterate over its attributes
                for key, value in vars(entity.metrics).items():
                    if not key.startswith('_'): # Ignore private attributes
                        metrics_dict[key] = value
        
        # Create a flat record
        record = {
            "interval_start": interval_start,
            "interval_end": interval_end,
            "handle_time_seconds": metrics_dict.get('handleTime', 0),
            "wait_time_seconds": metrics_dict.get('waitTime', 0),
            "resolution_count": metrics_dict.get('resolution', 0),
            "abandoned_count": metrics_dict.get('abandoned', 0),
            "answered_count": metrics_dict.get('answered', 0),
            "missed_count": metrics_dict.get('missed', 0),
            "offer_count": metrics_dict.get('offerCount', 0)
        }
        
        formatted_data.append(record)
    
    return formatted_data

Handling Metric Values:
Be aware that time-based metrics (handleTime, waitTime) are returned in seconds as floating-point numbers. Count-based metrics (resolution, abandoned) are integers. If you need minutes, divide the time metrics by 60.

Complete Working Example

Below is the complete, runnable script. It combines authentication, query construction, pagination, and data formatting.

import os
import sys
import json
from datetime import datetime, timedelta, timezone

from platformclientv2 import Configuration, AnalyticsApi, ConversationAggregatesQuery, ConversationAggregatesFilterBy, ConversationAggregatesGroupBy
from platformclientv2.rest import ApiException

def get_analytics_api_client():
    """Initializes and returns an authenticated Analytics API client."""
    client_id = os.getenv('GENESYS_CLIENT_ID')
    client_secret = os.getenv('GENESYS_CLIENT_SECRET')

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

    configuration = Configuration()
    configuration.client_id = client_id
    configuration.client_secret = client_secret

    analytics_api = AnalyticsApi(configuration)
    return analytics_api

def build_aggregate_query(queue_id: str, start_time: str, end_time: str, interval_size: str = "15m"):
    """Constructs the ConversationAggregatesQuery object."""
    
    date_range = {
        "startDate": start_time,
        "endDate": end_time
    }

    filter_by = ConversationAggregatesFilterBy()
    filter_by.queues = [queue_id]
    filter_by.media_types = ['voice']

    group_by = ConversationAggregatesGroupBy()
    group_by.interval_size = interval_size
    group_by.interval_type = "calendar"

    metrics = [
        "handleTime",
        "waitTime",
        "resolution",
        "abandoned",
        "offerCount",
        "answered",
        "missed"
    ]

    query = ConversationAggregatesQuery(
        date_range=date_range,
        filter_by=filter_by,
        group_by=group_by,
        metrics=metrics
    )
    return query

def fetch_all_aggregate_data(api_client: AnalyticsApi, query: ConversationAggregatesQuery):
    """Fetches all pages of aggregate data using pagination."""
    all_entities = []
    
    try:
        response = api_client.post_analytics_conversations_aggregates_query(body=query)
    except ApiException as e:
        print(f"Initial request failed: {e}")
        raise

    if response.entities:
        all_entities.extend(response.entities)

    while response.next_uri:
        try:
            response = api_client.post_analytics_conversations_aggregates_query(next_uri=response.next_uri)
            if response.entities:
                all_entities.extend(response.entities)
        except ApiException as e:
            print(f"Pagination failed: {e}")
            break

    return all_entities

def format_aggregate_data(entities):
    """Flattens the API response into a list of dictionaries."""
    formatted_data = []
    
    for entity in entities:
        # Handle SDK attribute naming variations
        interval_start = entity.from_ts if hasattr(entity, 'from_ts') else getattr(entity, 'from_', None)
        interval_end = entity.to_ts if hasattr(entity, 'to_ts') else getattr(entity, 'to_', None)
        
        metrics_dict = {}
        if entity.metrics:
            if isinstance(entity.metrics, dict):
                metrics_dict = entity.metrics
            else:
                for key, value in vars(entity.metrics).items():
                    if not key.startswith('_'):
                        metrics_dict[key] = value
        
        record = {
            "interval_start": interval_start,
            "interval_end": interval_end,
            "handle_time_sec": metrics_dict.get('handleTime', 0),
            "wait_time_sec": metrics_dict.get('waitTime', 0),
            "resolution": metrics_dict.get('resolution', 0),
            "abandoned": metrics_dict.get('abandoned', 0),
            "answered": metrics_dict.get('answered', 0),
            "missed": metrics_dict.get('missed', 0),
            "offer_count": metrics_dict.get('offerCount', 0)
        }
        formatted_data.append(record)
    
    return formatted_data

def main():
    # Configuration
    QUEUE_ID = os.getenv('GENESYS_QUEUE_ID', 'your-queue-id-here')
    INTERVAL_SIZE = "15m"
    
    # Define Date Range (Last 24 Hours)
    end_time = datetime.now(timezone.utc)
    start_time = end_time - timedelta(hours=24)
    
    # Format as ISO 8601
    start_iso = start_time.isoformat()
    end_iso = end_time.isoformat()

    print(f"Fetching data for Queue: {QUEUE_ID}")
    print(f"Date Range: {start_iso} to {end_iso}")
    print(f"Interval: {INTERVAL_SIZE}")

    try:
        # 1. Authenticate
        api_client = get_analytics_api_client()
        
        # 2. Build Query
        query = build_aggregate_query(QUEUE_ID, start_iso, end_iso, INTERVAL_SIZE)
        
        # 3. Fetch Data
        print("Fetching data from API...")
        entities = fetch_all_aggregate_data(api_client, query)
        
        if not entities:
            print("No data found for the specified criteria.")
            return

        print(f"Retrieved {len(entities)} interval records.")
        
        # 4. Format Data
        formatted_data = format_aggregate_data(entities)
        
        # 5. Output Results
        # Print first 5 records as JSON for verification
        print("\n--- Sample Data (First 5 Records) ---")
        for record in formatted_data[:5]:
            print(json.dumps(record, indent=2))
            
        # Optional: Save to CSV using pandas
        try:
            import pandas as pd
            df = pd.DataFrame(formatted_data)
            csv_filename = f"analytics_report_{QUEUE_ID}_{start_time.strftime('%Y%m%d')}.csv"
            df.to_csv(csv_filename, index=False)
            print(f"\nFull report saved to: {csv_filename}")
        except ImportError:
            print("\nPandas not installed. Skipping CSV export.")

    except Exception as e:
        print(f"An error occurred: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The GENESYS_CLIENT_ID or GENESYS_CLIENT_SECRET is incorrect, or the API Client has been disabled in the Genesys Cloud Admin Console.
• Fix: Verify the credentials in the Admin Console under Admin > Security > API Clients. Ensure the client is Active. Check that the environment variables are loaded correctly in your script.

Error: 403 Forbidden

• Cause: The API Client lacks the required OAuth scope analytics:conversation:view.
• Fix: In the Admin Console, edit the API Client. Scroll to the Scopes section and ensure analytics:conversation:view is checked. Save the changes. The change may take a few minutes to propagate.

Error: 429 Too Many Requests

• Cause: The Analytics API has strict rate limits. Aggregates queries are resource-intensive. If you are querying multiple queues in a loop, you will hit the limit.
• Fix: Implement exponential backoff. The SDK does not automatically retry 429s for all methods. You should wrap the fetch_all_aggregate_data call in a retry loop.

import time

def fetch_with_retry(api_client, query, max_retries=3):
    for attempt in range(max_retries):
        try:
            return fetch_all_aggregate_data(api_client, query)
        except ApiException as e:
            if e.status == 429:
                wait_time = 2 ** attempt # Exponential backoff: 1s, 2s, 4s
                print(f"Rate limited. Retrying in {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise
    raise Exception("Max retries exceeded for 429 error")

Error: Empty Result Set

• Cause: The date range is in the future, or there were no conversations in the specified queue during that time.
• Fix: Verify the startDate and endDate are in the past. Ensure the queue_id is correct and active. Check if the media_types filter is too restrictive (e.g., filtering for voice when the queue only handles chat).

Error: Metrics Return Zero or Null

• Cause: The metric is not applicable to the filter. For example, handleTime will be zero if there are no answered conversations. waitTime is only relevant for answered or abandoned conversations.
• Fix: Review the metric definitions in the Genesys Cloud documentation. Ensure the filter_by criteria actually result in conversations that possess the metric you are querying.

Official References

• Genesys Cloud Analytics Conversations Aggregates API Documentation
• Python SDK Reference: AnalyticsApi
• OAuth 2.0 Client Credentials Flow