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

What You Will Build

You will build a Python script that authenticates with the NICE CXone platform and retrieves real-time statistical data for specific queues, including agent availability and call volume. You will use the NICE CXone v2 Reporting API, specifically the GET /api/v2/interactionstats/queues endpoint. The implementation covers token acquisition, parameter construction, response parsing, and error handling using the requests library.

Prerequisites

• OAuth Client Type: Service Account (Client Credentials Flow) or User Account (Authorization Code Flow). This tutorial uses the Client Credentials Flow for server-to-server automation.
• Required Scopes: interactionstats:read is required to access queue statistics.
• SDK/API Version: NICE CXone API v2.
• Language/Runtime: Python 3.8+.
• External Dependencies: requests (HTTP library), python-dotenv (for secure credential management).

Install dependencies:

pip install requests python-dotenv

Authentication Setup

NICE CXone uses OAuth 2.0 for authentication. For backend integrations, the Client Credentials flow is standard. You must exchange your API Key and Secret for an access token. The token expires after 3600 seconds (1 hour), so your integration must handle token refresh or re-authentication.

Create a .env file in your project root:

CXONE_API_KEY=your_api_key_here
CXONE_API_SECRET=your_api_secret_here
CXONE_AUTH_URL=https://platform.niceincontact.com/oauth2/token
CXONE_API_BASE_URL=https://platform.niceincontact.com/api/v2

Authentication Code:

import os
import requests
from dotenv import load_dotenv

load_dotenv()

CXONE_AUTH_URL = os.getenv("CXONE_AUTH_URL")
CXONE_API_KEY = os.getenv("CXONE_API_KEY")
CXONE_API_SECRET = os.getenv("CXONE_API_SECRET")
CXONE_API_BASE_URL = os.getenv("CXONE_API_BASE_URL")

def get_access_token() -> str:
    """
    Authenticates with CXone and returns an OAuth access token.
    Raises an exception if authentication fails.
    """
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    data = {
        "grant_type": "client_credentials",
        "client_id": CXONE_API_KEY,
        "client_secret": CXONE_API_SECRET
    }

    try:
        response = requests.post(CXONE_AUTH_URL, headers=headers, data=data)
        response.raise_for_status()
        token_data = response.json()
        return token_data["access_token"]
    except requests.exceptions.HTTPError as e:
        print(f"Authentication failed: {e.response.status_code} - {e.response.text}")
        raise
    except requests.exceptions.RequestException as e:
        print(f"Network error during authentication: {e}")
        raise

# Example usage
# token = get_access_token()

Implementation

Step 1: Constructing the Request Parameters

The GET /api/v2/interactionstats/queues endpoint requires specific query parameters to filter the data. You cannot simply call the endpoint without parameters; it expects a list of queue IDs.

Critical Parameters:

• queueIds: A comma-separated string of Queue IDs. You must know these IDs beforehand. They are internal UUIDs, not the queue name.
• interval: The time interval for the stats. For real-time data, use realtime.
• metrics: A comma-separated list of metrics to retrieve. Common metrics include agentCount, availableAgentCount, busyAgentCount, callCount, queuePosition.

If you do not know the Queue IDs, you must first call GET /api/v2/queues to fetch them. This tutorial assumes you have the IDs or provides a helper to fetch them.

Fetching Queue IDs (Helper Function):

def get_queue_ids(token: str) -> list[str]:
    """
    Fetches all queues and returns a list of their IDs.
    Scope: interactionstats:read is usually sufficient, but queues:read is safer.
    """
    url = f"{CXONE_API_BASE_URL}/queues"
    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/json"
    }
    
    params = {
        "pageSize": 100,
        "pageNumber": 1
    }

    all_queue_ids = []
    page = 1
    
    while True:
        params["pageNumber"] = page
        response = requests.get(url, headers=headers, params=params)
        
        if response.status_code == 401:
            raise Exception("Token expired. Please re-authenticate.")
        response.raise_for_status()
        
        data = response.json()
        items = data.get("items", [])
        all_queue_ids.extend([q["id"] for q in items])
        
        # Check if there are more pages
        if page >= data.get("totalPageCount", 1):
            break
        page += 1
        
    return all_queue_ids

Step 2: Retrieving Real-Time Queue Stats

With the token and queue IDs, you can now query the stats endpoint. This endpoint returns a snapshot of the queue’s current state.

