Querying NICE CXone Agent State History via the Reporting API

What You Will Build

This tutorial demonstrates how to retrieve a historical log of agent state changes (such as Ready, Busy, Wrap-up, or Offline) for a specific agent over the last 24 hours. You will use the NICE CXone Reporting API (v2) to execute a custom query that filters by agent ID and timestamp range. The implementation is provided in Python using the requests library, focusing on robust error handling and pagination logic.

Prerequisites

• NICE CXone Tenant Access: You must have a valid NICE CXone tenant URL (e.g., https://platform.us2.niceincontact.com).
• OAuth 2.0 Credentials: A Client ID and Client Secret with the following scopes:
  • reporting:read (Required for accessing reporting data)
  • agents:read (Optional, if you need to resolve agent IDs from names)
• Python Environment: Python 3.8 or higher.
• Dependencies:
  • requests: For HTTP communication.
  • python-dateutil: For robust date parsing and manipulation.

Install dependencies via pip:

pip install requests python-dateutil

Authentication Setup

NICE CXone uses OAuth 2.0 Client Credentials flow for API access. You must exchange your Client ID and Client Secret for an access token before making any reporting queries.

The token endpoint is typically located at {tenant_url}/oauth/token.

Token Acquisition Code

import requests
import time
from typing import Optional

class CxoneAuth:
    def __init__(self, tenant_url: str, client_id: str, client_secret: str):
        self.tenant_url = tenant_url.rstrip('/')
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_endpoint = f"{self.tenant_url}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0

    def get_token(self) -> str:
        """
        Retrieves an OAuth access token.
        Implements basic caching to avoid unnecessary token refreshes.
        """
        if self.access_token and time.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_endpoint, data=payload, headers=headers)
            response.raise_for_status()
            data = response.json()
            
            self.access_token = data["access_token"]
            # Expire token 60 seconds before actual expiry to prevent race conditions
            self.token_expiry = time.time() + (data.get("expires_in", 3600) - 60)
            
            return self.access_token

        except requests.exceptions.HTTPError as e:
            print(f"Authentication failed: {e.response.status_code} - {e.response.text}")
            raise
        except requests.exceptions.RequestException as e:
            print(f"Network error during authentication: {e}")
            raise

# Initialize Auth
# Replace these with your actual credentials
CXONE_TENANT = "https://platform.us2.niceincontact.com"
CLIENT_ID = "your_client_id_here"
CLIENT_SECRET = "your_client_secret_here"

auth = CxoneAuth(CXONE_TENANT, CLIENT_ID, CLIENT_SECRET)
ACCESS_TOKEN = auth.get_token()

Implementation

Step 1: Constructing the Reporting Query

The CXone Reporting API v2 does not have a dedicated endpoint for “agent state history.” Instead, it uses a generic query execution endpoint (/api/v2/reporting/query) where you define the data source, filters, and columns in the request body.

To get agent state history, you must target the agent_activity or agent_state_history data source. The most reliable source for detailed state transitions is agent_activity.

Key parameters for the query body:

• sourceId: The ID of the data source. For agent activity, this is typically agent_activity.
• columns: The fields you want to retrieve (e.g., agentName, stateName, startTime, endTime).
• filters: A list of filter objects to constrain the results by agent ID and time range.

Step 2: Defining the Request Payload

We need to calculate the start and end times for the last 24 hours. The API expects timestamps in ISO 8601 format.

from datetime import datetime, timedelta, timezone
import json

def build_agent_state_query(agent_id: str) -> dict:
    """
    Constructs the JSON payload for the Reporting API query.
    
    Args:
        agent_id: The unique identifier of the agent (not the name).
        
    Returns:
        A dictionary representing the query payload.
    """
    now = datetime.now(timezone.utc)
    start_time = now - timedelta(hours=24)
    
    # Format timestamps as ISO 8601 with 'Z' suffix for UTC
    start_iso = start_time.strftime("%Y-%m-%dT%H:%M:%S.") + f"{start_time.microsecond // 1000:03d}Z"
    end_iso = now.strftime("%Y-%m-%dT%H:%M:%S.") + f"{now.microsecond // 1000:03d}Z"

    query_payload = {
        "sourceId": "agent_activity",
        "columns": [
            "agentName",
            "agentId",
            "stateName",
            "skillName",
            "startTime",
            "endTime",
            "duration"
        ],
        "filters": [
            {
                "field": "agentId",
                "operator": "eq",
                "value": agent_id
            },
            {
                "field": "startTime",
                "operator": "gte",
                "value": start_iso
            },
            {
                "field": "startTime",
                "operator": "lte",
                "value": end_iso
            }
        ],
        "sort": [
            {
                "field": "startTime",
                "order": "asc"
            }
        ],
        "pageSize": 100,
        "page": 1
    }
    
    return query_payload

