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

What You Will Build

• One sentence: This tutorial shows how to query the NICE CXone v2 Reporting API to retrieve current queue metrics, such as agents logged in, calls in queue, and average wait time.
• One sentence: The code uses the NICE CXone Python SDK (cxone-api) to handle authentication and API requests.
• One sentence: The implementation is written in Python 3.9+ using async/await for non-blocking I/O.

Prerequisites

• OAuth Client Type: Service Account (Client Credentials Grant) or User Account (Authorization Code Grant with PKCE). This tutorial uses Client Credentials for server-to-server communication.
• Required Scopes: reporting:view is mandatory for reading queue statistics. If you need agent-level details, agent:view may also be required depending on the specific data points.
• SDK Version: cxone-api v2.0.0 or later.
• Language/Runtime: Python 3.9+.
• External Dependencies:
  • cxone-api: The official NICE CXone Python SDK.
  • python-dotenv: For managing environment variables securely.
  • httpx: Used internally by the SDK for HTTP requests, but useful for debugging raw calls if needed.

Install the dependencies:

pip install cxone-api python-dotenv

Authentication Setup

NICE CXone uses OAuth 2.0 for authentication. The Python SDK handles the token exchange and refresh automatically once configured. You must provide your Client ID, Client Secret, and Tenant ID.

Create a .env file in your project root:

CXONE_CLIENT_ID=your_client_id_here
CXONE_CLIENT_SECRET=your_client_secret_here
CXONE_TENANT_ID=your_tenant_id_here

Initialize the SDK client. The cxone_api library requires you to set the global configuration before creating any API instances.

import os
from cxone_api import Configuration, ApiClient
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

def get_cxone_api_client() -> ApiClient:
    """
    Configures and returns an authenticated CXone API Client.
    """
    # 1. Create Configuration Object
    configuration = Configuration(
        client_id=os.getenv("CXONE_CLIENT_ID"),
        client_secret=os.getenv("CXONE_CLIENT_SECRET"),
        tenant_id=os.getenv("CXONE_TENANT_ID")
    )

    # 2. Set the OAuth Token URL explicitly if needed, 
    # though the SDK defaults to the correct CXone Auth endpoint.
    # configuration.host = "https://api.nicecxone.com" 

    # 3. Create the API Client
    # The ApiClient handles OAuth token caching and refreshing automatically.
    api_client = ApiClient(configuration)
    
    return api_client

Error Handling Note: If you receive a 401 Unauthorized error here, verify that your Client ID and Secret are correct and that the Service Account has not been disabled in the CXone Admin Portal.

Implementation

Step 1: Configure the Reporting API Client

The NICE CXone SDK organizes APIs by domain. Queue statistics reside in the ReportingApi module. We will instantiate this client using the ApiClient created in the previous step.

from cxone_api.reporting import ReportingApi
from cxone_api.reporting.models import QueryQueueStatsRequest

def get_reporting_api(api_client: ApiClient) -> ReportingApi:
    """
    Returns an instance of the ReportingApi.
    """
    return ReportingApi(api_client)

Step 2: Construct the Query Payload

The endpoint for real-time queue stats is POST /api/v2/analytics/conversations/details/query. However, for aggregated queue metrics, the specific endpoint is POST /api/v2/analytics/queues/stats/query.

We need to construct a QueryQueueStatsRequest object. This request requires:

1. from_date: The start of the time window.
2. to_date: The end of the time window.
3. interval: The granularity of the data (e.g., PT1M for 1-minute intervals).
4. queue_ids: A list of specific queue IDs to query.
5. metrics: The specific metrics you want to retrieve.

Common metrics for real-time monitoring include:

• agents_logged_in: Number of agents available.
• calls_in_queue: Number of interactions waiting.
• average_wait_time: Average time spent in queue.
• abandoned_count: Number of abandoned interactions.

Important: For “real-time” data, you must set the to_date to the current time and from_date to a short window in the past (e.g., 5 minutes ago). The interval should be small, such as PT1M.

from datetime import datetime, timedelta
from dateutil.tz import tzutc
import pytz

def build_queue_stats_request(queue_ids: list[str]) -> QueryQueueStatsRequest:
    """
    Constructs the request body for querying queue stats.
    
    Args:
        queue_ids: List of UUID strings representing the queues to query.
        
    Returns:
        QueryQueueStatsRequest object.
    """
    # Define time window: Last 5 minutes
    end_time = datetime.now(tzutc())
    start_time = end_time - timedelta(minutes=5)
    
    # Format dates as ISO 8601 strings with timezone
    from_date = start_time.isoformat()
    to_date = end_time.isoformat()
    
    # Define interval: 1 minute (PT1M)
    # Note: For true real-time snapshots, you might query a very small window 
    # or use the 'snapshot' logic if available in newer SDK versions.
    # However, standard analytics queries require an interval.
    interval = "PT1M"
    
    # Select metrics
    metrics = [
        "agents_logged_in",
        "agents_available",
        "calls_in_queue",
        "average_wait_time",
        "abandoned_count",
        "service_level"
    ]
    
    # Build the request object
    request_body = QueryQueueStatsRequest(
        from_date=from_date,
        to_date=to_date,
        interval=interval,
        queue_ids=queue_ids,
        metrics=metrics
    )
    
    return request_body

