CXone Admin API — how to bulk-update agent skill proficiencies via REST
What You Will Build
- You will build a script that retrieves a list of agents, updates their skill proficiency levels for a specific set of skills, and pushes those changes back to NICE CXone via the REST API.
- This tutorial uses the NICE CXone Admin API (specifically the Users and Skills endpoints).
- The implementation is provided in Python using the
requestslibrary, with HTTP/1.1 REST calls.
Prerequisites
- OAuth Client Type: Service Account (Client Credentials Grant) or User Access Token (Authorization Code Grant). Service accounts are recommended for bulk administrative tasks to avoid user session timeouts.
- Required OAuth Scopes:
admin:users:read(to retrieve agent profiles)admin:users:write(to update agent profiles)admin:skills:read(to retrieve skill IDs if not already known)
- SDK/API Version: CXone REST API v1 (Current stable version).
- Language/Runtime: Python 3.8+
- External Dependencies:
requests(for HTTP calls)python-dotenv(for secure credential management)
Authentication Setup
NICE CXone uses OAuth 2.0 for authentication. For bulk administrative operations, the Client Credentials Grant flow is the most robust method. It provides a long-lived access token (typically 1 hour) that can be refreshed without user interaction.
Step 1: Configure Environment Variables
Create a .env file in your project root. Replace the placeholder values with your actual CXone tenant details.
# .env
CXONE_TENANT_URL=https://your-tenant.nice-incontact.com
CXONE_CLIENT_ID=your_client_id_here
CXONE_CLIENT_SECRET=your_client_secret_here
Step 2: Implement Token Fetcher
The following Python class handles the OAuth token retrieval and caching. It ensures that you do not hit rate limits by repeatedly requesting new tokens.
import os
import time
import requests
from dotenv import load_dotenv
load_dotenv()
class CXoneAuth:
def __init__(self):
self.tenant_url = os.getenv("CXONE_TENANT_URL")
self.client_id = os.getenv("CXONE_CLIENT_ID")
self.client_secret = os.getenv("CXONE_CLIENT_SECRET")
self.token_url = f"{self.tenant_url}/api/oauth2/token"
self.access_token = None
self.token_expiry = 0
def get_token(self) -> str:
"""
Returns a valid OAuth access token.
If the current token is expired or does not exist, it fetches a new one.
"""
if self.access_token and time.time() < self.token_expiry:
return self.access_token
payload = {
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret
}
try:
response = requests.post(self.token_url, data=payload)
response.raise_for_status()
data = response.json()
self.access_token = data["access_token"]
# Set expiry to slightly before actual expiry to prevent race conditions
self.token_expiry = time.time() + (data["expires_in"] - 60)
return self.access_token
except requests.exceptions.HTTPError as e:
print(f"Authentication failed: {e.response.status_code} - {e.response.text}")
raise
except Exception as e:
print(f"Failed to fetch token: {str(e)}")
raise
Implementation
Step 1: Retrieve Target Agents and Skills
Before updating proficiencies, you must identify the unique IDs of the agents and the specific skills you intend to modify. CXone identifies agents by id and skills by skill.id.
OAuth Scope Required: admin:users:read, admin:skills:read
First, we define helper methods to fetch these IDs. We will use the /api/v2/users endpoint to list users and filter for agents, and /api/v1/skills to find the target skill.
import json
class CXoneAgentSkillUpdater:
def __init__(self, auth: CXoneAuth):
self.auth = auth
self.base_url = auth.tenant_url
def get_skill_id_by_name(self, skill_name: str) -> str:
"""
Fetches the skill ID for a given skill name.
Endpoint: GET /api/v1/skills
"""
headers = {
"Authorization": f"Bearer {self.auth.get_token()}",
"Content-Type": "application/json"
}
url = f"{self.base_url}/api/v1/skills"
params = {
"name": skill_name,
"pageSize": 1
}
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
data = response.json()
if not data.get("items"):
raise ValueError(f"Skill '{skill_name}' not found.")
return data["items"][0]["id"]
def get_agent_ids_by_email_suffix(self, email_suffix: str) -> list:
"""
Fetches agent IDs for users matching an email suffix.
Endpoint: GET /api/v2/users
"""
headers = {
"Authorization": f"Bearer {self.auth.get_token()}",
"Content-Type": "application/json"
}
url = f"{self.base_url}/api/v2/users"
params = {
"email": f"*{email_suffix}",
"pageSize": 200
}
agent_ids = []
while url:
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
data = response.json()
for user in data.get("items", []):
# Filter for agents only (users with a routing profile assigned)
if user.get("routingProfileId"):
agent_ids.append(user["id"])
# Handle pagination
url = data.get("nextPageUri")
# Reset params for subsequent calls if using nextPageUri
if url:
params = {}
return agent_ids
Step 2: Construct the Bulk Update Payload
CXone does not have a single “bulk update skills” endpoint that accepts a list of user IDs and a new skill level in one call. Instead, the standard pattern for bulk updates is:
- Fetch the current user profile.
- Modify the
skillsarray within the user object. - Send a
PUTrequest to/api/v2/users/{userId}.
To optimize performance and avoid rate limiting (429 errors), we will process these updates sequentially with a small delay or implement a simple queue. For this tutorial, we will implement a sequential update with error handling per agent.
OAuth Scope Required: admin:users:write
The proficiency level is represented by the proficiency field within the skill object. Valid values are typically integers ranging from 1 to 10, where 10 is the highest proficiency.
def update_agent_skill_proficiency(self, user_id: str, skill_id: str, proficiency: int) -> bool:
"""
Updates the proficiency of a specific skill for a specific user.
Endpoint: PUT /api/v2/users/{userId}
Args:
user_id: The unique ID of the agent.
skill_id: The unique ID of the skill.
proficiency: Integer from 1 to 10.
Returns:
True if successful, False otherwise.
"""
headers = {
"Authorization": f"Bearer {self.auth.get_token()}",
"Content-Type": "application/json"
}
# Step 1: Get current user profile
get_url = f"{self.base_url}/api/v2/users/{user_id}"
get_response = requests.get(get_url, headers=headers)
if get_response.status_code == 404:
print(f"User {user_id} not found.")
return False
elif get_response.status_code == 401:
# Token might have expired mid-batch, refresh and retry
self.auth.access_token = None
headers["Authorization"] = f"Bearer {self.auth.get_token()}"
get_response = requests.get(get_url, headers=headers)
get_response.raise_for_status()
user_data = get_response.json()
# Step 2: Modify the skills array
current_skills = user_data.get("skills", [])
updated = False
for skill_obj in current_skills:
if skill_obj.get("skillId") == skill_id:
# Update existing skill
skill_obj["proficiency"] = proficiency
updated = True
break
if not updated:
# Add new skill if not present
new_skill = {
"skillId": skill_id,
"proficiency": proficiency,
"primary": False # Set to True if this is the primary skill
}
current_skills.append(new_skill)
user_data["skills"] = current_skills
# Step 3: Send PUT request
put_url = f"{self.base_url}/api/v2/users/{user_id}"
try:
put_response = requests.put(put_url, headers=headers, json=user_data)
put_response.raise_for_status()
print(f"Successfully updated skill proficiency for User ID: {user_id}")
return True
except requests.exceptions.HTTPError as e:
if e.response.status_code == 409:
print(f"Conflict updating user {user_id}. The user may have been modified by another process.")
elif e.response.status_code == 429:
print("Rate limit exceeded. Please implement exponential backoff.")
else:
print(f"Error updating user {user_id}: {e.response.text}")
return False
Step 3: Execute the Bulk Operation
This method ties the previous steps together. It retrieves the list of agents, finds the target skill, and iterates through the agents to update their proficiencies.
def bulk_update_agents(self, email_suffix: str, skill_name: str, proficiency: int):
"""
Orchestrates the bulk update process.
"""
print(f"Starting bulk update for skill '{skill_name}' to proficiency {proficiency}...")
# 1. Get Skill ID
try:
skill_id = self.get_skill_id_by_name(skill_name)
print(f"Found Skill ID: {skill_id}")
except Exception as e:
print(f"Error finding skill: {e}")
return
# 2. Get Agent IDs
try:
agent_ids = self.get_agent_ids_by_email_suffix(email_suffix)
print(f"Found {len(agent_ids)} agents matching email suffix '{email_suffix}'")
except Exception as e:
print(f"Error fetching agents: {e}")
return
if not agent_ids:
print("No agents found to update.")
return
# 3. Iterate and Update
success_count = 0
fail_count = 0
for user_id in agent_ids:
# Small delay to respect rate limits (e.g., 100ms)
# CXone allows ~100 requests per second for most endpoints, but PUTs can be heavier.
time.sleep(0.1)
if self.update_agent_skill_proficiency(user_id, skill_id, proficiency):
success_count += 1
else:
fail_count += 1
print(f"\nBulk Update Complete.")
print(f"Successes: {success_count}")
print(f"Failures: {fail_count}")
Complete Working Example
Below is the full, copy-pasteable script. Save this as bulk_update_skills.py. Ensure you have installed requests and python-dotenv via pip.
import os
import time
import requests
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
class CXoneAuth:
def __init__(self):
self.tenant_url = os.getenv("CXONE_TENANT_URL")
self.client_id = os.getenv("CXONE_CLIENT_ID")
self.client_secret = os.getenv("CXONE_CLIENT_SECRET")
if not all([self.tenant_url, self.client_id, self.client_secret]):
raise EnvironmentError("Missing required environment variables: CXONE_TENANT_URL, CXONE_CLIENT_ID, CXONE_CLIENT_SECRET")
self.token_url = f"{self.tenant_url}/api/oauth2/token"
self.access_token = None
self.token_expiry = 0
def get_token(self) -> str:
if self.access_token and time.time() < self.token_expiry:
return self.access_token
payload = {
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret
}
try:
response = requests.post(self.token_url, data=payload)
response.raise_for_status()
data = response.json()
self.access_token = data["access_token"]
self.token_expiry = time.time() + (data["expires_in"] - 60)
return self.access_token
except requests.exceptions.HTTPError as e:
print(f"Authentication failed: {e.response.status_code} - {e.response.text}")
raise
except Exception as e:
print(f"Failed to fetch token: {str(e)}")
raise
class CXoneAgentSkillUpdater:
def __init__(self, auth: CXoneAuth):
self.auth = auth
self.base_url = auth.tenant_url
def get_skill_id_by_name(self, skill_name: str) -> str:
headers = {
"Authorization": f"Bearer {self.auth.get_token()}",
"Content-Type": "application/json"
}
url = f"{self.base_url}/api/v1/skills"
params = {
"name": skill_name,
"pageSize": 1
}
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
data = response.json()
if not data.get("items"):
raise ValueError(f"Skill '{skill_name}' not found.")
return data["items"][0]["id"]
def get_agent_ids_by_email_suffix(self, email_suffix: str) -> list:
headers = {
"Authorization": f"Bearer {self.auth.get_token()}",
"Content-Type": "application/json"
}
url = f"{self.base_url}/api/v2/users"
params = {
"email": f"*{email_suffix}",
"pageSize": 200
}
agent_ids = []
while url:
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
data = response.json()
for user in data.get("items", []):
if user.get("routingProfileId"):
agent_ids.append(user["id"])
url = data.get("nextPageUri")
if url:
params = {}
return agent_ids
def update_agent_skill_proficiency(self, user_id: str, skill_id: str, proficiency: int) -> bool:
headers = {
"Authorization": f"Bearer {self.auth.get_token()}",
"Content-Type": "application/json"
}
get_url = f"{self.base_url}/api/v2/users/{user_id}"
get_response = requests.get(get_url, headers=headers)
# Handle 401 by refreshing token
if get_response.status_code == 401:
self.auth.access_token = None
headers["Authorization"] = f"Bearer {self.auth.get_token()}"
get_response = requests.get(get_url, headers=headers)
if get_response.status_code == 404:
print(f"User {user_id} not found.")
return False
get_response.raise_for_status()
user_data = get_response.json()
current_skills = user_data.get("skills", [])
updated = False
for skill_obj in current_skills:
if skill_obj.get("skillId") == skill_id:
skill_obj["proficiency"] = proficiency
updated = True
break
if not updated:
new_skill = {
"skillId": skill_id,
"proficiency": proficiency,
"primary": False
}
current_skills.append(new_skill)
user_data["skills"] = current_skills
put_url = f"{self.base_url}/api/v2/users/{user_id}"
try:
put_response = requests.put(put_url, headers=headers, json=user_data)
put_response.raise_for_status()
print(f"Updated User ID: {user_id}")
return True
except requests.exceptions.HTTPError as e:
if e.response.status_code == 429:
print("Rate limit exceeded. Waiting 5 seconds...")
time.sleep(5)
return self.update_agent_skill_proficiency(user_id, skill_id, proficiency) # Retry once
print(f"Error updating user {user_id}: {e.response.text}")
return False
def bulk_update_agents(self, email_suffix: str, skill_name: str, proficiency: int):
print(f"Starting bulk update for skill '{skill_name}' to proficiency {proficiency}...")
try:
skill_id = self.get_skill_id_by_name(skill_name)
print(f"Found Skill ID: {skill_id}")
except Exception as e:
print(f"Error finding skill: {e}")
return
try:
agent_ids = self.get_agent_ids_by_email_suffix(email_suffix)
print(f"Found {len(agent_ids)} agents matching email suffix '{email_suffix}'")
except Exception as e:
print(f"Error fetching agents: {e}")
return
if not agent_ids:
print("No agents found to update.")
return
success_count = 0
fail_count = 0
for user_id in agent_ids:
time.sleep(0.1) # Rate limit protection
if self.update_agent_skill_proficiency(user_id, skill_id, proficiency):
success_count += 1
else:
fail_count += 1
print(f"\nBulk Update Complete.")
print(f"Successes: {success_count}")
print(f"Failures: {fail_count}")
if __name__ == "__main__":
try:
auth = CXoneAuth()
updater = CXoneAgentSkillUpdater(auth)
# Configuration for the bulk update
# Example: Update all agents with @company.com emails to have proficiency 10 in "Technical Support"
EMAIL_SUFFIX = "@company.com"
SKILL_NAME = "Technical Support"
PROFICIENCY_LEVEL = 10
updater.bulk_update_agents(EMAIL_SUFFIX, SKILL_NAME, PROFICIENCY_LEVEL)
except Exception as e:
print(f"Critical Error: {str(e)}")
Common Errors & Debugging
Error: 401 Unauthorized
Cause: The OAuth token has expired.
Fix: Ensure your CXoneAuth class checks the token_expiry timestamp. In the update_agent_skill_proficiency method, we explicitly check for 401 status codes and refresh the token before retrying the request.
Error: 403 Forbidden
Cause: The OAuth client does not have the required scopes.
Fix: Verify that your Service Account in the CXone Admin Console has the admin:users:write and admin:users:read scopes assigned. Check the scope field in the token response payload to confirm.
Error: 429 Too Many Requests
Cause: You are sending requests faster than the CXone API allows.
Fix: Implement exponential backoff or fixed delays. In the complete example, time.sleep(0.1) is used. For larger batches (1000+ agents), consider increasing this delay or implementing a queue-based worker with concurrency limits (e.g., using concurrent.futures.ThreadPoolExecutor with max_workers=5).
Error: 409 Conflict
Cause: Another admin modified the user profile while your script was reading it.
Fix: The CXone API uses optimistic locking. If you receive a 409, you must re-fetch the user profile (GET /api/v2/users/{id}), apply your changes to the fresh data, and attempt the PUT again. The code above handles this by logging the error, but in production, you should implement a retry loop for 409s.
Error: Skill Not Found
Cause: The skill name provided does not match exactly, or the skill is not visible to the OAuth client’s site/region.
Fix: Use the /api/v1/skills endpoint with pageSize increased to list all skills and verify the exact name and id. Ensure the skill is published and active.