How to extract real-time queue stats from NICE CXone using the v2 Reporting API

What You Will Build

• This tutorial builds a Python script that retrieves live queue metrics, including waiting calls, answered calls, and average wait times, for a specified list of queues.
• The solution utilizes the NICE CXone v2 Reporting API (/api/v2/reporting/queues/stats) to fetch aggregated real-time data.
• The implementation is written in Python 3.10+ using the requests library for HTTP handling and pydantic for response validation.

Prerequisites

• OAuth Client Type: Client Credentials Grant. You must have an API key configured in the CXone Admin Console with the necessary permissions.
• Required Scopes: reporting:queues:read. Without this scope, the API will return a 403 Forbidden error.
• SDK Version: This tutorial uses direct HTTP requests via the requests library to demonstrate the underlying mechanics, which is often more debuggable than the official SDK for complex reporting queries.
• Language/Runtime Requirements: Python 3.10 or higher.
• External Dependencies:
  • requests: For HTTP communication.
  • pydantic: For data validation and type safety.
  • python-dotenv: For secure credential management.

Install dependencies using pip:

pip install requests pydantic python-dotenv

Authentication Setup

NICE CXone uses OAuth 2.0 for authentication. The Client Credentials flow is the standard for server-to-server integrations where no user context is required. You must exchange your API Key and API Secret for an access token before making any reporting calls.

The token endpoint is https://platform.my.site.com/oauth/token. Note that platform.my.site.com is a placeholder; you must replace my.site with your specific CXone environment subdomain (e.g., platform.us-east-1.my.site.com).

Create a configuration file .env in your project root:

CXONE_API_KEY=your_client_id_here
CXONE_API_SECRET=your_client_secret_here
CXONE_BASE_URL=https://platform.us-east-1.my.site.com

Implement the authentication logic. This function handles the token exchange and caches the token with a slight buffer to prevent expiration during execution.

import os
import time
import requests
from dotenv import load_dotenv
from typing import Optional

load_dotenv()

class CXoneAuth:
    def __init__(self):
        self.api_key = os.getenv("CXONE_API_KEY")
        self.api_secret = os.getenv("CXONE_API_SECRET")
        self.base_url = os.getenv("CXONE_BASE_URL")
        self.token_url = f"{self.base_url}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0.0

    def get_access_token(self) -> str:
        """
        Retrieves an OAuth2 access token using Client Credentials Grant.
        Implements simple caching to avoid unnecessary token refreshes.
        """
        # Check if token is still valid (buffer 60 seconds)
        if self.access_token and time.time() < self.token_expiry - 60:
            return self.access_token

        payload = {
            "grant_type": "client_credentials",
            "client_id": self.api_key,
            "client_secret": self.api_secret
        }

        try:
            response = requests.post(self.token_url, data=payload, timeout=10)
            response.raise_for_status()
            
            token_data = response.json()
            self.access_token = token_data.get("access_token")
            expires_in = token_data.get("expires_in", 3600)
            self.token_expiry = time.time() + expires_in
            
            if not self.access_token:
                raise ValueError("No access_token in response")
                
            return self.access_token

        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise Exception("Authentication failed: Invalid API Key or Secret") from e
            elif response.status_code == 403:
                raise Exception("Authentication failed: Client does not have permission to request tokens") from e
            else:
                raise Exception(f"HTTP Error {response.status_code}: {response.text}") from e
        except requests.exceptions.RequestException as e:
            raise Exception(f"Network error during authentication: {e}") from e

Implementation

Step 1: Define the Request Payload and Endpoint

The CXone v2 Reporting API for queue stats requires a specific JSON payload to define the query. Unlike GET requests with query parameters, this endpoint uses a POST body to specify the grouping, metrics, and dateRange.

For real-time stats, you do not need a historical date range. You request the current state. The grouping determines how the data is aggregated (e.g., by Queue ID). The metrics array specifies exactly which data points you need.

Common metrics for queue monitoring include:

• waiting: Number of interactions currently waiting in the queue.
• answered: Number of interactions answered in the current interval.
• abandoned: Number of interactions abandoned.
• avgWaitTime: Average wait time for answered interactions.
• serviceLevel: Percentage of interactions answered within the defined service level threshold.

import json
from typing import List, Dict, Any

