Querying Agent State History via NICE CXone Reporting API (v2)

Querying Agent State History via NICE CXone Reporting API (v2)

What You Will Build

  • A script that retrieves the last 24 hours of agent state changes (Login, Logout, Wrap-up, Ready) for a specific user.
  • This tutorial uses the NICE CXone Reporting API (v2) endpoint /api/v2/reports/agent-state-history.
  • The implementation is in Python using the requests library for explicit HTTP control.

Prerequisites

  • OAuth Client: You need a CXone OAuth Client ID and Client Secret. The client must have the reporting capability enabled.
  • Required Scopes: reports:read is mandatory for accessing reporting data.
  • API Version: CXone Reporting API v2.
  • Runtime: Python 3.8 or higher.
  • Dependencies:
    pip install requests python-dateutil
    

Authentication Setup

CXone uses OAuth 2.0 Client Credentials flow for server-to-server API access. You must obtain an access token before querying reports. The token expires after 20 minutes (1200 seconds), so a robust implementation includes a refresh check.

The following Python class handles token acquisition and caching. It ensures that subsequent calls within the same session reuse the existing token until it expires.

import requests
import time
from datetime import datetime, timezone

class CxoneAuth:
    def __init__(self, tenant: str, client_id: str, client_secret: str):
        self.tenant = tenant
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = f"https://{tenant}.api.cxone.com"
        self.token_url = f"{self.base_url}/oauth/token"
        
        self.access_token = None
        self.token_expiry = 0

    def get_access_token(self) -> str:
        """
        Returns a valid access token. 
        Refreshes the token if it is expired or not yet obtained.
        """
        current_time = time.time()
        
        # If we have a token and it is not expired, return it
        if self.access_token and current_time < self.token_expiry:
            return self.access_token

        # Otherwise, fetch a new 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_url, data=payload, headers=headers)
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise Exception("Authentication failed: Invalid Client ID or Secret.") from e
            elif response.status_code == 403:
                raise Exception("Authentication failed: Client lacks permissions or is disabled.") from e
            else:
                raise Exception(f"OAuth Error: {response.status_code} - {response.text}") from e

        data = response.json()
        self.access_token = data['access_token']
        self.token_expiry = current_time + (data['expires_in'] - 60) # Subtract 60s for safety margin
        
        return self.access_token

Implementation

Step 1: Constructing the Report Query

The CXone Reporting API does not return data immediately upon request. It operates asynchronously. You must submit a query body that defines the metrics, filters, and time range. The API returns a jobId. You then poll this jobId until the data is ready.

For Agent State History, the critical parameters are:

  1. reportType: Must be agent-state-history.
  2. dateFrom / dateTo: ISO 8601 formatted strings.
  3. filter: A JSON structure defining which agents to query.

We will calculate the last 24 hours dynamically.

from datetime import datetime, timedelta, timezone

def generate_agent_state_query(user_id: str) -> dict:
    """
    Generates the JSON payload for the Agent State History report.
    """
    now = datetime.now(timezone.utc)
    twenty_four_hours_ago = now - timedelta(hours=24)

    # Format dates to ISO 8601 without timezone offset for CXone API compatibility
    # CXone expects UTC timestamps.
    date_to = now.strftime("%Y-%m-%dT%H:%M:%S.000Z")
    date_from = twenty_four_hours_ago.strftime("%Y-%m-%dT%H:%M:%S.000Z")

    query_body = {
        "reportType": "agent-state-history",
        "dateFrom": date_from,
        "dateTo": date_to,
        "filter": {
            "type": "user",
            "id": user_id
        },
        "groupBy": [
            "userId",
            "stateName",
            "skillName"
        ],
        "metrics": [
            "count"
        ]
    }
    
    return query_body

Why these parameters?

  • groupBy: Grouping by userId, stateName, and skillName allows you to see not just that the agent was “Ready”, but which skill they were ready for. This is crucial for multi-skill routing environments.
  • metrics: We request count to get the number of transitions or duration buckets, depending on the specific metric definition in the report type. For state history, count typically represents the number of state change events.

Step 2: Submitting the Report Job

Once the query body is constructed, send it to the /api/v2/reports endpoint. This is a POST request. The response will contain a jobId and a statusUrl.

