Querying Agent State History in NICE CXone Using Python and the Reporting API v2

What You Will Build

• You will build a Python script that retrieves the complete interaction and state history for a specific agent over the last 24 hours using the NICE CXone Reporting API.
• This tutorial uses the NICE CXone REST API v2 endpoint GET /api/v2/reporting/agents/{agentId}/history.
• The implementation is written in Python 3.9+ using the requests library for HTTP communication and JSON parsing.

Prerequisites

• OAuth Client: A NICE CXone OAuth client with the read:agent and read:reporting scopes.
• Agent ID: The unique UUID of the agent whose history you want to retrieve.
• Python Environment: Python 3.9 or higher.
• Dependencies:
  • requests: For making HTTP requests.
  • python-dotenv: For managing environment variables securely.

Install the dependencies using pip:

pip install requests python-dotenv

Authentication Setup

NICE CXone uses OAuth 2.0 for authentication. You must obtain an access token before making any API calls. This tutorial uses the Client Credentials Grant flow, which is suitable for server-to-server integrations.

Create a .env file in your project root with the following variables:

CXONE_CLIENT_ID=your_client_id
CXONE_CLIENT_SECRET=your_client_secret
CXONE_API_BASE_URL=https://api-us-01.nice-incontact.com
CXONE_AGENT_ID=your_agent_uuid

The following Python function handles the token acquisition and caching. In a production environment, you should implement token refresh logic or use a library that handles this automatically. For this tutorial, we assume the token is valid for the duration of the script execution.

import os
import requests
from dotenv import load_dotenv

load_dotenv()

CXONE_CLIENT_ID = os.getenv("CXONE_CLIENT_ID")
CXONE_CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET")
CXONE_API_BASE_URL = os.getenv("CXONE_API_BASE_URL")
CXONE_AGENT_ID = os.getenv("CXONE_AGENT_ID")

def get_access_token() -> str:
    """
    Retrieves an OAuth 2.0 access token from NICE CXone.
    
    Returns:
        str: The access token.
    """
    token_url = f"{CXONE_API_BASE_URL}/oauth/token"
    
    payload = {
        "grant_type": "client_credentials",
        "client_id": CXONE_CLIENT_ID,
        "client_secret": CXONE_CLIENT_SECRET
    }
    
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    
    response = requests.post(token_url, data=payload, headers=headers)
    
    if response.status_code != 200:
        raise Exception(f"Failed to obtain access token: {response.status_code} - {response.text}")
    
    token_data = response.json()
    return token_data["access_token"]

Implementation

Step 1: Define the Time Range and Query Parameters

The NICE CXone Reporting API requires a time range to filter data. You must specify a start and end timestamp in ISO 8601 format. For a 24-hour history, you calculate the current time and subtract 24 hours.

Additionally, the GET /api/v2/reporting/agents/{agentId}/history endpoint supports pagination. You should define the pageSize and page parameters to handle large datasets.

from datetime import datetime, timedelta, timezone

def get_time_range_iso() -> tuple[str, str]:
    """
    Calculates the start and end times for the last 24 hours in ISO 8601 format.
    
    Returns:
        tuple: (start_time_iso, end_time_iso)
    """
    now = datetime.now(timezone.utc)
    start_time = now - timedelta(hours=24)
    
    # Format as ISO 8601 with timezone offset
    start_iso = start_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
    end_iso = now.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
    
    return start_iso, end_iso

Step 2: Construct the API Request with Pagination Logic

The agent history endpoint returns a list of state changes. Each record includes the agent ID, the previous state, the new state, and the timestamp of the change. The API supports pagination, so you must loop through pages until no more records are returned.

The endpoint requires the read:reporting scope.

