Querying NICE CXone Agent State History via the Reporting API v2

What You Will Build

• This tutorial demonstrates how to retrieve a granular timeline of agent state changes (Ready, Busy, Break, Offline) for a specific user over the last 24 hours.
• The solution utilizes the NICE CXone Reporting API v2 endpoint /reporting/v2/agents/states/history.
• The implementation is provided in Python using the requests library for precise control over HTTP headers and pagination logic.

Prerequisites

• OAuth Client Type: Service Account (Client Credentials Grant).
• Required Scopes: reporting:read is mandatory. If you encounter 403 Forbidden errors, verify that the service account also has admin:users:read if you need to resolve user IDs from names, though this tutorial assumes you already possess the user_id.
• SDK/API Version: NICE CXone Reporting API v2.
• Language/Runtime: Python 3.8+
• External Dependencies:
  • requests (standard HTTP client)
  • python-dotenv (for secure credential management)
  • pytz (for timezone handling, though ISO 8601 is preferred)

Authentication Setup

The NICE CXone platform uses OAuth 2.0 Client Credentials flow for server-to-server communication. You must exchange your Client ID and Client Secret for a short-lived access token (typically valid for 1 hour).

This section provides a robust authentication helper that caches the token to avoid unnecessary API calls during script execution.

import requests
import time
import os
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str, base_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = base_url.rstrip('/')
        self.token = None
        self.token_expiry = 0
        self.token_url = f"{self.base_url}/oauth/token"

    def get_token(self) -> str:
        """
        Retrieves an OAuth access token.
        Returns a cached token if it is still valid, otherwise fetches a new one.
        """
        # Check if we have a valid token
        if self.token and time.time() < self.token_expiry:
            return self.token

        # Prepare the client credentials grant payload
        payload = {
            "grant_type": "client_credentials"
        }
        
        # Headers for the token request
        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }

        try:
            response = requests.post(
                self.token_url,
                data=payload,
                headers=headers,
                auth=(self.client_id, self.client_secret)
            )
            
            response.raise_for_status()
            token_data = response.json()
            
            self.token = token_data["access_token"]
            # Set expiry slightly before actual expiry to allow for clock skew
            self.token_expiry = time.time() + token_data["expires_in"] - 60
            
            return self.token

        except requests.exceptions.HTTPError as http_err:
            print(f"HTTP error occurred while fetching token: {http_err}")
            print(f"Response body: {response.text}")
            raise
        except Exception as err:
            print(f"An error occurred while fetching token: {err}")
            raise

Implementation

Step 1: Constructing the Query Parameters

The NICE CXone Reporting API v2 for agent state history relies heavily on query parameters to define the temporal window and the specific agent. Unlike the v1 API, v2 enforces strict ISO 8601 date formats and requires explicit timezone specification or UTC timestamps.

Key parameters:

• user_id: The unique identifier of the agent.
• start_date_time: The beginning of the query window (inclusive).
• end_date_time: The end of the query window (exclusive).
• timezone: The timezone for the date filters (e.g., America/New_York). If omitted, UTC is assumed.

from datetime import datetime, timedelta
from dateutil import tz

def get_last_24_hours_params(user_id: str, timezone_str: str = "UTC") -> dict:
    """
    Constructs the required query parameters for the last 24 hours.
    """
    now = datetime.now(tz.gettz(timezone_str))
    start_time = now - timedelta(hours=24)
    
    # Format to ISO 8601 with timezone offset
    # Example: 2023-10-27T10:00:00-04:00
    start_iso = start_time.isoformat()
    end_iso = now.isoformat()

    return {
        "user_id": user_id,
        "start_date_time": start_iso,
        "end_date_time": end_iso,
        "timezone": timezone_str
    }

Step 2: Executing the API Request with Pagination

The /reporting/v2/agents/states/history endpoint returns paginated results. The default page size is often 100 records. To retrieve the full 24-hour history, you must implement a loop that follows the next_page_token provided in the response headers or body.

