Bulk-Update Agent Skill Proficiencies in NICE CXone via REST

What You Will Build

• This tutorial demonstrates how to programmatically update the skill proficiency levels for multiple agents in NICE CXone using the Admin REST API.
• The solution utilizes the PUT /api/v2/users endpoint to perform batch updates on user profiles.
• The implementation is provided in Python 3.10+ using the requests library.

Prerequisites

• OAuth Client Type: Server-to-Server (Client Credentials) or User-to-Server (Authorization Code with PKCE). Server-to-Server is recommended for bulk administrative tasks.
• Required Scopes: admin:users:write and admin:users:read.
• API Version: CXone Admin API v2.
• Runtime Requirements: Python 3.10 or higher.
• External Dependencies: requests, tenacity (for robust retry logic).

Authentication Setup

NICE CXone uses OAuth 2.0 for authentication. For bulk operations, a Service Account (Client Credentials flow) is the most reliable method because it does not expire with user sessions and has higher rate limits than personal access tokens.

You must configure a Service Account in the CXone Admin Console under Security > OAuth > Clients. Ensure the client has the admin:users:write scope assigned.

The following Python code initializes the HTTP client and handles token acquisition. It includes a simple caching mechanism to avoid requesting a new token for every single API call, which helps mitigate 429 rate-limit errors.

import os
import requests
import time
from typing import Optional
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class CxoneAuthManager:
    def __init__(self, tenant_domain: str, client_id: str, client_secret: str):
        self.tenant_domain = tenant_domain
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = f"https://{tenant_domain}/api/v2/oauth/token"
        self.base_url = f"https://{tenant_domain}/api/v2"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0

    def _get_token(self) -> str:
        """
        Acquires a new OAuth2 access token using Client Credentials flow.
        """
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": "admin:users:write admin:users:read"
        }
        
        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }

        response = requests.post(self.token_url, data=payload, headers=headers)
        response.raise_for_status()
        
        token_data = response.json()
        self.access_token = token_data["access_token"]
        # Expire slightly before actual expiry to prevent mid-request failures
        self.token_expiry = time.time() + token_data["expires_in"] - 60
        return self.access_token

    def get_headers(self) -> dict:
        """
        Returns headers with a valid Bearer token. Refreshes if expired.
        """
        if not self.access_token or time.time() >= self.token_expiry:
            self._get_token()
        
        return {
            "Authorization": f"Bearer {self.access_token}",
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

# Usage Initialization
# Ensure these are set in environment variables for security
TENANT_DOMAIN = os.getenv("CXONE_TENANT_DOMAIN", "your-tenant.nicecvai.com")
CLIENT_ID = os.getenv("CXONE_CLIENT_ID")
CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET")

auth_manager = CxoneAuthManager(TENANT_DOMAIN, CLIENT_ID, CLIENT_SECRET)

Implementation

Step 1: Identify Agents and Current Skills

Before updating, you must know which users exist and what their current skill proficiencies are. CXone does not support a single “bulk update skills” endpoint that accepts a list of user IDs and skill objects simultaneously in a single atomic transaction. Instead, the standard pattern is to fetch the user details, modify the skills array in the payload, and send a PUT request to /api/v2/users/{userId}.

To optimize this, we will first fetch a list of users filtered by a specific queue or group, or simply iterate through a provided list of User IDs.

def fetch_user_details(user_id: str) -> dict:
    """
    Fetches detailed user information including current skills.
    
    Args:
        user_id: The UUID of the CXone user.
        
    Returns:
        dict: The user object from the API.
    """
    endpoint = f"{auth_manager.base_url}/users/{user_id}"
    headers = auth_manager.get_headers()
    
    try:
        response = requests.get(endpoint, headers=headers)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        if response.status_code == 404:
            print(f"User {user_id} not found.")
        elif response.status_code == 401:
            print("Authentication failed. Check token.")
        else:
            print(f"HTTP Error {response.status_code}: {e}")
        raise

Step 2: Construct the Update Payload

The PUT /api/v2/users/{userId} endpoint requires a full or partial user object. To update skills, you must include the skills attribute. The skills object contains a proficiencies array. Each proficiency object requires:

1. skill: The ID of the skill.
2. level: The proficiency level (e.g., 1-5 or custom labels).

Critical Note: If you send a partial update, you must include the skills wrapper correctly. If you omit the skills array entirely in the PUT body, it may reset the user’s skills to empty depending on the API version behavior. It is safer to fetch the user, modify the skills.proficiencies list, and send the updated object.

Below is a helper function to construct the update payload.

def build_update_payload(user_data: dict, new_skill_id: str, new_level: int) -> dict:
    """
    Modifies the user data to include a new or updated skill proficiency.
    
    Args:
        user_data: The existing user object from the API.
        new_skill_id: The ID of the skill to add/update.
        new_level: The proficiency level (integer).
        
    Returns:
        dict: The modified user object ready for PUT request.
    """
    # Ensure skills structure exists
    if "skills" not in user_data:
        user_data["skills"] = {"proficiencies": []}
    else:
        if "proficiencies" not in user_data["skills"]:
            user_data["skills"]["proficiencies"] = []
    
    proficiencies = user_data["skills"]["proficiencies"]
    
    # Check if skill already exists and update it, otherwise append
    updated = False
    for proficiency in proficiencies:
        if proficiency.get("skill") == new_skill_id:
            proficiency["level"] = new_level
            updated = True
            break
    
    if not updated:
        proficiencies.append({
            "skill": new_skill_id,
            "level": new_level
        })
    
    # Return only necessary fields to keep payload small and reduce risk of overwriting unrelated fields
    # However, for PUT /users, it is often safer to send the minimal required structure
    # depending on API strictness. CXone v2 is generally forgiving with partial updates
    # if the structure is correct.
    
    return {
        "skills": user_data["skills"]
    }

Step 3: Execute Bulk Updates with Rate Limiting

CXone enforces strict rate limits. Sending 100 requests in 1 second will result in 429 Too Many Requests. You must implement exponential backoff. The tenacity library is ideal for this.

We will iterate through a list of User IDs, fetch their data, modify the skills, and push the update.

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    retry=retry_if_exception_type(requests.exceptions.HTTPError)
)
def update_user_skills(user_id: str, skill_id: str, level: int) -> bool:
    """
    Updates a single user's skill proficiency with retry logic.
    
    Args:
        user_id: The UUID of the user.
        skill_id: The ID of the skill to update.
        level: The new proficiency level.
        
    Returns:
        bool: True if successful, False otherwise.
    """
    try:
        # 1. Fetch current user data
        user_data = fetch_user_details(user_id)
        
        # 2. Build update payload
        payload = build_update_payload(user_data, skill_id, level)
        
        # 3. Send PUT request
        endpoint = f"{auth_manager.base_url}/users/{user_id}"
        headers = auth_manager.get_headers()
        
        response = requests.put(endpoint, json=payload, headers=headers)
        
        # 4. Handle specific status codes
        if response.status_code == 200:
            return True
        elif response.status_code == 409:
            # Conflict usually means the user was modified by another process since we fetched it
            print(f"Conflict updating user {user_id}. Retry may resolve this.")
            raise requests.exceptions.HTTPError("Conflict")
        elif response.status_code == 429:
            # Tenacity will catch this if we raise HTTPError, but we can log it
            print(f"Rate limited for user {user_id}. Retrying...")
            raise requests.exceptions.HTTPError("Rate Limit")
        else:
            response.raise_for_status()
            
    except requests.exceptions.HTTPError as e:
        print(f"Failed to update user {user_id}: {e}")
        raise # Re-raise for tenacity to catch
    except Exception as e:
        print(f"Unexpected error updating user {user_id}: {e}")
        raise

