Querying NICE CXone Agent State History via Reporting API v2

Querying NICE CXone Agent State History via Reporting API v2

What You Will Build

  • You will build a script that retrieves the historical state transitions for a specific agent over the last 24 hours.
  • This solution uses the NICE CXone Reporting API v2 endpoint /api/v2/reporting/agents/state-history.
  • The implementation is provided in Python using the requests library and TypeScript using axios.

Prerequisites

OAuth Configuration

  • Client Type: Confidential Client (Client Credentials Grant).
  • Required Scope: reporting:read. This scope is mandatory for accessing any data from the /api/v2/reporting namespace.
  • Environment: You need a valid NICE CXone environment ID (e.g., us-east-1).

Runtime Dependencies

  • Python: Python 3.8+ with requests and python-dateutil.
  • TypeScript/Node.js: Node.js 18+ with axios and dotenv.

API Version

  • This tutorial targets the Reporting API v2. Note that CXone is actively migrating reporting endpoints to v2. Legacy v1 endpoints are deprecated and may return incomplete or inconsistent state data.

Authentication Setup

NICE CXone uses OAuth 2.0 Client Credentials flow for API access. You must obtain an access token before making any reporting queries. The token expires after a defined period (typically 1 hour), so your application must handle refresh logic.

Python Authentication Helper

import requests
import time
from typing import Optional

class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str, environment: str = "us-east-1"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.environment = environment
        self.token_url = f"https://{environment}.auth.niceincontact.com/oauth2/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0.0

    def get_token(self) -> str:
        """
        Retrieves an OAuth access token. Caches it until it expires.
        """
        current_time = time.time()
        if self.access_token and current_time < self.token_expiry:
            return self.access_token

        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }
        
        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }

        try:
            response = requests.post(self.token_url, data=payload, headers=headers)
            response.raise_for_status()
            data = response.json()
            
            self.access_token = data["access_token"]
            # Subtract 60 seconds to ensure we refresh before hard expiry
            self.token_expiry = current_time + (data["expires_in"] - 60)
            return self.access_token
            
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise Exception("Authentication failed: Invalid Client ID or Secret.")
            elif response.status_code == 403:
                raise Exception("Authentication failed: Client does not have permission to request tokens.")
            else:
                raise Exception(f"Authentication request failed with status {response.status_code}: {response.text}")

    def get_headers(self) -> dict:
        """
        Returns headers ready for API calls.
        """
        return {
            "Authorization": f"Bearer {self.get_token()}",
            "Content-Type": "application/json"
        }

TypeScript Authentication Helper

import axios, { AxiosResponse } from 'axios';

interface TokenResponse {
  access_token: string;
  expires_in: number;
}

class CXoneAuth {
  private clientId: string;
  private clientSecret: string;
  private environment: string;
  private tokenUrl: string;
  private accessToken: string | null = null;
  private tokenExpiry: number = 0;

  constructor(clientId: string, clientSecret: string, environment: string = 'us-east-1') {
    this.clientId = clientId;
    this.clientSecret = clientSecret;
    this.environment = environment;
    this.tokenUrl = `https://${environment}.auth.niceincontact.com/oauth2/token`;
  }

  async getToken(): Promise<string> {
    const now = Date.now();
    
    if (this.accessToken && now < this.tokenExpiry) {
      return this.accessToken;
    }

    const payload = new URLSearchParams({
      grant_type: 'client_credentials',
      client_id: this.clientId,
      client_secret: this.clientSecret
    });

    try {
      const response: AxiosResponse<TokenResponse> = await axios.post(this.tokenUrl, payload, {
        headers: {
          'Content-Type': 'application/x-www-form-urlencoded'
        }
      });

      this.accessToken = response.data.access_token;
      // Subtract 60 seconds for buffer
      this.tokenExpiry = now + (response.data.expires_in * 1000) - 60000;
      return this.accessToken;

    } catch (error) {
      if (axios.isAxiosError(error) && error.response) {
        if (error.response.status === 401) {
          throw new Error('Authentication failed: Invalid Client ID or Secret.');
        } else if (error.response.status === 403) {
          throw new Error('Authentication failed: Client lacks permission.');
        } else {
          throw new Error(`Auth request failed: ${error.response.status} ${error.response.statusText}`);
        }
      }
      throw error;
    }
  }

  getHeaders(): Record<string, string> {
    return {
      'Authorization': `Bearer ${this.getToken()}`,
      'Content-Type': 'application/json'
    };
  }
}

