Query Agent State History via NICE CXone Reporting API v2

Query Agent State History via NICE CXone Reporting API v2

What You Will Build

  • A Python script that authenticates with NICE CXone and retrieves a detailed timeline of agent login, logout, and status changes for the last 24 hours.
  • This tutorial uses the NICE CXone Reporting API v2 (/api/v2/reporting) to query the AgentStateHistory report.
  • The implementation uses Python 3.8+ with the requests library for HTTP handling.

Prerequisites

  • OAuth Client Type: Service Account or Client Credentials. You need a client ID and client secret with appropriate permissions.
  • Required Scopes: reporting:read is mandatory. If you need to identify specific agents by name rather than ID, user:read may be helpful, but this tutorial relies on Agent IDs.
  • API Version: CXone Reporting API v2.
  • Language/Runtime: Python 3.8 or higher.
  • External Dependencies:
    • requests: For HTTP requests.
    • python-dotenv: For managing environment variables securely.

Install dependencies:

pip install requests python-dotenv

Authentication Setup

NICE CXone uses OAuth 2.0 for authentication. The standard flow for server-to-server integrations is the Client Credentials Grant. You must exchange your client ID and secret for an access token before making any Reporting API calls.

The token endpoint is https://<your-cxone-domain>.api.nice.com/oauth/token.

Token Retrieval Code

Create a file named cxone_auth.py to handle authentication. This module provides a function to fetch and return a valid bearer token.

import requests
import os
from datetime import datetime, timedelta

# Load environment variables
# Ensure you have a .env file with:
# CXONE_DOMAIN=your-domain
# CXONE_CLIENT_ID=your-client-id
# CXONE_CLIENT_SECRET=your-client-secret

def get_access_token() -> str:
    """
    Retrieves an OAuth2 access token from NICE CXone.
    """
    domain = os.getenv("CXONE_DOMAIN")
    client_id = os.getenv("CXONE_CLIENT_ID")
    client_secret = os.getenv("CXONE_CLIENT_SECRET")

    if not all([domain, client_id, client_secret]):
        raise ValueError("Missing required environment variables: CXONE_DOMAIN, CXONE_CLIENT_ID, CXONE_CLIENT_SECRET")

    token_url = f"https://{domain}.api.nice.com/oauth/token"
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    data = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
        "scope": "reporting:read"
    }

    response = requests.post(token_url, headers=headers, data=data)

    if response.status_code != 200:
        raise Exception(f"Failed to get token: {response.status_code} - {response.text}")

    token_data = response.json()
    return token_data["access_token"]

Token Caching Strategy

OAuth tokens in CXone typically expire after one hour (3600 seconds). In a production application, you should cache the token and only request a new one when the current one is expired or close to expiration. For this tutorial, we assume a single run within the token validity window.

Implementation

The NICE CXone Reporting API v2 uses a request/response pattern for asynchronous reporting. Unlike synchronous REST endpoints that return data immediately, complex reports (like Agent State History) require two steps:

  1. Submit Request: POST a report specification to /api/v2/reporting. The API returns a reportId and a status of PENDING.
  2. Poll for Results: GET the report status using the reportId. When the status changes to SUCCESS, you download the result using the downloadUrl provided in the response.

Step 1: Define the Report Specification

We need to construct a JSON payload that defines the report type, date range, and filters. For Agent State History, the report type is AgentStateHistory.

The date range must be in ISO 8601 format. We want the last 24 hours.

from datetime import datetime, timezone, timedelta
import json

def get_agent_state_history_spec(agent_ids: list[str]) -> dict:
    """
    Constructs the report specification for Agent State History.
    
    Args:
        agent_ids: A list of Agent IDs (e.g., ["agent123", "agent456"])
    
    Returns:
        The JSON payload for the POST request.
    """
    # Calculate date range: Last 24 hours
    end_date = datetime.now(timezone.utc)
    start_date = end_date - timedelta(hours=24)

    # Format dates as ISO 8601 strings
    start_date_str = start_date.isoformat()
    end_date_str = end_date.isoformat()

    report_spec = {
        "reportType": "AgentStateHistory",
        "reportName": "Agent State History - Last 24 Hours",
        "dateRange": {
            "startDate": start_date_str,
            "endDate": end_date_str,
            "granularity": "DAY" # Granularity is less relevant for state history but required by schema
        },
        "filters": {
            "agentIds": agent_ids
        },
        # Optional: Specify columns if you do not want all default columns
        # "columns": ["agentId", "state", "startTime", "endTime"]
    }

    return report_spec