API Endpoint: GET /api/v2/interactionstats/queues

Required Scopes: interactionstats:read

Code Implementation:

def get_realtime_queue_stats(token: str, queue_ids: list[str]) -> dict:
    """
    Retrieves real-time statistics for a list of queues.
    
    Args:
        token: Valid OAuth access token.
        queue_ids: List of Queue UUIDs.
        
    Returns:
        Dictionary containing the stats response.
    """
    if not queue_ids:
        return {}

    # CXone expects a comma-separated string for queueIds
    queue_ids_param = ",".join(queue_ids)
    
    url = f"{CXONE_API_BASE_URL}/interactionstats/queues"
    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/json"
    }
    
    params = {
        "queueIds": queue_ids_param,
        "interval": "realtime",
        # Select specific metrics to reduce payload size
        "metrics": "agentCount,availableAgentCount,busyAgentCount,callCount,queuePosition,holdCount,waitingCount"
    }

    try:
        response = requests.get(url, headers=headers, params=params)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        if response.status_code == 429:
            print("Rate limit exceeded. Implement exponential backoff.")
        elif response.status_code == 401:
            print("Unauthorized. Token may be invalid or expired.")
        else:
            print(f"HTTP Error: {response.status_code} - {response.text}")
        raise
    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}")
        raise

Step 3: Processing Results and Handling Edge Cases

The response from CXone is nested. The structure typically looks like this:

{
  "items": [
    {
      "queueId": "abc-123-def",
      "queueName": "Sales Support",
      "metrics": {
        "agentCount": 10,
        "availableAgentCount": 2,
        "busyAgentCount": 8,
        "callCount": 5,
        "queuePosition": 0,
        "holdCount": 1,
        "waitingCount": 3
      }
    }
  ]
}

Processing Logic:

def process_queue_stats(stats_data: dict) -> list[dict]:
    """
    Parses the raw API response into a clean list of dictionaries.
    Handles cases where a queue might return null metrics.
    """
    processed_stats = []
    
    if not stats_data or "items" not in stats_data:
        return processed_stats

    for item in stats_data["items"]:
        queue_id = item.get("queueId")
        queue_name = item.get("queueName", "Unknown Queue")
        metrics = item.get("metrics", {})
        
        # Default to 0 if metric is missing or null
        clean_metrics = {
            "agent_count": metrics.get("agentCount", 0) or 0,
            "available_agents": metrics.get("availableAgentCount", 0) or 0,
            "busy_agents": metrics.get("busyAgentCount", 0) or 0,
            "total_calls": metrics.get("callCount", 0) or 0,
            "queue_position": metrics.get("queuePosition", 0) or 0,
            "on_hold": metrics.get("holdCount", 0) or 0,
            "waiting": metrics.get("waitingCount", 0) or 0
        }
        
        processed_stats.append({
            "queue_id": queue_id,
            "queue_name": queue_name,
            "stats": clean_metrics
        })
        
    return processed_stats

Complete Working Example

This script combines authentication, queue ID retrieval, and stats fetching into a single executable module.

import os
import time
import requests
from dotenv import load_dotenv

load_dotenv()

# Configuration
CXONE_AUTH_URL = os.getenv("CXONE_AUTH_URL")
CXONE_API_KEY = os.getenv("CXONE_API_KEY")
CXONE_API_SECRET = os.getenv("CXONE_API_SECRET")
CXONE_API_BASE_URL = os.getenv("CXONE_API_BASE_URL")

def get_access_token() -> str:
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    data = {
        "grant_type": "client_credentials",
        "client_id": CXONE_API_KEY,
        "client_secret": CXONE_API_SECRET
    }
    try:
        response = requests.post(CXONE_AUTH_URL, headers=headers, data=data)
        response.raise_for_status()
        return response.json()["access_token"]
    except requests.exceptions.RequestException as e:
        print(f"Authentication failed: {e}")
        raise

def get_queue_ids(token: str) -> list[str]:
    url = f"{CXONE_API_BASE_URL}/queues"
    headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
    params = {"pageSize": 100, "pageNumber": 1}
    
    all_queue_ids = []
    page = 1
    
    while True:
        params["pageNumber"] = page
        response = requests.get(url, headers=headers, params=params)
        response.raise_for_status()
        
        data = response.json()
        items = data.get("items", [])
        all_queue_ids.extend([q["id"] for q in items])
        
        if page >= data.get("totalPageCount", 1):
            break
        page += 1
    return all_queue_ids

