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

What You Will Build

• One sentence: The code retrieves live metrics for a specific CXone queue, including agents logged in, calls waiting, and average wait times.
• One sentence: This tutorial uses the NICE CXone v2 Reporting API endpoint /api/v2/reporting/queues/realtime.
• One sentence: The implementation is provided in Python using the requests library and in JavaScript using fetch.

Prerequisites

• OAuth Client Type: A CXone API Client with Client Credentials grant type.
• Required Scopes: reporting:queue:view is mandatory for accessing real-time queue data.
• SDK/API Version: CXone API v2 (Reporting).
• Language/Runtime: Python 3.8+ or Node.js 18+.
• External Dependencies:
  • Python: requests, python-dotenv
  • Node.js: node-fetch (if not using native fetch in Node 18+), dotenv

Authentication Setup

CXone uses OAuth 2.0 for authentication. You must obtain an access token before making any API calls. The token expires after a set period (typically one hour), so your application should cache the token or implement a refresh mechanism.

Below is the Python implementation for obtaining a token using the Client Credentials flow.

import requests
import os
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

CXONE_BASE_URL = "https://api-us-02.nice-incontact.com" # Adjust region as needed
CXONE_CLIENT_ID = os.getenv("CXONE_CLIENT_ID")
CXONE_CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET")
CXONE_TENANT_ID = os.getenv("CXONE_TENANT_ID")

def get_access_token() -> str:
    """
    Retrieves an OAuth2 access token from CXone.
    """
    token_url = f"{CXONE_BASE_URL}/oauth2/token"
    
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    
    # The body must be URL-encoded form data
    body = {
        "grant_type": "client_credentials",
        "client_id": CXONE_CLIENT_ID,
        "client_secret": CXONE_CLIENT_SECRET,
        "tenant_id": CXONE_TENANT_ID # Required for some CXone endpoints
    }

    try:
        response = requests.post(token_url, headers=headers, data=body)
        response.raise_for_status() # Raises HTTPError for bad responses (4xx or 5xx)
        
        token_data = response.json()
        return token_data["access_token"]
    
    except requests.exceptions.HTTPError as http_err:
        print(f"HTTP error occurred: {http_err}")
        print(f"Response content: {response.text}")
        raise
    except requests.exceptions.RequestException as err:
        print(f"An error occurred: {err}")
        raise

# Get the token
access_token = get_access_token()
print(f"Token acquired successfully.")

Implementation

Step 1: Construct the Real-Time Queue Request

The CXone v2 Reporting API provides real-time data via the /api/v2/reporting/queues/realtime endpoint. This endpoint accepts a POST request with a JSON body containing the queue ID and optional time filters. Unlike historical reports, real-time data does not require a timeRange object in the same way, but it does require the queueId to be specified.

You must identify the Queue ID first. If you do not have it, you can retrieve it using the /api/v2/queues endpoint, but for this tutorial, we assume you have the queueId.

Here is the JavaScript implementation using fetch. This example demonstrates how to structure the request body and handle the authorization header.

const CXONE_BASE_URL = "https://api-us-02.nice-incontact.com"; // Adjust region
const QUEUE_ID = "your-queue-id-here"; // Replace with actual Queue ID
const ACCESS_TOKEN = "your-access-token-here"; // In production, fetch this dynamically

async function getRealTimeQueueStats(queueId) {
    const endpoint = `${CXONE_BASE_URL}/api/v2/reporting/queues/realtime`;

    const requestBody = {
        "queueId": queueId,
        // Optional: You can filter by specific skill IDs if needed
        // "skillIds": ["skill-1", "skill-2"] 
    };

    const options = {
        method: 'POST',
        headers: {
            'Content-Type': 'application/json',
            'Authorization': `Bearer ${ACCESS_TOKEN}`
        },
        body: JSON.stringify(requestBody)
    };

    try {
        const response = await fetch(endpoint, options);

        if (!response.ok) {
            const errorText = await response.text();
            throw new Error(`HTTP Error: ${response.status} - ${errorText}`);
        }

        const data = await response.json();
        return data;

    } catch (error) {
        console.error("Failed to retrieve queue stats:", error);
        throw error;
    }
}

// Usage
getRealTimeQueueStats(QUEUE_ID)
    .then(stats => {
        console.log("Queue Stats:", JSON.stringify(stats, null, 2));
    })
    .catch(err => console.error(err));

Step 2: Parse the Response Structure