Critical Parameter Explanation:

  • reportType: Must be exactly AgentStateHistory. Case-sensitive.
  • filters.agentIds: If you omit this, the report will attempt to pull history for all agents in the organization, which can cause timeout errors or extremely long processing times. Always filter by specific agents if possible.
  • dateRange: The API is strict about time zones. Always use UTC.

Step 2: Submit the Report Request

Now we send the specification to the CXone API. This call is synchronous but returns metadata, not data.

import requests
import time

def submit_report_request(token: str, domain: str, report_spec: dict) -> dict:
    """
    Submits a report request to CXone Reporting API.
    
    Args:
        token: OAuth access token.
        domain: CXone domain (without .api.nice.com).
        report_spec: The report specification dictionary.
    
    Returns:
        The JSON response containing reportId and status.
    """
    url = f"https://{domain}.api.nice.com/api/v2/reporting"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    response = requests.post(url, headers=headers, json=report_spec)

    if response.status_code == 202:
        # 202 Accepted means the report is being processed
        return response.json()
    elif response.status_code == 401:
        raise Exception("Unauthorized: Check your OAuth token.")
    elif response.status_code == 403:
        raise Exception("Forbidden: Your client may lack 'reporting:read' scope.")
    elif response.status_code == 400:
        raise Exception(f"Bad Request: {response.text}")
    else:
        raise Exception(f"Unexpected status code: {response.status_code} - {response.text}")

Expected Response:

{
  "reportId": "a1b2c3d4-e5f6-7890-1234-567890abcdef",
  "status": "PENDING",
  "reportName": "Agent State History - Last 24 Hours",
  "createdDate": "2023-10-27T10:00:00Z"
}

Step 3: Poll for Report Completion

The report processing time varies based on data volume. We must poll the status endpoint until the status is SUCCESS or FAILED.

def poll_report_status(token: str, domain: str, report_id: str, max_retries: int = 60, wait_seconds: int = 5) -> dict:
    """
    Polls the report status until it is complete or fails.
    
    Args:
        token: OAuth access token.
        domain: CXone domain.
        report_id: The ID returned from submit_report_request.
        max_retries: Maximum number of polling attempts.
        wait_seconds: Seconds to wait between polls.
    
    Returns:
        The final report status JSON.
    """
    url = f"https://{domain}.api.nice.com/api/v2/reporting/{report_id}"
    headers = {
        "Authorization": f"Bearer {token}"
    }

    for i in range(max_retries):
        response = requests.get(url, headers=headers)
        
        if response.status_code != 200:
            raise Exception(f"Failed to get report status: {response.status_code} - {response.text}")
        
        status_data = response.json()
        current_status = status_data.get("status")

        if current_status == "SUCCESS":
            return status_data
        elif current_status == "FAILED":
            raise Exception(f"Report generation failed: {status_data.get('errorMessage', 'Unknown error')}")
        elif current_status in ["PENDING", "RUNNING"]:
            print(f"Report status: {current_status}. Waiting {wait_seconds} seconds...")
            time.sleep(wait_seconds)
        else:
            raise Exception(f"Unknown report status: {current_status}")

    raise Exception("Report polling timed out. The report is taking longer than expected.")

Step 4: Download and Process Results

Once the status is SUCCESS, the response contains a downloadUrl. This URL is temporary (usually valid for 1-2 hours) and contains the actual CSV or JSON data.

