Query NICE CXone Agent State History with Python and the Reporting API

Query NICE CXone Agent State History with Python and the Reporting API

What You Will Build

  • A Python script that retrieves a detailed timeline of agent state changes for a specific user over the last 24 hours.
  • This solution utilizes the NICE CXone Reporting API (v2) endpoint /api/v2/reporting/agent/realtime to fetch near-real-time performance metrics and state logs.
  • The implementation uses Python 3.9+ with the requests library and standard JSON parsing.

Prerequisites

  • OAuth Client Type: Machine-to-Machine (Client Credentials) or User-to-Machine (Authorization Code) with a valid access token.
  • Required Scopes: reporting:read and users:read.
  • API Version: NICE CXone Reporting API v2.
  • Runtime: Python 3.9 or higher.
  • Dependencies:
    • requests (for HTTP calls)
    • pyjwt (optional, for debugging token contents)

Install the required package:

pip install requests

Authentication Setup

NICE CXone uses OAuth 2.0. For this tutorial, we assume you have already obtained an Access Token. If you are using Client Credentials flow, the token is obtained by posting your client_id and client_secret to the authorization server.

The Reporting API requires the reporting:read scope. Ensure your token contains this scope. You can verify this by decoding the JWT payload.

import requests
import json
from datetime import datetime, timedelta, timezone
import base64

# Configuration
AUTH_SERVER = "https://<your-subdomain>.niceincontact.com/oauth2/token"
API_BASE_URL = "https://<your-subdomain>.niceincontact.com"
CLIENT_ID = "your_client_id"
CLIENT_SECRET = "your_client_secret"

def get_access_token() -> str:
    """
    Retrieves an OAuth2 access token using Client Credentials flow.
    """
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    data = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "scope": "reporting:read users:read"
    }

    response = requests.post(AUTH_SERVER, headers=headers, data=data)
    
    if response.status_code != 200:
        raise Exception(f"Authentication failed: {response.status_code} - {response.text}")
    
    token_data = response.json()
    return token_data["access_token"]

Implementation

Step 1: Define the Time Window and Parameters

The Reporting API v2 uses a specific query structure to define time windows. To get data for the last 24 hours, we must calculate the start and end times in UTC ISO 8601 format. The API endpoint /api/v2/reporting/agent/realtime is designed for near-real-time data, but it also supports historical queries within a limited window depending on your tenant configuration. For a strict 24-hour history, we often use the /api/v2/reporting/agent endpoint with a timeframe parameter, or the realtime endpoint with a lookback.

However, the most robust way to get state history (transitions between Ready, Busy, Break, etc.) is to query the Agent Realtime report or the Agent Detail report. The realtime endpoint is preferred for recent data (last 24-48 hours depending on retention).

We will construct the query payload. The realtime endpoint accepts a JSON body defining the metrics and timeframe.

def build_query_payload(user_id: str) -> dict:
    """
    Constructs the JSON payload for the Reporting API.
    """
    # Calculate time window: Last 24 hours in UTC
    end_time = datetime.now(timezone.utc)
    start_time = end_time - timedelta(hours=24)

    # Format as ISO 8601 with timezone info
    start_str = start_time.isoformat()
    end_str = end_time.isoformat()

    payload = {
        "timeframe": {
            "start": start_str,
            "end": end_str
        },
        "metrics": [
            "agent_id",
            "agent_name",
            "state",
            "state_description",
            "timestamp"
        ],
        "filters": {
            "agent_id": [user_id]
        },
        "groupings": [
            "agent_id"
        ]
    }
    
    return payload

Step 2: Execute the API Call

We will send a POST request to /api/v2/reporting/agent/realtime. This endpoint is asynchronous in some contexts, but for smaller datasets (single agent, 24 hours), it often returns synchronously. If the dataset is large, CXone may return a 202 Accepted with a report-id. For this tutorial, we will handle the synchronous response first, as it is the most common pattern for ad-hoc debugging scripts.

Note: If your tenant has high concurrency, you may need to implement the asynchronous polling pattern. We will focus on the synchronous path for clarity, but include error handling for 202.

