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

What You Will Build

• You will build a script that queries the NICE CXone Reporting API to retrieve current queue metrics, including wait times, agent availability, and call volumes.
• This implementation uses the NICE CXone v2 Reporting API (/api/v2/reporting/queue/realtime) to fetch live data without historical aggregation delays.
• The tutorial provides production-ready examples in Python and JavaScript, demonstrating authentication, parameter construction, and result parsing.

Prerequisites

• OAuth Client: An NICE CXone API Client with the reporting:queue:read scope. You must generate an OAuth 2.0 access token using the Client Credentials flow.
• API Version: NICE CXone API v2.
• Language/Runtime:
  • Python 3.8+
  • Node.js 16+
• Dependencies:
  • Python: requests, pyjwt (optional, for token introspection).
  • JavaScript: axios (via npm).

Authentication Setup

NICE CXone uses OAuth 2.0 for authentication. Before querying real-time stats, you must obtain a valid access token. The token expires after a set duration (typically 10 minutes), so your application must handle token refresh or regeneration.

Python Authentication Helper

This function retrieves a token using client credentials. Replace YOUR_CLIENT_ID, YOUR_CLIENT_SECRET, and YOUR_OAUTH_URL with your specific values. The OAuth URL is typically https://platform.nicecxone.com/oauth/token for production or the equivalent for sandbox environments.

import requests
import time

class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str, oauth_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.oauth_url = oauth_url
        self.access_token = None
        self.token_expiry = 0

    def get_token(self) -> str:
        """
        Retrieves an OAuth 2.0 access token.
        Caches the token until it expires.
        """
        # Check if token is still valid (add 30s buffer)
        if self.access_token and time.time() < (self.token_expiry - 30):
            return self.access_token

        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": "reporting:queue:read"
        }

        try:
            response = requests.post(self.oauth_url, data=payload)
            response.raise_for_status()
            data = response.json()
            
            self.access_token = data["access_token"]
            # 'expires_in' is in seconds
            self.token_expiry = time.time() + data["expires_in"]
            
            return self.access_token
        
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 401:
                raise Exception("Invalid client ID or secret.")
            elif e.response.status_code == 403:
                raise Exception("Insufficient scope. Ensure 'reporting:queue:read' is added to the client.")
            else:
                raise Exception(f"OAuth Error: {e.response.text}")
        except requests.exceptions.RequestException as e:
            raise Exception(f"Network error during authentication: {str(e)}")

JavaScript Authentication Helper

const axios = require('axios');

class CXoneAuth {
    constructor(clientId, clientSecret, oauthUrl) {
        this.clientId = clientId;
        this.clientSecret = clientSecret;
        this.oauthUrl = oauthUrl;
        this.accessToken = null;
        this.tokenExpiry = 0;
    }

    async getToken() {
        // Check if token is still valid (add 30s buffer)
        if (this.accessToken && Date.now() < (this.tokenExpiry - 30000)) {
            return this.accessToken;
        }

        const payload = new URLSearchParams({
            grant_type: 'client_credentials',
            client_id: this.clientId,
            client_secret: this.clientSecret,
            scope: 'reporting:queue:read'
        });

        try {
            const response = await axios.post(this.oauthUrl, payload, {
                headers: {
                    'Content-Type': 'application/x-www-form-urlencoded'
                }
            });

            this.accessToken = response.data.access_token;
            // 'expires_in' is in seconds, convert to milliseconds
            this.tokenExpiry = Date.now() + (response.data.expires_in * 1000);

            return this.accessToken;
        } catch (error) {
            if (error.response) {
                if (error.response.status === 401) {
                    throw new Error("Invalid client ID or secret.");
                } else if (error.response.status === 403) {
                    throw new Error("Insufficient scope. Ensure 'reporting:queue:read' is added to the client.");
                }
            }
            throw new Error(`Authentication failed: ${error.message}`);
        }
    }
}

module.exports = CXoneAuth;

Implementation

Step 1: Constructing the Real-Time Query