def download_report_results(token: str, download_url: str) -> list[dict]:
    """
    Downloads the report results from the provided URL.
    
    Args:
        token: OAuth access token.
        download_url: The downloadUrl from the SUCCESS status response.
    
    Returns:
        A list of dictionaries representing each row of the report.
    """
    headers = {
        "Authorization": f"Bearer {token}"
    }

    response = requests.get(download_url, headers=headers)

    if response.status_code != 200:
        raise Exception(f"Failed to download report: {response.status_code} - {response.text}")

    # CXone Reporting API v2 returns JSON by default if requested, 
    # but often defaults to CSV. We will assume JSON for easier parsing in Python.
    # If you receive CSV, you would use the csv module instead.
    
    try:
        data = response.json()
        return data
    except json.JSONDecodeError:
        # Fallback if the response is not JSON (e.g., CSV)
        print("Warning: Response was not JSON. Returning raw text.")
        return [{"raw_content": response.text}]

Data Structure Note:
The AgentStateHistory report returns an array of objects. Each object represents a state change event. Common fields include:

  • agentId: The unique identifier of the agent.
  • state: The state name (e.g., “Available”, “Auxiliary”, “Login”).
  • startTime: ISO 8601 timestamp when the state started.
  • endTime: ISO 8601 timestamp when the state ended. Null if the state is still active.
  • duration: The duration of the state in milliseconds.

Complete Working Example

Below is the full, copy-pasteable script. Save this as fetch_agent_state.py.

import os
import json
import time
import requests
from datetime import datetime, timezone, timedelta
from typing import List, Dict, Any

# Configuration
CXONE_DOMAIN = os.getenv("CXONE_DOMAIN")
CXONE_CLIENT_ID = os.getenv("CXONE_CLIENT_ID")
CXONE_CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET")
AGENT_IDS = os.getenv("AGENT_IDS", "").split(",")  # Comma-separated list of Agent IDs

def get_access_token() -> str:
    """Retrieves an OAuth2 access token from NICE CXone."""
    if not all([CXONE_DOMAIN, CXONE_CLIENT_ID, CXONE_CLIENT_SECRET]):
        raise ValueError("Missing required environment variables: CXONE_DOMAIN, CXONE_CLIENT_ID, CXONE_CLIENT_SECRET")

    token_url = f"https://{CXONE_DOMAIN}.api.nice.com/oauth/token"
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    data = {
        "grant_type": "client_credentials",
        "client_id": CXONE_CLIENT_ID,
        "client_secret": CXONE_CLIENT_SECRET,
        "scope": "reporting:read"
    }

    response = requests.post(token_url, headers=headers, data=data)
    if response.status_code != 200:
        raise Exception(f"Failed to get token: {response.status_code} - {response.text}")
    
    return response.json()["access_token"]

def get_agent_state_history_spec(agent_ids: List[str]) -> Dict[str, Any]:
    """Constructs the report specification for Agent State History."""
    end_date = datetime.now(timezone.utc)
    start_date = end_date - timedelta(hours=24)

    report_spec = {
        "reportType": "AgentStateHistory",
        "reportName": "Agent State History - Last 24 Hours",
        "dateRange": {
            "startDate": start_date.isoformat(),
            "endDate": end_date.isoformat(),
            "granularity": "DAY"
        },
        "filters": {
            "agentIds": agent_ids
        }
    }
    return report_spec

def submit_report_request(token: str, report_spec: Dict[str, Any]) -> Dict[str, Any]:
    """Submits a report request to CXone Reporting API."""
    url = f"https://{CXONE_DOMAIN}.api.nice.com/api/v2/reporting"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    response = requests.post(url, headers=headers, json=report_spec)

    if response.status_code == 202:
        return response.json()
    elif response.status_code == 401:
        raise Exception("Unauthorized: Check your OAuth token.")
    elif response.status_code == 403:
        raise Exception("Forbidden: Your client may lack 'reporting:read' scope.")
    elif response.status_code == 400:
        raise Exception(f"Bad Request: {response.text}")
    else:
        raise Exception(f"Unexpected status code: {response.status_code} - {response.text}")