def fetch_agent_history(access_token: str, agent_id: str, start_time: str, end_time: str) -> list[dict]:
    """
    Fetches agent history from NICE CXone with pagination.
    
    Args:
        access_token (str): OAuth 2.0 access token.
        agent_id (str): The UUID of the agent.
        start_time (str): Start time in ISO 8601 format.
        end_time (str): End time in ISO 8601 format.
        
    Returns:
        list[dict]: A list of agent history records.
    """
    base_url = f"{CXONE_API_BASE_URL}/api/v2/reporting/agents/{agent_id}/history"
    all_records = []
    page = 1
    page_size = 100  # Maximum allowed page size for this endpoint
    
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Accept": "application/json"
    }
    
    params = {
        "start": start_time,
        "end": end_time,
        "pageSize": page_size,
        "page": page
    }
    
    while True:
        try:
            response = requests.get(base_url, headers=headers, params=params)
            
            if response.status_code == 200:
                data = response.json()
                records = data.get("records", [])
                
                if not records:
                    break
                
                all_records.extend(records)
                
                # Check if there are more pages
                total_records = data.get("totalRecords", 0)
                if len(all_records) >= total_records:
                    break
                
                page += 1
                params["page"] = page
                
            elif response.status_code == 429:
                # Handle rate limiting
                retry_after = int(response.headers.get("Retry-After", 5))
                print(f"Rate limited. Waiting {retry_after} seconds...")
                import time
                time.sleep(retry_after)
                
            elif response.status_code == 401 or response.status_code == 403:
                raise Exception(f"Authentication/Authorization failed: {response.status_code} - {response.text}")
                
            else:
                raise Exception(f"API request failed: {response.status_code} - {response.text}")
                
        except requests.exceptions.RequestException as e:
            raise Exception(f"Network error: {str(e)}")
            
    return all_records

Step 3: Process and Format the Results

The raw response from the API contains a list of records. Each record includes metadata such as the agent ID, the state change details, and the timestamp. You may want to process this data to make it more readable or to aggregate statistics.

def process_agent_history(records: list[dict]) -> list[dict]:
    """
    Processes the raw agent history records into a more readable format.
    
    Args:
        records (list[dict]): Raw records from the API.
        
    Returns:
        list[dict]: Processed records with formatted timestamps and state names.
    """
    processed_records = []
    
    for record in records:
        timestamp = record.get("timestamp", "")
        state = record.get("state", {})
        
        # Extract state details
        state_name = state.get("name", "Unknown")
        state_id = state.get("id", "")
        
        # Format timestamp for readability
        try:
            dt_object = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
            formatted_time = dt_object.strftime("%Y-%m-%d %H:%M:%S UTC")
        except ValueError:
            formatted_time = timestamp
            
        processed_record = {
            "timestamp": formatted_time,
            "state_name": state_name,
            "state_id": state_id,
            "agent_id": record.get("agentId", ""),
            "previous_state": record.get("previousState", {}).get("name", "N/A")
        }
        
        processed_records.append(processed_record)
        
    return processed_records

Complete Working Example

The following script combines all the previous steps into a single, runnable module. It authenticates, fetches the agent history, processes the records, and prints the results.

import os
import requests
from datetime import datetime, timedelta, timezone
from dotenv import load_dotenv

load_dotenv()

CXONE_CLIENT_ID = os.getenv("CXONE_CLIENT_ID")
CXONE_CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET")
CXONE_API_BASE_URL = os.getenv("CXONE_API_BASE_URL")
CXONE_AGENT_ID = os.getenv("CXONE_AGENT_ID")

def get_access_token() -> str:
    """
    Retrieves an OAuth 2.0 access token from NICE CXone.
    """
    token_url = f"{CXONE_API_BASE_URL}/oauth/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": CXONE_CLIENT_ID,
        "client_secret": CXONE_CLIENT_SECRET
    }
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    
    response = requests.post(token_url, data=payload, headers=headers)
    if response.status_code != 200:
        raise Exception(f"Failed to obtain access token: {response.status_code} - {response.text}")
    
    return response.json()["access_token"]

def get_time_range_iso() -> tuple[str, str]:
    """
    Calculates the start and end times for the last 24 hours in ISO 8601 format.
    """
    now = datetime.now(timezone.utc)
    start_time = now - timedelta(hours=24)
    start_iso = start_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
    end_iso = now.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
    return start_iso, end_iso