Step 3: Execute the Query and Process Results

Now we combine the client and the request to fetch the data. The SDK method is post_analytics_queues_stats_query.

import asyncio
from typing import List, Dict, Any
from cxone_api.reporting.models import QueryQueueStatsResponse

async def fetch_queue_stats(api_client: ApiClient, queue_ids: List[str]) -> Dict[str, Any]:
    """
    Fetches real-time queue statistics for the given queue IDs.
    
    Args:
        api_client: Authenticated CXone API Client.
        queue_ids: List of queue UUIDs.
        
    Returns:
        Dictionary containing processed queue stats.
    """
    reporting_api = get_reporting_api(api_client)
    request_body = build_queue_stats_request(queue_ids)
    
    try:
        # Execute the API call
        # Note: The SDK method is synchronous in the base definition, 
        # but we wrap it in asyncio.to_thread for async compatibility 
        # if running in an async environment.
        response = await asyncio.to_thread(
            reporting_api.post_analytics_queues_stats_query, 
            body=request_body
        )
        
        return process_response(response)
        
    except Exception as e:
        print(f"Error fetching queue stats: {e}")
        raise

def process_response(response: QueryQueueStatsResponse) -> Dict[str, Any]:
    """
    Parses the API response into a usable dictionary.
    """
    results = {}
    
    if not response.entities:
        return results
        
    for entity in response.entities:
        queue_id = entity.queue_id
        results[queue_id] = {
            "name": entity.queue_name,
            "metrics": {}
        }
        
        # The response contains a 'data' array with time-series points
        if entity.data:
            # For real-time, we usually care about the latest data point
            latest_point = entity.data[-1] # Assuming data is sorted by time
            
            # Map metric keys to values
            for metric_key in latest_point.keys():
                results[queue_id]["metrics"][metric_key] = latest_point[metric_key]
                
    return results

Step 4: Handling Pagination and Large Datasets

The CXone Reporting API supports pagination via next_page_token. If you query many queues or a long time period, the response may be split. For real-time monitoring of a few queues, pagination is rarely triggered. However, for robustness, we should implement a loop.

async def fetch_queue_stats_with_pagination(api_client: ApiClient, queue_ids: List[str]) -> Dict[str, Any]:
    """
    Fetches queue stats handling potential pagination.
    """
    reporting_api = get_reporting_api(api_client)
    request_body = build_queue_stats_request(queue_ids)
    
    all_results = {}
    next_page_token = None
    
    max_pages = 10 # Safety limit to prevent infinite loops
    
    for _ in range(max_pages):
        try:
            # Execute the API call
            response = await asyncio.to_thread(
                reporting_api.post_analytics_queues_stats_query, 
                body=request_body
            )
            
            # Merge results
            if response.entities:
                for entity in response.entities:
                    queue_id = entity.queue_id
                    if queue_id not in all_results:
                        all_results[queue_id] = {
                            "name": entity.queue_name,
                            "metrics": {}
                        }
                    
                    if entity.data:
                        # Update with latest data point
                        latest_point = entity.data[-1]
                        for metric_key in latest_point.keys():
                            all_results[queue_id]["metrics"][metric_key] = latest_point[metric_key]
            
            # Check for next page
            if response.next_page_token:
                request_body.next_page_token = response.next_page_token
                continue
            else:
                break
                
        except Exception as e:
            print(f"Error during pagination fetch: {e}")
            raise

    return all_results

Complete Working Example

Below is the full, copy-pasteable script. Save this as cxone_queue_monitor.py.

import os
import asyncio
import sys
from datetime import datetime, timedelta
from dateutil.tz import tzutc
from typing import List, Dict, Any

from cxone_api import Configuration, ApiClient
from cxone_api.reporting import ReportingApi
from cxone_api.reporting.models import QueryQueueStatsRequest
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

def get_cxone_api_client() -> ApiClient:
    """
    Configures and returns an authenticated CXone API Client.
    """
    client_id = os.getenv("CXONE_CLIENT_ID")
    client_secret = os.getenv("CXONE_CLIENT_SECRET")
    tenant_id = os.getenv("CXONE_TENANT_ID")

    if not all([client_id, client_secret, tenant_id]):
        raise ValueError("Missing CXONE_CLIENT_ID, CXONE_CLIENT_SECRET, or CXONE_TENANT_ID in .env")

    configuration = Configuration(
        client_id=client_id,
        client_secret=client_secret,
        tenant_id=tenant_id
    )
    
    return ApiClient(configuration)

def build_queue_stats_request(queue_ids: List[str]) -> QueryQueueStatsRequest:
    """
    Constructs the request body for querying queue stats.
    """
    # Define time window: Last 5 minutes
    end_time = datetime.now(tzutc())
    start_time = end_time - timedelta(minutes=5)
    
    from_date = start_time.isoformat()
    to_date = end_time.isoformat()
    interval = "PT1M"
    
    metrics = [
        "agents_logged_in",
        "agents_available",
        "calls_in_queue",
        "average_wait_time",
        "abandoned_count",
        "service_level"
    ]
    
    return QueryQueueStatsRequest(
        from_date=from_date,
        to_date=to_date,
        interval=interval,
        queue_ids=queue_ids,
        metrics=metrics
    )