# Example Agent ID
AGENT_ID = "12345678-1234-1234-1234-123456789012"
QUERY_PAYLOAD = build_agent_state_query(AGENT_ID)
print(json.dumps(QUERY_PAYLOAD, indent=2))

Step 3: Executing the Query and Handling Pagination

The Reporting API returns paginated results. You must check the pageInfo in the response to determine if more data is available. If hasMore is true, increment the page number and resend the request.

def fetch_agent_state_history(agent_id: str, token: str, tenant_url: str) -> list:
    """
    Fetches agent state history for the last 24 hours with pagination support.
    
    Args:
        agent_id: The agent's unique ID.
        token: OAuth access token.
        tenant_url: Base URL of the CXone tenant.
        
    Returns:
        A list of dictionaries representing each state change record.
    """
    api_url = f"{tenant_url}/api/v2/reporting/query"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }
    
    all_records = []
    page = 1
    max_pages = 50 # Safety limit to prevent infinite loops
    
    while page <= max_pages:
        # Update the page number in the payload
        current_payload = build_agent_state_query(agent_id)
        current_payload["page"] = page
        
        try:
            response = requests.post(api_url, json=current_payload, headers=headers)
            
            # Handle Rate Limiting (429)
            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 1))
                print(f"Rate limited. Waiting {retry_after} seconds...")
                time.sleep(retry_after)
                continue
                
            response.raise_for_status()
            data = response.json()
            
            # Extract records
            records = data.get("records", [])
            all_records.extend(records)
            
            # Check pagination info
            page_info = data.get("pageInfo", {})
            has_more = page_info.get("hasMore", False)
            
            if not has_more:
                break
                
            page += 1
            
        except requests.exceptions.HTTPError as e:
            print(f"HTTP Error on page {page}: {e.response.status_code} - {e.response.text}")
            raise
        except requests.exceptions.RequestException as e:
            print(f"Network error on page {page}: {e}")
            raise

    return all_records

# Execute the fetch
try:
    history = fetch_agent_state_history(AGENT_ID, ACCESS_TOKEN, CXONE_TENANT)
    print(f"Retrieved {len(history)} state change records.")
    if history:
        print("Sample record:")
        print(json.dumps(history[0], indent=2))
except Exception as e:
    print(f"Failed to fetch history: {e}")

Complete Working Example

Below is the complete, consolidated script. Save this as cxone_agent_history.py and update the credentials at the top.

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