def fetch_agent_state_history(access_token: str, user_id: str) -> dict:
    """
    Queries the CXone Reporting API for agent state history.
    """
    url = f"{API_BASE_URL}/api/v2/reporting/agent/realtime"
    
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }

    payload = build_query_payload(user_id)

    # Send the request
    response = requests.post(url, headers=headers, json=payload)

    # Handle HTTP Status Codes
    if response.status_code == 200:
        return response.json()
    elif response.status_code == 202:
        # Asynchronous response: The report is being generated
        report_id = response.headers.get("Report-Id")
        print(f"Report generation started. Report ID: {report_id}")
        # In a production app, you would poll /api/v2/reporting/reports/{report_id}
        raise Exception("Asynchronous report generation triggered. Implement polling for production use.")
    elif response.status_code == 401:
        raise Exception("Unauthorized. Check your Access Token and Scopes.")
    elif response.status_code == 403:
        raise Exception("Forbidden. You likely lack the 'reporting:read' scope.")
    elif response.status_code == 400:
        raise Exception(f"Bad Request. Check your payload structure. Error: {response.text}")
    else:
        raise Exception(f"API Error: {response.status_code} - {response.text}")

Step 3: Process and Filter Results

The response from /api/v2/reporting/agent/realtime returns a structure containing data and metadata. The data array contains the state transitions. Each entry typically includes the state (numeric code) and state_description (human-readable string).

We need to parse this JSON and format it into a readable timeline.

def parse_state_history(raw_data: dict) -> list:
    """
    Parses the API response into a list of state transition events.
    """
    # The structure varies slightly by API version, but typically:
    # { "data": [ { "agent_id": "...", "state": 1, "timestamp": "..." } ] }
    
    if "data" not in raw_data:
        return []

    transitions = []
    
    for record in raw_data["data"]:
        # Extract relevant fields
        agent_id = record.get("agent_id")
        state_code = record.get("state")
        state_desc = record.get("state_description", "Unknown")
        timestamp_str = record.get("timestamp")
        
        # Parse timestamp
        try:
            ts = datetime.fromisoformat(timestamp_str.replace("Z", "+00:00"))
        except Exception:
            ts = None

        transitions.append({
            "agent_id": agent_id,
            "state_code": state_code,
            "state_description": state_desc,
            "timestamp": ts
        })

    # Sort by timestamp to ensure chronological order
    transitions.sort(key=lambda x: x["timestamp"] if x["timestamp"] else datetime.min.replace(tzinfo=timezone.utc))
    
    return transitions

Complete Working Example

Below is the full, copy-pasteable Python script. Replace the placeholder values for AUTH_SERVER, CLIENT_ID, CLIENT_SECRET, and USER_ID with your actual CXone tenant details.

import requests
import json
from datetime import datetime, timedelta, timezone
import sys

# --- Configuration ---
# Replace these with your actual CXone tenant details
CXONE_SUBDOMAIN = "your-subdomain"
AUTH_SERVER = f"https://{CXONE_SUBDOMAIN}.niceincontact.com/oauth2/token"
API_BASE_URL = f"https://{CXONE_SUBDOMAIN}.niceincontact.com"
CLIENT_ID = "your_client_id"
CLIENT_SECRET = "your_client_secret"
USER_ID = "target_agent_user_id"  # The UUID of the agent you want to query

# --- Authentication ---

def get_access_token() -> str:
    """
    Retrieves an OAuth2 access token using Client Credentials flow.
    """
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    data = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "scope": "reporting:read users:read"
    }

    try:
        response = requests.post(AUTH_SERVER, headers=headers, data=data, timeout=10)
        response.raise_for_status()
        token_data = response.json()
        return token_data["access_token"]
    except requests.exceptions.RequestException as e:
        print(f"Authentication failed: {e}")
        sys.exit(1)

# --- Reporting Logic ---

def build_query_payload(user_id: str) -> dict:
    """
    Constructs the JSON payload for the Reporting API.
    Queries the last 24 hours of agent state history.
    """
    end_time = datetime.now(timezone.utc)
    start_time = end_time - timedelta(hours=24)

    start_str = start_time.isoformat()
    end_str = end_time.isoformat()

    payload = {
        "timeframe": {
            "start": start_str,
            "end": end_str
        },
        "metrics": [
            "agent_id",
            "agent_name",
            "state",
            "state_description",
            "timestamp"
        ],
        "filters": {
            "agent_id": [user_id]
        },
        "groupings": [
            "agent_id"
        ]
    }
    return payload