The NICE CXone Real-Time Reporting API endpoint for queues is POST /api/v2/reporting/queue/realtime. Unlike historical reports, this endpoint accepts a specific JSON body to define which queues and metrics you require.

The request body requires:

1. viewId: The ID of the reporting view (usually the default view unless you have custom views).
2. metrics: An array of metric IDs you wish to retrieve (e.g., queue.waitTime, queue.agentsAvailable).
3. filter: Optional, but recommended to limit results to specific queues to reduce payload size.

Python Implementation

import requests

class CXoneQueueStats:
    def __init__(self, auth: CXoneAuth, base_url: str):
        self.auth = auth
        self.base_url = base_url
        self.api_version = "v2"
        self.reporting_endpoint = f"{self.base_url}/api/{self.api_version}/reporting/queue/realtime"

    def get_realtime_queue_stats(self, queue_ids: list, view_id: str = "default") -> dict:
        """
        Fetches real-time statistics for specified queues.
        
        Args:
            queue_ids: List of queue IDs to query.
            view_id: The reporting view ID. Use 'default' for standard views.
            
        Returns:
            Parsed JSON response from the API.
        """
        token = self.auth.get_token()
        
        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

        # Define the metrics we want to extract
        # Common metrics: 
        # queue.waitTime, queue.agentsAvailable, queue.agentsBusy, 
        # queue.callsWaiting, queue.callsHandled, queue.abandonedCalls
        metrics = [
            "queue.waitTime",
            "queue.agentsAvailable",
            "queue.agentsBusy",
            "queue.callsWaiting",
            "queue.callsHandled"
        ]

        # Construct the request body
        body = {
            "viewId": view_id,
            "metrics": metrics,
            "filter": {
                "and": [
                    {
                        "field": "queue.id",
                        "op": "in",
                        "values": queue_ids
                    }
                ]
            }
        }

        try:
            response = requests.post(
                self.reporting_endpoint,
                headers=headers,
                json=body
            )
            response.raise_for_status()
            return response.json()
        
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 400:
                raise Exception(f"Bad Request. Check metric names and filter syntax. Response: {e.response.text}")
            elif e.response.status_code == 403:
                raise Exception("Forbidden. The client may not have access to the specified queues or view.")
            elif e.response.status_code == 429:
                raise Exception("Rate Limited. Implement exponential backoff.")
            else:
                raise Exception(f"API Error: {e.response.status_code} - {e.response.text}")
        except requests.exceptions.RequestException as e:
            raise Exception(f"Network error: {str(e)}")

JavaScript Implementation

const axios = require('axios');

class CXoneQueueStats {
    constructor(auth, baseUrl) {
        this.auth = auth;
        this.baseUrl = baseUrl;
        this.apiVersion = 'v2';
        this.reportingEndpoint = `${this.baseUrl}/api/${this.apiVersion}/reporting/queue/realtime`;
    }

    async getRealtimeQueueStats(queueIds, viewId = 'default') {
        const token = await this.auth.getToken();

        const headers = {
            'Authorization': `Bearer ${token}`,
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        };

        // Define the metrics we want to extract
        const metrics = [
            'queue.waitTime',
            'queue.agentsAvailable',
            'queue.agentsBusy',
            'queue.callsWaiting',
            'queue.callsHandled'
        ];

        // Construct the request body
        const body = {
            viewId: viewId,
            metrics: metrics,
            filter: {
                and: [
                    {
                        field: 'queue.id',
                        op: 'in',
                        values: queueIds
                    }
                ]
            }
        };

        try {
            const response = await axios.post(this.reportingEndpoint, body, { headers });
            return response.data;
        } catch (error) {
            if (error.response) {
                if (error.response.status === 400) {
                    throw new Error(`Bad Request: ${error.response.data.message}`);
                } else if (error.response.status === 403) {
                    throw new Error('Forbidden: Insufficient permissions for queues or view.');
                } else if (error.response.status === 429) {
                    throw new Error('Rate Limited: Too many requests.');
                }
                throw new Error(`API Error ${error.response.status}: ${error.response.data.message}`);
            }
            throw new Error(`Network error: ${error.message}`);
        }
    }
}