In the NICE CXone API, pagination is typically handled via a page_token query parameter. The response will include a next_page_token if more data is available.

def fetch_agent_state_history(auth: CXoneAuth, user_id: str, timezone: str = "UTC") -> list:
    """
    Fetches all agent state history records for the given user ID over the last 24 hours.
    Handles pagination automatically.
    """
    api_endpoint = f"{auth.base_url}/reporting/v2/agents/states/history"
    all_records = []
    
    # Initial parameters
    params = get_last_24_hours_params(user_id, timezone)
    
    # Maximum iterations to prevent infinite loops in case of API bugs
    max_pages = 50
    current_page = 0

    while current_page < max_pages:
        current_page += 1
        
        try:
            # Make the GET request
            response = requests.get(
                api_endpoint,
                params=params,
                headers={"Authorization": f"Bearer {auth.get_token()}"}
            )
            
            # Handle HTTP Errors
            if response.status_code == 401:
                print("Authentication failed. Token may be expired or invalid.")
                raise Exception("401 Unauthorized")
            elif response.status_code == 403:
                print("Forbidden. Check if your service account has 'reporting:read' scope.")
                raise Exception("403 Forbidden")
            elif response.status_code == 429:
                # Rate limiting: Wait and retry
                retry_after = int(response.headers.get("Retry-After", 1))
                print(f"Rate limited. Waiting {retry_after} seconds...")
                time.sleep(retry_after)
                continue
            elif response.status_code >= 400:
                print(f"Error {response.status_code}: {response.text}")
                break
            
            response.raise_for_status()
            
            data = response.json()
            
            # The response structure typically contains a 'records' or 'data' array
            # Note: NICE API structures can vary slightly by endpoint. 
            # For v2 reporting, the data is usually in a 'records' list or directly in the JSON body.
            # Based on standard CXone v2 patterns, we look for 'records' or iterate if it's a list.
            
            records = data.get("records", [])
            if not records and isinstance(data, list):
                records = data
            
            all_records.extend(records)
            
            # Check for pagination token
            # CXone v2 often returns pagination info in the response body under 'pagination'
            # or uses a 'next_page_token' field.
            next_token = data.get("next_page_token") or data.get("pagination", {}).get("next_page_token")
            
            if not next_token:
                break # No more pages
            
            # Update params for the next request
            params["page_token"] = next_token

        except requests.exceptions.RequestException as e:
            print(f"Network error occurred: {e}")
            break

    return all_records

Step 3: Processing and Interpreting State Data

The raw API response contains technical state codes and timestamps. To make this data useful, we must map the internal state IDs to human-readable labels and calculate the duration of each state.

Common NICE CXone Agent States:

• READY: Agent is available to take interactions.
• BUSY: Agent is in an interaction or manually set to busy.
• BREAK: Agent is on break.
• OFFLINE: Agent is logged out or not logged in.
• PAUSED: Agent is paused (often synonymous with Busy depending on configuration).

def map_state_code(state_id: int) -> str:
    """
    Maps NICE CXone internal state IDs to readable labels.
    Note: These IDs can vary by tenant configuration, but these are standard defaults.
    """
    state_map = {
        1: "OFFLINE",
        2: "READY",
        3: "BUSY",
        4: "BREAK",
        5: "PAUSED",
        6: "NOT_READY",
        7: "QUEUE_READY",
        8: "QUEUE_BUSY"
    }
    return state_map.get(state_id, f"UNKNOWN({state_id})")