class CxoneReportingClient:
    def __init__(self, auth: CxoneAuth):
        self.auth = auth
        self.base_url = f"https://{auth.tenant}.api.cxone.com"

    def submit_report(self, query_body: dict) -> str:
        """
        Submits the report query and returns the jobId.
        """
        url = f"{self.base_url}/api/v2/reports"
        headers = {
            "Authorization": f"Bearer {self.auth.get_access_token()}",
            "Content-Type": "application/json"
        }

        try:
            response = requests.post(url, json=query_body, headers=headers)
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise Exception("Unauthorized: Token may be expired or invalid.") from e
            elif response.status_code == 403:
                raise Exception("Forbidden: Check if the client has 'reports:read' scope.") from e
            elif response.status_code == 429:
                raise Exception("Rate Limited: Too many requests. Wait before retrying.") from e
            else:
                raise Exception(f"API Error: {response.status_code} - {response.text}") from e

        result = response.json()
        job_id = result.get('id')
        
        if not job_id:
            raise Exception("Failed to retrieve jobId from response.")
            
        return job_id

Step 3: Polling for Results

The report generation is asynchronous. You must poll the jobId endpoint. The typical pattern is to check every 1-2 seconds. The status can be:

  • PENDING: The job is queued.
  • RUNNING: The job is processing.
  • COMPLETED: The data is ready.
  • FAILED: The job failed (check error details).
import time

def poll_report_status(self, job_id: str, max_retries: int = 30, retry_delay: int = 2) -> dict:
    """
    Polls the report job until completion or failure.
    """
    url = f"{self.base_url}/api/v2/reports/{job_id}"
    headers = {
        "Authorization": f"Bearer {self.auth.get_access_token()}"
    }

    for attempt in range(max_retries):
        try:
            response = requests.get(url, headers=headers)
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
            if response.status_code == 404:
                raise Exception(f"Job {job_id} not found.") from e
            elif response.status_code == 429:
                time.sleep(retry_delay * 2) # Backoff on rate limit
                continue
            else:
                raise Exception(f"Polling Error: {response.status_code} - {response.text}") from e

        result = response.json()
        status = result.get('status')
        
        if status == 'COMPLETED':
            return result
        elif status == 'FAILED':
            error_message = result.get('errorMessage', 'Unknown error')
            raise Exception(f"Report Job Failed: {error_message}")
        elif status in ['PENDING', 'RUNNING']:
            time.sleep(retry_delay)
        else:
            raise Exception(f"Unexpected status: {status}")

    raise Exception("Timeout: Report did not complete within the allotted time.")

Step 4: Processing the Results

When the job is COMPLETED, the response contains a data field. For agent-state-history, the data structure is a list of objects, each representing a state change event or an aggregated bucket depending on the groupBy configuration.

The response typically looks like this:

{
  "id": "job-12345",
  "status": "COMPLETED",
  "data": [
    {
      "userId": "user-abc-123",
      "userName": "John Doe",
      "stateName": "Ready",
      "stateId": "state-ready-001",
      "skillName": "English Support",
      "count": 1,
      "dateFrom": "2023-10-27T08:00:00.000Z",
      "dateTo": "2023-10-27T09:00:00.000Z"
    },
    {
      "userId": "user-abc-123",
      "userName": "John Doe",
      "stateName": "Wrap-up",
      "stateId": "state-wrap-002",
      "skillName": "English Support",
      "count": 1,
      "dateFrom": "2023-10-27T09:00:00.000Z",
      "dateTo": "2023-10-27T09:05:00.000Z"
    }
  ]
}

You should iterate through this list to extract the timeline of events.

Complete Working Example

This is the full, copy-pasteable script. Replace the TENANT, CLIENT_ID, CLIENT_SECRET, and USER_ID variables with your actual credentials.

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

# --- Configuration ---
TENANT = "your-tenant"
CLIENT_ID = "your-client-id"
CLIENT_SECRET = "your-client-secret"
USER_ID = "your-user-id" # The ID of the agent you want to query

class CxoneAuth:
    def __init__(self, tenant: str, client_id: str, client_secret: str):
        self.tenant = tenant
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = f"https://{tenant}.api.cxone.com"
        self.token_url = f"{self.base_url}/oauth/token"
        self.access_token = None
        self.token_expiry = 0

    def get_access_token(self) -> str:
        current_time = time.time()
        if self.access_token and current_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_url, data=payload, headers=headers)
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise Exception("Authentication failed: Invalid Client ID or Secret.") from e
            raise Exception(f"OAuth Error: {response.status_code} - {response.text}") from e

        data = response.json()
        self.access_token = data['access_token']
        self.token_expiry = current_time + (data['expires_in'] - 60)
        return self.access_token