module.exports = CXoneQueueStats;

Step 2: Understanding the Response Structure

The response from /api/v2/reporting/queue/realtime returns a structured JSON object. It is crucial to understand the nesting to extract data correctly.

Expected Response Format

{
  "viewId": "default",
  "metrics": [
    "queue.waitTime",
    "queue.agentsAvailable",
    "queue.callsWaiting"
  ],
  "data": [
    {
      "queueId": "12345678-1234-1234-1234-123456789012",
      "queueName": "Sales Support",
      "metrics": {
        "queue.waitTime": "00:00:15",
        "queue.agentsAvailable": 5,
        "queue.callsWaiting": 2
      }
    },
    {
      "queueId": "87654321-4321-4321-4321-210987654321",
      "queueName": "Technical Support",
      "metrics": {
        "queue.waitTime": "00:02:30",
        "queue.agentsAvailable": 0,
        "queue.callsWaiting": 12
      }
    }
  ],
  "totalResults": 2
}

Key observations:

• The data array contains one object per queue.
• Metrics are nested inside the metrics object of each queue entry.
• Time-based metrics (like queue.waitTime) are returned as strings in HH:MM:SS format.
• Numeric metrics are integers.

Step 3: Processing and Formatting Results

Raw API data often requires transformation for downstream use (e.g., displaying in a dashboard or triggering an alert). This step demonstrates how to parse the response and handle edge cases, such as missing data or zero values.

Python Processing Logic

def parse_queue_stats(api_response: dict) -> list:
    """
    Parses the raw API response into a simplified list of dictionaries.
    
    Args:
        api_response: The JSON response from the CXone API.
        
    Returns:
        A list of simplified queue stats dictionaries.
    """
    if 'data' not in api_response:
        return []

    parsed_stats = []
    
    for queue_data in api_response['data']:
        queue_id = queue_data.get('queueId', 'Unknown ID')
        queue_name = queue_data.get('queueName', 'Unknown Queue')
        metrics = queue_data.get('metrics', {})
        
        # Extract specific metrics with defaults
        wait_time_str = metrics.get('queue.waitTime', '00:00:00')
        agents_available = metrics.get('queue.agentsAvailable', 0)
        calls_waiting = metrics.get('queue.callsWaiting', 0)
        
        # Convert wait time string to seconds for easier comparison
        wait_time_seconds = cxone_time_to_seconds(wait_time_str)
        
        parsed_stats.append({
            'queue_id': queue_id,
            'queue_name': queue_name,
            'wait_time_seconds': wait_time_seconds,
            'agents_available': agents_available,
            'calls_waiting': calls_waiting
        })
        
    return parsed_stats

def cxone_time_to_seconds(time_str: str) -> int:
    """
    Converts a HH:MM:SS string to total seconds.
    
    Args:
        time_str: Time string in HH:MM:SS format.
        
    Returns:
        Total seconds as an integer.
    """
    try:
        parts = time_str.split(':')
        if len(parts) != 3:
            return 0
        hours = int(parts[0])
        minutes = int(parts[1])
        seconds = int(parts[2])
        return (hours * 3600) + (minutes * 60) + seconds
    except (ValueError, IndexError):
        return 0

JavaScript Processing Logic

function parseQueueStats(apiResponse) {
    if (!apiResponse || !apiResponse.data) {
        return [];
    }

    const parsedStats = apiResponse.data.map(queueData => {
        const queueId = queueData.queueId || 'Unknown ID';
        const queueName = queueData.queueName || 'Unknown Queue';
        const metrics = queueData.metrics || {};

        // Extract specific metrics with defaults
        const waitTimeStr = metrics['queue.waitTime'] || '00:00:00';
        const agentsAvailable = metrics['queue.agentsAvailable'] || 0;
        const callsWaiting = metrics['queue.callsWaiting'] || 0;

        // Convert wait time string to seconds
        const waitTimeSeconds = cxoneTimeToSeconds(waitTimeStr);

        return {
            queueId,
            queueName,
            waitTimeSeconds,
            agentsAvailable,
            callsWaiting
        };
    });

    return parsedStats;
}

