Querying NICE CXone Agent State History with the v2 Reporting API

Querying NICE CXone Agent State History with the v2 Reporting API

What You Will Build

  • A Python script that retrieves the login and state change history for a specific agent over the last 24 hours.
  • This uses the NICE CXone Reporting API v2, specifically the GET /v2/reporting/agents/{agentId}/state-history endpoint.
  • The programming language used is Python 3.9+ with the requests library for HTTP handling.

Prerequisites

  • OAuth Client: A CXone OAuth client with the Reporting scope. You need the client_id and client_secret.
  • API Version: CXone Reporting API v2.
  • Runtime: Python 3.9 or higher.
  • Dependencies:
    • requests: For making HTTP calls.
    • python-dateutil: For parsing ISO 8601 date strings.

Install dependencies via pip:

pip install requests python-dateutil

Authentication Setup

CXone uses standard OAuth 2.0 Client Credentials Grant. You must obtain an access token before making any Reporting API calls. The token is valid for a limited duration (usually 1 hour), so production code should cache and refresh tokens. For this tutorial, we will implement a simple function to fetch a fresh token.

The endpoint is https://platform.nice.incontact.com/oauth2/token (or your specific regional platform URL).

import requests
import json
from typing import Optional

# Configuration
OAUTH_CLIENT_ID = "YOUR_CLIENT_ID"
OAUTH_CLIENT_SECRET = "YOUR_CLIENT_SECRET"
PLATFORM_URL = "https://platform.nice.incontact.com"  # Replace with your region if different

