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

What You Will Build

This tutorial demonstrates how to retrieve live, real-time statistics for specific queues in NICE CXone using the v2 Reporting API. You will build a Python script that authenticates via OAuth 2.0, queries the /api/v2/reporting/queues/realtime endpoint, and parses the JSON response to display key metrics such as agents available, calls in queue, and average wait time. The solution includes robust error handling for rate limits and authentication failures.

Prerequisites

• OAuth Client Type: Machine-to-Machine (M2M) application.
• Required Scopes: reporting:read and icm:read.
• SDK/API Version: NICE CXone REST API v2.
• Language/Runtime: Python 3.8+ (uses requests library).
• External Dependencies: requests (pip install requests).
• Environment Variables: CXONE_CLIENT_ID, CXONE_CLIENT_SECRET, CXONE_DOMAIN, CXONE_QUEUE_ID.

Authentication Setup

NICE CXone uses OAuth 2.0 for authentication. For server-side integrations, the Client Credentials flow is the standard approach. You must obtain an access token before making any reporting calls. Tokens expire after 3600 seconds (1 hour), so production code should cache tokens and refresh them before expiration.

The following function handles the token acquisition. It sends a POST request to the /oauth/token endpoint with your Client ID and Client Secret.

import os
import requests
import time
import json
from typing import Optional, Dict, Any

# Configuration from environment variables
CLIENT_ID = os.getenv("CXONE_CLIENT_ID")
CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET")
DOMAIN = os.getenv("CXONE_DOMAIN")  # e.g., "api-us-01.niceincontact.com"

# Ensure domain does not have trailing slashes or protocol
if DOMAIN.startswith("https://"):
    DOMAIN = DOMAIN.replace("https://", "")
if DOMAIN.endswith("/"):
    DOMAIN = DOMAIN.rstrip("/")

def get_access_token() -> str:
    """
    Authenticates with NICE CXone and returns an OAuth 2.0 access token.
    
    Returns:
        str: The access token.
        
    Raises:
        requests.exceptions.HTTPError: If authentication fails.
    """
    auth_url = f"https://{DOMAIN}/oauth/token"
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    data = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET
    }

    response = requests.post(auth_url, headers=headers, data=data)
    
    if response.status_code != 200:
        raise Exception(f"Authentication failed with status {response.status_code}: {response.text}")
    
    token_data = response.json()
    return token_data["access_token"]

Implementation

Step 1: Constructing the Real-Time Queue Query

The v2 Reporting API for real-time queue stats requires a specific JSON payload to define the scope of the query. Unlike historical reports, real-time queries do not use date ranges. Instead, they require a list of queue IDs to monitor.

You must provide the queueIds array in the request body. If you omit this, the API may return an error or an empty result set depending on your tenant configuration.

The endpoint is POST /api/v2/reporting/queues/realtime.

def build_queue_query(queue_ids: list[str]) -> Dict[str, Any]:
    """
    Constructs the JSON payload for the real-time queue stats API.
    
    Args:
        queue_ids: A list of queue ID strings to monitor.
        
    Returns:
        Dict[str, Any]: The JSON payload ready for the API request.
    """
    payload = {
        "queueIds": queue_ids,
        "groupBy": "queue"  # Optional: groups results by queue. Default is often fine.
    }
    return payload

Step 2: Executing the API Request with Retry Logic

NICE CXone enforces strict rate limits. Hitting a 429 Too Many Requests error is common when polling frequently. A robust integration must handle these errors gracefully by implementing exponential backoff.

This step sends the query constructed in Step 1 to the API. It includes logic to detect 429 errors, parse the Retry-After header if present, and retry the request.

