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

What You Will Build

• A Python script that authenticates with NICE CXone and retrieves current wait times, agent counts, and interaction volumes for specific queues.
• This tutorial uses the NICE CXone REST API (specifically the v2 Reporting endpoints).
• The implementation uses Python 3.9+ with the requests library for HTTP handling and pydantic for data validation.

Prerequisites

• OAuth Client Type: Confidential Client (Client Credentials Grant). You need a CXone Integration with admin or reporting privileges.
• Required Scopes: reporting:read is mandatory for accessing queue statistics.
• SDK Version: Direct REST API usage (no official Python SDK is maintained by NICE for v2 reporting, so raw HTTP is the standard approach).
• Language/Runtime: Python 3.9 or higher.
• External Dependencies:
  • requests (for HTTP calls)
  • pydantic (for robust response parsing)
  • tenacity (for exponential backoff retry logic)

Install dependencies via pip:

pip install requests pydantic tenacity

Authentication Setup

NICE CXone uses OAuth 2.0 Client Credentials Grant for server-to-server communication. You must exchange your Client ID and Client Secret for an access token before making any API calls.

The token endpoint is located at https://login.nicecxone.com/oauth2/token.

Below is the authentication logic. This function caches the token to avoid unnecessary re-authentication within the token’s validity window.

import requests
import time
from typing import Optional

class CxoneAuth:
    def __init__(self, client_id: str, client_secret: str, environment: str = "prod"):
        """
        Initialize CXone Authentication.
        
        :param client_id: Your OAuth Client ID
        :param client_secret: Your OAuth Client Secret
        :param environment: 'prod' or 'dev' determines the login domain
        """
        self.client_id = client_id
        self.client_secret = client_secret
        self.environment = environment
        self._token: Optional[str] = None
        self._token_expiry: float = 0
        
        if environment == "dev":
            self.login_url = "https://login.nicecxone.com/oauth2/token"
            self.api_base = "https://api.nicecxone.com/v2"
        else:
            self.login_url = "https://login.nicecxone.com/oauth2/token"
            self.api_base = "https://api.nicecxone.com/v2"

    def get_access_token(self) -> str:
        """
        Retrieves an OAuth access token.
        Returns cached token if not expired.
        """
        if self._token and time.time() < self._token_expiry:
            return self._token

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

        try:
            response = requests.post(self.login_url, data=payload)
            response.raise_for_status()
            
            data = response.json()
            self._token = data["access_token"]
            # Expire 60 seconds before actual expiry to prevent edge-case 401s
            self._token_expiry = time.time() + (data["expires_in"] - 60)
            
            return self._token
            
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise Exception("Authentication failed: Invalid Client ID or Secret.") from e
            elif response.status_code == 400:
                raise Exception("Bad Request: Check grant_type and payload format.") from e
            else:
                raise Exception(f"Authentication error: {e}") from e
        except requests.exceptions.ConnectionError:
            raise Exception("Failed to connect to CXone Login Service.")

Implementation

Step 1: Configure the Reporting Endpoint

The NICE CXone v2 Reporting API exposes real-time queue metrics via the GET /reporting/queues/stats endpoint. Unlike historical reporting, this endpoint returns the current state of the system.

Key parameters:

• queueIds: A comma-separated list of queue IDs (UUIDs). If omitted, it returns stats for all queues the user has access to.
• interval: For real-time stats, this is often ignored or set to current. However, the API typically expects a time window for aggregation. For strict real-time snapshots, we rely on the default behavior of the stats endpoint which aggregates over the last few seconds/minutes depending on the metric.

OAuth Scope Required: reporting:read

import requests
from typing import List, Dict, Any