class CxoneReportingClient:
    def __init__(self, tenant_url: str, client_id: str, client_secret: str):
        self.tenant_url = tenant_url.rstrip('/')
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_endpoint = f"{self.tenant_url}/oauth/token"
        self.api_endpoint = f"{self.tenant_url}/api/v2/reporting/query"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0

    def _get_token(self) -> str:
        """Retrieves OAuth token with caching."""
        if self.access_token and time.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"}

        response = requests.post(self.token_endpoint, data=payload, headers=headers)
        response.raise_for_status()
        data = response.json()
        
        self.access_token = data["access_token"]
        self.token_expiry = time.time() + (data.get("expires_in", 3600) - 60)
        return self.access_token

    def get_agent_state_history(self, agent_id: str, hours: int = 24) -> List[Dict]:
        """
        Queries the Reporting API for agent state history.
        
        Args:
            agent_id: The UUID of the agent.
            hours: Number of hours to look back (default 24).
            
        Returns:
            List of state history records.
        """
        token = self._get_token()
        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

        now = datetime.now(timezone.utc)
        start_time = now - timedelta(hours=hours)
        
        start_iso = start_time.strftime("%Y-%m-%dT%H:%M:%S.") + f"{start_time.microsecond // 1000:03d}Z"
        end_iso = now.strftime("%Y-%m-%dT%H:%M:%S.") + f"{now.microsecond // 1000:03d}Z"

        base_payload = {
            "sourceId": "agent_activity",
            "columns": [
                "agentName",
                "agentId",
                "stateName",
                "skillName",
                "startTime",
                "endTime",
                "duration"
            ],
            "filters": [
                {"field": "agentId", "operator": "eq", "value": agent_id},
                {"field": "startTime", "operator": "gte", "value": start_iso},
                {"field": "startTime", "operator": "lte", "value": end_iso}
            ],
            "sort": [{"field": "startTime", "order": "asc"}],
            "pageSize": 100,
            "page": 1
        }

        all_records = []
        page = 1
        max_pages = 100 # Prevent infinite loops

        while page <= max_pages:
            current_payload = base_payload.copy()
            current_payload["page"] = page

            try:
                response = requests.post(self.api_endpoint, json=current_payload, headers=headers)
                
                if response.status_code == 429:
                    retry_after = int(response.headers.get("Retry-After", 5))
                    print(f"Rate limited (429). Retrying in {retry_after}s...")
                    time.sleep(retry_after)
                    continue
                
                response.raise_for_status()
                data = response.json()
                
                records = data.get("records", [])
                all_records.extend(records)
                
                page_info = data.get("pageInfo", {})
                if not page_info.get("hasMore", False):
                    break
                    
                page += 1

            except requests.exceptions.HTTPError as e:
                print(f"HTTP Error: {e.response.status_code}")
                print(e.response.text)
                raise
            except requests.exceptions.RequestException as e:
                print(f"Request failed: {e}")
                raise

        return all_records

if __name__ == "__main__":
    # Configuration
    CXONE_TENANT = "https://platform.us2.niceincontact.com"
    CLIENT_ID = "your_client_id"
    CLIENT_SECRET = "your_client_secret"
    TARGET_AGENT_ID = "12345678-1234-1234-1234-123456789012"

    client = CxoneReportingClient(CXONE_TENANT, CLIENT_ID, CLIENT_SECRET)
    
    try:
        history = client.get_agent_state_history(TARGET_AGENT_ID, hours=24)
        print(f"Total records found: {len(history)}")
        for record in history[:5]: # Print first 5 records
            print(json.dumps(record, indent=2))
    except Exception as e:
        print(f"Error: {e}")

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token is expired, invalid, or the Client ID/Secret is incorrect.
• Fix: Verify your credentials. Ensure the get_token method is called before every batch of requests. Check that the token has not expired (the code above handles refresh, but manual testing may require a fresh token).

Error: 403 Forbidden

• Cause: The OAuth client lacks the reporting:read scope.
• Fix: Go to the CXone Admin Console > Platform > OAuth Clients. Edit your client and ensure reporting:read is checked in the scopes list. Re-authorize the application if necessary.

Error: 400 Bad Request (Invalid Query)

• Cause: The sourceId is incorrect, or the filter fields do not exist in the agent_activity data source.
• Fix: Verify that sourceId is exactly agent_activity. Check the CXone Reporting API documentation for the exact field names in the agent_activity source. Common typos include using agent_id instead of agentId.

Error: Empty Results

• Cause: The agent was offline during the entire 24-hour window, or the time zone conversion is incorrect.
• Fix: Ensure the timestamps are in UTC. The code uses datetime.now(timezone.utc). If you are testing with an agent who is currently active, reduce the hours parameter to 1 to see recent data. Also, verify that the agent_id is correct and belongs to an agent who has logged in recently.

Error: 429 Too Many Requests

• Cause: You have exceeded the rate limit for the Reporting API.
• Fix: The provided code includes a retry loop with Retry-After header parsing. For high-volume applications, implement exponential backoff and cache results where possible. Avoid polling the API more than once every few seconds per agent.

Official References

• NICE CXone Reporting API Documentation
• OAuth 2.0 Client Credentials Flow
• Agent Activity Data Source Schema