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

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

What You Will Build

  • This tutorial demonstrates how to programmatically update the proficiency levels for multiple agents across multiple skills in a single transaction.
  • It utilizes the NICE CXone Admin REST API, specifically the User Skills endpoint with batch processing capabilities.
  • The implementation uses Python with the requests library for HTTP handling.

Prerequisites

  • OAuth Client: A CXone Admin API Client ID and Client Secret with the User.ReadWrite scope.
  • API Version: CXone Admin API (v2).
  • Language/Runtime: Python 3.8+ installed.
  • External Dependencies:
    • requests (HTTP library)
    • pyjwt (optional, for token debugging, not strictly required for this tutorial)

Install dependencies:

pip install requests

Authentication Setup

NICE CXone uses OAuth 2.0 Client Credentials Grant for server-to-server communication. You must obtain an access token before calling any Admin API endpoints. The token expires after one hour, so production code should implement caching or refresh logic.

The following function handles the authentication handshake. It returns the access token string.

import requests
import time
from typing import Optional

# Configuration
CXONE_DOMAIN = "api-us-02.nice-incontact.com"  # Adjust for your region (api-eu-01, api-ap-01, etc.)
CLIENT_ID = "your_client_id"
CLIENT_SECRET = "your_client_secret"
SCOPES = "User.ReadWrite"

def get_access_token() -> str:
    """
    Obtains an OAuth 2.0 access token from CXone.
    In production, implement caching to avoid calling this on every request.
    """
    url = f"https://{CXONE_DOMAIN}/oauth/token"
    
    payload = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "scope": SCOPES
    }

    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }

    response = requests.post(url, data=payload, headers=headers)
    
    if response.status_code != 200:
        raise Exception(f"Authentication failed with status {response.status_code}: {response.text}")
    
    data = response.json()
    return data["access_token"]

# Cache token globally for this session
_token_cache: Optional[str] = None
_token_expiry: float = 0

def get_cached_token() -> str:
    global _token_cache, _token_expiry
    
    # If token exists and has not expired (minus 60s buffer), return it
    if _token_cache and time.time() < _token_expiry:
        return _token_cache
    
    # Fetch new token
    token = get_access_token()
    
    # Parse expiry from token payload (standard JWT) or assume 1 hour
    # For simplicity in this tutorial, we assume standard 1 hour expiry
    _token_cache = token
    _token_expiry = time.time() + 3500 # 58 minutes
    
    return token

Required Scope: User.ReadWrite
Endpoint: POST https://{domain}/oauth/token

Implementation

Step 1: Identify Target Agents and Skills

Before updating proficiencies, you must know the exact id (UUID) of the agents and the id (UUID) of the skills. You cannot use names or external IDs directly in the proficiency update payload.

We will write a helper function to search for users by name. The CXone API supports fuzzy search on the /users endpoint.

def find_user_by_name(token: str, name_query: str) -> Optional[dict]:
    """
    Searches for a user by name and returns the first match.
    Returns None if no user is found.
    """
    url = f"https://{CXONE_DOMAIN}/api/v2/users"
    params = {
        "query": name_query,
        "pageSize": 10,
        "pageNumber": 1
    }
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/json"
    }

    response = requests.get(url, headers=headers, params=params)
    
    if response.status_code != 200:
        raise Exception(f"User search failed: {response.text}")
    
    data = response.json()
    if data["count"] > 0:
        # Return the first user that matches reasonably well
        for user in data["users"]:
            if name_query.lower() in user["name"].lower():
                return user
    return None

def find_skill_by_name(token: str, skill_name: str) -> Optional[dict]:
    """
    Searches for a skill by name.
    Note: Skill search is less robust than user search. 
    If you have many skills, consider caching the skill map.
    """
    url = f"https://{CXONE_DOMAIN}/api/v2/skills"
    params = {
        "pageSize": 100, # Fetch more to find the skill
        "pageNumber": 1
    }
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/json"
    }

    response = requests.get(url, headers=headers, params=params)
    
    if response.status_code != 200:
        raise Exception(f"Skill search failed: {response.text}")
    
    data = response.json()
    for skill in data["skills"]:
        if skill["name"].lower() == skill_name.lower():
            return skill
    return None

