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

What You Will Build

• A Python script that retrieves a list of agents, identifies those requiring skill updates, and applies bulk proficiency changes using the NICE CXone Admin API.
• This tutorial uses the NICE CXone REST API endpoints for User Management and Skill Proficiencies.
• The implementation is written in Python using the requests library and httpx for async operations where appropriate.

Prerequisites

• OAuth Client Type: Confidential Client (Client Credentials Flow).
• Required Scopes:
  • admin:users:read (to retrieve agent list)
  • admin:users:write (to update user attributes)
  • admin:skills:read (to validate skill IDs)
  • admin:skillproficiencies:write (to update proficiency levels)
• SDK/API Version: NICE CXone REST API (v2).
• Language/Runtime: Python 3.8+.
• External Dependencies:
  • requests>=2.28.0
  • httpx>=0.24.0
  • pydantic>=2.0.0 (for data validation)

Authentication Setup

NICE CXone uses OAuth 2.0 for API authentication. For server-to-server integrations like bulk updates, the Client Credentials flow is standard. You must obtain an access token before making any API calls.

The token endpoint is typically https://platform.cxone.com/oauth2/token.

import requests
import os
import time
from typing import Dict, Optional

class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str, realm: str = "global"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.realm = realm
        self.token_url = f"https://platform.cxone.com/oauth2/token"
        self.access_token: Optional[str] = None
        self.token_expiry: Optional[float] = None

    def get_token(self) -> str:
        """
        Retrieves an OAuth access token using Client Credentials flow.
        Implements basic caching to avoid unnecessary requests.
        """
        if self.access_token and self.token_expiry 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,
            "realm": self.realm
        }

        try:
            response = requests.post(self.token_url, headers=headers, data=data)
            response.raise_for_status()
            token_data = response.json()
            
            self.access_token = token_data["access_token"]
            # Set expiry slightly before actual expiration to prevent edge-case failures
            self.token_expiry = time.time() + (token_data["expires_in"] - 60)
            
            return self.access_token
        
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise Exception("Authentication failed: Invalid client credentials or realm.") from e
            raise Exception(f"OAuth token request failed with status {response.status_code}: {e}") from e
        except Exception as e:
            raise Exception(f"Failed to retrieve token: {e}") from e

    def get_headers(self) -> Dict[str, str]:
        """Returns headers required for API calls."""
        token = self.get_token()
        return {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

Implementation

Step 1: Retrieve Target Agents and Skills

Before updating proficiencies, you must identify the unique identifiers for both the agents and the skills. NICE CXone does not allow bulk updates by name; you must use userId and skillId.

We will use the GET /api/v2/users endpoint to fetch users. Note that this endpoint supports pagination. We will implement a generator to handle large datasets efficiently.

import requests
from typing import Generator, Dict, Any, List

class CXoneClient:
    def __init__(self, auth: CXoneAuth, base_url: str = "https://platform.cxone.com"):
        self.auth = auth
        self.base_url = base_url.rstrip("/")

    def _make_request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
        """Helper method to make authenticated requests."""
        headers = self.auth.get_headers()
        url = f"{self.base_url}{endpoint}"
        
        # Retry logic for 429 Too Many Requests
        retries = 3
        for attempt in range(retries):
            response = requests.request(method, url, headers=headers, **kwargs)
            
            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 5))
                time.sleep(retry_after)
                continue
            
            return response
        
        raise Exception(f"Max retries exceeded for {url}")

    def get_users(self, page_size: int = 100, search_query: Optional[str] = None) -> Generator[Dict[str, Any], None, None]:
        """
        Generator that yields user objects from the CXone API.
        Handles pagination automatically.
        
        :param page_size: Number of users per page (max 1000)
        :param search_query: Optional query to filter users (e.g., "type:agent")
        """
        page = 1
        has_more = True
        
        while has_more:
            params = {
                "pageSize": page_size,
                "page": page,
                "expand": "skills,proficiencies" # Expand related data if needed later
            }
            
            if search_query:
                params["q"] = search_query

            try:
                response = self._make_request("GET", "/api/v2/users", params=params)
                response.raise_for_status()
                data = response.json()
                
                if "entities" not in data or not data["entities"]:
                    break
                
                for user in data["entities"]:
                    yield user
                
                # Check if there are more pages
                has_more = data.get("hasNextPage", False)
                page += 1

            except requests.exceptions.HTTPError as e:
                raise Exception(f"Failed to fetch users: {e}") from e

    def get_skills(self, page_size: int = 100) -> Dict[str, str]:
        """
        Retrieves a mapping of skill names to skill IDs.
        
        :return: Dictionary mapping skill name to skill ID
        """
        skills_map = {}
        page = 1
        has_more = True
        
        while has_more:
            params = {"pageSize": page_size, "page": page}
            try:
                response = self._make_request("GET", "/api/v2/skills", params=params)
                response.raise_for_status()
                data = response.json()
                
                for skill in data.get("entities", []):
                    skills_map[skill["name"]] = skill["id"]
                
                has_more = data.get("hasNextPage", False)
                page += 1
                
            except requests.exceptions.HTTPError as e:
                raise Exception(f"Failed to fetch skills: {e}") from e
                
        return skills_map