Implementation

Step 1: Constructing the Query Parameters

The Reporting API v2 uses a query-based model. You must specify the date range and the target agent. Unlike v1, v2 requires ISO 8601 date strings for precision.

Key Parameters:

  • startDate: ISO 8601 string (e.g., 2023-10-27T00:00:00.000Z).
  • endDate: ISO 8601 string.
  • agentId: The UUID of the agent.
  • groupBy: For state history, we often group by agent or skill, but for raw history, we may leave this empty or specify agent depending on the desired granularity.

Python Example: Date Calculation

from datetime import datetime, timedelta, timezone

def get_last_24_hours_range() -> tuple[str, str]:
    """
    Returns a tuple of (start_date, end_date) as ISO 8601 strings 
    representing the last 24 hours.
    """
    now = datetime.now(timezone.utc)
    start = now - timedelta(hours=24)
    
    # Format as ISO 8601 with 'Z' suffix for UTC
    start_iso = start.isoformat().replace("+00:00", "Z")
    end_iso = now.isoformat().replace("+00:00", "Z")
    
    return start_iso, end_iso

TypeScript Example: Date Calculation

function getLast24HoursRange(): [string, string] {
  const now = new Date();
  const start = new Date(now.getTime() - 24 * 60 * 60 * 1000);
  
  //toISOString() returns UTC format ending in Z
  return [start.toISOString(), now.toISOString()];
}

Step 2: Executing the State History Query

The endpoint is GET /api/v2/reporting/agents/state-history. This endpoint returns a list of state change events. Each event indicates when an agent entered a specific state (e.g., Ready, Busy, Break).

Python Implementation

def fetch_agent_state_history(auth: CXoneAuth, agent_id: str, environment: str) -> dict:
    """
    Fetches agent state history for the last 24 hours.
    
    Args:
        auth: CXoneAuth instance
        agent_id: UUID of the agent
        environment: CXone environment (e.g., 'us-east-1')
        
    Returns:
        Parsed JSON response from the API
    """
    start_date, end_date = get_last_24_hours_range()
    
    # Construct the API URL
    base_url = f"https://{environment}.api.niceincontact.com"
    endpoint = "/api/v2/reporting/agents/state-history"
    full_url = f"{base_url}{endpoint}"
    
    params = {
        "startDate": start_date,
        "endDate": end_date,
        "agentId": agent_id,
        # Optional: limit results if you expect high volume
        # "limit": 1000 
    }
    
    headers = auth.get_headers()
    
    print(f"Requesting state history for Agent {agent_id}...")
    print(f"Date Range: {start_date} to {end_date}")
    
    try:
        response = requests.get(full_url, params=params, headers=headers)
        
        # Handle Rate Limiting (429)
        if response.status_code == 429:
            retry_after = int(response.headers.get('Retry-After', 60))
            print(f"Rate limited. Retrying after {retry_after} seconds...")
            time.sleep(retry_after)
            response = requests.get(full_url, params=params, headers=headers)
            
        response.raise_for_status()
        
        return response.json()
        
    except requests.exceptions.HTTPError as e:
        if response.status_code == 401:
            print("Error: Unauthorized. Token may be expired.")
        elif response.status_code == 403:
            print("Error: Forbidden. Check if 'reporting:read' scope is granted.")
        elif response.status_code == 404:
            print("Error: Not Found. Agent ID may be invalid or endpoint unavailable.")
        else:
            print(f"HTTP Error: {response.status_code} - {response.text}")
        raise

TypeScript Implementation

async function fetchAgentStateHistory(
  auth: CXoneAuth,
  agentId: string,
  environment: string
): Promise<any> {
  const [startDate, endDate] = getLast24HoursRange();

  const baseUrl = `https://${environment}.api.niceincontact.com`;
  const endpoint = '/api/v2/reporting/agents/state-history';
  const fullUrl = `${baseUrl}${endpoint}`;

  const params = new URLSearchParams({
    startDate,
    endDate,
    agentId
  });

  const headers = await auth.getHeaders();

  console.log(`Requesting state history for Agent ${agentId}...`);
  console.log(`Date Range: ${startDate} to ${endDate}`);

  try {
    const response = await axios.get(fullUrl, {
      params,
      headers
    });

    return response.data;

  } catch (error) {
    if (axios.isAxiosError(error) && error.response) {
      const status = error.response.status;
      
      if (status === 429) {
        const retryAfter = parseInt(error.response.headers['retry-after'] || '60');
        console.warn(`Rate limited. Retrying after ${retryAfter} seconds...`);
        await new Promise(resolve => setTimeout(resolve, retryAfter * 1000));
        
        // Retry once
        return await axios.get(fullUrl, { params, headers }).then(res => res.data);
      }

      if (status === 401) {
        throw new Error('Unauthorized: Token expired or invalid.');
      } else if (status === 403) {
        throw new Error('Forbidden: Missing reporting:read scope.');
      } else if (status === 404) {
        throw new Error('Not Found: Agent ID invalid or endpoint missing.');
      } else {
        throw new Error(`HTTP Error ${status}: ${error.response.statusText}`);
      }
    }
    throw error;
  }
}