def get_realtime_queue_stats(token: str, queue_ids: list[str]) -> dict:
    if not queue_ids:
        return {}
    
    queue_ids_param = ",".join(queue_ids)
    url = f"{CXONE_API_BASE_URL}/interactionstats/queues"
    headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
    params = {
        "queueIds": queue_ids_param,
        "interval": "realtime",
        "metrics": "agentCount,availableAgentCount,busyAgentCount,callCount,queuePosition,holdCount,waitingCount"
    }
    
    try:
        response = requests.get(url, headers=headers, params=params)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        print(f"API Error: {response.status_code} - {response.text}")
        raise
    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}")
        raise

def main():
    print("Step 1: Authenticating...")
    token = get_access_token()
    print("Authentication successful.")
    
    print("Step 2: Fetching Queue IDs...")
    queue_ids = get_queue_ids(token)
    print(f"Found {len(queue_ids)} queues.")
    
    if not queue_ids:
        print("No queues found. Exiting.")
        return

    print("Step 3: Retrieving Real-Time Stats...")
    # Limit to first 5 queues for demo purposes to avoid payload size issues
    demo_queue_ids = queue_ids[:5] 
    stats_data = get_realtime_queue_stats(token, demo_queue_ids)
    
    print("Step 4: Processing Results...")
    processed_stats = []
    if stats_data and "items" in stats_data:
        for item in stats_data["items"]:
            metrics = item.get("metrics", {})
            processed_stats.append({
                "queue_name": item.get("queueName"),
                "available_agents": metrics.get("availableAgentCount", 0) or 0,
                "busy_agents": metrics.get("busyAgentCount", 0) or 0,
                "waiting_calls": metrics.get("waitingCount", 0) or 0
            })
    
    print("\n--- Real-Time Queue Statistics ---")
    print(f"{'Queue Name':<20} | {'Avail Agents':<12} | {'Busy Agents':<12} | {'Waiting':<10}")
    print("-" * 60)
    for stat in processed_stats:
        print(f"{stat['queue_name']:<20} | {stat['available_agents']:<12} | {stat['busy_agents']:<12} | {stat['waiting_calls']:<10}")
    
    print("\nData retrieval complete.")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

Cause: The access token is expired, invalid, or missing. OAuth tokens in CXone expire after 1 hour.
Fix: Implement a token cache. Check the token’s expiry time before making API calls. If the token is expired, call get_access_token() again.

# Simple token caching logic
TOKEN_CACHE = {"token": None, "expiry": 0}

def get_cached_token() -> str:
    if TOKEN_CACHE["token"] and time.time() < TOKEN_CACHE["expiry"]:
        return TOKEN_CACHE["token"]
    
    new_token = get_access_token()
    # Token expires in 3600s, cache for 3500s to be safe
    TOKEN_CACHE["token"] = new_token
    TOKEN_CACHE["expiry"] = time.time() + 3500
    return new_token

Error: 429 Too Many Requests

Cause: You have exceeded the API rate limit. CXone enforces strict rate limits per tenant.
Fix: Implement exponential backoff. Do not retry immediately. Wait 1 second, then 2, then 4, etc.

import time

def fetch_with_retry(url, headers, params, max_retries=3):
    for attempt in range(max_retries):
        response = requests.get(url, headers=headers, params=params)
        if response.status_code == 429:
            wait_time = 2 ** attempt
            print(f"Rate limited. Waiting {wait_time} seconds...")
            time.sleep(wait_time)
            continue
        response.raise_for_status()
        return response.json()
    raise Exception("Max retries exceeded")

Error: Empty “items” array

Cause: The queueIds parameter contains invalid IDs, or the queue is inactive/deleted.
Fix: Verify the Queue IDs by calling GET /api/v2/queues/{id} individually. Ensure the IDs are valid UUIDs. Check that the queue status is Active.

Error: 403 Forbidden

Cause: The OAuth client does not have the interactionstats:read scope.
Fix: Go to the CXone Admin Portal > Security > API Keys. Edit your API Key and ensure the interactionstats:read permission is checked. Regenerate the token after updating scopes.

Official References

• NICE CXone Interaction Stats API Documentation
• NICE CXone OAuth Authentication Guide
• NICE CXone Python SDK (Optional)