Step 2: Construct Proficiency Update Payloads

The NICE CXone API for updating user proficiencies is not a single “bulk update all” endpoint. Instead, it uses a specific endpoint to update the proficiency of a user for a specific skill.

Endpoint: PUT /api/v2/users/{userId}/proficiencies/{skillId}

To perform a “bulk” update, we must construct a list of requests. We will create a helper function that prepares these payloads based on a configuration list.

OAuth Scope Required: admin:skillproficiencies:write

from dataclasses import dataclass
from typing import List

@dataclass
class ProficiencyUpdate:
    user_id: str
    skill_id: str
    proficiency_level: str  # e.g., "expert", "intermediate", "novice", "untrained"

def prepare_bulk_updates(
    users: List[Dict[str, Any]], 
    skills_map: Dict[str, str], 
    target_skill_name: str, 
    new_level: str
) -> List[ProficiencyUpdate]:
    """
    Prepares a list of ProficiencyUpdate objects for agents who need updating.
    
    :param users: List of user dictionaries from the API
    :param skills_map: Dictionary mapping skill names to IDs
    :param target_skill_name: The name of the skill to update
    :param new_level: The new proficiency level
    :return: List of ProficiencyUpdate objects
    """
    if target_skill_name not in skills_map:
        raise ValueError(f"Skill '{target_skill_name}' not found in available skills.")
    
    target_skill_id = skills_map[target_skill_name]
    updates = []
    
    for user in users:
        # Only process agents (not supervisors or admins if not needed)
        if user.get("type") != "agent":
            continue
            
        # Check current proficiency to avoid unnecessary API calls
        # Note: 'proficiencies' might not be expanded in GET /users depending on performance.
        # For robustness, we might need to GET specific user proficiencies, but for bulk
        # updates, we often assume the change is required or check a subset.
        
        # Here we assume we want to update ALL agents for this skill to the new level.
        # If you need to check current state, you would loop and call GET /users/{id}/proficiencies/{skillId}
        
        updates.append(ProficiencyUpdate(
            user_id=user["id"],
            skill_id=target_skill_id,
            proficiency_level=new_level
        ))
        
    return updates

Step 3: Execute Bulk Updates with Throttling

Executing hundreds of PUT requests sequentially is slow. Executing them all in parallel will trigger 429 Too Many Requests errors. We will use httpx for asynchronous concurrent requests with a semaphore to control concurrency.

Important: NICE CXone rate limits are typically around 100 requests per second per client, but this can vary. A safe concurrency level is 10-20 simultaneous requests.

import httpx
from typing import Tuple, List
import asyncio

class CXoneBulkUpdater:
    def __init__(self, auth: CXoneAuth, base_url: str = "https://platform.cxone.com"):
        self.auth = auth
        self.base_url = base_url.rstrip("/")
        self.client = httpx.AsyncClient(
            headers=auth.get_headers(),
            timeout=30.0
        )

    async def update_single_proficiency(self, update: ProficiencyUpdate) -> Tuple[str, bool, str]:
        """
        Updates a single user's proficiency for a specific skill.
        
        :param update: ProficiencyUpdate object
        :return: Tuple of (user_id, success_bool, message)
        """
        # Refresh token if needed (httpx client doesn't auto-refresh, so we check)
        # In a real app, you might inject a header middleware for token refresh
        headers = self.auth.get_headers()
        self.client.headers.update(headers)

        endpoint = f"/api/v2/users/{update.user_id}/proficiencies/{update.skill_id}"
        url = f"{self.base_url}{endpoint}"
        
        payload = {
            "proficiencyLevel": update.proficiency_level
        }

        try:
            response = await self.client.put(url, json=payload)
            
            if response.status_code == 204:
                return (update.user_id, True, "Updated successfully")
            elif response.status_code == 404:
                return (update.user_id, False, "User or Skill not found")
            elif response.status_code == 409:
                return (update.user_id, False, "Conflict: User may not be allowed to have this proficiency")
            else:
                return (update.user_id, False, f"HTTP {response.status_code}: {response.text}")
                
        except httpx.HTTPStatusError as e:
            return (update.user_id, False, f"HTTP Error: {e}")
        except Exception as e:
            return (update.user_id, False, f"Exception: {str(e)}")

    async def run_bulk_update(self, updates: List[ProficiencyUpdate], concurrency: int = 10) -> List[Tuple[str, bool, str]]:
        """
        Runs the bulk update with controlled concurrency.
        
        :param updates: List of ProficiencyUpdate objects
        :param concurrency: Maximum number of simultaneous requests
        :return: List of results
        """
        semaphore = asyncio.Semaphore(concurrency)
        results = []
        
        async def bounded_update(update: ProficiencyUpdate):
            async with semaphore:
                result = await self.update_single_proficiency(update)
                results.append(result)
                return result

        # Create tasks for all updates
        tasks = [bounded_update(u) for u in updates]
        
        # Run all tasks concurrently
        await asyncio.gather(*tasks)
        
        return results

    async def close(self):
        await self.client.aclose()