The response from /api/v2/reporting/queues/realtime is a nested JSON object. The key metrics you likely need are located within the queue object. The structure includes:

• queueId: The ID of the queue.
• name: The name of the queue.
• agentsLoggedOn: Number of agents currently logged in.
• agentsAvailable: Number of agents ready to take calls.
• agentsBusy: Number of agents currently on a call or ACW (After Call Work).
• callsWaiting: Number of calls currently in the queue.
• averageWaitTime: Average wait time in milliseconds for calls currently in the queue.
• serviceLevel: The percentage of calls answered within the service level threshold.

Here is the Python code to parse this response and extract the critical metrics. It also includes error handling for cases where the queue might not exist or returns no data.

import requests
import json

def fetch_queue_stats(access_token: str, queue_id: str, base_url: str) -> dict:
    """
    Fetches and parses real-time stats for a specific CXone queue.
    """
    endpoint = f"{base_url}/api/v2/reporting/queues/realtime"
    
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {access_token}"
    }
    
    payload = {
        "queueId": queue_id
    }

    try:
        response = requests.post(endpoint, json=payload, headers=headers)
        
        # Check for 429 Too Many Requests
        if response.status_code == 429:
            retry_after = response.headers.get("Retry-After", 1)
            print(f"Rate limited. Please wait {retry_after} seconds.")
            return {}
        
        response.raise_for_status()
        
        data = response.json()
        
        # The API returns a list of queue objects, even if only one ID is requested
        if not data.get("queues"):
            print("No queue data returned.")
            return {}
            
        queue_data = data["queues"][0]
        
        # Extract key metrics
        stats = {
            "queue_id": queue_data.get("queueId"),
            "queue_name": queue_data.get("name"),
            "agents_logged_on": queue_data.get("agentsLoggedOn", 0),
            "agents_available": queue_data.get("agentsAvailable", 0),
            "agents_busy": queue_data.get("agentsBusy", 0),
            "calls_waiting": queue_data.get("callsWaiting", 0),
            "average_wait_time_ms": queue_data.get("averageWaitTime", 0),
            "service_level": queue_data.get("serviceLevel", 0.0)
        }
        
        return stats

    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 401:
            print("Unauthorized. Token may be expired or invalid.")
        elif e.response.status_code == 403:
            print("Forbidden. Check if the client has 'reporting:queue:view' scope.")
        elif e.response.status_code == 404:
            print(f"Queue ID {queue_id} not found.")
        else:
            print(f"HTTP Error: {e}")
        return {}
    except Exception as e:
        print(f"Unexpected error: {e}")
        return {}

# Example usage
# stats = fetch_queue_stats(access_token, "your-queue-id", CXONE_BASE_URL)
# print(json.dumps(stats, indent=2))

Step 3: Handle Pagination and Multiple Queues

If you need stats for multiple queues, you can pass an array of queueIds in the request body. The API will return a list of queue objects in the response. However, be aware of rate limits. CXone enforces strict rate limiting on reporting endpoints. If you query too frequently, you will receive a 429 status code.

To handle multiple queues efficiently in Python, you can batch them. The CXone API allows up to 100 queue IDs per request.

def fetch_multiple_queue_stats(access_token: str, queue_ids: list, base_url: str) -> list:
    """
    Fetches stats for multiple queues in a single request.
    """
    endpoint = f"{base_url}/api/v2/reporting/queues/realtime"
    
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {access_token}"
    }
    
    # Ensure we do not exceed the limit of 100 IDs per request
    if len(queue_ids) > 100:
        raise ValueError("CXone API supports a maximum of 100 queue IDs per request.")

    payload = {
        "queueIds": queue_ids
    }

    try:
        response = requests.post(endpoint, json=payload, headers=headers)
        response.raise_for_status()
        
        data = response.json()
        queues = data.get("queues", [])
        
        # Process each queue
        results = []
        for q in queues:
            stats = {
                "queue_id": q.get("queueId"),
                "queue_name": q.get("name"),
                "agents_logged_on": q.get("agentsLoggedOn", 0),
                "calls_waiting": q.get("callsWaiting", 0),
                "average_wait_time_ms": q.get("averageWaitTime", 0)
            }
            results.append(stats)
            
        return results

    except requests.exceptions.HTTPError as e:
        print(f"HTTP Error: {e.response.status_code} - {e.response.text}")
        return []
    except Exception as e:
        print(f"Error fetching multiple queues: {e}")
        return []

Complete Working Example

Below is a complete, runnable Python script. It handles token acquisition, error handling, and data extraction. You must install requests and python-dotenv and create a .env file with your credentials.

