Bulk-Update NICE CXone Agent Skill Proficiencies via REST
What You Will Build
- A Python script that retrieves a list of agents, modifies their skill proficiencies for specific languages and technical domains, and pushes those updates back to the NICE CXone platform.
- This tutorial uses the NICE CXone Admin API (
/api/v2/usersand/api/v2/users/{userId}) and the REST client pattern. - The implementation is written in Python 3.9+ using
requestsandhttpx.
Prerequisites
OAuth Client and Scopes
You must have a NICE CXone OAuth Client ID and Secret with the following scopes assigned:
admin:agent:read— To fetch user details and current skill sets.admin:agent:update— To modify user profiles and skill proficiencies.admin:skill:read— To resolve skill IDs if you are mapping by name.
SDK/API Version
- API Version: v2 (The v1 API is deprecated for user management).
- Base URL:
https://api-us-2.nicecxone.com/api/v2(Adjust for your region:api-eu-1,api-au-1, etc.).
Dependencies
Install the required Python packages:
pip install requests httpx
Authentication Setup
NICE CXone uses OAuth 2.0 Client Credentials flow for server-to-server API access. You must obtain a bearer token before making any API calls.
Token Retrieval Code
import httpx
import time
from typing import Optional
class CxoneAuth:
def __init__(self, client_id: str, client_secret: str, region: str = "us-2"):
self.client_id = client_id
self.client_secret = client_secret
self.region = region
self.token_url = f"https://auth-{region}.nicecxone.com/oauth/token"
self.access_token: Optional[str] = None
self.token_expiry: float = 0
def get_token(self) -> str:
"""
Retrieves an OAuth token. Handles caching if the token is still valid.
"""
if self.access_token and time.time() < self.token_expiry:
return self.access_token
headers = {
"Content-Type": "application/x-www-form-urlencoded"
}
data = {
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret
}
with httpx.Client() as client:
response = client.post(self.token_url, headers=headers, data=data)
response.raise_for_status()
token_data = response.json()
self.access_token = token_data["access_token"]
# Expires in is in seconds; add to current time
self.token_expiry = time.time() + token_data["expires_in"] - 60 # Buffer 60s
return self.access_token
def get_headers(self) -> dict:
return {
"Authorization": f"Bearer {self.get_token()}",
"Accept": "application/json",
"Content-Type": "application/json"
}
Implementation
Step 1: Identify Skills by Name
NICE CXone APIs require Skill IDs, not names. If your bulk update input is based on skill names (e.g., “English”, “Billing”), you must resolve these to IDs first.
Endpoint: GET /api/v2/skills
Scope: admin:skill:read
import httpx
from typing import Dict, List
class CxoneSkillManager:
def __init__(self, auth: CxoneAuth):
self.auth = auth
self.base_url = f"https://api-{auth.region}.nicecxone.com/api/v2"
def get_skill_id_by_name(self, skill_name: str) -> Optional[str]:
"""
Searches the skills list to find the ID for a given skill name.
Note: This endpoint supports pagination. For simplicity, we fetch the first page.
In production, iterate through 'nextPageUri' if skills exceed 100.
"""
headers = self.auth.get_headers()
params = {
"skillName": skill_name,
"pageSize": 1
}
with httpx.Client() as client:
try:
response = client.get(
f"{self.base_url}/skills",
headers=headers,
params=params
)
response.raise_for_status()
data = response.json()
if data.get("items") and len(data["items"]) > 0:
return data["items"][0]["id"]
except httpx.HTTPStatusError as e:
print(f"Error fetching skill '{skill_name}': {e.response.text}")
return None
def resolve_skill_ids(self, skill_names: List[str]) -> Dict[str, str]:
"""
Maps a list of skill names to their IDs.
Returns { skill_name: skill_id }
"""
mapping = {}
for name in skill_names:
sid = self.get_skill_id_by_name(name)
if sid:
mapping[name] = sid
else:
print(f"Warning: Skill '{name}' not found.")
return mapping
Step 2: Fetch Agents and Current Proficiencies
You cannot blindly overwrite an agent’s skill set without knowing their existing proficiencies unless you intend to wipe all other skills. The standard pattern is: Read → Modify → Write.
Endpoint: GET /api/v2/users
Scope: admin:agent:read
import httpx
from typing import List, Dict, Any
class CxoneAgentManager:
def __init__(self, auth: CxoneAuth):
self.auth = auth
self.base_url = f"https://api-{auth.region}.nicecxone.com/api/v2"
def get_agents(self, query: str = "") -> List[Dict[str, Any]]:
"""
Retrieves a list of users. Supports querying by name or email.
Handles pagination automatically.
"""
all_agents = []
headers = self.auth.get_headers()
params = {
"pageSize": 100,
"page": 1
}
if query:
params["query"] = query
with httpx.Client() as client:
while True:
response = client.get(
f"{self.base_url}/users",
headers=headers,
params=params
)
response.raise_for_status()
data = response.json()
all_agents.extend(data.get("items", []))
# Check for next page
if data.get("nextPageUri"):
# The nextPageUri is a full URL, but we need to keep headers
# It is safer to increment page number if the URI structure is consistent
# or parse the URI. For simplicity, we increment page.
params["page"] += 1
else:
break
return all_agents
Step 3: Update Agent Skill Proficiencies
The core logic. The user object in CXone has a skills array. Each item in this array contains:
skillId: The UUID of the skill.proficiency: An integer (typically 0-100, or 0-5 depending on your skill configuration, but 0-100 is standard for most bulk operations).status: Usually"ACTIVE".
Endpoint: PUT /api/v2/users/{userId}
Scope: admin:agent:update
Critical Note: When updating a user via PUT, you must send the entire user object back. If you omit fields like firstName, lastName, or other attributes, they may be reset to null or default values depending on the API version behavior. Always merge changes into the existing object.
import httpx
from typing import List, Dict, Any
class CxoneAgentManager:
# ... (previous methods)
def update_agent_skills(
self,
agent: Dict[str, Any],
skill_updates: Dict[str, int]
) -> bool:
"""
Updates an agent's skill proficiencies.
Args:
agent: The full user object retrieved from GET /users.
skill_updates: A dict mapping { skill_id: proficiency_value }.
Example: { "abc-123": 80, "def-456": 60 }
Returns:
True if successful, False otherwise.
"""
user_id = agent.get("id")
if not user_id:
return False
# Deep copy to avoid mutating the original list reference if reused
import copy
updated_agent = copy.deepcopy(agent)
# Ensure the skills list exists
if "skills" not in updated_agent or updated_agent["skills"] is None:
updated_agent["skills"] = []
# Map existing skills by ID for quick lookup
existing_skills_map = {s["skillId"]: s for s in updated_agent["skills"]}
for skill_id, proficiency in skill_updates.items():
if skill_id in existing_skills_map:
# Update existing skill
existing_skills_map[skill_id]["proficiency"] = proficiency
else:
# Add new skill
new_skill = {
"skillId": skill_id,
"proficiency": proficiency,
"status": "ACTIVE" # Default status
}
updated_agent["skills"].append(new_skill)
# Remove skills that are no longer in the update list?
# Option A: Keep existing skills not in skill_updates (Safe)
# Option B: Wipe all skills not in skill_updates (Destructive)
# This implementation uses Option A (Merge).
headers = self.auth.get_headers()
with httpx.Client() as client:
try:
response = client.put(
f"{self.base_url}/users/{user_id}",
headers=headers,
json=updated_agent
)
# 204 No Content is standard for successful PUT updates in CXone
if response.status_code == 204:
return True
else:
print(f"Failed to update user {user_id}: {response.status_code} - {response.text}")
return False
except httpx.HTTPStatusError as e:
if e.response.status_code == 409:
print(f"Conflict updating user {user_id}. Another process may have modified this user.")
elif e.response.status_code == 429:
print(f"Rate limit hit. Retry after backoff.")
else:
print(f"HTTP Error {e.response.status_code}: {e.response.text}")
return False
Complete Working Example
This script combines the above components. It reads a CSV-like structure (defined in memory for this example) containing Agent Email, Skill Name, and Desired Proficiency, then applies the updates.
import httpx
import time
import json
import sys
from typing import List, Dict, Any, Optional
# --- Configuration ---
CLIENT_ID = "your_client_id"
CLIENT_SECRET = "your_client_secret"
REGION = "us-2" # Change to eu-1, au-1, etc.
# --- Mock Input Data: List of updates to apply ---
# In production, load this from CSV/Excel/Database
BULK_UPDATES = [
{
"agent_email": "john.doe@company.com",
"skill_name": "English",
"proficiency": 90
},
{
"agent_email": "john.doe@company.com",
"skill_name": "Billing",
"proficiency": 70
},
{
"agent_email": "jane.smith@company.com",
"skill_name": "Spanish",
"proficiency": 85
}
]
class CxoneBulkSkillUpdater:
def __init__(self, client_id: str, client_secret: str, region: str):
self.auth = CxoneAuth(client_id, client_secret, region)
self.skill_mgr = CxoneSkillManager(self.auth)
self.agent_mgr = CxoneAgentManager(self.auth)
self.base_url = f"https://api-{region}.nicecxone.com/api/v2"
def run(self):
print("1. Resolving Skill Names to IDs...")
# Extract unique skill names
unique_skills = list(set([u["skill_name"] for u in BULK_UPDATES]))
skill_map = self.skill_mgr.resolve_skill_ids(unique_skills)
if not skill_map:
print("No skills found. Aborting.")
return
print("2. Fetching Agents...")
# We need to find agents by email.
# CXone /users endpoint does not support direct email lookup in query param easily
# without fetching all or using a specific search.
# For this example, we will fetch all active users (or a large subset)
# and build a local lookup map.
# Optimization: If you know specific User IDs, skip this step and fetch by ID.
all_agents = self.agent_mgr.get_agents()
# Build email -> agent mapping
agent_by_email = {agent["email"]: agent for agent in all_agents}
print(f"Found {len(all_agents)} agents. Processing {len(BULK_UPDATES)} updates...")
# Group updates by agent to minimize API calls
# Structure: { email: { skill_id: proficiency } }
updates_by_agent: Dict[str, Dict[str, int]] = {}
for update in BULK_UPDATES:
email = update["agent_email"]
skill_name = update["skill_name"]
proficiency = update["proficiency"]
skill_id = skill_map.get(skill_name)
if not skill_id:
print(f"Skipping update for {email}: Skill '{skill_name}' not found.")
continue
if email not in agent_by_email:
print(f"Skipping update: Agent with email '{email}' not found.")
continue
if email not in updates_by_agent:
updates_by_agent[email] = {}
updates_by_agent[email][skill_id] = proficiency
# Apply Updates
success_count = 0
fail_count = 0
for email, skill_updates in updates_by_agent.items():
agent_obj = agent_by_email[email]
print(f"Updating skills for {email}...")
# Rate Limiting: CXone typically allows ~100 req/s but varies.
# Adding a small delay to be safe in bulk operations.
time.sleep(0.1)
if self.agent_mgr.update_agent_skills(agent_obj, skill_updates):
success_count += 1
else:
fail_count += 1
print(f"\nCompleted. Success: {success_count}, Failed: {fail_count}")
if __name__ == "__main__":
if CLIENT_ID == "your_client_id":
print("Error: Please configure CLIENT_ID and CLIENT_SECRET in the script.")
sys.exit(1)
updater = CxoneBulkSkillUpdater(CLIENT_ID, CLIENT_SECRET, REGION)
updater.run()
Common Errors & Debugging
Error: 403 Forbidden
Cause: The OAuth client lacks the admin:agent:update scope.
Fix: Log into the NICE CXone Admin UI, navigate to Administration > Security > OAuth Clients, edit your client, and ensure admin:agent:update is checked.
Error: 400 Bad Request - “Invalid Skill ID”
Cause: The skillId provided does not exist in the tenant, or it is a soft-deleted skill.
Fix: Verify the skill ID exists by calling GET /api/v2/skills/{skillId}. Ensure the skill is active.
Error: 409 Conflict
Cause: Another process (or admin user) modified the agent’s profile between your GET and PUT calls. CXone uses optimistic locking.
Fix: Implement a retry mechanism with exponential backoff. Re-fetch the user object (GET /users/{id}) immediately before retrying the PUT.
def update_with_retry(self, agent, skill_updates, max_retries=3):
for attempt in range(max_retries):
success = self.agent_mgr.update_agent_skills(agent, skill_updates)
if success:
return True
# If failed, re-fetch the latest version
print(f"Conflict detected. Re-fetching user data (Attempt {attempt + 1})...")
# Re-fetch logic here
# agent = self.agent_mgr.get_user_by_id(agent["id"])
time.sleep(2 ** attempt)
return False
Error: 429 Too Many Requests
Cause: You are exceeding the rate limit for the users endpoint.
Fix: Add a delay between requests. The time.sleep(0.1) in the complete example is a conservative starting point. Monitor the Retry-After header in 429 responses if available.