class CxoneQueueReporter:
    def __init__(self, auth: CxoneAuth):
        self.auth = auth
        self.headers = {
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

    def get_queue_stats(self, queue_ids: List[str] = None) -> Dict[str, Any]:
        """
        Fetches real-time statistics for specified queues.
        
        :param queue_ids: List of Queue UUIDs. If None, fetches all accessible queues.
        :return: Dictionary containing queue statistics.
        """
        endpoint = f"{self.auth.api_base}/reporting/queues/stats"
        
        params = {}
        if queue_ids:
            # CXone API expects a comma-separated string for IDs
            params["queueIds"] = ",".join(queue_ids)
            
        # Explicitly request current state if supported by specific sub-endpoints,
        # though the main stats endpoint handles this implicitly.
        
        token = self.auth.get_access_token()
        self.headers["Authorization"] = f"Bearer {token}"

        try:
            response = requests.get(endpoint, headers=self.headers, params=params, timeout=10)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            if response.status_code == 403:
                raise Exception("Forbidden: Ensure your integration has 'reporting:read' scope.") from e
            elif response.status_code == 401:
                raise Exception("Unauthorized: Token may be expired or invalid.") from e
            else:
                raise Exception(f"API Error: {e}") from e

Step 2: Parse and Normalize the Response

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

{
  "results": [
    {
      "queueId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
      "queueName": "Sales Support",
      "metrics": {
        "waitTime": 12.5,
        "waitTimePercentile90": 45.0,
        "activeAgents": 5,
        "availableAgents": 2,
        "interactionsInQueue": 3,
        "interactionsHandled": 150
      }
    }
  ]
}

We must parse this into a usable Python object. We will use pydantic to define the structure and ensure type safety.

from pydantic import BaseModel, Field
from typing import List, Optional

class QueueMetrics(BaseModel):
    waitTime: float = 0.0
    waitTimePercentile90: float = 0.0
    activeAgents: int = 0
    availableAgents: int = 0
    interactionsInQueue: int = 0
    interactionsHandled: int = 0
    abandonedInteractions: int = 0

class QueueStat(BaseModel):
    queueId: str
    queueName: str
    metrics: QueueMetrics

class CxoneQueueResponse(BaseModel):
    results: List[QueueStat]

def parse_queue_stats(raw_json: Dict[str, Any]) -> List[QueueStat]:
    """
    Validates and parses the raw JSON response into Pydantic models.
    """
    try:
        response_model = CxoneQueueResponse(**raw_json)
        return response_model.results
    except Exception as e:
        raise ValueError(f"Failed to parse CXone response: {e}") from e

Step 3: Implement Retry Logic for Rate Limiting

CXone enforces rate limits. Hitting a 429 (Too Many Requests) is common when polling multiple queues or running frequent checks. We will wrap the call in a retry decorator using tenacity.

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import requests.exceptions

class ResilientCxoneReporter(CxoneQueueReporter):
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type(requests.exceptions.HTTPError)
    )
    def get_queue_stats_with_retry(self, queue_ids: List[str] = None) -> List[QueueStat]:
        """
        Fetches queue stats with automatic retry on 429/5xx errors.
        """
        raw_data = self.get_queue_stats(queue_ids)
        return parse_queue_stats(raw_data)

Complete Working Example

This script combines authentication, API calling, and data parsing into a single executable module. It demonstrates how to fetch stats for a specific queue and print the key metrics.

import os
import sys
import time
import requests
from typing import List, Dict, Any, Optional
from pydantic import BaseModel
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

# --- Models ---

class QueueMetrics(BaseModel):
    waitTime: float = 0.0
    waitTimePercentile90: float = 0.0
    activeAgents: int = 0
    availableAgents: int = 0
    interactionsInQueue: int = 0
    interactionsHandled: int = 0

class QueueStat(BaseModel):
    queueId: str
    queueName: str
    metrics: QueueMetrics

class CxoneQueueResponse(BaseModel):
    results: List[QueueStat]

# --- Authentication ---

