Query NICE CXone Agent State History via REST API

Query NICE CXone Agent State History via REST API

What You Will Build

  • A Python script that retrieves detailed agent state changes (login, logout, ready, not ready, wrap-up) for a specific agent over the last 24 hours.
  • This tutorial uses the NICE CXone Reporting API (v2) specifically the agent-state-history endpoint.
  • The implementation covers Python using the requests library, including proper OAuth 2.0 authentication, pagination handling, and error resilience.

Prerequisites

Before writing code, ensure you have the following configured in your NICE CXone instance:

  • OAuth Client: You need an OAuth client with the reporting scope. This is typically a “Client Credentials” grant type client.
  • Required Scopes: The reporting scope is mandatory. If you need to filter by specific queues or skills, ensure the client has read access to those resources, though state history generally requires only reporting permissions.
  • Agent ID: You must know the agentId (UUID) of the agent whose history you wish to retrieve.
  • Python Environment: Python 3.8+ with pip.
  • Dependencies: Install the requests library.
    pip install requests
    

Authentication Setup

NICE CXone uses OAuth 2.0 for authentication. For backend integrations like reporting queries, the Client Credentials Grant is the standard flow. This flow exchanges your client ID and secret for an access token.

The token expires after a set duration (typically one hour). In a production environment, you should cache the token and refresh it before it expires. For this tutorial, we will implement a simple function to fetch a fresh token.

Endpoint: POST https://{your-domain}.niceincontact.com/oauth/token
Content-Type: application/x-www-form-urlencoded

import requests
import time
from typing import Optional

# Configuration - Replace with your actual values
CXONE_DOMAIN = "your-instance"  # e.g., "mycompany"
CLIENT_ID = "your_client_id"
CLIENT_SECRET = "your_client_secret"

def get_access_token() -> str:
    """
    Retrieves an OAuth 2.0 access token using Client Credentials grant.
    """
    token_url = f"https://{CXONE_DOMAIN}.niceincontact.com/oauth/token"
    
    payload = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET
    }
    
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    
    try:
        response = requests.post(token_url, data=payload, headers=headers)
        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

if __name__ == "__main__":
    token = get_access_token()
    print(f"Token acquired (first 10 chars): {token[:10]}...")

Critical Note: The access_token returned is a JWT. You will use this in the Authorization: Bearer <token> header for all subsequent API calls.

Implementation

Step 1: Constructing the Date Range and Request Parameters

The CXone Reporting API requires specific date formats and parameter structures. To query the last 24 hours, we must calculate the startTime and endTime in ISO 8601 format (YYYY-MM-DDTHH:mm:ssZ).

The endpoint for agent state history is:
GET https://{your-domain}.niceincontact.com/api/reporting/v2/agent-state-history

Required Query Parameters:

  • startTime: The start of the reporting period (ISO 8601).
  • endTime: The end of the reporting period (ISO 8601).
  • agentId: The UUID of the agent.
  • interval: The aggregation interval. For detailed history, MINUTE is common, but for state changes, SECOND or no aggregation (depending on specific sub-endpoint behavior) might be needed. However, the agent-state-history endpoint typically returns discrete events. We will use MINUTE as a safe default for aggregation if events are dense, but note that state history often returns a list of state transitions.

Optional but Recommended Parameters:

  • pageSize: Number of records per page (max 1000).
  • pageNumber: For pagination.
from datetime import datetime, timedelta, timezone
import json

def get_last_24_hours_range():
    """
    Calculates the start and end times for the last 24 hours in ISO 8601 format.
    """
    end_time = datetime.now(timezone.utc)
    start_time = end_time - timedelta(hours=24)
    
    # Format to ISO 8601 with 'Z' suffix for UTC
    start_iso = start_time.strftime("%Y-%m-%dT%H:%M:%SZ")
    end_iso = end_time.strftime("%Y-%m-%dT%H:%M:%SZ")
    
    return start_iso, end_iso

# Example Usage
start, end = get_last_24_hours_range()
print(f"Start: {start}")
print(f"End: {end}")

Step 2: Executing the API Call with Pagination

The CXone Reporting API uses cursor-based or offset-based pagination. The agent-state-history endpoint typically returns a nextPageToken or requires manual page incrementing. We will implement a loop to fetch all pages until no more data is returned.

Endpoint: GET https://{your-domain}.niceincontact.com/api/reporting/v2/agent-state-history
Headers:

  • Authorization: Bearer <access_token>
  • Content-Type: application/json

Request Body: Some reporting endpoints accept POST bodies for complex filters, but agent-state-history is primarily GET with query parameters. We will use GET.

import requests

AGENT_ID = "00000000-0000-0000-0000-000000000000" # Replace with actual Agent UUID

