Querying Genesys Cloud Analytics Aggregations by Queue and Media Type

What You Will Build

• You will build a script that queries the Genesys Cloud Analytics API to retrieve aggregated conversation metrics grouped by queue ID and media type.
• This tutorial uses the Genesys Cloud /api/v2/analytics/conversations/details/query endpoint via the Python SDK.
• The code is written in Python using the genesyscloud package.

Prerequisites

• OAuth Client Type: Service Account or JWT Application.
• Required Scopes: analytics:conversation:view and routing:queue:view.
• SDK Version: genesyscloud >= 160.0.0.
• Language/Runtime: Python 3.8+.
• External Dependencies: pip install genesyscloud.

Authentication Setup

Genesys Cloud API access requires a valid OAuth access token. For programmatic access, the Service Account flow is the most common pattern. The token must be refreshed automatically when it expires to prevent 401 Unauthorized errors during long-running analytics queries.

The following code demonstrates how to initialize the PureCloudPlatformClientV2 with automatic token handling.

import os
from genesyscloud.rest import Configuration
from genesyscloud.platform.client_v2 import PureCloudPlatformClientV2

def get_platform_client():
    """
    Initializes and returns a configured Genesys Cloud Platform Client.
    Uses environment variables for sensitive credentials.
    """
    # Load configuration from environment variables
    # In production, use a secrets manager instead of env vars
    config = Configuration()
    config.host = os.getenv("GENESYS_CLOUD_HOST", "https://api.mypurecloud.com")
    config.api_key['Authorization'] = os.getenv("GENESYS_CLOUD_ACCESS_TOKEN")
    
    # If you do not have a pre-generated token, use the service account flow
    # This requires setting up the client with client_id, client_secret, etc.
    # However, for this tutorial, we assume a token is available or 
    # you will use the built-in auth helper.
    
    if not config.api_key['Authorization']:
        # Fallback: Service Account Auth
        config.client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
        config.client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
        config.auth_type = "oauth2"
        config.auth_url = f"{config.host}/oauth/token"
        
    client = PureCloudPlatformClientV2(config)
    return client

# Initialize the client
platform_client = get_platform_client()

Important Note on Scopes: If your token lacks analytics:conversation:view, the API will return a 403 Forbidden response. Ensure your application in the Genesys Cloud admin console has this scope enabled.

Implementation

Step 1: Define the Analytics Query Body

The core of this tutorial is constructing the ConversationsDetailsQuery body. This object defines the time window, the metrics to aggregate, and the groupings. To group by queue and media type, you must specify the interval and the group_by array.

The interval parameter determines the granularity of the time buckets. Common values are PT1H (1 hour), PT1D (1 day), or P1W (1 week). For a comprehensive view, PT1H is often useful to see intraday patterns.

from datetime import datetime, timedelta
from genesyscloud.analytics.models import ConversationsDetailsQuery

def build_analytics_query():
    """
    Constructs the analytics query body.
    Groups data by Queue ID and Media Type.
    """
    # Define the time window: Last 7 days
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(days=7)
    
    # Format timestamps as ISO 8601 strings
    start_time_str = start_time.isoformat() + "Z"
    end_time_str = end_time.isoformat() + "Z"
    
    # Define the query object
    query_body = ConversationsDetailsQuery(
        interval="PT1H",  # Group by 1-hour intervals
        start_time=start_time_str,
        end_time=end_time_str,
        group_by=["queueId", "mediaType"],  # The core requirement
        metrics=[
            "totalHandleTime",
            "totalWaitTime",
            "conversationCount",
            "abandonedCount"
        ],
        view="agent", # 'agent', 'supervisor', or 'all'. 'all' is safest for queue-level data
        filter_by=[
            {
                "type": "metric",
                "path": "queueId",
                "op": "in",
                "values": [] # We will populate this in Step 2
            }
        ]
    )
    
    return query_body

Why view="all"? The view parameter controls which conversations are included. agent view only includes conversations handled by agents. supervisor includes supervisor-assisted conversations. all includes every conversation that touched the queue, regardless of disposition. For queue-level analytics, all provides the most complete picture.

Step 2: Resolve Queue IDs and Refine the Query

The group_by parameter expects queueId. However, you likely want to filter by specific queues or ensure you are querying valid queues. It is best practice to fetch the list of queues first. This also allows you to filter out queues with no activity or exclude specific queues (e.g., test queues).

from genesyscloud.routing.api import RoutingApi