class CxoneReportingClient:
    def __init__(self, auth: CxoneAuth):
        self.auth = auth
        self.base_url = f"https://{auth.tenant}.api.cxone.com"

    def submit_report(self, query_body: dict) -> str:
        url = f"{self.base_url}/api/v2/reports"
        headers = {
            "Authorization": f"Bearer {self.auth.get_access_token()}",
            "Content-Type": "application/json"
        }
        try:
            response = requests.post(url, json=query_body, headers=headers)
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise Exception("Unauthorized: Token may be expired or invalid.") from e
            if response.status_code == 403:
                raise Exception("Forbidden: Check if the client has 'reports:read' scope.") from e
            if response.status_code == 429:
                raise Exception("Rate Limited: Too many requests. Wait before retrying.") from e
            raise Exception(f"API Error: {response.status_code} - {response.text}") from e

        result = response.json()
        job_id = result.get('id')
        if not job_id:
            raise Exception("Failed to retrieve jobId from response.")
        return job_id

    def poll_report_status(self, job_id: str, max_retries: int = 30, retry_delay: int = 2) -> dict:
        url = f"{self.base_url}/api/v2/reports/{job_id}"
        headers = {"Authorization": f"Bearer {self.auth.get_access_token()}"}

        for attempt in range(max_retries):
            try:
                response = requests.get(url, headers=headers)
                response.raise_for_status()
            except requests.exceptions.HTTPError as e:
                if response.status_code == 404:
                    raise Exception(f"Job {job_id} not found.") from e
                if response.status_code == 429:
                    time.sleep(retry_delay * 2)
                    continue
                raise Exception(f"Polling Error: {response.status_code} - {response.text}") from e

            result = response.json()
            status = result.get('status')
            
            if status == 'COMPLETED':
                return result
            elif status == 'FAILED':
                error_message = result.get('errorMessage', 'Unknown error')
                raise Exception(f"Report Job Failed: {error_message}")
            elif status in ['PENDING', 'RUNNING']:
                time.sleep(retry_delay)
            else:
                raise Exception(f"Unexpected status: {status}")
        raise Exception("Timeout: Report did not complete within the allotted time.")

    def get_agent_state_history(self, user_id: str) -> list:
        now = datetime.now(timezone.utc)
        twenty_four_hours_ago = now - timedelta(hours=24)
        
        date_to = now.strftime("%Y-%m-%dT%H:%M:%S.000Z")
        date_from = twenty_four_hours_ago.strftime("%Y-%m-%dT%H:%M:%S.000Z")

        query_body = {
            "reportType": "agent-state-history",
            "dateFrom": date_from,
            "dateTo": date_to,
            "filter": {
                "type": "user",
                "id": user_id
            },
            "groupBy": ["userId", "stateName", "skillName"],
            "metrics": ["count"]
        }

        print(f"Submitting report for user {user_id} from {date_from} to {date_to}...")
        job_id = self.submit_report(query_body)
        print(f"Job submitted. Job ID: {job_id}. Polling for results...")
        
        result = self.poll_report_status(job_id)
        
        data = result.get('data', [])
        print(f"Report completed. Found {len(data)} state history records.")
        return data

def main():
    try:
        auth = CxoneAuth(TENANT, CLIENT_ID, CLIENT_SECRET)
        client = CxoneReportingClient(auth)
        
        history = client.get_agent_state_history(USER_ID)
        
        # Pretty print the results
        for record in history:
            print(json.dumps(record, indent=2))
            
    except Exception as e:
        print(f"Error: {e}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

Cause: The OAuth Client does not have the reports:read scope, or the tenant does not have the Reporting module licensed.

Fix:

  1. Log in to the CXone Admin Console.
  2. Navigate to Admin > Security > OAuth Clients.
  3. Edit your client.
  4. Ensure Reports is checked under Capabilities.
  5. Ensure reports:read is selected under Scopes.
  6. Update the client and generate a new token.

Error: 422 Unprocessable Entity

Cause: The dateFrom or dateTo format is incorrect, or the filter structure is invalid. CXone requires strict ISO 8601 format with Z suffix for UTC.

Fix:
Verify the date string format in the generate_agent_state_query function. Ensure it matches YYYY-MM-DDTHH:mm:ss.SSSZ. Do not include timezone offsets like +00:00. Use Z explicitly.

Error: Empty Data Array

Cause: The agent did not change states in the last 24 hours, or the USER_ID is incorrect.

Fix:

  1. Verify the USER_ID is the internal CXone ID (starts with user-), not the email address.
  2. Check if the agent was active during the requested window. If the agent was inactive, no state history will be generated.
  3. Expand the time window to 7 days to verify if data exists at all.

Error: 429 Too Many Requests

Cause: You are polling too frequently or submitting too many report jobs in a short period.

Fix:
Implement exponential backoff in the poll_report_status method. The code above uses a simple 2-second delay. If you hit 429, increase the delay to 5 seconds and retry. Do not submit more than 10 report jobs per minute per client.

Official References