def fetch_agent_state_history(agent_id: str, start_time: str, end_time: str, token: str, max_pages: int = 100) -> list:
    """
    Fetches agent state history with pagination support.
    
    Args:
        agent_id: The UUID of the agent.
        start_time: ISO 8601 start time.
        end_time: ISO 8601 end time.
        token: OAuth access token.
        max_pages: Safety break to prevent infinite loops.
        
    Returns:
        List of state history records.
    """
    base_url = f"https://{CXONE_DOMAIN}.niceincontact.com/api/reporting/v2/agent-state-history"
    
    all_records = []
    page_number = 1
    page_size = 1000
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    params = {
        "startTime": start_time,
        "endTime": end_time,
        "agentId": agent_id,
        "pageSize": page_size,
        "pageNumber": page_number
    }
    
    print(f"Starting fetch for Agent {agent_id} from {start_time} to {end_time}")
    
    while page_number <= max_pages:
        try:
            response = requests.get(base_url, headers=headers, params=params)
            
            # Handle HTTP Errors
            if response.status_code == 401:
                raise Exception("Unauthorized: Token may be expired. Refresh token.")
            elif response.status_code == 403:
                raise Exception("Forbidden: Client lacks 'reporting' scope.")
            elif response.status_code == 404:
                print("No data found for this agent in the specified time range.")
                break
            elif response.status_code == 429:
                # Rate Limiting: Wait and retry
                wait_time = int(response.headers.get('Retry-After', 5))
                print(f"Rate limited. Waiting {wait_time} seconds...")
                time.sleep(wait_time)
                continue
            else:
                response.raise_for_status()
                
            data = response.json()
            
            # Check if there are records
            records = data.get("records", [])
            
            if not records:
                print("No more records found.")
                break
                
            all_records.extend(records)
            print(f"Fetched page {page_number}: {len(records)} records. Total: {len(all_records)}")
            
            # Check for next page
            # CXone v2 API often returns 'nextPageToken' or relies on pageNumber increment
            # If the response has fewer records than page_size, we are likely at the end
            if len(records) < page_size:
                print("Last page detected (fewer records than page_size).")
                break
                
            # Prepare next page
            page_number += 1
            params["pageNumber"] = page_number
            
        except requests.exceptions.HTTPError as e:
            print(f"HTTP Error on page {page_number}: {e}")
            break
        except requests.exceptions.RequestException as e:
            print(f"Network error: {e}")
            break
            
    return all_records

Step 3: Processing and Interpreting Results

The response body contains a records array. Each record represents a state change event. Key fields include:

  • timestamp: When the state change occurred.
  • state: The new state (e.g., READY, NOT_READY, WRAP_UP, LOGIN, LOGOUT).
  • subState: Optional detailed state (e.g., reason codes for NOT_READY).
  • duration: How long the agent was in the previous state (in milliseconds).
def process_state_records(records: list):
    """
    Parses and prints a summary of agent state history.
    """
    if not records:
        print("No records to process.")
        return

    print("\n--- Agent State History Summary ---")
    for record in records:
        timestamp = record.get("timestamp", "Unknown")
        state = record.get("state", "Unknown")
        sub_state = record.get("subState", "N/A")
        duration_ms = record.get("duration", 0)
        
        # Convert duration to seconds for readability
        duration_sec = duration_ms / 1000
        
        print(f"Time: {timestamp} | State: {state:10} | SubState: {sub_state:15} | Duration: {duration_sec:.2f}s")

    # Calculate total time in specific states
    total_ready_time = 0
    total_not_ready_time = 0
    
    for record in records:
        state = record.get("state")
        duration_sec = record.get("duration", 0) / 1000
        
        if state == "READY":
            total_ready_time += duration_sec
        elif state == "NOT_READY":
            total_not_ready_time += duration_sec
            
    print("\n--- Aggregated Totals ---")
    print(f"Total Ready Time: {total_ready_time:.2f} seconds")
    print(f"Total Not Ready Time: {total_not_ready_time:.2f} seconds")

Complete Working Example

Below is the complete, copy-pasteable Python script. Replace the placeholder values at the top with your actual CXone credentials and Agent ID.

import requests
import time
from datetime import datetime, timedelta, timezone

# ================= CONFIGURATION =================
CXONE_DOMAIN = "your-instance"  # Replace with your CXone domain
CLIENT_ID = "your_client_id"    # Replace with your OAuth Client ID
CLIENT_SECRET = "your_client_secret" # Replace with your OAuth Client Secret
AGENT_ID = "00000000-0000-0000-0000-000000000000" # Replace with Agent UUID
# =================================================

def get_access_token() -> str:
    """
    Retrieves an OAuth 2.0 access token using Client Credentials grant.
    """
    token_url = f"https://{CXONE_DOMAIN}.niceincontact.com/oauth/token"
    
    payload = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET
    }
    
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    
    try:
        response = requests.post(token_url, data=payload, headers=headers, timeout=10)
        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

def get_last_24_hours_range():
    """
    Calculates the start and end times for the last 24 hours in ISO 8601 format.
    """
    end_time = datetime.now(timezone.utc)
    start_time = end_time - timedelta(hours=24)
    
    # Format to ISO 8601 with 'Z' suffix for UTC
    start_iso = start_time.strftime("%Y-%m-%dT%H:%M:%SZ")
    end_iso = end_time.strftime("%Y-%m-%dT%H:%M:%SZ")
    
    return start_iso, end_iso

