Extract Real-Time Queue Statistics from NICE CXone v2 Reporting API

What You Will Build

• You will build a script that queries the NICE CXone v2 Reporting API to retrieve real-time queue performance metrics.
• The code uses the niceincontact Python SDK to handle authentication and API requests.
• The tutorial covers Python 3.9+ and demonstrates handling of OAuth2 client credentials and complex query parameters.

Prerequisites

• OAuth Client Type: You need a CXone Integration (API Key) configured with the Reporting scope. Specifically, ensure the client has the reporting:read scope enabled in the CXone Admin UI under Integrations.
• SDK Version: niceincontact Python SDK version 10.0.0 or later.
• Language/Runtime: Python 3.9 or higher.
• External Dependencies:
  • niceincontact: The official NICE CXone Python SDK.
  • requests: Included in the SDK dependencies, but useful for debugging raw HTTP calls.

Authentication Setup

NICE CXone uses OAuth 2.0 Client Credentials flow for server-to-server communication. The SDK handles the token exchange, caching, and refresh automatically, provided you configure the ClientCredentials object correctly.

You must store your credentials securely. For this tutorial, we assume environment variables are set:

• CXONE_CLIENT_ID
• CXONE_CLIENT_SECRET
• CXONE_BASE_URL (e.g., https://api-us-02.niceincontact.com)

Step 1: Initialize the API Client

The first step is to instantiate the ApiClient and configure it with your OAuth credentials. The SDK uses the ClientCredentials class to manage the token lifecycle.

import os
import sys
from niceincontact import ApiClient, Configuration, ClientCredentials
from niceincontact.apis import ReportingApi
from niceincontact.exceptions import ApiException

def get_reporting_api_client() -> ReportingApi:
    """
    Initializes and returns a configured ReportingApi client.
    Handles OAuth2 token acquisition automatically.
    """
    # Load credentials from environment variables
    client_id = os.getenv("CXONE_CLIENT_ID")
    client_secret = os.getenv("CXONE_CLIENT_SECRET")
    base_url = os.getenv("CXONE_BASE_URL", "https://api-us-02.niceincontact.com")

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

    # Create the OAuth configuration
    # The SDK will automatically fetch and refresh the token
    oauth_config = Configuration(
        host=base_url,
        access_token=None, # Will be populated by the SDK
        client_id=client_id,
        client_secret=client_secret,
        grant_type="client_credentials",
        scopes=["reporting:read"] # Required scope for Reporting API
    )

    # Create the API Client
    api_client = ApiClient(oauth_config)

    # Create the Reporting API instance
    reporting_api = ReportingApi(api_client)

    return reporting_api

Key Details:

• The scopes parameter is critical. If you omit reporting:read, the token exchange will succeed, but subsequent API calls will return 403 Forbidden.
• The Configuration object handles the initial POST to /oauth/token. It caches the token and refreshes it before expiration.

Implementation

Step 2: Construct the Query Parameters

The v2 Reporting API is query-based. To get real-time queue stats, you do not pull a static snapshot. Instead, you submit a query that defines the metrics, dimensions, and time window.

For “real-time” data, you must use the Realtime query type. The API endpoint is POST /api/v2/reporting/queries.

You need to define:

1. Metrics: What numbers do you want? (e.g., Queue:WaitTime, Queue:AbandonRate, Queue:AvailableAgents).
2. Dimensions: How do you want to group the data? (e.g., Queue).
3. Filter: Which queues do you want? (e.g., Queue.ID equals a specific list).

Building the Query Object

The SDK provides model classes for these objects. You must construct a RealtimeQuery object.

from niceincontact.models import RealtimeQuery, Metric, Dimension, Filter, FilterItem

def build_realtime_queue_query(queue_ids: list[str]) -> RealtimeQuery:
    """
    Constructs a RealtimeQuery object for queue statistics.
    
    Args:
        queue_ids: A list of string IDs for the queues to monitor.
        
    Returns:
        A configured RealtimeQuery object.
    """
    
    # 1. Define Metrics
    # Metric names are case-sensitive and must match the API documentation exactly
    metrics = [
        Metric(name="Queue:WaitTime", alias="avg_wait_time", unit="seconds"),
        Metric(name="Queue:AbandonRate", alias="abandon_rate", unit="percent"),
        Metric(name="Queue:AvailableAgents", alias="available_agents", unit="count"),
        Metric(name="Queue:CallsInQueue", alias="calls_in_queue", unit="count")
    ]

    # 2. Define Dimensions
    # Grouping by Queue allows us to see stats for each queue individually
    dimensions = [
        Dimension(name="Queue", alias="queue_info")
    ]

    # 3. Define Filters
    # We filter by Queue ID to limit the scope of the report
    # Note: The v2 API often requires specific filter structures for dimensions
    filter_items = [
        FilterItem(
            dimension="Queue",
            operator="IN",
            values=queue_ids
        )
    ]
    
    query_filter = Filter(items=filter_items)

    # 4. Assemble the Query
    # The 'id' field is optional but recommended for tracking query execution
    query_id = f"realtime_queue_stats_{len(queue_ids)}"
    
    realtime_query = RealtimeQuery(
        id=query_id,
        metrics=metrics,
        dimensions=dimensions,
        filter=query_filter
    )

    return realtime_query

Critical Parameter Explanation:

• Metric Names: Queue:WaitTime returns the average wait time in seconds. Queue:AbandonRate returns the percentage of calls abandoned. These are standard CXone metric keys.
• Alias: You can assign an alias to metrics for easier parsing in the response, but the SDK will also return the original name.
• Filter Operator: Using IN allows you to query multiple queues in a single request. If you use EQ, you can only query one queue.

Step 3: Submit the Query and Handle Pagination

The Reporting API is asynchronous for large datasets, but for small real-time queries, it often returns results immediately. However, the pattern is always: Submit Query → Get Query ID → Get Results.

For real-time queries, the SDK provides a convenient method post_reporting_query_realtime which wraps the submit and fetch logic.

def fetch_realtime_stats(reporting_api: ReportingApi, queue_ids: list[str]) -> dict:
    """
    Submits the realtime query and retrieves the results.
    
    Args:
        reporting_api: The configured ReportingApi client.
        queue_ids: List of queue IDs to query.
        
    Returns:
        A dictionary containing the query results.
    """
    query = build_realtime_queue_query(queue_ids)

    try:
        # The SDK method post_reporting_query_realtime submits the query
        # and waits for the results.
        # It returns a RealtimeQueryResponse object.
        response = reporting_api.post_reporting_query_realtime(query)
        
        return response
        
    except ApiException as e:
        print(f"Exception when calling ReportingApi->post_reporting_query_realtime: {e}\n")
        if e.status == 429:
            print("Rate limit exceeded. Implement exponential backoff.")
        elif e.status == 403:
            print("Forbidden. Check if your client has 'reporting:read' scope.")
        raise

Step 4: Parse and Process Results

The response object contains a results list. Each item in the list corresponds to a dimension combination (in this case, one item per queue).

import json
from niceincontact.models import RealtimeQueryResponse

def process_results(response: RealtimeQueryResponse) -> list[dict]:
    """
    Parses the RealtimeQueryResponse into a clean list of dictionaries.
    
    Args:
        response: The response object from the API call.
        
    Returns:
        A list of dictionaries, one per queue, with parsed metric values.
    """
    if not response or not response.results:
        print("No results returned from the query.")
        return []

    processed_data = []

    for result in response.results:
        # Each result has a 'dimensions' dict and a 'metrics' dict
        queue_info = result.dimensions.get("Queue", {})
        queue_id = queue_info.get("id")
        queue_name = queue_info.get("name")

        metrics = result.metrics
        
        # Extract specific metrics
        # The metric value is usually in the 'value' field
        avg_wait = metrics.get("avg_wait_time", {}).get("value")
        abandon_rate = metrics.get("abandon_rate", {}).get("value")
        available_agents = metrics.get("available_agents", {}).get("value")
        calls_in_queue = metrics.get("calls_in_queue", {}).get("value")

        queue_stat = {
            "queue_id": queue_id,
            "queue_name": queue_name,
            "avg_wait_time_sec": avg_wait,
            "abandon_rate_pct": abandon_rate,
            "available_agents": available_agents,
            "calls_in_queue": calls_in_queue
        }
        
        processed_data.append(queue_stat)

    return processed_data

Complete Working Example

This script combines all steps into a single executable module. Save this as cxone_queue_stats.py.

"""
NICE CXone Real-Time Queue Statistics Extractor
Author: Senior Developer Advocate
Description: Demonstrates how to use the CXone v2 Reporting API to fetch real-time queue metrics.
"""

import os
import sys
import json
from typing import List, Dict, Optional

from niceincontact import ApiClient, Configuration, ClientCredentials
from niceincontact.apis import ReportingApi
from niceincontact.models import RealtimeQuery, Metric, Dimension, Filter, FilterItem, RealtimeQueryResponse
from niceincontact.exceptions import ApiException


def setup_api_client() -> ReportingApi:
    """Initializes the CXone Reporting API client with OAuth2 credentials."""
    client_id = os.getenv("CXONE_CLIENT_ID")
    client_secret = os.getenv("CXONE_CLIENT_SECRET")
    base_url = os.getenv("CXONE_BASE_URL", "https://api-us-02.niceincontact.com")

    if not client_id or not client_secret:
        raise EnvironmentError("Missing required environment variables: CXONE_CLIENT_ID, CXONE_CLIENT_SECRET")

    config = Configuration(
        host=base_url,
        client_id=client_id,
        client_secret=client_secret,
        grant_type="client_credentials",
        scopes=["reporting:read"]
    )

    api_client = ApiClient(config)
    return ReportingApi(api_client)


def create_realtime_query(queue_ids: List[str]) -> RealtimeQuery:
    """
    Constructs a RealtimeQuery object targeting specific queue metrics.
    """
    metrics = [
        Metric(name="Queue:WaitTime", alias="wait_time"),
        Metric(name="Queue:AbandonRate", alias="abandon_rate"),
        Metric(name="Queue:AvailableAgents", alias="avail_agents"),
        Metric(name="Queue:CallsInQueue", alias="in_queue")
    ]

    dimensions = [
        Dimension(name="Queue", alias="queue_dim")
    ]

    filters = Filter(items=[
        FilterItem(dimension="Queue", operator="IN", values=queue_ids)
    ])

    return RealtimeQuery(
        id="rt_queue_stats_v1",
        metrics=metrics,
        dimensions=dimensions,
        filter=filters
    )


def fetch_and_parse_stats(reporting_api: ReportingApi, queue_ids: List[str]) -> List[Dict]:
    """
    Executes the query and parses the response.
    """
    query = create_realtime_query(queue_ids)

    try:
        response: RealtimeQueryResponse = reporting_api.post_reporting_query_realtime(query)
    except ApiException as e:
        print(f"API Error: {e.status} - {e.reason}")
        if e.body:
            print(f"Response Body: {e.body}")
        raise

    if not response or not response.results:
        print("Query executed successfully but returned no results.")
        return []

    output = []
    for result in response.results:
        dim = result.dimensions.get("queue_dim", {})
        met = result.metrics

        record = {
            "id": dim.get("id"),
            "name": dim.get("name"),
            "metrics": {
                "wait_time_sec": met.get("wait_time", {}).get("value"),
                "abandon_rate": met.get("abandon_rate", {}).get("value"),
                "avail_agents": met.get("avail_agents", {}).get("value"),
                "in_queue": met.get("in_queue", {}).get("value")
            }
        }
        output.append(record)

    return output


def main():
    """
    Main entry point.
    """
    # Example Queue IDs - Replace with actual IDs from your CXone instance
    # You can find Queue IDs by navigating to Admin -> Routing -> Queues
    # or by querying GET /api/v2/routing/queues
    example_queue_ids = os.getenv("CXONE_QUEUE_IDS", "12345678-1234-1234-1234-123456789abc").split(",")
    
    print(f"Fetching real-time stats for {len(example_queue_ids)} queue(s)...")

    try:
        api = setup_api_client()
        stats = fetch_and_parse_stats(api, example_queue_ids)

        if stats:
            print(json.dumps(stats, indent=2))
        else:
            print("No data found.")

    except Exception as e:
        print(f"Fatal Error: {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

Cause: The OAuth token was issued, but it lacks the necessary scope.
Fix: Ensure your Integration (API Key) in CXone Admin UI has the reporting:read scope checked. If you recently added the scope, you must rotate the API key or wait for the token to expire and refresh.
Code Check: Verify scopes=["reporting:read"] in the Configuration object.

Error: 400 Bad Request - Invalid Metric Name

Cause: The metric name string does not match the CXone schema exactly.
Fix: Check the CXone API Documentation for the exact metric key. Common mistakes include case sensitivity (Queue:waittime vs Queue:WaitTime) or typos.
Debugging: Print the raw request body before sending. The SDK logs can help if you enable debug mode.

Error: Empty Results

Cause: The filter is too restrictive, or the queues specified do not exist or have no activity.
Fix:

1. Verify the queue_ids are valid. You can validate them by calling GET /api/v2/routing/queues/{id}.
2. Ensure the queues are active and have agents assigned or calls flowing.
3. Check if the FilterItem dimension matches the Dimension name exactly.

Error: 429 Too Many Requests

Cause: You are exceeding the rate limit for the Reporting API.
Fix: Implement exponential backoff. The CXone API returns Retry-After headers.
Code Implementation:

import time

def safe_fetch(api, query, max_retries=3):
    for attempt in range(max_retries):
        try:
            return api.post_reporting_query_realtime(query)
        except ApiException as e:
            if e.status == 429:
                wait_time = 2 ** attempt
                print(f"Rate limited. Waiting {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise
    raise Exception("Max retries exceeded for 429 error.")

Official References

• NICE CXone v2 Reporting API Documentation
• niceincontact Python SDK on PyPI
• OAuth2 Client Credentials Flow
• Reporting Metrics Reference