Step 2: Construct the Bulk Update Payload

The CXone API allows updating user skills via PUT to /api/v2/users/{userId}/skills. However, doing this individually for 100 agents is inefficient and risks rate limiting.

A more efficient pattern for “bulk” updates in CXone is to iterate through the agents and construct a batch of requests, or use the PATCH method if supported for partial updates. Currently, the most reliable method for setting specific proficiency levels is to use the PUT endpoint for each user but execute them in parallel or a tight loop.

Note: CXone does not have a single “bulk update all agents” endpoint for skills. You must update per user. However, we can optimize this by preparing the data structure correctly.

The payload for /api/v2/users/{userId}/skills requires an array of skill objects. Each object needs the id of the skill and the proficiency level (0-100).

Critical Detail: When you PUT to this endpoint, you are replacing the entire list of skills for that user. If you only want to update one skill, you must first GET the user’s current skills, modify the target skill, and then PUT the entire list back.

Here is the logic to safely update proficiencies:

def update_agent_skills(token: str, user_id: str, skill_updates: list[dict]) -> dict:
    """
    Updates the skills for a specific user.
    
    Args:
        token: OAuth Bearer token
        user_id: UUID of the user
        skill_updates: List of dicts with 'skill_id' and 'proficiency'
        
    Returns:
        Response JSON
    """
    url = f"https://{CXONE_DOMAIN}/api/v2/users/{user_id}/skills"
    
    # Step 1: Get current skills to preserve existing ones
    get_headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/json"
    }
    
    get_response = requests.get(url, headers=get_headers)
    if get_response.status_code != 200:
        if get_response.status_code == 404:
            raise Exception(f"User {user_id} not found or has no skills endpoint accessible.")
        raise Exception(f"Failed to get user skills: {get_response.text}")
    
    current_skills = get_response.json().get("skills", [])
    
    # Step 2: Merge updates into current skills
    skill_map = {s["id"]: s for s in current_skills}
    
    for update in skill_updates:
        sid = update["skill_id"]
        prof = update["proficiency"]
        
        # Ensure proficiency is within valid range
        if not (0 <= prof <= 100):
            raise ValueError(f"Proficiency {prof} is out of range [0, 100]")
            
        if sid in skill_map:
            skill_map[sid]["proficiency"] = prof
        else:
            # Add new skill
            skill_map[sid] = {
                "id": sid,
                "proficiency": prof
            }
    
    # Step 3: Prepare the PUT payload
    # The API expects a list of Skill objects
    payload = list(skill_map.values())
    
    put_headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }
    
    # Step 4: Execute the update
    response = requests.put(url, json=payload, headers=put_headers)
    
    if response.status_code not in [200, 204]:
        raise Exception(f"Failed to update skills for user {user_id}: {response.status_code} - {response.text}")
    
    return response.json()

Step 3: Execute Bulk Updates with Rate Limit Handling

To process multiple agents, we will create a driver function. CXone APIs enforce rate limits (typically 10-30 requests per second depending on the endpoint and subscription tier). We must implement a small delay or use a semaphore to avoid 429 Too Many Requests errors.

import time

