Bulk-Update Agent Skill Proficiencies via NICE CXone Admin REST API

What You Will Build

• A Python script that authenticates to NICE CXone, retrieves current agent skill assignments, applies batch proficiency updates, and persists the changes via the REST API.
• This tutorial uses the NICE CXone Admin API (/api/v2/admin/users) and the associated skill assignment endpoints.
• The implementation is provided in Python 3.10+ using the requests library for direct HTTP interaction, ensuring full control over batching and error handling.

Prerequisites

• OAuth Client Credentials: You need a NICE CXone OAuth client with the admin role. Specifically, you require the admin:users:write and admin:users:read scopes.
• API Endpoint: The base URL for your CXone environment (e.g., https://api.nicecxone.com or your specific region like https://api.usw.nicecxone.com).
• Python Environment: Python 3.10 or higher.
• Dependencies: Install the requests library if not already present.

  pip install requests
  

Authentication Setup

NICE CXone uses standard OAuth 2.0 Client Credentials flow. The token expires after a short duration (typically 15 minutes), so the code below includes a simple token cache mechanism to avoid unnecessary re-authentication during batch operations.

import requests
import time
import json
from typing import Optional

class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str, base_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = base_url.rstrip('/')
        self.token_url = f"{self.base_url}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0

    def get_token(self) -> str:
        """
        Retrieves an OAuth access token. Returns a cached token if valid.
        """
        if self.access_token and time.time() < self.token_expiry - 60:
            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()
            data = response.json()
            
            self.access_token = data["access_token"]
            # Store expiry time (current time + expires_in seconds)
            self.token_expiry = time.time() + data.get("expires_in", 900)
            
            return self.access_token
        except requests.exceptions.HTTPError as e:
            raise Exception(f"Authentication failed: {e.response.text}") from e
        except requests.exceptions.RequestException as e:
            raise Exception(f"Network error during authentication: {e}") from e

Implementation

Step 1: Retrieve Current Agent Skill Assignments

Before updating proficiencies, it is often safer to read the current state to ensure you are not overwriting unrelated metadata or to verify the target users exist. We will fetch a list of users based on a group or simply iterate through a provided list of User IDs.

For this tutorial, we assume you have a list of user_ids and skill_ids you wish to update. However, to demonstrate the read capability, here is how you fetch a specific user’s details, which includes their skill assignments.

Endpoint: GET /api/v2/admin/users/{userId}
Scope: admin:users:read

class CXoneAdminClient:
    def __init__(self, auth: CXoneAuth):
        self.auth = auth
        self.base_url = auth.base_url

    def get_headers(self) -> dict:
        token = self.auth.get_token()
        return {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }

    def get_user_details(self, user_id: str) -> dict:
        """
        Fetches detailed information for a single user, including skill assignments.
        """
        url = f"{self.base_url}/api/v2/admin/users/{user_id}"
        headers = self.get_headers()

        try:
            response = requests.get(url, headers=headers)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            if response.status_code == 404:
                raise ValueError(f"User {user_id} not found.") from e
            raise Exception(f"Failed to fetch user {user_id}: {e}") from e

Step 2: Construct the Bulk Update Payload

NICE CXone does not have a single “bulk update skill proficiency” endpoint that takes a list of arbitrary updates in one call. Instead, the standard pattern for bulk operations is to iterate through your dataset and issue individual PATCH requests to the user endpoint, or use the bulk user update endpoint if available for specific attributes.

However, the most robust and common method for updating Skill Proficiencies specifically is via the PATCH /api/v2/admin/users/{userId} endpoint, modifying the skillAssignments array within the user object.

To optimize this, we will create a helper that structures the JSON body correctly. The skillAssignments object typically looks like this:

{
  "skillAssignments": {
    "skillId123": {
      "proficiency": 5,
      "skillGroupId": "groupId456"
    }
  }
}

Important: When using PATCH, you only send the fields you wish to change. If you send the entire user object, you risk overwriting other fields. We will send a minimal payload containing only the skillAssignments updates.

def create_skill_update_payload(skill_id: str, proficiency: int, skill_group_id: str) -> dict:
    """
    Creates the JSON body for a PATCH request to update a single skill proficiency.
    
    Args:
        skill_id: The ID of the skill to update.
        proficiency: The new proficiency level (integer, typically 1-5).
        skill_group_id: The ID of the skill group this skill belongs to.
        
    Returns:
        A dictionary representing the JSON body for the PATCH request.
    """
    return {
        "skillAssignments": {
            skill_id: {
                "proficiency": proficiency,
                "skillGroupId": skill_group_id
            }
        }
    }

Step 3: Execute Bulk Updates with Retry Logic

We will now implement the core loop that iterates through a list of updates. This function includes basic retry logic for transient errors (like 429 Too Many Requests) which are common when performing bulk operations.

Endpoint: PATCH /api/v2/admin/users/{userId}
Scope: admin:users:write

import time

class CXoneSkillUpdater:
    def __init__(self, client: CXoneAdminClient):
        self.client = client

    def update_agent_skill(self, user_id: str, skill_id: str, proficiency: int, skill_group_id: str) -> bool:
        """
        Updates the proficiency of a specific skill for a specific user.
        
        Args:
            user_id: The ID of the agent/user.
            skill_id: The ID of the skill.
            proficiency: The new proficiency level.
            skill_group_id: The ID of the skill group.
            
        Returns:
            True if successful, False otherwise.
        """
        url = f"{self.client.base_url}/api/v2/admin/users/{user_id}"
        headers = self.client.get_headers()
        payload = create_skill_update_payload(skill_id, proficiency, skill_group_id)
        
        max_retries = 3
        for attempt in range(max_retries):
            try:
                response = requests.patch(url, headers=headers, json=payload)
                
                # Handle Success
                if response.status_code == 200 or response.status_code == 204:
                    return True
                
                # Handle Rate Limiting (429)
                if response.status_code == 429:
                    wait_time = 2 ** attempt  # Exponential backoff
                    time.sleep(wait_time)
                    continue
                
                # Handle Other Errors
                response.raise_for_status()
                
            except requests.exceptions.HTTPError as e:
                # Log the error but do not retry on 400/403/404
                if response.status_code in [400, 403, 404]:
                    print(f"Error updating user {user_id} skill {skill_id}: {response.text}")
                    return False
                raise Exception(f"HTTP Error for user {user_id}: {e}") from e
            except requests.exceptions.RequestException as e:
                # Network errors, retry
                if attempt == max_retries - 1:
                    raise Exception(f"Network failure for user {user_id} after {max_retries} attempts: {e}")
                time.sleep(1)

        return False

    def bulk_update_skills(self, updates: list) -> dict:
        """
        Processes a list of skill updates.
        
        Args:
            updates: A list of tuples/dicts containing (user_id, skill_id, proficiency, skill_group_id).
            
        Returns:
            A dictionary with success and failure counts.
        """
        results = {"success": 0, "failed": 0, "errors": []}
        
        for update in updates:
            user_id = update["user_id"]
            skill_id = update["skill_id"]
            proficiency = update["proficiency"]
            skill_group_id = update["skill_group_id"]
            
            try:
                success = self.update_agent_skill(user_id, skill_id, proficiency, skill_group_id)
                if success:
                    results["success"] += 1
                else:
                    results["failed"] += 1
                    results["errors"].append(f"Failed to update user {user_id} skill {skill_id}")
            except Exception as e:
                results["failed"] += 1
                results["errors"].append(f"Exception for user {user_id}: {str(e)}")
                
        return results

Complete Working Example

This script ties everything together. It defines a mock dataset of updates, initializes the authentication, and executes the bulk update.

import requests
import time
import json
from typing import Optional, List, Dict

# --- Configuration ---
CLIENT_ID = "your_client_id"
CLIENT_SECRET = "your_client_secret"
BASE_URL = "https://api.usw.nicecxone.com" # Change to your region

# --- Mock Data: List of updates to apply ---
# In a real scenario, this would come from a CSV, Excel file, or database query.
BULK_UPDATES = [
    {
        "user_id": "user-id-12345",
        "skill_id": "skill-id-abcde",
        "proficiency": 5,
        "skill_group_id": "group-id-11111"
    },
    {
        "user_id": "user-id-12346",
        "skill_id": "skill-id-abcde",
        "proficiency": 4,
        "skill_group_id": "group-id-11111"
    },
    # Add more updates here
]

class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str, base_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = base_url.rstrip('/')
        self.token_url = f"{self.base_url}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0

    def get_token(self) -> str:
        if self.access_token and time.time() < self.token_expiry - 60:
            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()
            data = response.json()
            self.access_token = data["access_token"]
            self.token_expiry = time.time() + data.get("expires_in", 900)
            return self.access_token
        except requests.exceptions.HTTPError as e:
            raise Exception(f"Authentication failed: {e.response.text}") from e

class CXoneAdminClient:
    def __init__(self, auth: CXoneAuth):
        self.auth = auth
        self.base_url = auth.base_url

    def get_headers(self) -> dict:
        token = self.auth.get_token()
        return {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }

def create_skill_update_payload(skill_id: str, proficiency: int, skill_group_id: str) -> dict:
    return {
        "skillAssignments": {
            skill_id: {
                "proficiency": proficiency,
                "skillGroupId": skill_group_id
            }
        }
    }

def update_agent_skill(client: CXoneAdminClient, user_id: str, skill_id: str, proficiency: int, skill_group_id: str) -> bool:
    url = f"{client.base_url}/api/v2/admin/users/{user_id}"
    headers = client.get_headers()
    payload = create_skill_update_payload(skill_id, proficiency, skill_group_id)
    
    max_retries = 3
    for attempt in range(max_retries):
        try:
            response = requests.patch(url, headers=headers, json=payload)
            
            if response.status_code in [200, 204]:
                return True
            
            if response.status_code == 429:
                time.sleep(2 ** attempt)
                continue
            
            response.raise_for_status()
            
        except requests.exceptions.HTTPError as e:
            if response.status_code in [400, 403, 404]:
                print(f"Error updating user {user_id} skill {skill_id}: {response.text}")
                return False
            raise Exception(f"HTTP Error for user {user_id}: {e}") from e
            
    return False

def main():
    # Initialize Authentication
    auth = CXoneAuth(CLIENT_ID, CLIENT_SECRET, BASE_URL)
    
    # Initialize Client
    client = CXoneAdminClient(auth)
    
    # Execute Bulk Updates
    success_count = 0
    failed_count = 0
    
    print(f"Starting bulk update for {len(BULK_UPDATES)} agents...")
    
    for update in BULK_UPDATES:
        try:
            success = update_agent_skill(
                client,
                update["user_id"],
                update["skill_id"],
                update["proficiency"],
                update["skill_group_id"]
            )
            if success:
                success_count += 1
            else:
                failed_count += 1
        except Exception as e:
            print(f"Unexpected error for {update['user_id']}: {e}")
            failed_count += 1
            
    print(f"Completed. Success: {success_count}, Failed: {failed_count}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

• Cause: The OAuth client lacks the admin:users:write scope, or the user account associated with the client does not have the necessary administrative permissions in CXone.
• Fix: Verify the OAuth client configuration in the CXone Admin Portal. Ensure the admin:users:write scope is granted. Check that the client is associated with an admin role that can modify user profiles.

Error: 400 Bad Request

• Cause: The JSON payload is malformed, or the skill_id does not belong to the specified skill_group_id.
• Fix: Validate the JSON structure. Ensure that the skillGroupId in the payload matches the group to which the skillId actually belongs. You can verify this by fetching the skill details via GET /api/v2/admin/skills/{skillId}.

Error: 404 Not Found

• Cause: The user_id provided does not exist in the CXone instance.
• Fix: Verify the User ID. Use the GET /api/v2/admin/users/{userId} endpoint to confirm the user exists before attempting the update.

Error: 429 Too Many Requests

• Cause: The rate limit for the admin/users endpoint has been exceeded.
• Fix: Implement exponential backoff (as shown in the code). Reduce the concurrency if you are running multiple threads. CXone typically allows ~100-200 requests per minute for admin APIs, but this can vary by contract.

Official References

• NICE CXone Admin API - Users
• NICE CXone OAuth 2.0 Documentation
• User Object Schema