def build_queue_stats_payload(queue_ids: List[str]) -> Dict[str, Any]:
    """
    Constructs the JSON payload for the /api/v2/reporting/queues/stats endpoint.
    
    Args:
        queue_ids: List of queue IDs to query.
        
    Returns:
        Dictionary containing the query parameters.
    """
    # Define the metrics we want to extract
    metrics = [
        "waiting",
        "answered",
        "abandoned",
        "avgWaitTime",
        "serviceLevel",
        "maxWaitTime",
        "totalInteractions"
    ]
    
    payload = {
        "grouping": [
            {
                "type": "queueId"
            }
        ],
        "metrics": metrics,
        "dateRange": {
            "type": "last24Hours", 
            # Note: For strict "real-time" snapshots, some implementations use 
            # a very short recent window. However, CXone's real-time endpoint 
            # often defaults to current state if no historical range is strictly enforced 
            # or if using the specific 'realtime' flag if available in newer SDKs.
            # The v2 reporting API typically aggregates over the specified range.
            # To get a pure snapshot, we rely on the 'waiting' metric which is instantaneous.
        },
        "filters": [
            {
                "filterType": "queueId",
                "values": queue_ids
            }
        ]
    }
    
    return payload

Step 2: Execute the API Call with Error Handling

Now, combine the authentication and payload generation into a client class. This step handles the actual HTTP POST request to /api/v2/reporting/queues/stats.

Key considerations:

1. Headers: Must include Authorization: Bearer <token> and Content-Type: application/json.
2. Timeouts: Always set timeouts to prevent hanging threads.
3. Status Codes: Handle 401 (expired token), 403 (scope issues), 429 (rate limiting), and 5xx (server errors).

from datetime import datetime
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class CXoneQueueStatsClient:
    def __init__(self, auth: CXoneAuth):
        self.auth = auth
        self.base_url = auth.base_url
        self.endpoint = f"{self.base_url}/api/v2/reporting/queues/stats"

    def get_queue_stats(self, queue_ids: List[str]) -> Dict[str, Any]:
        """
        Fetches real-time statistics for a list of queues.
        
        Args:
            queue_ids: List of queue IDs.
            
        Returns:
            Parsed JSON response containing queue statistics.
        """
        token = self.auth.get_access_token()
        payload = build_queue_stats_payload(queue_ids)
        
        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

        try:
            logger.info(f"Fetching stats for {len(queue_ids)} queues...")
            response = requests.post(
                self.endpoint,
                headers=headers,
                json=payload,
                timeout=30
            )
            
            # Handle HTTP Errors
            if response.status_code == 401:
                logger.warning("Token expired or invalid. Refreshing...")
                self.auth.access_token = None # Force refresh
                return self.get_queue_stats(queue_ids) # Retry once
            elif response.status_code == 403:
                raise PermissionError("Access denied. Check if 'reporting:queues:read' scope is assigned.")
            elif response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 5))
                logger.warning(f"Rate limited. Retrying after {retry_after} seconds.")
                time.sleep(retry_after)
                return self.get_queue_stats(queue_ids)
            elif response.status_code >= 500:
                raise Exception(f"Server error: {response.status_code} - {response.text}")
            
            response.raise_for_status()
            return response.json()

        except requests.exceptions.Timeout:
            raise Exception("Request timed out. The CXone API may be under high load.")
        except requests.exceptions.ConnectionError:
            raise Exception("Network connection error. Check your internet connection.")
        except Exception as e:
            logger.error(f"Unexpected error: {e}")
            raise

Step 3: Process and Normalize Results

The response from CXone reporting APIs can be nested and complex. It typically returns a results array where each item corresponds to a grouping key (in this case, a Queue ID). Each result contains a metrics array with the values.

You need to map these raw metrics back to a readable structure.

from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime

@dataclass
class QueueMetric:
    queue_id: str
    queue_name: Optional[str]
    waiting: int
    answered: int
    abandoned: int
    avg_wait_time_ms: float
    service_level_pct: float
    timestamp: str