def bulk_update_skills(user_ids: list, skill_id: str, level: int):
    """
    Iterates through a list of user IDs and updates their skill proficiency.
    
    Args:
        user_ids: List of User UUIDs.
        skill_id: The Skill UUID to apply.
        level: The proficiency level.
    """
    success_count = 0
    fail_count = 0
    
    print(f"Starting bulk update for {len(user_ids)} users...")
    
    for user_id in user_ids:
        try:
            if update_user_skills(user_id, skill_id, level):
                success_count += 1
                print(f"Successfully updated user: {user_id}")
        except Exception as e:
            fail_count += 1
            print(f"Permanently failed to update user: {user_id}. Error: {e}")
            
        # Optional: Small fixed delay between users to be polite to the API
        # even if not rate-limited, to spread load.
        time.sleep(0.2)

    print(f"Bulk update complete. Success: {success_count}, Failed: {fail_count}")

Complete Working Example

Below is the complete, copy-pasteable Python script. It combines authentication, data fetching, payload construction, and bulk execution with robust error handling.

Prerequisites:

1. Install dependencies: pip install requests tenacity
2. Set environment variables: CXONE_TENANT_DOMAIN, CXONE_CLIENT_ID, CXONE_CLIENT_SECRET.

import os
import requests
import time
from typing import Optional, List
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

# --- Configuration ---
TENANT_DOMAIN = os.getenv("CXONE_TENANT_DOMAIN")
CLIENT_ID = os.getenv("CXONE_CLIENT_ID")
CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET")

if not all([TENANT_DOMAIN, CLIENT_ID, CLIENT_SECRET]):
    raise ValueError("Missing environment variables: CXONE_TENANT_DOMAIN, CXONE_CLIENT_ID, CXONE_CLIENT_SECRET")