Step 3: Processing the Results

The response from /api/v2/reporting/agents/state-history is structured as a list of state events. Each event contains:

  • agentId: The agent UUID.
  • stateId: The unique ID of the state (e.g., ready, break, busy).
  • stateName: Human-readable name (e.g., Ready, Lunch Break).
  • startTime: ISO 8601 timestamp when the agent entered the state.
  • endTime: ISO 8601 timestamp when the agent left the state (or null if currently in state).
  • durationMs: Duration of the state in milliseconds.

Python: Parsing and Formatting

def parse_state_history(data: dict) -> list[dict]:
    """
    Parses the API response and formats the state history for readability.
    """
    # The response structure typically wraps the results in a 'results' or 'data' key
    # depending on the specific v2 reporting sub-endpoint. 
    # For state-history, it often returns a direct list or an object with 'items'.
    
    items = data.get('items', data.get('results', []))
    
    formatted_history = []
    for item in items:
        start_dt = datetime.fromisoformat(item['startTime'].replace('Z', '+00:00'))
        end_dt = None
        if item.get('endTime'):
            end_dt = datetime.fromisoformat(item['endTime'].replace('Z', '+00:00'))
            
        duration_sec = item.get('durationMs', 0) / 1000.0
        
        formatted_history.append({
            'state': item['stateName'],
            'start': start_dt.strftime('%Y-%m-%d %H:%M:%S UTC'),
            'end': end_dt.strftime('%Y-%m-%d %H:%M:%S UTC') if end_dt else 'Current',
            'duration_sec': round(duration_sec, 2)
        })
        
    # Sort by start time ascending
    formatted_history.sort(key=lambda x: x['start'])
    return formatted_history

TypeScript: Parsing and Formatting

interface StateEvent {
  stateName: string;
  startTime: string;
  endTime: string | null;
  durationMs: number;
}

function parseStateHistory(data: any): StateEvent[] {
  // Handle potential wrapper objects
  const items: StateEvent[] = data.items || data.results || data;

  return items.map((item: StateEvent) => ({
    stateName: item.stateName,
    startTime: item.startTime,
    endTime: item.endTime,
    durationMs: item.durationMs || 0
  })).sort((a, b) => new Date(a.startTime).getTime() - new Date(b.startTime).getTime());
}

Complete Working Example

Below is the complete Python script. Save this as cxone_agent_history.py.

import requests
import time
import sys
from datetime import datetime, timedelta, timezone
from typing import Optional

# --- Authentication Module ---

class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str, environment: str = "us-east-1"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.environment = environment
        self.token_url = f"https://{environment}.auth.niceincontact.com/oauth2/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0.0

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

        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }
        
        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }

        try:
            response = requests.post(self.token_url, data=payload, headers=headers)
            response.raise_for_status()
            data = response.json()
            
            self.access_token = data["access_token"]
            self.token_expiry = current_time + (data["expires_in"] - 60)
            return self.access_token
            
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise Exception("Authentication failed: Invalid Client ID or Secret.")
            elif response.status_code == 403:
                raise Exception("Authentication failed: Client lacks permission.")
            else:
                raise Exception(f"Authentication request failed: {response.text}")

    def get_headers(self) -> dict:
        return {
            "Authorization": f"Bearer {self.get_token()}",
            "Content-Type": "application/json"
        }

# --- Reporting Logic ---

def get_last_24_hours_range() -> tuple[str, str]:
    now = datetime.now(timezone.utc)
    start = now - timedelta(hours=24)
    start_iso = start.isoformat().replace("+00:00", "Z")
    end_iso = now.isoformat().replace("+00:00", "Z")
    return start_iso, end_iso