function cxoneTimeToSeconds(timeStr) {
    try {
        const parts = timeStr.split(':');
        if (parts.length !== 3) return 0;
        const hours = parseInt(parts[0], 10);
        const minutes = parseInt(parts[1], 10);
        const seconds = parseInt(parts[2], 10);
        return (hours * 3600) + (minutes * 60) + seconds;
    } catch (e) {
        return 0;
    }
}

module.exports = { parseQueueStats, cxoneTimeToSeconds };

Complete Working Example

Below is a complete Python script that ties authentication, API querying, and data processing together. This script can be run directly after installing dependencies and configuring credentials.

import time
import sys
from cxone_auth import CXoneAuth
from cxone_queue_stats import CXoneQueueStats, parse_queue_stats, cxone_time_to_seconds

# Configuration
CLIENT_ID = "your_client_id"
CLIENT_SECRET = "your_client_secret"
OAUTH_URL = "https://platform.nicecxone.com/oauth/token"
BASE_URL = "https://platform.nicecxone.com"
QUEUE_IDS = ["12345678-1234-1234-1234-123456789012"] # Replace with actual Queue IDs

def main():
    try:
        # 1. Initialize Authentication
        auth = CXoneAuth(CLIENT_ID, CLIENT_SECRET, OAUTH_URL)
        
        # 2. Initialize Queue Stats Client
        queue_client = CXoneQueueStats(auth, BASE_URL)
        
        # 3. Fetch Real-Time Stats
        print("Fetching real-time queue statistics...")
        raw_response = queue_client.get_realtime_queue_stats(QUEUE_IDS)
        
        # 4. Process Results
        stats = parse_queue_stats(raw_response)
        
        # 5. Output Results
        print("\n--- Queue Statistics ---")
        for stat in stats:
            print(f"Queue: {stat['queue_name']}")
            print(f"  Agents Available: {stat['agents_available']}")
            print(f"  Calls Waiting: {stat['calls_waiting']}")
            print(f"  Wait Time: {stat['wait_time_seconds']} seconds")
            
            # Example Alert Logic
            if stat['calls_waiting'] > 10:
                print(f"  [ALERT] High wait volume detected for {stat['queue_name']}!")
            print("-" * 20)
            
        print("Success.")
        
    except Exception as e:
        print(f"Error: {str(e)}", file=sys.stderr)
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

Cause: The OAuth client lacks the reporting:queue:read scope, or the client does not have access to the specific queues or view requested.

Fix:

1. Verify the client scope in the NICE CXone Admin Portal under Administration > Security > API Clients.
2. Ensure the scope parameter in the OAuth request includes reporting:queue:read.
3. Check if the queues exist and are visible to the client’s organizational unit.

Error: 400 Bad Request

Cause: Invalid metric names, incorrect filter syntax, or malformed JSON.

Fix:

1. Verify metric names against the official NICE CXone API documentation. Common typos include queue.waitTime vs queue.waittime.
2. Ensure the filter object follows the strict JSON structure required by the API.
3. Print the raw request body before sending to validate JSON syntax.

Error: 429 Too Many Requests

Cause: Exceeding the API rate limit.

Fix:

1. Implement exponential backoff in your retry logic.
2. Cache results where possible. Real-time stats do not change every millisecond; polling every 5-10 seconds is usually sufficient for dashboards.
3. Reduce the number of metrics requested per call.

Error: Empty Data Array

Cause: The filter criteria do not match any active queues, or the queues are not currently receiving traffic.

Fix:

1. Verify the queue_ids provided are correct and active.
2. Remove the filter to see if any data is returned for all queues.
3. Check if the queues have agents assigned and are configured to receive interactions.

Official References

• NICE CXone Reporting API Documentation
• OAuth 2.0 Client Credentials Flow
• Queue Real-Time Metrics Reference