File: cxone_queue_stats.py

import requests
import os
import time
import json
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

# Configuration
CXONE_BASE_URL = os.getenv("CXONE_BASE_URL", "https://api-us-02.nice-incontact.com")
CXONE_CLIENT_ID = os.getenv("CXONE_CLIENT_ID")
CXONE_CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET")
CXONE_TENANT_ID = os.getenv("CXONE_TENANT_ID")
TARGET_QUEUE_ID = os.getenv("TARGET_QUEUE_ID")

class CXoneClient:
    def __init__(self, base_url, client_id, client_secret, tenant_id):
        self.base_url = base_url
        self.client_id = client_id
        self.client_secret = client_secret
        self.tenant_id = tenant_id
        self.access_token = None
        self.token_expiry = 0

    def _get_token(self):
        """Fetches a new OAuth2 token."""
        token_url = f"{self.base_url}/oauth2/token"
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        body = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "tenant_id": self.tenant_id
        }
        
        response = requests.post(token_url, headers=headers, data=body)
        response.raise_for_status()
        token_data = response.json()
        self.access_token = token_data["access_token"]
        # Tokens usually expire in 3600 seconds (1 hour)
        self.token_expiry = time.time() + 3500 # Refresh slightly before expiry
        return self.access_token

    def ensure_token(self):
        """Ensures we have a valid token."""
        if not self.access_token or time.time() > self.token_expiry:
            self._get_token()

    def get_queue_stats(self, queue_id):
        """Fetches real-time stats for a specific queue."""
        self.ensure_token()
        
        endpoint = f"{self.base_url}/api/v2/reporting/queues/realtime"
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.access_token}"
        }
        payload = {"queueId": queue_id}

        try:
            response = requests.post(endpoint, json=payload, headers=headers)
            
            if response.status_code == 429:
                print("Rate limited. Waiting 5 seconds...")
                time.sleep(5)
                return self.get_queue_stats(queue_id) # Retry
            
            response.raise_for_status()
            data = response.json()
            
            if not data.get("queues"):
                return None
            
            q = data["queues"][0]
            return {
                "queue_id": q.get("queueId"),
                "name": q.get("name"),
                "agents_logged_on": q.get("agentsLoggedOn", 0),
                "agents_available": q.get("agentsAvailable", 0),
                "calls_waiting": q.get("callsWaiting", 0),
                "avg_wait_time_ms": q.get("averageWaitTime", 0),
                "service_level": q.get("serviceLevel", 0)
            }
            
        except requests.exceptions.HTTPError as e:
            print(f"HTTP Error: {e}")
            return None
        except Exception as e:
            print(f"Error: {e}")
            return None

def main():
    if not all([CXONE_CLIENT_ID, CXONE_CLIENT_SECRET, CXONE_TENANT_ID, TARGET_QUEUE_ID]):
        print("Missing environment variables. Please check your .env file.")
        return

    client = CXoneClient(CXONE_BASE_URL, CXONE_CLIENT_ID, CXONE_CLIENT_SECRET, CXONE_TENANT_ID)
    
    print(f"Fetching stats for Queue ID: {TARGET_QUEUE_ID}")
    stats = client.get_queue_stats(TARGET_QUEUE_ID)
    
    if stats:
        print(json.dumps(stats, indent=2))
    else:
        print("Failed to retrieve stats.")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The access token is expired, invalid, or missing.
• Fix: Ensure your _get_token method is called before each API request or that you are caching the token correctly. Check that your Client ID and Secret are correct.

Error: 403 Forbidden

• Cause: The OAuth client does not have the required scope.
• Fix: Log into the CXone Admin Portal, navigate to Administration > Security > API Clients, and ensure the client has the reporting:queue:view scope assigned.

Error: 429 Too Many Requests

• Cause: You have exceeded the rate limit for the Reporting API. CXone has strict limits on real-time endpoints (often around 10-20 requests per minute per client).
• Fix: Implement exponential backoff. Do not poll every second. Poll every 30-60 seconds unless you have a specific high-frequency requirement approved by NICE.

Error: 404 Not Found

• Cause: The Queue ID provided does not exist or is not visible to the current tenant.
• Fix: Verify the Queue ID using the /api/v2/queues endpoint. Ensure the queue is active.

Official References

• CXone v2 Reporting API Documentation
• CXone OAuth 2.0 Authentication Guide
• CXone Real-Time Queue Metrics Reference