def fetch_agent_history(access_token: str, agent_id: str, start_time: str, end_time: str) -> list[dict]:
    """
    Fetches agent history from NICE CXone with pagination.
    """
    base_url = f"{CXONE_API_BASE_URL}/api/v2/reporting/agents/{agent_id}/history"
    all_records = []
    page = 1
    page_size = 100
    
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Accept": "application/json"
    }
    
    params = {
        "start": start_time,
        "end": end_time,
        "pageSize": page_size,
        "page": page
    }
    
    while True:
        try:
            response = requests.get(base_url, headers=headers, params=params)
            
            if response.status_code == 200:
                data = response.json()
                records = data.get("records", [])
                
                if not records:
                    break
                
                all_records.extend(records)
                
                total_records = data.get("totalRecords", 0)
                if len(all_records) >= total_records:
                    break
                
                page += 1
                params["page"] = page
                
            elif response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 5))
                print(f"Rate limited. Waiting {retry_after} seconds...")
                import time
                time.sleep(retry_after)
                
            elif response.status_code in [401, 403]:
                raise Exception(f"Authentication/Authorization failed: {response.status_code}")
                
            else:
                raise Exception(f"API request failed: {response.status_code} - {response.text}")
                
        except requests.exceptions.RequestException as e:
            raise Exception(f"Network error: {str(e)}")
            
    return all_records

def process_agent_history(records: list[dict]) -> list[dict]:
    """
    Processes the raw agent history records.
    """
    processed_records = []
    for record in records:
        timestamp = record.get("timestamp", "")
        state = record.get("state", {})
        state_name = state.get("name", "Unknown")
        state_id = state.get("id", "")
        previous_state = record.get("previousState", {}).get("name", "N/A")
        
        try:
            dt_object = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
            formatted_time = dt_object.strftime("%Y-%m-%d %H:%M:%S UTC")
        except ValueError:
            formatted_time = timestamp
            
        processed_records.append({
            "timestamp": formatted_time,
            "state_name": state_name,
            "state_id": state_id,
            "agent_id": record.get("agentId", ""),
            "previous_state": previous_state
        })
    return processed_records

def main():
    """
    Main execution function.
    """
    try:
        # Step 1: Authenticate
        print("Authenticating with NICE CXone...")
        access_token = get_access_token()
        print("Authentication successful.")
        
        # Step 2: Define time range
        start_time, end_time = get_time_range_iso()
        print(f"Fetching history from {start_time} to {end_time}")
        
        # Step 3: Fetch data
        print("Fetching agent history...")
        raw_history = fetch_agent_history(access_token, CXONE_AGENT_ID, start_time, end_time)
        print(f"Retrieved {len(raw_history)} records.")
        
        # Step 4: Process data
        processed_history = process_agent_history(raw_history)
        
        # Step 5: Display results
        print("\n--- Agent State History ---")
        for record in processed_history:
            print(f"[{record['timestamp']}] State: {record['state_name']} (from: {record['previous_state']})")
            
    except Exception as e:
        print(f"Error: {str(e)}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The access token is invalid, expired, or missing. This can also occur if the OAuth client credentials are incorrect.
• Fix: Verify that CXONE_CLIENT_ID and CXONE_CLIENT_SECRET are correct in your .env file. Ensure the token request returns a 200 status code. Check that the token is being included in the Authorization header as Bearer <token>.

Error: 403 Forbidden

• Cause: The OAuth client does not have the required scopes (read:reporting or read:agent) or the user associated with the client does not have permission to view the agent’s data.
• Fix: Check the OAuth client configuration in the NICE CXone admin console. Ensure the read:reporting scope is enabled. Verify that the agent ID exists and that the client has access to that agent’s data.

Error: 429 Too Many Requests

• Cause: You have exceeded the API rate limit. NICE CXone enforces rate limits to ensure fair usage.
• Fix: Implement exponential backoff and retry logic. The response header Retry-After indicates how many seconds to wait before retrying. The code example above includes basic handling for this error.

Error: Empty Records

• Cause: The agent may not have had any state changes in the specified time range, or the time range is incorrect.
• Fix: Verify the start and end timestamps. Ensure the agent was active during the queried period. Check the totalRecords field in the API response to confirm if there are indeed no records.

Official References

• NICE CXone Reporting API Documentation
• Agent History Endpoint Reference
• OAuth 2.0 Client Credentials Grant