def poll_report_status(token: str, report_id: str, max_retries: int = 60, wait_seconds: int = 5) -> Dict[str, Any]:
    """Polls the report status until it is complete or fails."""
    url = f"https://{CXONE_DOMAIN}.api.nice.com/api/v2/reporting/{report_id}"
    headers = {"Authorization": f"Bearer {token}"}

    for i in range(max_retries):
        response = requests.get(url, headers=headers)
        
        if response.status_code != 200:
            raise Exception(f"Failed to get report status: {response.status_code} - {response.text}")
        
        status_data = response.json()
        current_status = status_data.get("status")

        if current_status == "SUCCESS":
            return status_data
        elif current_status == "FAILED":
            raise Exception(f"Report generation failed: {status_data.get('errorMessage', 'Unknown error')}")
        elif current_status in ["PENDING", "RUNNING"]:
            print(f"[{i+1}/{max_retries}] Report status: {current_status}. Waiting {wait_seconds} seconds...")
            time.sleep(wait_seconds)
        else:
            raise Exception(f"Unknown report status: {current_status}")

    raise Exception("Report polling timed out.")

def download_report_results(token: str, download_url: str) -> List[Dict[str, Any]]:
    """Downloads the report results from the provided URL."""
    headers = {"Authorization": f"Bearer {token}"}
    response = requests.get(download_url, headers=headers)

    if response.status_code != 200:
        raise Exception(f"Failed to download report: {response.status_code} - {response.text}")

    try:
        return response.json()
    except json.JSONDecodeError:
        raise Exception("Failed to parse JSON response. Ensure the API returned JSON.")

def main():
    """Main execution flow."""
    try:
        print("Step 1: Authenticating...")
        token = get_access_token()
        
        if not AGENT_IDS:
            print("Warning: No Agent IDs provided. Please set AGENT_IDS in environment variables.")
            return

        print(f"Step 2: Submitting report for agents: {AGENT_IDS}")
        spec = get_agent_state_history_spec(AGENT_IDS)
        report_response = submit_report_request(token, spec)
        report_id = report_response["reportId"]
        print(f"Report submitted with ID: {report_id}")

        print("Step 3: Polling for report completion...")
        final_status = poll_report_status(token, report_id)
        
        if final_status["status"] == "SUCCESS":
            print("Step 4: Downloading results...")
            download_url = final_status["downloadUrl"]
            results = download_report_results(token, download_url)
            
            print(f"Success! Retrieved {len(results)} state history records.")
            
            # Example: Print first 5 records
            for record in results[:5]:
                print(json.dumps(record, indent=2))
            
            # Save to file
            with open("agent_state_history.json", "w") as f:
                json.dump(results, f, indent=2)
            print("Results saved to agent_state_history.json")
        else:
            print(f"Report did not complete successfully. Status: {final_status['status']}")

    except Exception as e:
        print(f"Error: {e}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden on Report Submission

Cause: The OAuth client used to generate the token does not have the reporting:read scope.

Fix:

  1. Log in to the CXone Admin Portal.
  2. Navigate to Admin > Security > OAuth Clients.
  3. Select your client.
  4. Ensure reporting:read is checked in the Scopes section.
  5. Regenerate the token.

Error: 400 Bad Request - “Invalid date range”

Cause: The startDate is after the endDate, or the date format is not ISO 8601.

Fix:
Ensure your Python code uses datetime.now(timezone.utc).isoformat(). Do not use local time zones unless you explicitly convert them to UTC. CXone APIs enforce UTC.

Error: Report Polling Times Out

Cause: The report is too large. Querying state history for hundreds of agents over 24 hours generates massive datasets.

Fix:

  1. Reduce the number of agents in the agentIds filter.
  2. Reduce the date range (e.g., last 1 hour instead of 24 hours) for testing.
  3. Increase max_retries in the poll_report_status function if the data volume is legitimately high.

Error: Empty Results

Cause: The agents specified did not change state during the requested time window, or the agentIds are incorrect.

Fix:

  1. Verify the agentIds are valid by checking the Admin Portal.
  2. Ensure the agents were actually logged in or active during the last 24 hours.
  3. Check if the report returned successfully but with an empty array []. This is valid if no events occurred.

Official References