def fetch_agent_state_history(access_token: str, user_id: str) -> dict:
    """
    Queries the CXone Reporting API for agent state history.
    """
    url = f"{API_BASE_URL}/api/v2/reporting/agent/realtime"
    
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }

    payload = build_query_payload(user_id)

    try:
        response = requests.post(url, headers=headers, json=payload, timeout=30)
        
        if response.status_code == 200:
            return response.json()
        elif response.status_code == 202:
            report_id = response.headers.get("Report-Id")
            print(f"Warning: Asynchronous report generation triggered (ID: {report_id}).")
            print("This script does not implement polling. For large datasets, implement polling against /api/v2/reporting/reports/{report_id}")
            return None
        elif response.status_code == 401:
            print("Error: Unauthorized. Check your Access Token and Scopes.")
            return None
        elif response.status_code == 403:
            print("Error: Forbidden. Ensure the token has 'reporting:read' scope.")
            return None
        elif response.status_code == 400:
            print(f"Error: Bad Request. {response.text}")
            return None
        else:
            print(f"Error: Unexpected status code {response.status_code}. Response: {response.text}")
            return None
    except requests.exceptions.RequestException as e:
        print(f"Network error: {e}")
        return None

def parse_and_print_history(raw_data: dict) -> None:
    """
    Parses the API response and prints a formatted timeline.
    """
    if not raw_data or "data" not in raw_data:
        print("No data found or invalid response structure.")
        return

    transitions = []
    
    for record in raw_data["data"]:
        agent_id = record.get("agent_id")
        state_code = record.get("state")
        state_desc = record.get("state_description", "Unknown State")
        timestamp_str = record.get("timestamp")
        
        if not timestamp_str:
            continue
            
        try:
            # Handle ISO 8601 parsing
            ts = datetime.fromisoformat(timestamp_str.replace("Z", "+00:00"))
        except ValueError:
            continue

        transitions.append({
            "timestamp": ts,
            "state_code": state_code,
            "state_description": state_desc
        })

    if not transitions:
        print("No state transitions found in the last 24 hours.")
        return

    # Sort chronologically
    transitions.sort(key=lambda x: x["timestamp"])

    print(f"\n--- Agent State History for {USER_ID} (Last 24 Hours) ---")
    print(f"{'Timestamp (UTC)':<25} | {'State Code':<10} | {'Description'}")
    print("-" * 60)
    
    for event in transitions:
        ts_str = event["timestamp"].strftime("%Y-%m-%d %H:%M:%S")
        print(f"{ts_str:<25} | {str(event['state_code']):<10} | {event['state_description']}")

    print("-" * 60)
    print(f"Total transitions: {len(transitions)}")

# --- Main Execution ---

if __name__ == "__main__":
    print("Starting CXone Agent State History Query...")
    
    # Step 1: Authenticate
    token = get_access_token()
    if not token:
        sys.exit(1)
    
    print("Authentication successful.")

    # Step 2: Fetch Data
    raw_data = fetch_agent_state_history(token, USER_ID)

    # Step 3: Process and Display
    if raw_data:
        parse_and_print_history(raw_data)
    else:
        print("Failed to retrieve data.")

Common Errors & Debugging

Error: 401 Unauthorized

  • Cause: The Access Token is expired, invalid, or missing.
  • Fix: Verify that get_access_token() is returning a valid JWT. Check that the Authorization header is formatted exactly as Bearer <token> with a space after “Bearer”.
  • Code Check: Ensure the scope in the token request includes reporting:read.

Error: 403 Forbidden

  • Cause: The OAuth client does not have the required permissions.
  • Fix: In the CXone Admin Portal, navigate to Security > OAuth Clients. Edit your client and ensure the reporting:read scope is checked. Also verify that the user associated with the token (if using User-to-Machine) has access to the Reporting module.

Error: 400 Bad Request

  • Cause: The JSON payload is malformed or the time window is invalid.
  • Fix: Ensure start and end times are in ISO 8601 format. Ensure start is before end. The filters object must use arrays for values (e.g., "agent_id": ["uuid"]), not strings.

Error: 202 Accepted (Asynchronous Response)

  • Cause: The query is too large for synchronous processing (e.g., querying multiple agents over a long period).
  • Fix: The script above detects this and prints a warning. In production, you must implement a polling loop:
    1. Extract Report-Id from the response headers.
    2. Poll GET /api/v2/reporting/reports/{report-id} every 2-5 seconds.
    3. Check the status field in the response. When status is completed, download the data using the downloadUrl provided.

Error: Empty Data

  • Cause: The agent was not logged in during the queried timeframe, or the user ID is incorrect.
  • Fix: Verify the USER_ID is the correct UUID for the agent. Check the Admin Console to confirm the agent was active in the last 24 hours.

Official References