Query Agent State History for the Last 24 Hours Using the CXone Reporting API v2

What You Will Build

• A Python script that authenticates via OAuth 2.0, submits a reporting query for agent state history over a rolling 24-hour window, and parses the returned JSON into a structured list of state durations.
• This uses the NICE CXone Reporting API v2 endpoint POST /api/v2/reporting/agents/state-history.
• The tutorial uses Python 3.9+ with the requests library and includes production-grade retry, pagination, and error handling.

Prerequisites

• OAuth 2.0 Client Credentials grant type with the reporting:view scope
• CXone Reporting API v2 (current stable version)
• Python 3.9+ runtime
• requests library installed via pip install requests
• Valid CXone environment URL (for example, api-us-01.niceincontact.com or api-eu-01.niceincontact.com)
• A registered OAuth client with reporting permissions enabled

Authentication Setup

CXone uses the OAuth 2.0 Client Credentials flow. The client ID and client secret are sent as HTTP Basic Auth credentials to the /oauth/token endpoint. The response contains an access token and an expires_in value in seconds. You must cache the token and refresh it before expiration to avoid 401 errors during long-running queries.

The required scope for all reporting endpoints is reporting:view.

import requests
from datetime import datetime, timedelta
import time
import json
from typing import Optional

class CXoneReportingClient:
    def __init__(self, client_id: str, client_secret: str, environment: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = f"https://{environment}"
        self.token_url = f"{self.base_url}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0.0

    def _get_token(self) -> str:
        """Fetches or returns a cached OAuth token."""
        if self.access_token and datetime.now().timestamp() < self.token_expiry:
            return self.access_token

        payload = {
            "grant_type": "client_credentials",
            "scope": "reporting:view"
        }
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        response = requests.post(
            self.token_url,
            data=payload,
            headers=headers,
            auth=(self.client_id, self.client_secret)
        )
        response.raise_for_status()
        data = response.json()
        
        if "access_token" not in data:
            raise ValueError("OAuth response missing access_token")
            
        self.access_token = data["access_token"]
        expires_in = data.get("expires_in", 3600)
        # Subtract 60 seconds to provide a refresh buffer
        self.token_expiry = datetime.now().timestamp() + (expires_in - 60)
        return self.access_token

Implementation

Step 1: Build the 24-Hour Time Range and Query Payload

The Reporting API v2 expects a JSON body defining the time window, groupings, metrics, and pagination size. Agent state history tracks how long agents spend in specific states (for example, available, away, wrapup). You must group by agentId and stateId to retrieve meaningful history. The metrics array requests the aggregate duration in milliseconds.

    def _build_query_payload(self, hours_ago: int = 24, page_size: int = 100) -> dict:
        """Constructs the reporting query payload for the last N hours."""
        now = datetime.utcnow()
        start_time = now - timedelta(hours=hours_ago)
        
        # CXone requires ISO 8601 with millisecond precision
        time_range = {
            "from": start_time.strftime("%Y-%m-%dT%H:%M:%S.") + f"{start_time.microsecond // 1000:03d}Z",
            "to": now.strftime("%Y-%m-%dT%H:%M:%S.") + f"{now.microsecond // 1000:03d}Z"
        }
        
        payload = {
            "timeRange": time_range,
            "groupings": ["agentId", "stateId"],
            "metrics": ["duration"],
            "pageSize": page_size
        }
        return payload

Step 2: Execute the Reporting Query with Retry and Pagination

The POST /api/v2/reporting/agents/state-history endpoint returns paginated results. You must handle the continuationToken to fetch all pages. The CXone API enforces strict rate limits. A 429 response requires exponential backoff. The code below implements a retry loop for 429 errors and processes pagination until continuationToken is null.

    def query_agent_state_history(self, hours_ago: int = 24) -> list[dict]:
        """Fetches agent state history with pagination and 429 retry logic."""
        endpoint = f"{self.base_url}/api/v2/reporting/agents/state-history"
        headers = {
            "Authorization": f"Bearer {self._get_token()}",
            "Content-Type": "application/json"
        }
        
        payload = self._build_query_payload(hours_ago)
        all_results = []
        max_retries = 3
        retry_delay = 2.0
        
        while True:
            retries = 0
            while retries < max_retries:
                response = requests.post(endpoint, json=payload, headers=headers)
                
                if response.status_code == 429:
                    retries += 1
                    time.sleep(retry_delay)
                    retry_delay *= 2  # Exponential backoff
                    continue
                    
                if response.status_code == 401:
                    self.access_token = None  # Force token refresh
                    headers["Authorization"] = f"Bearer {self._get_token()}"
                    continue
                    
                response.raise_for_status()
                break
            else:
                raise RuntimeError(f"Max retries exceeded for 429 rate limit")
                
            data = response.json()
            results = data.get("results", [])
            all_results.extend(results)
            
            continuation_token = data.get("continuationToken")
            if not continuation_token:
                break
                
            payload["continuationToken"] = continuation_token
            
        return all_results

Step 3: Process and Flatten the Response

The API returns a nested structure. Each item contains a groupings object and a metrics object. You must extract the agent ID, state ID, and duration, then convert the millisecond duration into a human-readable format. This step also filters out zero-duration entries that may appear due to reporting aggregation thresholds.

    @staticmethod
    def parse_state_results(raw_results: list[dict]) -> list[dict]:
        """Flattens nested reporting output into a readable list."""
        parsed = []
        for item in raw_results:
            groupings = item.get("groupings", {})
            metrics = item.get("metrics", {})
            
            duration_ms = metrics.get("duration", 0)
            if duration_ms <= 0:
                continue
                
            duration_seconds = duration_ms / 1000.0
            duration_minutes = duration_seconds / 60.0
            
            parsed.append({
                "agent_id": groupings.get("agentId", "unknown"),
                "state_id": groupings.get("stateId", "unknown"),
                "duration_ms": duration_ms,
                "duration_minutes": round(duration_minutes, 2)
            })
        return parsed

Complete Working Example

The following script combines authentication, query execution, pagination, retry logic, and result parsing. Replace the placeholder credentials and environment URL before execution.

import requests
from datetime import datetime, timedelta
import time
import json
from typing import Optional

class CXoneReportingClient:
    def __init__(self, client_id: str, client_secret: str, environment: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = f"https://{environment}"
        self.token_url = f"{self.base_url}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0.0

    def _get_token(self) -> str:
        if self.access_token and datetime.now().timestamp() < self.token_expiry:
            return self.access_token

        payload = {
            "grant_type": "client_credentials",
            "scope": "reporting:view"
        }
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        response = requests.post(
            self.token_url,
            data=payload,
            headers=headers,
            auth=(self.client_id, self.client_secret)
        )
        response.raise_for_status()
        data = response.json()
        
        if "access_token" not in data:
            raise ValueError("OAuth response missing access_token")
            
        self.access_token = data["access_token"]
        expires_in = data.get("expires_in", 3600)
        self.token_expiry = datetime.now().timestamp() + (expires_in - 60)
        return self.access_token

    def _build_query_payload(self, hours_ago: int = 24, page_size: int = 100) -> dict:
        now = datetime.utcnow()
        start_time = now - timedelta(hours=hours_ago)
        
        time_range = {
            "from": start_time.strftime("%Y-%m-%dT%H:%M:%S.") + f"{start_time.microsecond // 1000:03d}Z",
            "to": now.strftime("%Y-%m-%dT%H:%M:%S.") + f"{now.microsecond // 1000:03d}Z"
        }
        
        return {
            "timeRange": time_range,
            "groupings": ["agentId", "stateId"],
            "metrics": ["duration"],
            "pageSize": page_size
        }

    def query_agent_state_history(self, hours_ago: int = 24) -> list[dict]:
        endpoint = f"{self.base_url}/api/v2/reporting/agents/state-history"
        headers = {
            "Authorization": f"Bearer {self._get_token()}",
            "Content-Type": "application/json"
        }
        
        payload = self._build_query_payload(hours_ago)
        all_results = []
        max_retries = 3
        retry_delay = 2.0
        
        while True:
            retries = 0
            while retries < max_retries:
                response = requests.post(endpoint, json=payload, headers=headers)
                
                if response.status_code == 429:
                    retries += 1
                    time.sleep(retry_delay)
                    retry_delay *= 2
                    continue
                    
                if response.status_code == 401:
                    self.access_token = None
                    headers["Authorization"] = f"Bearer {self._get_token()}"
                    continue
                    
                response.raise_for_status()
                break
            else:
                raise RuntimeError("Max retries exceeded for 429 rate limit")
                
            data = response.json()
            results = data.get("results", [])
            all_results.extend(results)
            
            continuation_token = data.get("continuationToken")
            if not continuation_token:
                break
                
            payload["continuationToken"] = continuation_token
            
        return all_results

    @staticmethod
    def parse_state_results(raw_results: list[dict]) -> list[dict]:
        parsed = []
        for item in raw_results:
            groupings = item.get("groupings", {})
            metrics = item.get("metrics", {})
            
            duration_ms = metrics.get("duration", 0)
            if duration_ms <= 0:
                continue
                
            duration_minutes = (duration_ms / 1000.0) / 60.0
            
            parsed.append({
                "agent_id": groupings.get("agentId", "unknown"),
                "state_id": groupings.get("stateId", "unknown"),
                "duration_ms": duration_ms,
                "duration_minutes": round(duration_minutes, 2)
            })
        return parsed

if __name__ == "__main__":
    CLIENT_ID = "your_client_id"
    CLIENT_SECRET = "your_client_secret"
    ENVIRONMENT = "api-us-01.niceincontact.com"
    
    client = CXoneReportingClient(CLIENT_ID, CLIENT_SECRET, ENVIRONMENT)
    
    try:
        raw_data = client.query_agent_state_history(hours_ago=24)
        formatted_data = client.parse_state_results(raw_data)
        print(json.dumps(formatted_data, indent=2))
    except requests.exceptions.HTTPError as e:
        print(f"HTTP Error: {e}")
        if e.response is not None:
            print(f"Response Body: {e.response.text}")
    except Exception as e:
        print(f"Execution Error: {e}")

Common Errors & Debugging

Error: 400 Bad Request

• Cause: The time range exceeds CXone retention limits, the grouping combination is unsupported, or the timeRange format lacks millisecond precision.
• Fix: Verify the from and to timestamps use ISO 8601 with milliseconds ending in Z. Ensure groupings contains exactly ["agentId", "stateId"]. Reduce the query window if retention policies block older data.
• Code Fix: The _build_query_payload method enforces correct timestamp formatting. Add logging to print the exact payload sent to the API.

Error: 401 Unauthorized

• Cause: The OAuth token expired during pagination, or the client credentials lack the reporting:view scope.
• Fix: The _get_token method automatically refreshes the token when token_expiry passes. If 401 persists, verify the OAuth client in the CXone admin console has the reporting:view scope assigned.
• Code Fix: The retry loop in query_agent_state_history detects 401, clears the cached token, and forces a refresh before retrying.

Error: 429 Too Many Requests

• Cause: You exceeded the CXone reporting API rate limit (typically 10 requests per second per environment).
• Fix: Implement exponential backoff. The code includes a retry loop that doubles the delay between attempts. Reduce pageSize if processing large datasets, as smaller pages reduce per-request payload size and may improve throughput.
• Code Fix: The retries < max_retries loop handles 429 responses with time.sleep(retry_delay) and retry_delay *= 2.

Error: Empty Results Array

• Cause: No agents changed states during the requested window, or the environment has no active workforce management data.
• Fix: Verify the time window aligns with business hours. Check the CXone admin console to confirm agents were logged in. Expand the hours_ago parameter to 48 or 72 to capture off-peak state transitions.
• Code Fix: The parse_state_results method filters out zero-duration entries. If the final list remains empty, the API correctly returned no matching data.

Official References

• NICE CXone Reporting API v2 Documentation
• CXone OAuth 2.0 Client Credentials Flow
• CXone Python SDK Reference