def process_state_history(records: list) -> list:
    """
    Transforms raw API records into a structured list of state events with durations.
    """
    processed_events = []
    
    for record in records:
        # Extract key fields
        start_time = record.get("start_time")
        end_time = record.get("end_time")
        state_id = record.get("state_id")
        
        # Convert ISO strings to datetime objects for duration calculation
        if start_time and end_time:
            start_dt = datetime.fromisoformat(start_time.replace("Z", "+00:00"))
            end_dt = datetime.fromisoformat(end_time.replace("Z", "+00:00"))
            duration_seconds = (end_dt - start_dt).total_seconds()
        else:
            start_dt = None
            end_dt = None
            duration_seconds = 0

        processed_events.append({
            "start_time": start_time,
            "end_time": end_time,
            "state": map_state_code(state_id),
            "state_id": state_id,
            "duration_seconds": duration_seconds,
            "duration_formatted": f"{int(duration_seconds // 60)}m {int(duration_seconds % 60)}s"
        })
        
    return processed_events

Complete Working Example

Below is the complete, runnable Python script. Save this as cxone_agent_history.py. Ensure you have a .env file in the same directory with the following variables:
CXONE_CLIENT_ID=your_client_id
CXONE_CLIENT_SECRET=your_client_secret
CXONE_BASE_URL=https://api-us-1.cxone.com (or your region)
CXONE_USER_ID=the_agent_user_id

import requests
import time
import os
from datetime import datetime, timedelta
from dateutil import tz
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str, base_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = base_url.rstrip('/')
        self.token = None
        self.token_expiry = 0
        self.token_url = f"{self.base_url}/oauth/token"

    def get_token(self) -> str:
        if self.token and time.time() < self.token_expiry:
            return self.token

        payload = {"grant_type": "client_credentials"}
        headers = {"Content-Type": "application/x-www-form-urlencoded"}

        try:
            response = requests.post(
                self.token_url,
                data=payload,
                headers=headers,
                auth=(self.client_id, self.client_secret)
            )
            response.raise_for_status()
            token_data = response.json()
            
            self.token = token_data["access_token"]
            self.token_expiry = time.time() + token_data["expires_in"] - 60
            return self.token

        except requests.exceptions.HTTPError as http_err:
            print(f"HTTP error occurred while fetching token: {http_err}")
            raise
        except Exception as err:
            print(f"An error occurred while fetching token: {err}")
            raise

def get_last_24_hours_params(user_id: str, timezone_str: str = "UTC") -> dict:
    now = datetime.now(tz.gettz(timezone_str))
    start_time = now - timedelta(hours=24)
    
    start_iso = start_time.isoformat()
    end_iso = now.isoformat()

    return {
        "user_id": user_id,
        "start_date_time": start_iso,
        "end_date_time": end_iso,
        "timezone": timezone_str
    }

def fetch_agent_state_history(auth: CXoneAuth, user_id: str, timezone: str = "UTC") -> list:
    api_endpoint = f"{auth.base_url}/reporting/v2/agents/states/history"
    all_records = []
    
    params = get_last_24_hours_params(user_id, timezone)
    
    max_pages = 50
    current_page = 0

    while current_page < max_pages:
        current_page += 1
        
        try:
            response = requests.get(
                api_endpoint,
                params=params,
                headers={"Authorization": f"Bearer {auth.get_token()}"}
            )
            
            if response.status_code == 401:
                raise Exception("401 Unauthorized")
            elif response.status_code == 403:
                raise Exception("403 Forbidden: Check scopes")
            elif response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 1))
                time.sleep(retry_after)
                continue
            elif response.status_code >= 400:
                print(f"Error {response.status_code}: {response.text}")
                break
            
            response.raise_for_status()
            
            data = response.json()
            
            # Handle potential variations in response structure
            records = data.get("records", [])
            if not records and isinstance(data, list):
                records = data
            
            all_records.extend(records)
            
            next_token = data.get("next_page_token") or data.get("pagination", {}).get("next_page_token")
            
            if not next_token:
                break
            
            params["page_token"] = next_token

        except requests.exceptions.RequestException as e:
            print(f"Network error occurred: {e}")
            break

    return all_records

