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
requestslibrary for explicit HTTP control.
Prerequisites
- OAuth Client: You need a CXone OAuth Client ID and Client Secret. The client must have the
reportingcapability enabled. - Required Scopes:
reports:readis 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:
reportType: Must beagent-state-history.dateFrom/dateTo: ISO 8601 formatted strings.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 byuserId,stateName, andskillNameallows 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 requestcountto get the number of transitions or duration buckets, depending on the specific metric definition in the report type. For state history,counttypically 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:
- Log in to the CXone Admin Console.
- Navigate to Admin > Security > OAuth Clients.
- Edit your client.
- Ensure Reports is checked under Capabilities.
- Ensure reports:read is selected under Scopes.
- 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:
- Verify the
USER_IDis the internal CXone ID (starts withuser-), not the email address. - Check if the agent was active during the requested window. If the agent was inactive, no state history will be generated.
- 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.