def parse_queue_response(raw_response: Dict[str, Any]) -> List[QueueMetric]:
    """
    Parses the raw JSON response from CXone into a list of QueueMetric objects.
    
    Args:
        raw_response: The JSON dictionary returned by the API.
        
    Returns:
        List of QueueMetric dataclass instances.
    """
    results = raw_response.get("results", [])
    parsed_metrics = []
    
    # Extract metadata if available for mapping IDs to Names
    # Note: The reporting API often returns IDs. You may need a separate call 
    # to /api/v2/queues to map IDs to Names if not included in the result.
    
    for result in results:
        grouping = result.get("grouping", {})
        queue_id = grouping.get("queueId")
        
        # Initialize metrics with defaults
        waiting = 0
        answered = 0
        abandoned = 0
        avg_wait_time = 0.0
        service_level = 0.0
        
        # Map metrics from the response array
        for metric in result.get("metrics", []):
            metric_name = metric.get("name")
            metric_value = metric.get("value")
            
            if metric_name == "waiting":
                waiting = int(metric_value) if metric_value else 0
            elif metric_name == "answered":
                answered = int(metric_value) if metric_value else 0
            elif metric_name == "abandoned":
                abandoned = int(metric_value) if metric_value else 0
            elif metric_name == "avgWaitTime":
                # CXone often returns time in milliseconds or seconds depending on config
                # Assuming milliseconds for this example
                avg_wait_time = float(metric_value) if metric_value else 0.0
            elif metric_name == "serviceLevel":
                service_level = float(metric_value) if metric_value else 0.0

        parsed_metrics.append(QueueMetric(
            queue_id=queue_id,
            queue_name=None, # Would require enrichment from Queue API
            waiting=waiting,
            answered=answered,
            abandoned=abandoned,
            avg_wait_time_ms=avg_wait_time,
            service_level_pct=service_level,
            timestamp=datetime.utcnow().isoformat()
        ))
        
    return parsed_metrics

Complete Working Example

This script combines all previous steps into a single executable file. It fetches the token, queries the stats, parses the results, and prints them in a human-readable format.

import os
import sys
import time
import requests
import logging
from typing import List, Dict, Any
from dataclasses import dataclass
from datetime import datetime
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# --- Data Models ---

@dataclass
class QueueMetric:
    queue_id: str
    queue_name: str
    waiting: int
    answered: int
    abandoned: int
    avg_wait_time_ms: float
    service_level_pct: float
    timestamp: str

# --- Authentication Module ---

class CXoneAuth:
    def __init__(self):
        self.api_key = os.getenv("CXONE_API_KEY")
        self.api_secret = os.getenv("CXONE_API_SECRET")
        self.base_url = os.getenv("CXONE_BASE_URL")
        
        if not all([self.api_key, self.api_secret, self.base_url]):
            raise ValueError("Missing required environment variables: CXONE_API_KEY, CXONE_API_SECRET, CXONE_BASE_URL")
            
        self.token_url = f"{self.base_url}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0.0

    def get_access_token(self) -> str:
        if self.access_token and time.time() < self.token_expiry - 60:
            return self.access_token

        payload = {
            "grant_type": "client_credentials",
            "client_id": self.api_key,
            "client_secret": self.api_secret
        }

        try:
            response = requests.post(self.token_url, data=payload, timeout=10)
            response.raise_for_status()
            
            token_data = response.json()
            self.access_token = token_data.get("access_token")
            expires_in = token_data.get("expires_in", 3600)
            self.token_expiry = time.time() + expires_in
            
            return self.access_token

        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise Exception("Authentication failed: Invalid API Key or Secret") from e
            raise Exception(f"HTTP Error {response.status_code}: {response.text}") from e
        except Exception as e:
            raise Exception(f"Authentication error: {e}") from e

# --- API Client Module ---