def fetch_agent_state_history(auth: CXoneAuth, agent_id: str, environment: str) -> list[dict]:
    start_date, end_date = get_last_24_hours_range()
    base_url = f"https://{environment}.api.niceincontact.com"
    endpoint = "/api/v2/reporting/agents/state-history"
    full_url = f"{base_url}{endpoint}"
    
    params = {
        "startDate": start_date,
        "endDate": end_date,
        "agentId": agent_id
    }
    
    headers = auth.get_headers()
    
    print(f"Querying CXone Reporting API for Agent: {agent_id}")
    print(f"Range: {start_date} to {end_date}")
    
    try:
        response = requests.get(full_url, params=params, headers=headers)
        
        if response.status_code == 429:
            retry_after = int(response.headers.get('Retry-After', 60))
            print(f"Rate limited. Waiting {retry_after}s...")
            time.sleep(retry_after)
            response = requests.get(full_url, params=params, headers=headers)
            
        response.raise_for_status()
        data = response.json()
        
        # Extract items from response
        items = data.get('items', data.get('results', []))
        
        history = []
        for item in items:
            start_dt = datetime.fromisoformat(item['startTime'].replace('Z', '+00:00'))
            end_dt = None
            if item.get('endTime'):
                end_dt = datetime.fromisoformat(item['endTime'].replace('Z', '+00:00'))
                
            duration_sec = item.get('durationMs', 0) / 1000.0
            
            history.append({
                'state': item['stateName'],
                'start': start_dt.strftime('%Y-%m-%d %H:%M:%S UTC'),
                'end': end_dt.strftime('%Y-%m-%d %H:%M:%S UTC') if end_dt else 'Current',
                'duration_sec': round(duration_sec, 2)
            })
            
        history.sort(key=lambda x: x['start'])
        return history
        
    except requests.exceptions.HTTPError as e:
        print(f"API Error: {response.status_code} - {response.text}")
        sys.exit(1)

# --- Main Execution ---

if __name__ == "__main__":
    # Configuration
    CLIENT_ID = "YOUR_CLIENT_ID"
    CLIENT_SECRET = "YOUR_CLIENT_SECRET"
    ENVIRONMENT = "us-east-1"
    AGENT_ID = "YOUR_AGENT_UUID"
    
    if CLIENT_ID == "YOUR_CLIENT_ID":
        print("Please configure CLIENT_ID, CLIENT_SECRET, ENVIRONMENT, and AGENT_ID in the script.")
        sys.exit(1)
        
    auth = CXoneAuth(CLIENT_ID, CLIENT_SECRET, ENVIRONMENT)
    
    try:
        history = fetch_agent_state_history(auth, AGENT_ID, ENVIRONMENT)
        
        print("\n--- Agent State History (Last 24 Hours) ---")
        print(f"{'State':<15} | {'Start Time':<25} | {'End Time':<25} | {'Duration (s)':<10}")
        print("-" * 80)
        
        for event in history:
            print(f"{event['state']:<15} | {event['start']:<25} | {event['end']:<25} | {event['duration_sec']:<10}")
            
    except Exception as e:
        print(f"Failed to retrieve history: {e}")
        sys.exit(1)

Common Errors & Debugging

Error: 401 Unauthorized

  • Cause: The access token is expired, invalid, or not included in the header.
  • Fix: Ensure the CXoneAuth class is correctly refreshing the token. Check that Authorization: Bearer <token> is present in the headers. Verify the Client ID and Secret are correct in the Auth URL.

Error: 403 Forbidden

  • Cause: The OAuth client does not have the reporting:read scope.
  • Fix: Log in to the CXone Admin Console. Navigate to Integrations > OAuth Clients. Edit your client and ensure the Reporting scope is checked. You must re-authorize the client if you change scopes.

Error: 429 Too Many Requests

  • Cause: You have exceeded the rate limit for the Reporting API. CXone enforces strict rate limits to protect system performance.
  • Fix: Implement exponential backoff. The Retry-After header indicates how many seconds to wait. The code examples above include a basic retry mechanism for 429 errors.

Error: 404 Not Found

  • Cause: The agentId provided is invalid, or the agent has no state history in the requested time range.
  • Fix: Verify the Agent UUID. Ensure the agent was active in the last 24 hours. If the agent was never logged in, the API may return an empty list rather than a 404, depending on the specific endpoint implementation.

Error: Empty Results

  • Cause: The date range does not contain any state changes, or the agent was inactive.
  • Fix: Check the startDate and endDate values. Ensure they are in the past. Verify the agent was logged in during this period.

Official References