CXone Admin API — how to bulk-update agent skill proficiencies via REST

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 requests library, 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:

  1. Fetch the current user profile.
  2. Modify the skills array within the user object.
  3. Send a PUT request 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.

Official References