def get_access_token() -> str:
    """
    Fetches an OAuth2 access token using Client Credentials flow.
    
    Returns:
        str: The JWT access token.
    """
    token_url = f"{PLATFORM_URL}/oauth2/token"
    
    # The body for client credentials grant
    payload = {
        "grant_type": "client_credentials",
        "client_id": OAUTH_CLIENT_ID,
        "client_secret": OAUTH_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 http_err:
        if response.status_code == 401:
            raise RuntimeError("Authentication failed. Check CLIENT_ID and CLIENT_SECRET.") from http_err
        else:
            raise RuntimeError(f"HTTP error occurred: {http_err}") from http_err
    except requests.exceptions.RequestException as err:
        raise RuntimeError(f"Network error occurred: {err}") from err

Implementation

Step 1: Construct the Time Range and Query Parameters

The CXone Reporting API v2 requires specific time boundaries. To get data for the “last 24 hours,” you must calculate the start and end timestamps in ISO 8601 format. The API operates in UTC.

We also need to identify the agent. You can use the agent’s id (UUID) or external_id (if configured). For this example, we assume you have the agent’s UUID.

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

def get_last_24_hours_range() -> tuple[str, str]:
    """
    Calculates the start and end timestamps for the last 24 hours in ISO 8601 UTC format.
    
    Returns:
        tuple: (start_time, end_time) as ISO 8601 strings.
    """
    end_time = datetime.now(tzutc())
    start_time = end_time - timedelta(hours=24)
    
    # Format as ISO 8601 with 'Z' for UTC
    return start_time.isoformat().replace("+00:00", "Z"), end_time.isoformat().replace("+00:00", "Z")

Step 2: Build the API Request with Pagination Support

The state-history endpoint supports pagination via limit and offset. By default, it may return a small set of records. To ensure we get all state changes for the last 24 hours, we must implement a loop that continues fetching until no more records are returned.

Required OAuth Scope: Reporting

Endpoint: GET /v2/reporting/agents/{agentId}/state-history

Query Parameters:

  • start: ISO 8601 timestamp (UTC).
  • end: ISO 8601 timestamp (UTC).
  • limit: Number of records to return per page (max usually 1000).
  • offset: Number of records to skip.
import requests
from typing import List, Dict, Any

def fetch_agent_state_history(agent_id: str, token: str) -> List[Dict[str, Any]]:
    """
    Fetches all agent state history records for the last 24 hours with pagination.
    
    Args:
        agent_id: The UUID of the agent.
        token: The OAuth2 access token.
        
    Returns:
        List of dictionaries containing state history records.
    """
    base_url = f"{PLATFORM_URL}/v2/reporting/agents/{agent_id}/state-history"
    
    start_time, end_time = get_last_24_hours_range()
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    params = {
        "start": start_time,
        "end": end_time,
        "limit": 100,  # Fetch in batches of 100
        "offset": 0
    }
    
    all_records = []
    
    while True:
        try:
            response = requests.get(base_url, headers=headers, params=params)
            response.raise_for_status()
            
            data = response.json()
            
            # The response structure typically contains a list of records under 'data' or directly as a list
            # CXone v2 Reporting API usually returns { "data": [ ... ], "pagination": { ... } }
            records = data.get("data", [])
            
            if not records:
                break
            
            all_records.extend(records)
            
            # Check if we need to paginate further
            pagination = data.get("pagination", {})
            total_count = pagination.get("total", 0)
            current_offset = params["offset"]
            limit = params["limit"]
            
            if current_offset + limit >= total_count:
                break
                
            # Update offset for next iteration
            params["offset"] += limit
            
        except requests.exceptions.HTTPError as http_err:
            if response.status_code == 401:
                raise RuntimeError("Token expired or invalid. Refresh token.") from http_err
            elif response.status_code == 404:
                raise RuntimeError(f"Agent {agent_id} not found or no permissions.") from http_err
            else:
                raise RuntimeError(f"HTTP error: {http_err}") from http_err
        except requests.exceptions.RequestException as err:
            raise RuntimeError(f"Network error: {err}") from err
            
    return all_records

Step 3: Processing and Formatting Results

The raw JSON response contains machine-readable codes for states (e.g., LoggedIn, Available, Busy). To make this useful, we should map these codes to human-readable labels and sort the events chronologically.

def format_state_records(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """
    Formats raw API records into a cleaner, human-readable structure.
    
    Args:
        records: List of raw state history dictionaries.
        
    Returns:
        List of formatted dictionaries.
    """
    # Map of common CXone state codes to readable names
    STATE_MAP = {
        "LoggedIn": "Logged In",
        "Available": "Available",
        "Busy": "Busy",
        "WrapUp": "Wrap Up",
        "NotReady": "Not Ready",
        "Pause": "Paused",
        "Offline": "Offline",
        "Unknown": "Unknown"
    }
    
    formatted_records = []
    
    for record in records:
        state_code = record.get("state", "Unknown")
        readable_state = STATE_MAP.get(state_code, state_code)
        
        formatted_record = {
            "timestamp": record.get("timestamp"),
            "state": readable_state,
            "state_code": state_code,
            "reason": record.get("reason", ""),
            "skill": record.get("skill", ""), # Some state changes are skill-specific
            "queue": record.get("queue", "")   # Some state changes are queue-specific
        }
        formatted_records.append(formatted_record)
    
    # Sort by timestamp ascending
    formatted_records.sort(key=lambda x: x["timestamp"])
    
    return formatted_records

Complete Working Example

This is the full, copy-pasteable script. Replace YOUR_CLIENT_ID, YOUR_CLIENT_SECRET, and AGENT_UUID with your actual values.

import requests
import json
from datetime import datetime, timedelta
from dateutil.tz import tzutc
from typing import List, Dict, Any, Optional

# === CONFIGURATION ===
OAUTH_CLIENT_ID = "YOUR_CLIENT_ID"
OAUTH_CLIENT_SECRET = "YOUR_CLIENT_SECRET"
PLATFORM_URL = "https://platform.nice.incontact.com"  # Adjust for your region (e.g., eu, ap)
AGENT_UUID = "YOUR_AGENT_UUID"  # The UUID of the agent to query

# === AUTHENTICATION ===

def get_access_token() -> str:
    """
    Fetches an OAuth2 access token using Client Credentials flow.
    """
    token_url = f"{PLATFORM_URL}/oauth2/token"
    
    payload = {
        "grant_type": "client_credentials",
        "client_id": OAUTH_CLIENT_ID,
        "client_secret": OAUTH_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 http_err:
        if response.status_code == 401:
            raise RuntimeError("Authentication failed. Check CLIENT_ID and CLIENT_SECRET.") from http_err
        else:
            raise RuntimeError(f"HTTP error occurred: {http_err}") from http_err
    except requests.exceptions.RequestException as err:
        raise RuntimeError(f"Network error occurred: {err}") from err

# === DATA RETRIEVAL ===

def get_last_24_hours_range() -> tuple[str, str]:
    """
    Calculates the start and end timestamps for the last 24 hours in ISO 8601 UTC format.
    """
    end_time = datetime.now(tzutc())
    start_time = end_time - timedelta(hours=24)
    
    # Format as ISO 8601 with 'Z' for UTC
    return start_time.isoformat().replace("+00:00", "Z"), end_time.isoformat().replace("+00:00", "Z")

def fetch_agent_state_history(agent_id: str, token: str) -> List[Dict[str, Any]]:
    """
    Fetches all agent state history records for the last 24 hours with pagination.
    """
    base_url = f"{PLATFORM_URL}/v2/reporting/agents/{agent_id}/state-history"
    
    start_time, end_time = get_last_24_hours_range()
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    params = {
        "start": start_time,
        "end": end_time,
        "limit": 100,
        "offset": 0
    }
    
    all_records = []
    
    while True:
        try:
            response = requests.get(base_url, headers=headers, params=params)
            response.raise_for_status()
            
            data = response.json()
            
            records = data.get("data", [])
            
            if not records:
                break
            
            all_records.extend(records)
            
            pagination = data.get("pagination", {})
            total_count = pagination.get("total", 0)
            current_offset = params["offset"]
            limit = params["limit"]
            
            if current_offset + limit >= total_count:
                break
                
            params["offset"] += limit
            
        except requests.exceptions.HTTPError as http_err:
            if response.status_code == 401:
                raise RuntimeError("Token expired or invalid. Refresh token.") from http_err
            elif response.status_code == 404:
                raise RuntimeError(f"Agent {agent_id} not found or no permissions.") from http_err
            else:
                raise RuntimeError(f"HTTP error: {http_err}") from http_err
        except requests.exceptions.RequestException as err:
            raise RuntimeError(f"Network error: {err}") from err
            
    return all_records

# === POST-PROCESSING ===

def format_state_records(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """
    Formats raw API records into a cleaner, human-readable structure.
    """
    STATE_MAP = {
        "LoggedIn": "Logged In",
        "Available": "Available",
        "Busy": "Busy",
        "WrapUp": "Wrap Up",
        "NotReady": "Not Ready",
        "Pause": "Paused",
        "Offline": "Offline",
        "Unknown": "Unknown"
    }
    
    formatted_records = []
    
    for record in records:
        state_code = record.get("state", "Unknown")
        readable_state = STATE_MAP.get(state_code, state_code)
        
        formatted_record = {
            "timestamp": record.get("timestamp"),
            "state": readable_state,
            "state_code": state_code,
            "reason": record.get("reason", ""),
            "skill": record.get("skill", ""),
            "queue": record.get("queue", "")
        }
        formatted_records.append(formatted_record)
    
    # Sort by timestamp ascending
    formatted_records.sort(key=lambda x: x["timestamp"])
    
    return formatted_records

# === MAIN EXECUTION ===

def main():
    print(f"Fetching state history for Agent: {AGENT_UUID}")
    
    try:
        # 1. Authenticate
        print("Authenticating...")
        token = get_access_token()
        print("Authentication successful.")
        
        # 2. Fetch Data
        print("Fetching state history...")
        raw_records = fetch_agent_state_history(AGENT_UUID, token)
        print(f"Retrieved {len(raw_records)} raw records.")
        
        # 3. Format and Display
        if not raw_records:
            print("No state history found for the last 24 hours.")
            return
            
        formatted_records = format_state_records(raw_records)
        
        print("\n--- Agent State History (Last 24 Hours) ---")
        print(f"{'Timestamp':<25} {'State':<15} {'Reason':<20} {'Skill/Queue'}")
        print("-" * 80)
        
        for rec in formatted_records:
            ts = rec["timestamp"]
            state = rec["state"]
            reason = rec["reason"]
            extra = f"{rec['skill']} / {rec['queue']}" if rec['skill'] or rec['queue'] else ""
            
            print(f"{ts:<25} {state:<15} {reason:<20} {extra}")
            
        print("-" * 80)
        print(f"Total events: {len(formatted_records)}")
        
    except Exception as e:
        print(f"Error: {e}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

  • Cause: The OAuth token is invalid, expired, or the client credentials are incorrect.
  • Fix: Verify OAUTH_CLIENT_ID and OAUTH_CLIENT_SECRET are correct. Ensure the token was fetched recently. If the script runs for a long time, the token may expire mid-execution. For long-running processes, implement a token refresh mechanism.

Error: 403 Forbidden

  • Cause: The OAuth client does not have the Reporting scope, or the user associated with the client does not have permission to view the agent’s data.
  • Fix:
    1. Log in to the CXone Admin Portal.
    2. Navigate to Integrations > OAuth Clients.
    3. Edit your client and ensure the Reporting scope is checked.
    4. Ensure the client is associated with a user who has the “View Reports” permission for the relevant agent.

Error: 404 Not Found

  • Cause: The AGENT_UUID is invalid, or the agent does not exist in the CXone instance.
  • Fix: Verify the agent ID. You can find the agent ID by querying the GET /v2/agents endpoint or by checking the URL when viewing the agent in the CXone Admin Portal.

Error: Empty Response

  • Cause: The agent did not change states in the last 24 hours, or the time range calculation is incorrect.
  • Fix:
    1. Check the agent’s login history manually in the CXone Admin Portal to confirm activity.
    2. Verify that the start and end times are in UTC. The API strictly uses UTC. If your local time is not UTC, ensure you are converting correctly.
    3. Ensure the agent is not “Deleted” or “Inactive” in a way that prevents historical reporting.

Official References