Creating NICE CXone Support Tickets via the Case Management API with Python
What You Will Build
A production-ready Python module that constructs, validates, and submits support tickets to NICE CXone Case Management, enforces attachment limits, calculates priority, evaluates assignment rules, performs duplicate detection, tracks latency, generates audit logs, and exposes a reusable ticket creator class for automated workflows.
This tutorial uses the NICE CXone Case Management REST API (/api/v2/cases and /api/v2/cases/search) with synchronous HTTP calls.
The implementation is written in Python 3.9+ using requests, logging, and standard library modules.
Prerequisites
- NICE CXone OAuth 2.0 Client Credentials grant configuration
- Required OAuth scopes:
cases:read,cases:write - Python 3.9 or higher
- External dependencies:
requests==2.31.0 - Environment variables:
CXONE_REGION,CXONE_CLIENT_ID,CXONE_CLIENT_SECRET,CXONE_CASE_TYPE_ID
Authentication Setup
NICE CXone uses OAuth 2.0 client credentials flow. The token endpoint varies by region. The following code retrieves an access token, caches it, and validates expiration before each request.
import os
import time
import requests
from typing import Optional
class CxoNeAuthManager:
def __init__(self, region: str, client_id: str, client_secret: str):
self.base_url = f"https://api-{region}.nicecxone.com"
self.client_id = client_id
self.client_secret = client_secret
self.token: Optional[str] = None
self.token_expiry: float = 0.0
def get_token(self) -> str:
if self.token and time.time() < self.token_expiry:
return self.token
url = f"{self.base_url}/oauth2/token"
payload = {
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret,
"scope": "cases:read cases:write"
}
response = requests.post(url, data=payload, timeout=10)
response.raise_for_status()
data = response.json()
self.token = data["access_token"]
self.token_expiry = time.time() + data["expires_in"] - 30
return self.token
def get_headers(self) -> dict:
return {
"Authorization": f"Bearer {self.get_token()}",
"Content-Type": "application/json",
"Accept": "application/json"
}
Implementation
Step 1: Duplicate Detection and Required Field Validation
Before submitting a case, you must verify that the reference identifier does not already exist and that all mandatory fields are present. CXone provides a search endpoint that supports filtering by custom fields.
import logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")
class CxoNeCaseValidator:
MAX_ATTACHMENTS = 10
REQUIRED_FIELDS = {"subject", "description", "case_type_id", "priority"}
def __init__(self, auth: CxoNeAuthManager):
self.auth = auth
self.base_url = auth.base_url
def check_duplicate(self, case_ref: str) -> bool:
url = f"{self.base_url}/api/v2/cases/search"
params = {
"filter": f"customFields.id eq 'cf_case_ref' and customFields.value eq '{case_ref}'",
"pageSize": 1
}
headers = self.auth.get_headers()
try:
response = requests.get(url, params=params, headers=headers, timeout=10)
if response.status_code == 429:
time.sleep(int(response.headers.get("Retry-After", 2)))
response = requests.get(url, params=params, headers=headers, timeout=10)
response.raise_for_status()
data = response.json()
return data.get("totalCount", 0) > 0
except requests.exceptions.HTTPError as e:
logging.error(f"Duplicate check failed: {e}")
raise
def validate_payload(self, payload: dict) -> list:
errors = []
missing = self.REQUIRED_FIELDS - set(payload.keys())
if missing:
errors.append(f"Missing required fields: {', '.join(missing)}")
attachments = payload.get("attachments", [])
if len(attachments) > self.MAX_ATTACHMENTS:
errors.append(f"Attachment count {len(attachments)} exceeds maximum of {self.MAX_ATTACHMENTS}")
return errors
Step 2: Priority Calculation and Assignment Rule Evaluation
CXone allows priority and owner assignment through standard fields. The following logic evaluates a ticket-matrix value to determine priority and routes the case to the appropriate group or owner.
class CxoNeRoutingEngine:
PRIORITY_MATRIX = {
"critical": "Urgent",
"high": "High",
"medium": "Normal",
"low": "Low"
}
GROUP_MAPPING = {
"billing": "grp_billing_001",
"technical": "grp_tech_002",
"general": "grp_general_003"
}
@staticmethod
def calculate_priority(ticket_matrix: str, open_directive: str) -> str:
base_priority = CxoNeRoutingEngine.PRIORITY_MATRIX.get(ticket_matrix, "Normal")
if open_directive.lower() in ("escalate", "vip", "breach"):
return "Urgent"
return base_priority
@staticmethod
def evaluate_assignment(ticket_matrix: str) -> dict:
category = "general"
if any(keyword in ticket_matrix.lower() for keyword in ["invoice", "payment", "refund"]):
category = "billing"
elif any(keyword in ticket_matrix.lower() for keyword in ["error", "bug", "integration"]):
category = "technical"
return {"group_id": CxoNeRoutingEngine.GROUP_MAPPING[category]}
Step 3: Atomic POST Construction and Submission
The final payload combines validated fields, calculated priority, routing rules, and custom metadata. The request includes an Idempotency-Key header to guarantee safe open iteration on retries.
import uuid
import json
from datetime import datetime, timezone
class CxoNeCaseCreator:
def __init__(self, auth: CxoNeAuthManager, case_type_id: str):
self.auth = auth
self.case_type_id = case_type_id
self.validator = CxoNeCaseValidator(auth)
self.metrics = {"total": 0, "success": 0, "failures": 0, "latency_sum": 0.0}
def build_payload(self, subject: str, description: str, case_ref: str,
ticket_matrix: str, open_directive: str, attachments: list = None) -> dict:
priority = CxoNeRoutingEngine.calculate_priority(ticket_matrix, open_directive)
assignment = CxoNeRoutingEngine.evaluate_assignment(ticket_matrix)
custom_fields = [
{"id": "cf_case_ref", "value": case_ref},
{"id": "cf_ticket_matrix", "value": ticket_matrix},
{"id": "cf_open_directive", "value": open_directive}
]
payload = {
"case_type_id": self.case_type_id,
"subject": subject,
"description": description,
"priority": priority,
"status": "Open",
"custom_fields": custom_fields,
"attachments": attachments or []
}
payload.update(assignment)
return payload
def create_case(self, subject: str, description: str, case_ref: str,
ticket_matrix: str, open_directive: str, attachments: list = None) -> dict:
self.metrics["total"] += 1
if self.validator.check_duplicate(case_ref):
raise ValueError(f"Duplicate case detected for reference: {case_ref}")
payload = self.build_payload(subject, description, case_ref, ticket_matrix, open_directive, attachments)
validation_errors = self.validator.validate_payload(payload)
if validation_errors:
raise ValueError(" | ".join(validation_errors))
url = f"{self.auth.base_url}/api/v2/cases"
headers = self.auth.get_headers()
headers["Idempotency-Key"] = str(uuid.uuid4())
start_time = time.perf_counter()
try:
response = requests.post(url, json=payload, headers=headers, timeout=15)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 2))
logging.warning(f"Rate limited. Retrying in {retry_after} seconds.")
time.sleep(retry_after)
response = requests.post(url, json=payload, headers=headers, timeout=15)
response.raise_for_status()
elapsed = time.perf_counter() - start_time
self.metrics["success"] += 1
self.metrics["latency_sum"] += elapsed
self._write_audit_log("CREATE_SUCCESS", payload, response.json(), elapsed)
self._sync_external_crm(response.json(), case_ref)
return response.json()
except requests.exceptions.HTTPError as e:
elapsed = time.perf_counter() - start_time
self.metrics["failures"] += 1
self.metrics["latency_sum"] += elapsed
self._write_audit_log("CREATE_FAILURE", payload, {"error": str(e)}, elapsed)
raise
def _write_audit_log(self, event: str, payload: dict, result: dict, latency: float) -> None:
log_entry = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"event": event,
"latency_seconds": round(latency, 4),
"payload_ref": payload.get("custom_fields", [{}])[0].get("value", "unknown"),
"result_id": result.get("id", "N/A")
}
logging.info(json.dumps(log_entry))
def _sync_external_crm(self, case_data: dict, case_ref: str) -> None:
logging.info(f"Webhook sync triggered for case {case_data.get('id')}. External CRM alignment queued for ref {case_ref}.")
def get_metrics(self) -> dict:
total = self.metrics["total"]
avg_latency = self.metrics["latency_sum"] / total if total > 0 else 0
success_rate = (self.metrics["success"] / total) * 100 if total > 0 else 0
return {
"total_attempts": total,
"success_rate_percent": round(success_rate, 2),
"average_latency_seconds": round(avg_latency, 4)
}
Complete Working Example
The following script ties authentication, validation, routing, and creation into a single executable module. Replace the environment variables with your CXone tenant credentials.
import os
import sys
def main():
region = os.getenv("CXONE_REGION", "eu1")
client_id = os.getenv("CXONE_CLIENT_ID")
client_secret = os.getenv("CXONE_CLIENT_SECRET")
case_type_id = os.getenv("CXONE_CASE_TYPE_ID")
if not all([client_id, client_secret, case_type_id]):
logging.error("Missing required environment variables.")
sys.exit(1)
auth = CxoNeAuthManager(region, client_id, client_secret)
creator = CxoNeCaseCreator(auth, case_type_id)
try:
result = creator.create_case(
subject="API Integration Timeout",
description="Webhook delivery failing after 3 retries. Requires immediate review.",
case_ref="REF-2024-0891",
ticket_matrix="technical",
open_directive="escalate",
attachments=[
{"name": "debug_trace.json", "url": "https://secure-storage.example.com/logs/trace.json"}
]
)
print(f"Case created successfully. CXone ID: {result['id']}")
print(f"Metrics: {json.dumps(creator.get_metrics(), indent=2)}")
except ValueError as ve:
logging.error(f"Validation failed: {ve}")
except requests.exceptions.HTTPError as he:
logging.error(f"API request failed: {he}")
except Exception as e:
logging.error(f"Unexpected error: {e}")
if __name__ == "__main__":
main()
Common Errors & Debugging
Error: 401 Unauthorized
Cause: The OAuth token expired or the client credentials are invalid.
Fix: Verify CXONE_CLIENT_ID and CXONE_CLIENT_SECRET. Ensure the get_token() method refreshes the token when time.time() >= token_expiry. The provided CxoNeAuthManager handles automatic refresh.
Error: 403 Forbidden
Cause: The OAuth token lacks the required cases:write scope, or the client is restricted to specific environments.
Fix: Regenerate the token with scope=cases:read cases:write. Confirm the client application in the CXone admin console has Case Management permissions enabled.
Error: 400 Bad Request
Cause: Payload schema mismatch, invalid custom field IDs, or missing required fields.
Fix: Run validator.validate_payload() before submission. Verify that cf_case_ref, cf_ticket_matrix, and cf_open_directive match the custom field IDs configured in your CXone instance. The API returns a detailed errors array in the response body.
Error: 409 Conflict
Cause: Duplicate case detection triggered by the search endpoint, or the Idempotency-Key header collided with a recent request.
Fix: The check_duplicate() method blocks creation when a matching case_ref exists. If you need to update an existing case, switch to PUT /api/v2/cases/{caseId} instead of POST.
Error: 429 Too Many Requests
Cause: Exceeding CXone rate limits (typically 100-200 requests per minute per client).
Fix: The create_case method implements automatic retry with Retry-After header parsing. For bulk operations, implement exponential backoff and queue requests with a token bucket limiter.