def fetch_realtime_queue_stats(token: str, queue_ids: list[str], max_retries: int = 3) -> Dict[str, Any]:
    """
    Fetches real-time statistics for specified queues.
    
    Args:
        token: Valid OAuth access token.
        queue_ids: List of queue IDs to query.
        max_retries: Maximum number of retries for 429 errors.
        
    Returns:
        Dict[str, Any]: The parsed JSON response from the API.
    """
    url = f"https://{DOMAIN}/api/v2/reporting/queues/realtime"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    payload = build_queue_query(queue_ids)

    retries = 0
    while retries <= max_retries:
        try:
            response = requests.post(url, headers=headers, json=payload)
            
            # Handle Success
            if response.status_code == 200:
                return response.json()
            
            # Handle Rate Limiting (429)
            if response.status_code == 429:
                if retries < max_retries:
                    retry_after = int(response.headers.get("Retry-After", 5))
                    print(f"Rate limited. Waiting {retry_after} seconds before retry {retries + 1}/{max_retries}...")
                    time.sleep(retry_after)
                    retries += 1
                    continue
                else:
                    raise Exception("Max retries exceeded due to rate limiting.")
            
            # Handle Other Errors (401, 403, 5xx)
            if response.status_code == 401:
                raise Exception("Invalid or expired token. Please refresh authentication.")
            if response.status_code == 403:
                raise Exception("Forbidden: Check OAuth scopes (reporting:read, icm:read).")
            
            # Generic Error
            raise Exception(f"API Error {response.status_code}: {response.text}")
            
        except requests.exceptions.RequestException as e:
            raise Exception(f"Network error: {str(e)}")

    return {}

Step 3: Processing and Parsing the Results

The response from /api/v2/reporting/queues/realtime is a nested JSON object. The key data resides in the queueDetails array. Each element in this array corresponds to a queue ID you requested.

Key fields to extract:

• queueId: The unique identifier of the queue.
• agentsAvailable: Number of agents currently available to take calls.
• callsInQueue: Number of calls currently waiting.
• averageWaitTime: Average time calls have been waiting (in seconds).
• agentsBusy: Number of agents currently on a call.

The following function parses this structure into a more developer-friendly format.

def parse_queue_stats(response_data: Dict[str, Any]) -> list[Dict[str, Any]]:
    """
    Parses the API response into a simplified list of queue metrics.
    
    Args:
        response_data: The raw JSON response from the API.
        
    Returns:
        list[Dict[str, Any]]: A list of dictionaries containing key metrics per queue.
    """
    results = []
    
    # The API returns a structure like: {"queueDetails": [...]}
    if "queueDetails" not in response_data:
        print("Warning: 'queueDetails' key not found in response.")
        return results
    
    queue_details = response_data.get("queueDetails", [])
    
    for queue in queue_details:
        queue_id = queue.get("queueId")
        
        # Extract metrics, defaulting to 0 if missing
        metrics = {
            "queueId": queue_id,
            "agentsAvailable": queue.get("agentsAvailable", 0),
            "agentsBusy": queue.get("agentsBusy", 0),
            "callsInQueue": queue.get("callsInQueue", 0),
            "averageWaitTime": queue.get("averageWaitTime", 0),
            "totalHandled": queue.get("totalHandled", 0)
        }
        results.append(metrics)
        
    return results

Complete Working Example

The following script combines authentication, querying, and parsing into a single executable module. It polls a specific queue every 30 seconds, demonstrating a typical monitoring use case.

import os
import time
import requests
import json
from typing import Optional, Dict, Any, List

# --- Configuration ---
CLIENT_ID = os.getenv("CXONE_CLIENT_ID")
CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET")
DOMAIN = os.getenv("CXONE_DOMAIN")
QUEUE_ID = os.getenv("CXONE_QUEUE_ID")  # Single queue ID for this example

# Validate environment variables
if not all([CLIENT_ID, CLIENT_SECRET, DOMAIN, QUEUE_ID]):
    raise EnvironmentError("Missing required environment variables: CXONE_CLIENT_ID, CXONE_CLIENT_SECRET, CXONE_DOMAIN, CXONE_QUEUE_ID")

# Clean domain
if DOMAIN.startswith("https://"):
    DOMAIN = DOMAIN.replace("https://", "")
if DOMAIN.endswith("/"):
    DOMAIN = DOMAIN.rstrip("/")

# --- Authentication Module ---
def get_access_token() -> str:
    auth_url = f"https://{DOMAIN}/oauth/token"
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    data = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET
    }

    response = requests.post(auth_url, headers=headers, data=data)
    
    if response.status_code != 200:
        raise Exception(f"Authentication failed with status {response.status_code}: {response.text}")
    
    return response.json()["access_token"]