def fetch_agent_state_history(agent_id: str, start_time: str, end_time: str, token: str, max_pages: int = 100) -> list:
    """
    Fetches agent state history with pagination support.
    """
    base_url = f"https://{CXONE_DOMAIN}.niceincontact.com/api/reporting/v2/agent-state-history"
    
    all_records = []
    page_number = 1
    page_size = 1000
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    params = {
        "startTime": start_time,
        "endTime": end_time,
        "agentId": agent_id,
        "pageSize": page_size,
        "pageNumber": page_number
    }
    
    print(f"Starting fetch for Agent {agent_id} from {start_time} to {end_time}")
    
    while page_number <= max_pages:
        try:
            response = requests.get(base_url, headers=headers, params=params, timeout=30)
            
            # Handle HTTP Errors
            if response.status_code == 401:
                raise Exception("Unauthorized: Token may be expired.")
            elif response.status_code == 403:
                raise Exception("Forbidden: Client lacks 'reporting' scope.")
            elif response.status_code == 404:
                print("No data found for this agent in the specified time range.")
                break
            elif response.status_code == 429:
                wait_time = int(response.headers.get('Retry-After', 5))
                print(f"Rate limited. Waiting {wait_time} seconds...")
                time.sleep(wait_time)
                continue
            else:
                response.raise_for_status()
                
            data = response.json()
            
            # Check if there are records
            records = data.get("records", [])
            
            if not records:
                print("No more records found.")
                break
                
            all_records.extend(records)
            print(f"Fetched page {page_number}: {len(records)} records. Total: {len(all_records)}")
            
            # Check for next page
            if len(records) < page_size:
                print("Last page detected.")
                break
                
            # Prepare next page
            page_number += 1
            params["pageNumber"] = page_number
            
        except requests.exceptions.HTTPError as e:
            print(f"HTTP Error on page {page_number}: {e}")
            break
        except requests.exceptions.RequestException as e:
            print(f"Network error: {e}")
            break
            
    return all_records

def process_state_records(records: list):
    """
    Parses and prints a summary of agent state history.
    """
    if not records:
        print("No records to process.")
        return

    print("\n--- Agent State History Summary ---")
    for record in records:
        timestamp = record.get("timestamp", "Unknown")
        state = record.get("state", "Unknown")
        sub_state = record.get("subState", "N/A")
        duration_ms = record.get("duration", 0)
        
        duration_sec = duration_ms / 1000
        
        print(f"Time: {timestamp} | State: {state:10} | SubState: {sub_state:15} | Duration: {duration_sec:.2f}s")

    # Calculate total time in specific states
    total_ready_time = 0
    total_not_ready_time = 0
    
    for record in records:
        state = record.get("state")
        duration_sec = record.get("duration", 0) / 1000
        
        if state == "READY":
            total_ready_time += duration_sec
        elif state == "NOT_READY":
            total_not_ready_time += duration_sec
            
    print("\n--- Aggregated Totals ---")
    print(f"Total Ready Time: {total_ready_time:.2f} seconds")
    print(f"Total Not Ready Time: {total_not_ready_time:.2f} seconds")

if __name__ == "__main__":
    try:
        # 1. Authenticate
        print("Fetching access token...")
        token = get_access_token()
        
        # 2. Calculate Time Range
        start_time, end_time = get_last_24_hours_range()
        
        # 3. Fetch Data
        records = fetch_agent_state_history(AGENT_ID, start_time, end_time, token)
        
        # 4. Process Results
        process_state_records(records)
        
    except Exception as e:
        print(f"Application error: {e}")

Common Errors & Debugging

Error: 401 Unauthorized

Cause: The access token is invalid, expired, or malformed.
Fix: Ensure you are using the correct CLIENT_ID and CLIENT_SECRET. Verify that the token was obtained successfully before making the reporting call. If running for a long time, implement token refresh logic.

Error: 403 Forbidden

Cause: The OAuth client does not have the reporting scope, or the client does not have access to the specific agent’s data (if using role-based access control restrictions).
Fix: Log into the CXone Admin console. Navigate to Applications > OAuth Clients. Edit your client and ensure the reporting scope is checked. Save and regenerate credentials if necessary.

Error: 422 Unprocessable Entity

Cause: Incorrect date format or invalid agentId.
Fix: Ensure startTime and endTime are in strict ISO 8601 format (YYYY-MM-DDTHH:mm:ssZ). Verify the agentId is a valid UUID. Check that startTime is earlier than endTime.

Error: 429 Too Many Requests

Cause: You have exceeded the API rate limits.
Fix: Implement exponential backoff. The code above includes a basic retry mechanism for 429 responses by reading the Retry-After header. In production, consider using a library like tenacity for robust retry logic.

Error: Empty Records

Cause: The agent was not logged in during the specified time range, or the time range is too far in the past (CXone retains reporting data for a specific period, typically 30-90 days depending on your license).
Fix: Verify the agent was active. Check the startTime and endTime. Ensure the agent ID is correct.

Official References