Complete Working Example

This script combines all previous steps into a runnable module. It assumes you have set environment variables for your credentials.

import os
import asyncio
import json
from typing import List, Tuple

# Import classes defined above
# from cxone_auth import CXoneAuth
# from cxone_client import CXoneClient
# from cxone_bulk_updater import CXoneBulkUpdater, ProficiencyUpdate
# from prepare_updates import prepare_bulk_updates

def main():
    # 1. Setup Authentication
    client_id = os.getenv("CXONE_CLIENT_ID")
    client_secret = os.getenv("CXONE_CLIENT_SECRET")
    realm = os.getenv("CXONE_REALM", "global")

    if not client_id or not client_secret:
        raise EnvironmentError("CXONE_CLIENT_ID and CXONE_CLIENT_SECRET environment variables are required.")

    auth = CXoneAuth(client_id, client_secret, realm)
    
    # 2. Initialize Client
    client = CXoneClient(auth)

    # 3. Fetch Skills Map
    print("Fetching skills...")
    skills_map = client.get_skills()
    target_skill_name = "English" # Example skill
    new_proficiency = "expert"   # Example level

    if target_skill_name not in skills_map:
        print(f"Skill '{target_skill_name}' not found. Available skills: {list(skills_map.keys())[:10]}...")
        return

    # 4. Fetch Users (Agents only)
    print("Fetching agents...")
    users = list(client.get_users(search_query="type:agent"))
    print(f"Found {len(users)} agents.")

    if not users:
        print("No agents found.")
        return

    # 5. Prepare Updates
    print(f"Preparing updates for skill '{target_skill_name}' to level '{new_proficiency}'...")
    updates = prepare_bulk_updates(users, skills_map, target_skill_name, new_proficiency)
    print(f"Prepared {len(updates)} updates.")

    if not updates:
        print("No updates to perform.")
        return

    # 6. Execute Bulk Update
    async def run_async_updates():
        updater = CXoneBulkUpdater(auth)
        try:
            results = await updater.run_bulk_update(updates, concurrency=10)
            return results
        finally:
            await updater.close()

    print("Starting bulk update process...")
    results = asyncio.run(run_async_updates())

    # 7. Report Results
    successes = sum(1 for _, success, _ in results if success)
    failures = len(results) - successes
    
    print(f"\n--- Update Complete ---")
    print(f"Total Attempts: {len(results)}")
    print(f"Successes: {successes}")
    print(f"Failures: {failures}")

    if failures > 0:
        print("\nFailed Updates:")
        for user_id, success, message in results:
            if not success:
                print(f"  User {user_id}: {message}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token has expired, or the client credentials are invalid.
• Fix: Ensure your CXoneAuth class is refreshing the token before each batch of requests. Check that the client_id and client_secret match the API credentials generated in the CXone Admin Portal.
• Code Check: Verify the Authorization: Bearer <token> header is present in every request.

Error: 403 Forbidden

• Cause: The OAuth token does not have the required scope (admin:skillproficiencies:write).
• Fix: Regenerate your API credentials in the CXone Admin Portal and ensure you select the correct scopes during creation. Specifically, check that admin:skillproficiencies:write is included.
• Debugging: Print the scope claim from the decoded JWT token to verify permissions.

Error: 429 Too Many Requests

• Cause: You are exceeding the rate limit of the CXone API.
• Fix: Implement exponential backoff or reduce the concurrency level in CXoneBulkUpdater. The CXoneClient._make_request method includes basic retry logic for 429s, but the async updater should also respect Retry-After headers.
• Code Fix: In CXoneBulkUpdater.update_single_proficiency, catch httpx.HTTPStatusError with status 429 and sleep before retrying.

Error: 404 Not Found

• Cause: The userId or skillId is invalid.
• Fix: Verify that the users retrieved in Step 1 are valid agents. Ensure the skill name used in prepare_bulk_updates matches exactly (case-sensitive) with a skill in the system.
• Debugging: Log the skillId being used and verify it exists in the /api/v2/skills response.

Error: 409 Conflict

• Cause: The user is not allowed to have this proficiency level, or the skill is not available in the user’s skill group.
• Fix: Check the user’s skill groups and ensure the target skill is assigned to them. Some proficiency levels may be restricted by role.

Official References

• NICE CXone API Documentation: Users
• NICE CXone API Documentation: Skill Proficiencies
• NICE CXone OAuth 2.0 Guide
• Python Requests Library
• HTTPX Async Client