def map_state_code(state_id: int) -> str:
    state_map = {
        1: "OFFLINE",
        2: "READY",
        3: "BUSY",
        4: "BREAK",
        5: "PAUSED",
        6: "NOT_READY",
        7: "QUEUE_READY",
        8: "QUEUE_BUSY"
    }
    return state_map.get(state_id, f"UNKNOWN({state_id})")

def process_state_history(records: list) -> list:
    processed_events = []
    
    for record in records:
        start_time = record.get("start_time")
        end_time = record.get("end_time")
        state_id = record.get("state_id")
        
        if start_time and end_time:
            try:
                start_dt = datetime.fromisoformat(start_time.replace("Z", "+00:00"))
                end_dt = datetime.fromisoformat(end_time.replace("Z", "+00:00"))
                duration_seconds = (end_dt - start_dt).total_seconds()
            except ValueError:
                duration_seconds = 0
        else:
            duration_seconds = 0

        processed_events.append({
            "start_time": start_time,
            "end_time": end_time,
            "state": map_state_code(state_id),
            "state_id": state_id,
            "duration_seconds": duration_seconds,
            "duration_formatted": f"{int(duration_seconds // 60)}m {int(duration_seconds % 60)}s"
        })
        
    return processed_events

def main():
    client_id = os.getenv("CXONE_CLIENT_ID")
    client_secret = os.getenv("CXONE_CLIENT_SECRET")
    base_url = os.getenv("CXONE_BASE_URL", "https://api-us-1.cxone.com")
    user_id = os.getenv("CXONE_USER_ID")

    if not all([client_id, client_secret, user_id]):
        print("Missing environment variables. Please check your .env file.")
        return

    auth = CXoneAuth(client_id, client_secret, base_url)
    
    print(f"Fetching agent state history for User ID: {user_id}")
    raw_records = fetch_agent_state_history(auth, user_id, timezone="America/New_York")
    
    if not raw_records:
        print("No state history records found for the last 24 hours.")
        return

    processed_events = process_state_history(raw_records)
    
    # Sort by start_time to ensure chronological order
    processed_events.sort(key=lambda x: x["start_time"] if x["start_time"] else "")

    print("\n--- Agent State History (Last 24 Hours) ---")
    print(f"{'Start Time':<25} {'State':<15} {'Duration':<15}")
    print("-" * 55)
    
    for event in processed_events:
        start_str = event["start_time"][:19] if event["start_time"] else "N/A"
        print(f"{start_str:<25} {event['state']:<15} {event['duration_formatted']:<15}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

• Cause: The OAuth token lacks the necessary scope, or the service account does not have permission to view reporting data for the specified user.
• Fix:
  1. Verify the Client Credentials grant includes the reporting:read scope.
  2. In the NICE CXone Admin Console, navigate to Security > Roles. Ensure the role assigned to the service account includes the “Reporting” permission set.
  3. Confirm that the user_id provided is not for a suspended or deleted user.

Error: 400 Bad Request (Invalid Date Format)

• Cause: The start_date_time or end_date_time parameters do not conform to strict ISO 8601 standards, or the timezone string is invalid.
• Fix: Ensure timestamps include the timezone offset (e.g., 2023-10-27T10:00:00-04:00) or use Z for UTC. Do not use ambiguous local times without the timezone query parameter.

Error: Empty Response

• Cause: The agent has not changed states in the last 24 hours, or the agent was not logged in during the query window.
• Fix:
  1. Verify the user_id is correct.
  2. Expand the date range to test if data exists at all.
  3. Check if the agent is actually active in the system. State history is only generated for logged-in agents who have interacted with the state machine.

Error: 429 Too Many Requests

• Cause: You have exceeded the rate limit for the Reporting API.
• Fix: The provided code includes basic retry logic. For production workloads, implement an exponential backoff strategy. The Retry-After header indicates how many seconds to wait before the next request.

Official References

• NICE CXone Reporting API v2 Documentation
• Agent State History Endpoint
• OAuth 2.0 Client Credentials Flow