def bulk_update_agent_proficiencies(
    token: str, 
    agent_names: list[str], 
    skill_name: str, 
    new_proficiency: int
) -> list[dict]:
    """
    Iterates through a list of agent names, finds their IDs, finds the skill ID,
    and updates their proficiency.
    
    Args:
        token: OAuth Bearer token
        agent_names: List of agent names to update
        skill_name: Name of the skill to update
        new_proficiency: Integer 0-100
        
    Returns:
        List of results for each agent
    """
    
    # 1. Resolve Skill ID once
    skill_obj = find_skill_by_name(token, skill_name)
    if not skill_obj:
        raise Exception(f"Skill '{skill_name}' not found.")
    
    skill_id = skill_obj["id"]
    print(f"Resolved Skill ID: {skill_id} for '{skill_name}'")
    
    results = []
    total_agents = len(agent_names)
    
    for i, agent_name in enumerate(agent_names):
        try:
            print(f"[{i+1}/{total_agents}] Processing agent: {agent_name}")
            
            # 2. Resolve User ID
            user_obj = find_user_by_name(token, agent_name)
            if not user_obj:
                print(f"  WARNING: Agent '{agent_name}' not found. Skipping.")
                results.append({
                    "agent_name": agent_name,
                    "status": "error",
                    "message": "User not found"
                })
                continue
            
            user_id = user_obj["id"]
            
            # 3. Update Skills
            # We pass a list of updates. Here we only update one skill, 
            # but the function supports multiple.
            updates = [
                {
                    "skill_id": skill_id,
                    "proficiency": new_proficiency
                }
            ]
            
            update_agent_skills(token, user_id, updates)
            
            results.append({
                "agent_name": agent_name,
                "user_id": user_id,
                "status": "success"
            })
            
            # 4. Rate Limit Mitigation
            # CXone typically allows ~10-20 req/sec for user updates.
            # Adding a small sleep ensures stability.
            time.sleep(0.1) 
            
        except Exception as e:
            print(f"  ERROR processing {agent_name}: {str(e)}")
            results.append({
                "agent_name": agent_name,
                "status": "error",
                "message": str(e)
            })
            
    return results

Complete Working Example

This script combines authentication, lookup, and update logic into a single runnable file.

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

# --- Configuration ---
CXONE_DOMAIN = "api-us-02.nice-incontact.com" # Change to your region
CLIENT_ID = "YOUR_CLIENT_ID"
CLIENT_SECRET = "YOUR_CLIENT_SECRET"

# --- Authentication ---

def get_access_token() -> str:
    url = f"https://{CXONE_DOMAIN}/oauth/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "scope": "User.ReadWrite"
    }
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    response = requests.post(url, data=payload, headers=headers)
    if response.status_code != 200:
        raise Exception(f"Auth failed: {response.text}")
    return response.json()["access_token"]

# --- Helper Functions ---

def find_user_by_name(token: str, name_query: str) -> Optional[Dict]:
    url = f"https://{CXONE_DOMAIN}/api/v2/users"
    params = {"query": name_query, "pageSize": 10, "pageNumber": 1}
    headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
    response = requests.get(url, headers=headers, params=params)
    if response.status_code != 200:
        raise Exception(f"User search failed: {response.text}")
    data = response.json()
    for user in data.get("users", []):
        if name_query.lower() in user["name"].lower():
            return user
    return None

def find_skill_by_name(token: str, skill_name: str) -> Optional[Dict]:
    url = f"https://{CXONE_DOMAIN}/api/v2/skills"
    params = {"pageSize": 100, "pageNumber": 1}
    headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
    response = requests.get(url, headers=headers, params=params)
    if response.status_code != 200:
        raise Exception(f"Skill search failed: {response.text}")
    data = response.json()
    for skill in data.get("skills", []):
        if skill["name"].lower() == skill_name.lower():
            return skill
    return None