class CxoneAuth:
    def __init__(self, client_id: str, client_secret: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self._token: Optional[str] = None
        self._token_expiry: float = 0
        self.login_url = "https://login.nicecxone.com/oauth2/token"
        self.api_base = "https://api.nicecxone.com/v2"

    def get_access_token(self) -> str:
        if self._token and time.time() < self._token_expiry:
            return self._token

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

        try:
            response = requests.post(self.login_url, data=payload, timeout=10)
            response.raise_for_status()
            data = response.json()
            self._token = data["access_token"]
            self._token_expiry = time.time() + (data["expires_in"] - 60)
            return self._token
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise Exception("Auth Failed: Check Client ID/Secret") from e
            raise Exception(f"Auth Error: {e}") from e

# --- Reporting Client ---

class CxoneQueueReporter:
    def __init__(self, auth: CxoneAuth):
        self.auth = auth
        self.headers = {
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

    def _get_base_headers(self) -> Dict[str, str]:
        token = self.auth.get_access_token()
        return {**self.headers, "Authorization": f"Bearer {token}"}

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type(requests.exceptions.HTTPError)
    )
    def get_queue_stats(self, queue_ids: List[str] = None) -> List[QueueStat]:
        endpoint = f"{self.auth.api_base}/reporting/queues/stats"
        params = {}
        if queue_ids:
            params["queueIds"] = ",".join(queue_ids)

        try:
            response = requests.get(
                endpoint, 
                headers=self._get_base_headers(), 
                params=params, 
                timeout=15
            )
            response.raise_for_status()
            raw_json = response.json()
            
            # Parse using Pydantic
            parsed_response = CxoneQueueResponse(**raw_json)
            return parsed_response.results
            
        except requests.exceptions.HTTPError as e:
            if response.status_code == 403:
                print("Error: 403 Forbidden. Ensure 'reporting:read' scope is assigned to the integration.")
            elif response.status_code == 429:
                print("Warning: Rate limited (429). Retrying...")
            raise e

# --- Main Execution ---

def main():
    # Load credentials from environment variables
    CLIENT_ID = os.getenv("CXONE_CLIENT_ID")
    CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET")
    
    # Example Queue ID (Replace with a real Queue ID from your environment)
    # To find a queue ID: Admin > IVR > Queues > Select Queue > Copy ID from URL or API
    TARGET_QUEUE_ID = os.getenv("TARGET_QUEUE_ID") 

    if not CLIENT_ID or not CLIENT_SECRET:
        print("Error: Set CXONE_CLIENT_ID and CXONE_CLIENT_SECRET environment variables.")
        sys.exit(1)

    try:
        # 1. Initialize Auth
        auth = CxoneAuth(CLIENT_ID, CLIENT_SECRET)
        
        # 2. Initialize Reporter
        reporter = CxoneQueueReporter(auth)
        
        # 3. Fetch Stats
        # If TARGET_QUEUE_ID is set, fetch only that queue. Otherwise fetch all.
        queue_ids = [TARGET_QUEUE_ID] if TARGET_QUEUE_ID else None
        
        print(f"Fetching real-time stats for {'specific queue' if queue_ids else 'all queues'}...")
        stats = reporter.get_queue_stats(queue_ids)
        
        # 4. Display Results
        if not stats:
            print("No queue statistics returned. Check if the queue IDs are valid and accessible.")
            return

        for stat in stats:
            print("-" * 50)
            print(f"Queue: {stat.queueName} (ID: {stat.queueId})")
            print(f"  Agents Available: {stat.metrics.availableAgents}")
            print(f"  Agents Active:    {stat.metrics.activeAgents}")
            print(f"  Interactions in Queue: {stat.metrics.interactionsInQueue}")
            print(f"  Avg Wait Time (sec):   {stat.metrics.waitTime:.2f}")
            print(f"  90th %ile Wait (sec):  {stat.metrics.waitTimePercentile90:.2f}")
            print(f"  Total Handled:         {stat.metrics.interactionsHandled}")
            print("-" * 50)

    except Exception as e:
        print(f"Fatal Error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

What causes it:
The OAuth integration used to generate the token does not have the reporting:read scope. Alternatively, the integration is not assigned to an organization or site that has access to the queue.

How to fix it:

1. Log into the CXone Admin Console.
2. Navigate to Integrations.
3. Select your integration.
4. Go to the Scopes tab.
5. Ensure reporting:read is checked.
6. Save and regenerate the Client Secret if necessary (though scope changes usually apply immediately for new tokens).

Error: 422 Unprocessable Entity

What causes it:
The queueIds parameter contains invalid UUIDs or a queue ID that does not exist in the tenant.

How to fix it:

1. Validate that the string passed to queueIds is a valid UUID format.
2. Use the GET /api/v2/queues endpoint to verify the queue ID exists.
3. Ensure the integration has access to the specific site containing the queue.

Error: Empty Results Array

What causes it:
The API call succeeded, but no data was returned. This happens if:

1. The specified queue is empty and has no historical data for the requested window (though real-time stats usually return zeroed metrics).
2. The integration lacks visibility into the queue’s site.

How to fix it:

1. Check the Sites assignment of the integration.
2. Verify the queue is active.
3. If querying specific IDs, double-check the ID string for typos.

Official References

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