def get_queue_ids(platform_client: PureCloudPlatformClientV2):
    """
    Fetches all queue IDs from the Routing API.
    Handles pagination automatically.
    """
    routing_api = RoutingApi(platform_client)
    queue_ids = []
    
    try:
        # get_routing_queues fetches all queues. It handles pagination internally.
        queues_response = routing_api.get_routing_queues(
            expand=["members", "skills"]
        )
        
        if queues_response.entities:
            for queue in queues_response.entities:
                # Optional: Filter out queues with empty names or specific labels
                if queue.name and "TEST" not in queue.name.upper():
                    queue_ids.append(queue.id)
                    
    except Exception as e:
        print(f"Error fetching queues: {e}")
        return []
        
    return queue_ids

def update_query_with_queues(query_body: ConversationsDetailsQuery, queue_ids: list):
    """
    Updates the filter_by section of the query to include specific queue IDs.
    """
    if not queue_ids:
        return query_body
        
    # Update the filter to include the fetched queue IDs
    for filter_obj in query_body.filter_by:
        if filter_obj.type == "metric" and filter_obj.path == "queueId":
            filter_obj.values = queue_ids
            
    return query_body

Edge Case: Empty Queue List If the get_routing_queues call returns an empty list, the analytics query will return no data. The code above handles this by returning the original query body, which will then query all queues if the filter is removed, or return empty results if the filter remains strict. For this tutorial, we assume you want data for all active queues.

Step 3: Execute the Query and Process Results

The post_analytics_conversations_details_query endpoint is synchronous but can take several seconds to return for large date ranges. You must handle the response structure, which contains entities (the data buckets) and metadata (pagination info, though for this endpoint, it is usually a single page unless specified otherwise).

from genesyscloud.analytics.api import AnalyticsApi
import json

def run_analytics_query(platform_client: PureCloudPlatformClientV2, query_body: ConversationsDetailsQuery):
    """
    Executes the analytics query and returns the result entities.
    """
    analytics_api = AnalyticsApi(platform_client)
    
    try:
        # Execute the query
        # The 'async' parameter is False by default, meaning it waits for completion
        response = analytics_api.post_analytics_conversations_details_query(
            body=query_body
        )
        
        return response
        
    except Exception as e:
        # Handle API errors
        status_code = getattr(e, 'status', None)
        if status_code == 401:
            print("Authentication error. Token may be expired.")
        elif status_code == 403:
            print("Forbidden. Check OAuth scopes: analytics:conversation:view")
        elif status_code == 429:
            print("Rate limited. Implement exponential backoff.")
        else:
            print(f"API Error: {e}")
        return None

def process_results(response):
    """
    Parses the response and prints a readable summary.
    """
    if not response or not response.entities:
        print("No data returned.")
        return
        
    print(f"Total buckets returned: {len(response.entities)}")
    print("-" * 80)
    print(f"{'Start Time':<20} {'Queue ID':<36} {'Media Type':<10} {'Conv Count':<10} {'Handle Time (s)':<15}")
    print("-" * 80)
    
    for entity in response.entities:
        # Extract group keys
        queue_id = entity.queue_id or "N/A"
        media_type = entity.media_type or "N/A"
        start_time = entity.start_time or "N/A"
        
        # Extract metrics
        # Note: Metrics are nested under the 'metrics' key in the entity
        metrics = entity.metrics
        
        conv_count = metrics.get("conversationCount", {}).get("value", 0)
        handle_time = metrics.get("totalHandleTime", {}).get("value", 0)
        
        # Format handle time to 2 decimal places
        handle_time_str = f"{handle_time:.2f}"
        
        print(f"{start_time:<20} {queue_id:<36} {media_type:<10} {conv_count:<10} {handle_time_str:<15}")
        
    print("-" * 80)

Understanding the Response Structure: Each entity in the entities list represents a unique combination of the group_by parameters (Queue ID, Media Type) and the time interval. The metrics object contains the calculated values for each metric requested.

Complete Working Example

Below is the full, copy-pasteable Python script. Save this as gen_analytics_queue_media.py.

import os
import sys
from datetime import datetime, timedelta
from genesyscloud.rest import Configuration
from genesyscloud.platform.client_v2 import PureCloudPlatformClientV2
from genesyscloud.analytics.models import ConversationsDetailsQuery
from genesyscloud.routing.api import RoutingApi
from genesyscloud.analytics.api import AnalyticsApi

def get_platform_client():
    """Initializes the Genesys Cloud Platform Client."""
    config = Configuration()
    config.host = os.getenv("GENESYS_CLOUD_HOST", "https://api.mypurecloud.com")
    
    # Prefer pre-existing token
    token = os.getenv("GENESYS_CLOUD_ACCESS_TOKEN")
    if token:
        config.api_key['Authorization'] = token
    else:
        # Fallback to Service Account
        config.client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
        config.client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
        config.auth_type = "oauth2"
        config.auth_url = f"{config.host}/oauth/token"
        
    if not config.client_id and not token:
        raise ValueError("Please set GENESYS_CLOUD_ACCESS_TOKEN or GENESYS_CLOUD_CLIENT_ID/SECRET")

    return PureCloudPlatformClientV2(config)

