Querying NICE CXone Agent State History with the Reporting API (v2)
What You Will Build
You will build a Python script that retrieves a detailed timeline of agent presence changes (login, logout, state changes) for specific users over the last 24 hours. You will use the NICE CXone Reporting API v2 endpoint /api/v2/reporting/query with a structured JSON body. You will use Python 3.9+ with the requests library.
Prerequisites
OAuth Configuration
You need a valid NICE CXone OAuth 2.0 Client ID and Secret. The client must have the Reporting role or specific reporting permissions assigned.
Required OAuth Scope:
reporting:read
Environment Setup
- Python: 3.9 or higher.
- Dependencies:
requests,python-dotenv(for secure credential management).
Install dependencies via pip:
pip install requests python-dotenv
Base URL
Identify your NICE CXone instance base URL. For most production instances, this follows the pattern https://api-us-1.cxone.com or https://api-eu-1.cxone.com. Replace [YOUR_INSTANCE] in the code below with your actual domain.
Authentication Setup
NICE CXone uses standard OAuth 2.0 Client Credentials flow. You must exchange your Client ID and Secret for an access token before making any API calls. The token expires after a short duration (typically 1 hour), so robust implementations cache tokens or refresh them automatically. For this tutorial, we will implement a simple helper function to acquire the token.
import requests
import json
from datetime import datetime, timedelta, timezone
from typing import Dict, Optional, Any
class CXoneReportingClient:
def __init__(self, client_id: str, client_secret: str, base_url: str):
self.client_id = client_id
self.client_secret = client_secret
self.base_url = base_url.rstrip('/')
self.access_token: Optional[str] = None
self.token_expiry: Optional[datetime] = None
def get_access_token(self) -> str:
"""
Retrieves an OAuth2 access token.
Implements simple caching to avoid unnecessary token requests.
"""
if self.access_token and self.token_expiry and datetime.now(timezone.utc) < self.token_expiry:
return self.access_token
token_url = f"{self.base_url}/oauth/token"
payload = {
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret,
"scope": "reporting:read"
}
headers = {
"Content-Type": "application/x-www-form-urlencoded"
}
try:
response = requests.post(token_url, data=payload, headers=headers)
response.raise_for_status()
except requests.exceptions.HTTPError as http_err:
if response.status_code == 400:
raise ValueError("Invalid client credentials or scope.") from http_err
raise http_err
token_data = response.json()
self.access_token = token_data["access_token"]
# Parse expiry time. NICE usually returns 'expires_in' in seconds.
expires_in = int(token_data.get("expires_in", 3600))
self.token_expiry = datetime.now(timezone.utc) + timedelta(seconds=expires_in)
return self.access_token
Implementation
Step 1: Constructing the Query Payload
The NICE CXone Reporting API v2 does not use query parameters for filtering data dimensions. Instead, it uses a POST body with a specific JSON structure. To query agent state history, you must target the AgentPresence view.
Key fields in the request body:
viewName: Must be"AgentPresence".groupBy: An array of dimensions you want to slice the data by. For state history,["user"]is essential. You may also include["state"]to see the specific state changed to.metrics: An array of metrics. For presence,["presenceDuration"]is common, but often you just want the timeline, so metrics can be empty or used to filter duration.filter: The critical component. This defines the time window and specific users.
Time Window Logic:
The API expects time ranges in ISO 8601 format. We will calculate the start time as exactly 24 hours ago from the current UTC time.
def build_agent_presence_payload(self, user_ids: list[str]) -> Dict[str, Any]:
"""
Constructs the JSON payload for the AgentPresence query.
Args:
user_ids: List of NICE CXone User IDs (strings).
Returns:
Dictionary representing the request body.
"""
now = datetime.now(timezone.utc)
start_time = now - timedelta(hours=24)
# Format times as ISO 8601 with timezone offset
start_iso = start_time.isoformat()
end_iso = now.isoformat()
# Define the filter structure
# NICE CXone filters use a specific syntax for time ranges and lists
filter_obj = {
"timeRange": {
"start": start_iso,
"end": end_iso
},
"userIds": user_ids # Filters for specific agents
}
payload = {
"viewName": "AgentPresence",
"groupBy": [
"user",
"state", # Group by state to see transitions clearly
"timestamp" # Group by timestamp to get the history/timeline
],
"metrics": [
"presenceDuration"
],
"filter": filter_obj,
"format": "json"
}
return payload
Step 2: Executing the Query and Handling Pagination
The /api/v2/reporting/query endpoint supports pagination via the pageToken. If the result set is large (e.g., many agents with frequent state changes), a single request may not return all data. You must check the pageToken in the response and continue fetching until it is null.
Additionally, the API may return a 202 Accepted status initially if the report generation takes time, though for simple presence queries, it often returns 200 OK immediately. We will handle the immediate 200 case and include logic for basic error handling.
def query_agent_state_history(self, user_ids: list[str], max_pages: int = 10) -> list[Dict[str, Any]]:
"""
Queries the AgentPresence view for the last 24 hours.
Handles pagination automatically.
Args:
user_ids: List of user IDs to query.
max_pages: Safety limit to prevent infinite loops.
Returns:
A list of dictionaries, each representing a state record.
"""
token = self.get_access_token()
url = f"{self.base_url}/api/v2/reporting/query"
payload = self.build_agent_presence_payload(user_ids)
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
"Accept": "application/json"
}
all_records = []
page_token = None
page_count = 0
while page_count < max_pages:
# If we have a page token, we append it to the payload or use query params?
# NICE CXone Reporting API v2 typically uses the pageToken in the request body
# or as a query parameter. The standard practice for v2 reporting is often
# including it in the body if the SDK supports it, but the REST API spec
# usually places it in the body for the initial post and subsequent posts.
# However, the most reliable method for the raw REST API is often appending
# it to the body if the endpoint supports it, OR using it in the next request.
# Let's check the standard pattern: Usually, the response contains a 'pageToken'.
# For subsequent requests, you include that token.
request_payload = payload.copy()
if page_token:
request_payload["pageToken"] = page_token
try:
response = requests.post(url, json=request_payload, headers=headers)
if response.status_code == 429:
# Rate limited. Wait and retry.
retry_after = int(response.headers.get("Retry-After", 5))
import time
time.sleep(retry_after)
continue
response.raise_for_status()
data = response.json()
# Extract results
# The structure is usually: { "results": [ ... ], "pageToken": "..." }
results = data.get("results", [])
all_records.extend(results)
page_token = data.get("pageToken")
page_count += 1
# If no more page tokens, we are done
if not page_token:
break
except requests.exceptions.HTTPError as e:
if response.status_code == 401:
# Token expired or invalid. Refresh and retry one more time.
if page_count > 0:
raise Exception("Token expired during pagination. Please restart the query.") from e
self.access_token = None # Force refresh on next loop iteration
continue
elif response.status_code == 403:
raise PermissionError("Insufficient permissions. Ensure 'reporting:read' scope is active.") from e
else:
raise Exception(f"API Error: {response.status_code} - {response.text}") from e
if page_count >= max_pages:
print(f"Warning: Reached max pages ({max_pages}). Results may be truncated.")
return all_records
Step 3: Processing and Formatting Results
The raw response from NICE CXone contains nested objects. Each record in results typically looks like this:
{
"user": {
"id": "12345678-abcd-efgh-ijkl-1234567890ab",
"name": "John Doe"
},
"state": {
"id": "12345678-abcd-efgh-ijkl-1234567890ac",
"name": "Ready",
"type": "Available"
},
"timestamp": "2023-10-27T10:00:00Z",
"metrics": {
"presenceDuration": 3600000
}
}
We will create a helper method to flatten this into a more usable format for logging or database insertion.
def format_results(self, records: list[Dict[str, Any]]) -> list[Dict[str, Any]]:
"""
Flattens the nested API response into a cleaner list of dictionaries.
"""
formatted = []
for record in records:
user_info = record.get("user", {})
state_info = record.get("state", {})
metrics = record.get("metrics", {})
formatted_record = {
"user_id": user_info.get("id"),
"user_name": user_info.get("name"),
"state_id": state_info.get("id"),
"state_name": state_info.get("name"),
"state_type": state_info.get("type"), # e.g., Available, Unavailable, Offline
"timestamp": record.get("timestamp"),
"duration_ms": metrics.get("presenceDuration", 0)
}
formatted.append(formatted_record)
# Sort by timestamp for chronological history
formatted.sort(key=lambda x: x["timestamp"])
return formatted
Complete Working Example
Below is the complete, runnable script. Save this as cxone_agent_history.py.
import os
import sys
import json
import requests
from datetime import datetime, timedelta, timezone
from typing import Dict, Optional, Any, List
# Attempt to load environment variables from .env file
try:
from dotenv import load_dotenv
load_dotenv()
except ImportError:
print("Warning: python-dotenv not installed. Ensure environment variables are set in OS.")
class CXoneReportingClient:
def __init__(self, client_id: str, client_secret: str, base_url: str):
self.client_id = client_id
self.client_secret = client_secret
self.base_url = base_url.rstrip('/')
self.access_token: Optional[str] = None
self.token_expiry: Optional[datetime] = None
def get_access_token(self) -> str:
"""
Retrieves an OAuth2 access token.
Implements simple caching to avoid unnecessary token requests.
"""
if self.access_token and self.token_expiry and datetime.now(timezone.utc) < self.token_expiry:
return self.access_token
token_url = f"{self.base_url}/oauth/token"
payload = {
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret,
"scope": "reporting:read"
}
headers = {
"Content-Type": "application/x-www-form-urlencoded"
}
try:
response = requests.post(token_url, data=payload, headers=headers)
response.raise_for_status()
except requests.exceptions.HTTPError as http_err:
if response.status_code == 400:
raise ValueError("Invalid client credentials or scope.") from http_err
raise http_err
token_data = response.json()
self.access_token = token_data["access_token"]
# Parse expiry time. NICE usually returns 'expires_in' in seconds.
expires_in = int(token_data.get("expires_in", 3600))
self.token_expiry = datetime.now(timezone.utc) + timedelta(seconds=expires_in)
return self.access_token
def build_agent_presence_payload(self, user_ids: List[str]) -> Dict[str, Any]:
"""
Constructs the JSON payload for the AgentPresence query.
"""
now = datetime.now(timezone.utc)
start_time = now - timedelta(hours=24)
start_iso = start_time.isoformat()
end_iso = now.isoformat()
filter_obj = {
"timeRange": {
"start": start_iso,
"end": end_iso
},
"userIds": user_ids
}
payload = {
"viewName": "AgentPresence",
"groupBy": [
"user",
"state",
"timestamp"
],
"metrics": [
"presenceDuration"
],
"filter": filter_obj,
"format": "json"
}
return payload
def query_agent_state_history(self, user_ids: List[str], max_pages: int = 10) -> List[Dict[str, Any]]:
"""
Queries the AgentPresence view for the last 24 hours.
Handles pagination automatically.
"""
token = self.get_access_token()
url = f"{self.base_url}/api/v2/reporting/query"
payload = self.build_agent_presence_payload(user_ids)
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
"Accept": "application/json"
}
all_records = []
page_token = None
page_count = 0
while page_count < max_pages:
request_payload = payload.copy()
if page_token:
request_payload["pageToken"] = page_token
try:
response = requests.post(url, json=request_payload, headers=headers)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 5))
import time
time.sleep(retry_after)
continue
response.raise_for_status()
data = response.json()
results = data.get("results", [])
all_records.extend(results)
page_token = data.get("pageToken")
page_count += 1
if not page_token:
break
except requests.exceptions.HTTPError as e:
if response.status_code == 401:
if page_count > 0:
raise Exception("Token expired during pagination.") from e
self.access_token = None
continue
elif response.status_code == 403:
raise PermissionError("Insufficient permissions.") from e
else:
raise Exception(f"API Error: {response.status_code} - {response.text}") from e
if page_count >= max_pages:
print(f"Warning: Reached max pages ({max_pages}). Results may be truncated.")
return all_records
def format_results(self, records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Flattens the nested API response into a cleaner list of dictionaries.
"""
formatted = []
for record in records:
user_info = record.get("user", {})
state_info = record.get("state", {})
metrics = record.get("metrics", {})
formatted_record = {
"user_id": user_info.get("id"),
"user_name": user_info.get("name"),
"state_id": state_info.get("id"),
"state_name": state_info.get("name"),
"state_type": state_info.get("type"),
"timestamp": record.get("timestamp"),
"duration_ms": metrics.get("presenceDuration", 0)
}
formatted.append(formatted_record)
formatted.sort(key=lambda x: x["timestamp"])
return formatted
def main():
# Configuration
CLIENT_ID = os.getenv("CXONE_CLIENT_ID")
CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET")
BASE_URL = os.getenv("CXONE_BASE_URL", "https://api-us-1.cxone.com") # Default to US-1
# Example User IDs (Replace with actual IDs from your instance)
# You can find these in the Admin Console > People > Users
TARGET_USER_IDS = [
"12345678-abcd-efgh-ijkl-1234567890ab",
"87654321-dcba-hgfe-lkji-0987654321ba"
]
if not CLIENT_ID or not CLIENT_SECRET:
print("Error: CXONE_CLIENT_ID and CXONE_CLIENT_SECRET must be set in environment variables.")
sys.exit(1)
try:
client = CXoneReportingClient(CLIENT_ID, CLIENT_SECRET, BASE_URL)
print(f"Querying agent state history for {len(TARGET_USER_IDS)} users...")
raw_records = client.query_agent_state_history(TARGET_USER_IDS)
if not raw_records:
print("No records found for the specified users in the last 24 hours.")
return
formatted_records = client.format_results(raw_records)
print(f"\nFound {len(formatted_records)} state change records.\n")
print("-" * 80)
print(f"{'Timestamp':<25} | {'User Name':<20} | {'State Name':<15} | {'Type':<12} | {'Duration (ms)':<12}")
print("-" * 80)
for rec in formatted_records:
ts = rec['timestamp'].replace('T', ' ').replace('Z', '')[:19] if rec['timestamp'] else "N/A"
print(f"{ts:<25} | {rec['user_name'] or 'N/A':<20} | {rec['state_name'] or 'N/A':<15} | {rec['state_type'] or 'N/A':<12} | {rec['duration_ms']}")
print("-" * 80)
# Optional: Save to JSON file
output_file = "agent_state_history.json"
with open(output_file, 'w') as f:
json.dump(formatted_records, f, indent=2)
print(f"\nResults saved to {output_file}")
except Exception as e:
print(f"An error occurred: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
Common Errors & Debugging
Error: 401 Unauthorized
Cause: The OAuth token is expired, invalid, or the Authorization header is malformed.
Fix: Ensure your get_access_token method is called before every batch of requests. Check that your Client ID and Secret are correct. Verify that the scope includes reporting:read.
Error: 403 Forbidden
Cause: The OAuth client does not have permission to access the Reporting API.
Fix: Log in to the NICE CXone Admin Console. Navigate to Integrations > OAuth Clients. Select your client and ensure the Reporting role is assigned. If using a custom role, verify it includes the reporting:read permission.
Error: 400 Bad Request - “Invalid Filter”
Cause: The timeRange format is incorrect or the viewName is misspelled.
Fix: Ensure viewName is exactly "AgentPresence". Ensure timestamps are in ISO 8601 format with timezone offsets (e.g., 2023-10-27T10:00:00+00:00). The requests library .isoformat() on a timezone-aware datetime object handles this correctly.
Error: Empty Results
Cause: The specified userIds are invalid, or the agents have not been active in the last 24 hours.
Fix: Verify the User IDs in the Admin Console. Try querying without the userIds filter (by removing "userIds": user_ids from the filter object) to see if any data exists for the instance in that time window. Note that querying all users may trigger pagination or rate limits.
Error: 429 Too Many Requests
Cause: You have exceeded the rate limit for the Reporting API.
Fix: The code above includes a basic retry mechanism with Retry-After header parsing. If you are querying many users, consider batching the userIds list into smaller chunks (e.g., 10 users per request) to stay within rate limits.