# --- API Interaction Module ---
def fetch_realtime_queue_stats(token: str, queue_ids: List[str], max_retries: int = 3) -> Dict[str, Any]:
    url = f"https://{DOMAIN}/api/v2/reporting/queues/realtime"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    payload = {
        "queueIds": queue_ids,
        "groupBy": "queue"
    }

    retries = 0
    while retries <= max_retries:
        try:
            response = requests.post(url, headers=headers, json=payload)
            
            if response.status_code == 200:
                return response.json()
            
            if response.status_code == 429:
                if retries < max_retries:
                    retry_after = int(response.headers.get("Retry-After", 5))
                    print(f"[INFO] Rate limited. Retrying in {retry_after}s...")
                    time.sleep(retry_after)
                    retries += 1
                    continue
                else:
                    raise Exception("Max retries exceeded for 429.")
            
            if response.status_code == 401:
                raise Exception("Token expired or invalid.")
            if response.status_code == 403:
                raise Exception("Forbidden: Check scopes.")
            
            raise Exception(f"Unexpected status {response.status_code}: {response.text}")
            
        except requests.exceptions.RequestException as e:
            raise Exception(f"Network error: {str(e)}")
    
    return {}

# --- Data Processing Module ---
def parse_queue_stats(response_data: Dict[str, Any]) -> List[Dict[str, Any]]:
    results = []
    queue_details = response_data.get("queueDetails", [])
    
    for queue in queue_details:
        results.append({
            "queueId": queue.get("queueId"),
            "agentsAvailable": queue.get("agentsAvailable", 0),
            "agentsBusy": queue.get("agentsBusy", 0),
            "callsInQueue": queue.get("callsInQueue", 0),
            "averageWaitTime": queue.get("averageWaitTime", 0),
            "timestamp": queue.get("timestamp", "")
        })
    return results

# --- Main Execution ---
def main():
    print("Starting NICE CXone Real-Time Queue Monitor...")
    print(f"Monitoring Queue ID: {QUEUE_ID}")
    print("-" * 50)

    try:
        # 1. Authenticate
        print("Authenticating...")
        token = get_access_token()
        print("Authentication successful.")

        # 2. Loop for polling (Example: Poll every 30 seconds)
        while True:
            try:
                # 3. Fetch Stats
                raw_data = fetch_realtime_queue_stats(token, [QUEUE_ID])
                
                # 4. Parse and Display
                stats = parse_queue_stats(raw_data)
                
                if stats:
                    for stat in stats:
                        print(f"\n[Queue: {stat['queueId']}]")
                        print(f"  Agents Available: {stat['agentsAvailable']}")
                        print(f"  Agents Busy:      {stat['agentsBusy']}")
                        print(f"  Calls in Queue:   {stat['callsInQueue']}")
                        print(f"  Avg Wait Time:    {stat['averageWaitTime']}s")
                        print(f"  Snapshot Time:    {stat['timestamp']}")
                else:
                    print("No data returned for the specified queue ID.")
                    
            except Exception as e:
                print(f"\nError fetching data: {str(e)}")
                # If token is expired, re-authenticate
                if "expired" in str(e).lower():
                    print("Refreshing token...")
                    token = get_access_token()
                else:
                    print("Continuing loop despite error...")
            
            # Wait before next poll
            time.sleep(30)

    except KeyboardInterrupt:
        print("\nMonitor stopped by user.")
    except Exception as e:
        print(f"\nFatal error: {str(e)}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token is expired, invalid, or not included in the Authorization header.
• Fix: Ensure your get_access_token function is called before the API request. If polling over a long period, check if the token has expired (standard validity is 3600 seconds) and re-fetch it.
• Code Fix: Implement a token cache with a TTL (Time-To-Live) or refresh the token immediately upon receiving a 401 response.

Error: 403 Forbidden

• Cause: The OAuth client lacks the required scopes.
• Fix: Verify that the application in the NICE CXone Admin Console has the reporting:read and icm:read scopes assigned.
• Debug Step: Print the scopes returned in the initial token response (if available) or check the Admin Console under “Integrations” > “Applications”.

Error: 429 Too Many Requests

• Cause: You have exceeded the rate limit for the Reporting API. Real-time endpoints have lower limits than historical endpoints.
• Fix: Implement exponential backoff. Respect the Retry-After header in the response.
• Code Fix: The provided example includes a max_retries loop with time.sleep based on the Retry-After header.

Error: Empty queueDetails Array

• Cause: The queueId provided does not exist, is not active, or has no recent activity/data snapshots available.
• Fix: Verify the Queue ID in the NICE CXone Admin Console. Ensure the queue is in “Active” status. Real-time data is sampled; if no calls have occurred recently, some metrics may be zero or missing.

Official References

• NICE CXone v2 Reporting API Documentation
• NICE CXone OAuth 2.0 Authentication
• Python Requests Library Documentation