def get_queue_ids(platform_client: PureCloudPlatformClientV2):
    """Fetches all queue IDs."""
    routing_api = RoutingApi(platform_client)
    queue_ids = []
    try:
        queues_response = routing_api.get_routing_queues()
        if queues_response.entities:
            for queue in queues_response.entities:
                if queue.name and "TEST" not in queue.name.upper():
                    queue_ids.append(queue.id)
    except Exception as e:
        print(f"Error fetching queues: {e}")
    return queue_ids

def build_query(queue_ids: list):
    """Constructs the analytics query body."""
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(days=7)
    
    query_body = ConversationsDetailsQuery(
        interval="PT1D",  # Daily intervals for 7-day view
        start_time=start_time.isoformat() + "Z",
        end_time=end_time.isoformat() + "Z",
        group_by=["queueId", "mediaType"],
        metrics=[
            "totalHandleTime",
            "totalWaitTime",
            "conversationCount",
            "abandonedCount"
        ],
        view="all"
    )
    
    # Add filter if queue IDs are present
    if queue_ids:
        query_body.filter_by = [
            {
                "type": "metric",
                "path": "queueId",
                "op": "in",
                "values": queue_ids
            }
        ]
        
    return query_body

def main():
    print("Initializing Genesys Cloud Client...")
    platform_client = get_platform_client()
    
    print("Fetching Queue IDs...")
    queue_ids = get_queue_ids(platform_client)
    print(f"Found {len(queue_ids)} queues.")
    
    if not queue_ids:
        print("No queues found. Exiting.")
        sys.exit(1)
        
    print("Building Analytics Query...")
    query_body = build_query(queue_ids)
    
    print("Executing Query...")
    analytics_api = AnalyticsApi(platform_client)
    
    try:
        response = analytics_api.post_analytics_conversations_details_query(body=query_body)
        
        if not response.entities:
            print("No analytics data returned for the specified period.")
            return
            
        print(f"\n{'Start Time':<20} {'Queue ID':<36} {'Media Type':<10} {'Conversations':<12} {'Handle Time (s)':<15}")
        print("-" * 95)
        
        for entity in response.entities:
            queue_id = entity.queue_id or "N/A"
            media_type = entity.media_type or "N/A"
            start_time = entity.start_time.split('T')[0] if entity.start_time else "N/A" # Just date
            
            metrics = entity.metrics
            conv_count = metrics.get("conversationCount", {}).get("value", 0)
            handle_time = metrics.get("totalHandleTime", {}).get("value", 0)
            
            print(f"{start_time:<20} {queue_id:<36} {media_type:<10} {conv_count:<12} {handle_time:<15.2f}")
            
    except Exception as e:
        print(f"Error executing query: {e}")
        if hasattr(e, 'status'):
            print(f"Status Code: {e.status}")
        if hasattr(e, 'reason'):
            print(f"Reason: {e.reason}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 400 Bad Request

• Cause: The start_time or end_time format is invalid, or the interval is not supported for the date range.
• Fix: Ensure timestamps end with “Z” (UTC). For date ranges larger than 30 days, Genesys Cloud may require larger intervals (e.g., PT1D instead of PT1H).
• Code Check: Verify start_time.isoformat() + "Z" produces a valid ISO 8601 string.

Error: 403 Forbidden

• Cause: Missing OAuth scopes.
• Fix: Add analytics:conversation:view to your application’s OAuth scopes in the Genesys Cloud admin console.
• Debugging: Run a simple GET /api/v2/me to verify the token is valid, then check the scope claims in the decoded JWT.

Error: 429 Too Many Requests

• Cause: Exceeding the rate limit for the Analytics API. Analytics queries are expensive and have lower rate limits than CRUD operations.
• Fix: Implement exponential backoff. If you receive a 429, wait for the Retry-After header value or a fixed delay (e.g., 5 seconds) before retrying.
• Code Fix: Wrap the API call in a retry loop.

import time

def retryable_query(analytics_api, query_body, max_retries=3):
    for attempt in range(max_retries):
        try:
            return analytics_api.post_analytics_conversations_details_query(body=query_body)
        except Exception as e:
            if hasattr(e, 'status') and 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 e
    raise Exception("Max retries exceeded")

Error: Empty Entities List

• Cause: No conversations match the filter criteria in the specified time window.
• Fix: Verify that the queues selected have had activity in the last 7 days. Check the view parameter. If view="agent", ensure agents were logged in and handled calls.

Official References

• Genesys Cloud Analytics Conversations API Documentation
• Python SDK Reference
• OAuth Scopes Reference