class CXoneQueueStatsClient:
    def __init__(self, auth: CXoneAuth):
        self.auth = auth
        self.base_url = auth.base_url
        self.endpoint = f"{self.base_url}/api/v2/reporting/queues/stats"

    def get_queue_stats(self, queue_ids: List[str]) -> List[QueueMetric]:
        token = self.auth.get_access_token()
        payload = self._build_payload(queue_ids)
        
        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

        try:
            logger.info(f"Fetching stats for queues: {queue_ids}")
            response = requests.post(
                self.endpoint,
                headers=headers,
                json=payload,
                timeout=30
            )
            
            if response.status_code == 401:
                logger.warning("Token expired. Refreshing...")
                self.auth.access_token = None
                return self.get_queue_stats(queue_ids)
            
            if response.status_code != 200:
                logger.error(f"API Error {response.status_code}: {response.text}")
                return []
            
            return self._parse_response(response.json())

        except requests.exceptions.RequestException as e:
            logger.error(f"Request failed: {e}")
            return []

    def _build_payload(self, queue_ids: List[str]) -> Dict[str, Any]:
        return {
            "grouping": [{"type": "queueId"}],
            "metrics": [
                "waiting",
                "answered",
                "abandoned",
                "avgWaitTime",
                "serviceLevel"
            ],
            "dateRange": {
                "type": "last24Hours" 
            },
            "filters": [
                {
                    "filterType": "queueId",
                    "values": queue_ids
                }
            ]
        }

    def _parse_response(self, raw_response: Dict[str, Any]) -> List[QueueMetric]:
        results = raw_response.get("results", [])
        metrics = []
        
        for result in results:
            grouping = result.get("grouping", {})
            queue_id = grouping.get("queueId", "Unknown")
            
            waiting = 0
            answered = 0
            abandoned = 0
            avg_wait = 0.0
            sl = 0.0
            
            for m in result.get("metrics", []):
                name = m.get("name")
                val = m.get("value")
                
                if name == "waiting": waiting = int(val) if val else 0
                elif name == "answered": answered = int(val) if val else 0
                elif name == "abandoned": abandoned = int(val) if val else 0
                elif name == "avgWaitTime": avg_wait = float(val) if val else 0.0
                elif name == "serviceLevel": sl = float(val) if val else 0.0

            metrics.append(QueueMetric(
                queue_id=queue_id,
                queue_name="N/A", # Enrich with Queue API if needed
                waiting=waiting,
                answered=answered,
                abandoned=abandoned,
                avg_wait_time_ms=avg_wait,
                service_level_pct=sl,
                timestamp=datetime.utcnow().isoformat()
            ))
            
        return metrics

# --- Main Execution ---

def main():
    try:
        # Initialize Auth
        auth = CXoneAuth()
        
        # Initialize Client
        client = CXoneQueueStatsClient(auth)
        
        # Define Queue IDs to monitor
        # Replace these with actual Queue IDs from your CXone environment
        TARGET_QUEUE_IDS = [
            "12345678-1234-1234-1234-123456789012",
            "87654321-4321-4321-4321-210987654321"
        ]
        
        if not TARGET_QUEUE_IDS:
            logger.error("No queue IDs defined. Update TARGET_QUEUE_IDS in the script.")
            return

        # Fetch Stats
        queue_metrics = client.get_queue_stats(TARGET_QUEUE_IDS)
        
        # Display Results
        if not queue_metrics:
            logger.info("No metrics returned.")
            return

        print("\n--- CXone Real-Time Queue Stats ---")
        print(f"{'Queue ID':<35} {'Waiting':<10} {'Answered':<10} {'Abandoned':<10} {'Avg Wait (ms)':<15} {'SL %':<10}")
        print("-" * 100)
        
        for m in queue_metrics:
            print(
                f"{m.queue_id:<35} {m.waiting:<10} {m.answered:<10} {m.abandoned:<10} {m.avg_wait_time_ms:<15.2f} {m.service_level_pct:<10.2f}"
            )
            
    except Exception as e:
        logger.critical(f"Fatal error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

• Cause: The OAuth token does not have the required scope.
• Fix: Go to the CXone Admin Console > Security > API Keys. Edit your API key and ensure the reporting:queues:read scope is checked. You must re-generate the token after adding scopes.

Error: 401 Unauthorized

• Cause: The API Key or Secret is incorrect, or the token has expired.
• Fix: Verify the CXONE_API_KEY and CXONE_API_SECRET in your .env file. Ensure there are no trailing spaces. The code above includes a retry mechanism for token expiration, but if it fails repeatedly, check the validity of the credentials.

Error: Empty Results

• Cause: The Queue IDs provided do not exist, or the queues have had no activity in the specified date range.
• Fix: Verify the Queue IDs using the /api/v2/queues endpoint. If the queues are new, they may not have historical data. For real-time waiting counts, ensure there are actually calls in the queue.

Error: 429 Too Many Requests

• Cause: You have exceeded the rate limit for the reporting API.
• Fix: Implement exponential backoff. The code above includes a basic retry with Retry-After header parsing. For high-frequency polling, consider increasing the interval between requests or using CXone Webhooks for event-driven updates instead of polling.

Official References

• NICE CXone Reporting API Documentation
• OAuth 2.0 Client Credentials Flow
• Python Requests Library