def update_user_skills(token: str, user_id: str, skill_updates: List[Dict]) -> None:
    """
    Updates user skills by fetching current list, merging updates, and PUTting back.
    """
    url = f"https://{CXONE_DOMAIN}/api/v2/users/{user_id}/skills"
    headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
    
    # Get current skills
    get_resp = requests.get(url, headers=headers)
    if get_resp.status_code == 404:
        # User has no skills yet, start with empty list
        current_skills = []
    elif get_resp.status_code != 200:
        raise Exception(f"GET skills failed: {get_resp.text}")
    else:
        current_skills = get_resp.json().get("skills", [])
    
    # Merge
    skill_map = {s["id"]: s for s in current_skills}
    for upd in skill_updates:
        sid = upd["skill_id"]
        prof = upd["proficiency"]
        if not (0 <= prof <= 100):
            raise ValueError(f"Proficiency {prof} invalid")
        
        if sid in skill_map:
            skill_map[sid]["proficiency"] = prof
        else:
            skill_map[sid] = {"id": sid, "proficiency": prof}
    
    # Put back
    payload = list(skill_map.values())
    put_headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }
    
    resp = requests.put(url, json=payload, headers=put_headers)
    if resp.status_code not in [200, 204]:
        raise Exception(f"PUT skills failed for {user_id}: {resp.status_code} {resp.text}")

# --- Main Execution ---

def main():
    # 1. Authenticate
    print("Authenticating...")
    token = get_access_token()
    
    # 2. Define Targets
    # Example: Update "English" skill to 80 for these agents
    target_agents = ["John Doe", "Jane Smith", "Alice Johnson"]
    target_skill = "English"
    target_proficiency = 80
    
    # 3. Find Skill ID
    print(f"Finding skill: {target_skill}")
    skill_obj = find_skill_by_name(token, target_skill)
    if not skill_obj:
        print("Error: Skill not found.")
        sys.exit(1)
    skill_id = skill_obj["id"]
    print(f"Found Skill ID: {skill_id}")
    
    # 4. Process Agents
    results = []
    for i, agent_name in enumerate(target_agents):
        print(f"[{i+1}/{len(target_agents)}] Updating {agent_name}...")
        try:
            user_obj = find_user_by_name(token, agent_name)
            if not user_obj:
                print(f"  Skip: User not found.")
                results.append({"agent": agent_name, "status": "error", "msg": "Not found"})
                continue
            
            user_id = user_obj["id"]
            
            # Perform update
            update_user_skills(token, user_id, [{"skill_id": skill_id, "proficiency": target_proficiency}])
            results.append({"agent": agent_name, "status": "success"})
            
            # Rate limit delay
            time.sleep(0.1)
            
        except Exception as e:
            print(f"  Error: {e}")
            results.append({"agent": agent_name, "status": "error", "msg": str(e)})
    
    # 5. Summary
    print("\n--- Results ---")
    for r in results:
        print(f"{r['agent']}: {r['status']}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

  • Cause: The OAuth token is invalid, expired, or missing.
  • Fix: Ensure your CLIENT_ID and CLIENT_SECRET are correct. Check that the token is being passed in the Authorization: Bearer <token> header. If using cached tokens, ensure the cache logic checks for expiry.

Error: 403 Forbidden

  • Cause: The OAuth client does not have the User.ReadWrite scope, or the client is not authorized to modify the specific users (e.g., admin-only restrictions).
  • Fix: Verify the scopes granted to the API Client in the CXone Admin Console under Settings > API Clients. Ensure the scope User.ReadWrite is selected.

Error: 404 Not Found (User)

  • Cause: The agent name search returned no results, or the UUID is incorrect.
  • Fix: Use the find_user_by_name helper to debug. Print the count from the search response. Ensure the agent name is spelled exactly as it appears in CXone. Note that search is case-insensitive but exact match is not required for the query parameter.

Error: 409 Conflict

  • Cause: The skill ID provided does not exist, or there is a version conflict (rare for skills).
  • Fix: Validate the skill_id against the output of find_skill_by_name. Ensure the skill is active.

Error: 429 Too Many Requests

  • Cause: You are sending requests faster than CXone allows.
  • Fix: Increase the time.sleep() duration in the loop. For large batches (1000+ agents), consider implementing exponential backoff or using a queue-based consumer pattern.

Official References