# --- Authentication Manager ---
class CxoneAuthManager:
    def __init__(self, tenant_domain: str, client_id: str, client_secret: str):
        self.tenant_domain = tenant_domain
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = f"https://{tenant_domain}/api/v2/oauth/token"
        self.base_url = f"https://{tenant_domain}/api/v2"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0

    def _get_token(self) -> str:
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": "admin:users:write admin:users:read"
        }
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        
        response = requests.post(self.token_url, data=payload, headers=headers)
        response.raise_for_status()
        
        token_data = response.json()
        self.access_token = token_data["access_token"]
        self.token_expiry = time.time() + token_data["expires_in"] - 60
        return self.access_token

    def get_headers(self) -> dict:
        if not self.access_token or time.time() >= self.token_expiry:
            self._get_token()
        return {
            "Authorization": f"Bearer {self.access_token}",
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

# --- API Logic ---
def fetch_user_details(auth: CxoneAuthManager, user_id: str) -> dict:
    endpoint = f"{auth.base_url}/users/{user_id}"
    headers = auth.get_headers()
    response = requests.get(endpoint, headers=headers)
    response.raise_for_status()
    return response.json()

def build_update_payload(user_data: dict, new_skill_id: str, new_level: int) -> dict:
    if "skills" not in user_data:
        user_data["skills"] = {"proficiencies": []}
    else:
        if "proficiencies" not in user_data["skills"]:
            user_data["skills"]["proficiencies"] = []
    
    proficiencies = user_data["skills"]["proficiencies"]
    updated = False
    
    for proficiency in proficiencies:
        if proficiency.get("skill") == new_skill_id:
            proficiency["level"] = new_level
            updated = True
            break
    
    if not updated:
        proficiencies.append({
            "skill": new_skill_id,
            "level": new_level
        })
    
    return {"skills": user_data["skills"]}

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    retry=retry_if_exception_type(requests.exceptions.HTTPError)
)
def update_single_user(auth: CxoneAuthManager, user_id: str, skill_id: str, level: int) -> bool:
    try:
        user_data = fetch_user_details(auth, user_id)
        payload = build_update_payload(user_data, skill_id, level)
        
        endpoint = f"{auth.base_url}/users/{user_id}"
        headers = auth.get_headers()
        
        response = requests.put(endpoint, json=payload, headers=headers)
        
        if response.status_code == 200:
            return True
        elif response.status_code == 409:
            raise requests.exceptions.HTTPError("Conflict - User modified by another process")
        else:
            response.raise_for_status()
    except requests.exceptions.HTTPError as e:
        raise e
    except Exception as e:
        raise requests.exceptions.HTTPError(f"Unexpected error: {e}")

def run_bulk_update(auth: CxoneAuthManager, user_ids: List[str], skill_id: str, level: int):
    success_count = 0
    fail_count = 0
    
    print(f"Starting bulk update for {len(user_ids)} users...")
    
    for user_id in user_ids:
        try:
            if update_single_user(auth, user_id, skill_id, level):
                success_count += 1
                print(f"[OK] User {user_id}")
            else:
                fail_count += 1
        except Exception as e:
            fail_count += 1
            print(f"[FAIL] User {user_id}: {str(e)}")
        
        # Polite delay
        time.sleep(0.2)

    print(f"Completed. Success: {success_count}, Failed: {fail_count}")

# --- Execution ---
if __name__ == "__main__":
    # Initialize Auth
    auth = CxoneAuthManager(TENANT_DOMAIN, CLIENT_ID, CLIENT_SECRET)
    
    # Example: Update these 3 users with Skill ID "abc-123-def" to Level 5
    # Replace these with real User UUIDs from your tenant
    TARGET_USERS = [
        "550e8400-e29b-41d4-a716-446655440000", 
        "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
        "6ba7b811-9dad-11d1-80b4-00c04fd430c8"
    ]
    TARGET_SKILL_ID = "your-skill-uuid-here"
    TARGET_LEVEL = 5
    
    run_bulk_update(auth, TARGET_USERS, TARGET_SKILL_ID, TARGET_LEVEL)

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token has expired, or the Client ID/Secret is incorrect.
• Fix: Ensure the CxoneAuthManager correctly refreshes the token. Check that the Service Account in CXone Admin has the admin:users:write scope. Verify the tenant_domain does not include https:// or trailing slashes.

Error: 403 Forbidden

• Cause: The OAuth client lacks the required scope.
• Fix: Go to Security > OAuth > Clients in the CXone Admin Console. Edit your client and ensure admin:users:write is checked. Changes may take up to 5 minutes to propagate.

Error: 409 Conflict

• Cause: The user object was modified by another user or process after you fetched it but before you sent the PUT request. CXone uses optimistic locking.
• Fix: The provided code uses tenacity to retry on 409 errors. This is the correct approach. If it fails after 5 retries, the user is being actively edited by an administrator.

Error: 429 Too Many Requests

• Cause: You are sending requests faster than the API allows.
• Fix: The code includes time.sleep(0.2) between requests and exponential backoff via tenacity. If you still hit 429s, increase the sleep duration or reduce the concurrency. Do not use multithreading for this specific bulk update pattern unless you implement a sophisticated semaphore-based rate limiter.

Error: 400 Bad Request

• Cause: The payload structure is invalid.
• Fix: Ensure the skills.proficiencies array contains objects with both skill (UUID) and level (integer) keys. If the skill ID is invalid, the API may return 400 or 404. Validate that the TARGET_SKILL_ID exists in your tenant.

Official References

• NICE CXone Admin API: Users
• NICE CXone OAuth2 Authentication
• Python Requests Library