async def fetch_queue_stats(api_client: ApiClient, queue_ids: List[str]) -> Dict[str, Any]:
    """
    Fetches real-time queue statistics for the given queue IDs.
    """
    reporting_api = ReportingApi(api_client)
    request_body = build_queue_stats_request(queue_ids)
    
    try:
        # Execute the API call
        response = await asyncio.to_thread(
            reporting_api.post_analytics_queues_stats_query, 
            body=request_body
        )
        
        return process_response(response)
        
    except Exception as e:
        print(f"Error fetching queue stats: {e}")
        raise

def process_response(response) -> Dict[str, Any]:
    """
    Parses the API response into a usable dictionary.
    """
    results = {}
    
    if not response.entities:
        return results
        
    for entity in response.entities:
        queue_id = entity.queue_id
        results[queue_id] = {
            "name": entity.queue_name,
            "metrics": {}
        }
        
        if entity.data:
            # For real-time, we usually care about the latest data point
            latest_point = entity.data[-1] 
            
            # Map metric keys to values
            for metric_key in latest_point.keys():
                results[queue_id]["metrics"][metric_key] = latest_point[metric_key]
                
    return results

async def main():
    # Example Queue IDs (Replace with actual UUIDs from your CXone instance)
    # You can find these in the Admin Portal > IVR > Queues
    QUEUE_IDS = [
        "00000000-0000-0000-0000-000000000001", 
        "00000000-0000-0000-0000-000000000002"
    ]
    
    # If no IDs provided, exit with instruction
    if QUEUE_IDS[0].startswith("00000000"):
        print("Please replace QUEUE_IDS in the code with actual CXone Queue UUIDs.")
        sys.exit(1)

    try:
        api_client = get_cxone_api_client()
        stats = await fetch_queue_stats(api_client, QUEUE_IDS)
        
        print("=== CXone Queue Real-Time Stats ===")
        for q_id, data in stats.items():
            print(f"\nQueue: {data['name']} (ID: {q_id})")
            metrics = data['metrics']
            print(f"  Agents Logged In: {metrics.get('agents_logged_in', 'N/A')}")
            print(f"  Agents Available: {metrics.get('agents_available', 'N/A')}")
            print(f"  Calls in Queue:   {metrics.get('calls_in_queue', 'N/A')}")
            print(f"  Avg Wait Time:    {metrics.get('average_wait_time', 'N/A')} seconds")
            print(f"  Abandoned Calls:  {metrics.get('abandoned_count', 'N/A')}")
            
    except Exception as e:
        print(f"Fatal error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    asyncio.run(main())

Common Errors & Debugging

Error: 401 Unauthorized

What causes it: Invalid Client ID, Client Secret, or Tenant ID. The Service Account may be disabled.
How to fix it:

1. Check your .env file for typos.
2. Log into the CXone Admin Portal.
3. Go to Admin > Security > OAuth Clients.
4. Verify the Client ID matches.
5. Regenerate the Client Secret if it was compromised or expired.
6. Ensure the Service Account has the reporting:view scope assigned.

Error: 403 Forbidden

What causes it: The Service Account lacks the required permissions.
How to fix it:

1. In the CXone Admin Portal, navigate to Admin > Security > OAuth Clients.
2. Select your client.
3. Go to the Scopes tab.
4. Ensure reporting:view is checked.
5. If the error persists, check if the Service Account has access to the specific Queues. Some tenants restrict data access by region or department.

Error: 429 Too Many Requests

What causes it: You are hitting the API rate limit. CXone enforces strict rate limits on reporting APIs.
How to fix it:

1. Implement exponential backoff in your retry logic.
2. Cache the results locally. Real-time queue stats do not change every second. Polling every 30-60 seconds is usually sufficient.
3. Reduce the number of queues queried per request. Batch requests if possible, but respect the payload size limit.

Code Example for Retry Logic:

import time

async def fetch_with_retry(api_client, queue_ids, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await fetch_queue_stats(api_client, queue_ids)
        except Exception as e:
            if "429" in str(e) and attempt < max_retries - 1:
                wait_time = 2 ** attempt # Exponential backoff: 1s, 2s, 4s
                print(f"Rate limited. Waiting {wait_time} seconds...")
                await asyncio.sleep(wait_time)
            else:
                raise

Error: Empty Response (entities is null)

What causes it:

1. The Queue IDs are invalid.
2. The time window (from_date to to_date) contains no data (e.g., querying a queue that had no activity in the last 5 minutes).
3. The interval is too small for the data granularity.
   How to fix it:
4. Verify Queue UUIDs using the GET /api/v2/queues endpoint.
5. Expand the time window to PT1H (1 hour) to see if data exists.
6. Ensure the queues are active and not paused.

Official References

• NICE CXone Reporting API Documentation
• Query Queue Stats Endpoint
• CXone Python SDK on PyPI
• OAuth 2.0 Client Credentials Flow