Bulk-Update Agent Skill Proficiencies via NICE CXone Admin API

Bulk-Update Agent Skill Proficiencies via NICE CXone Admin API

What You Will Build

  • A script that retrieves a list of agents and updates their skill proficiencies in bulk using the NICE CXone Admin API.
  • This uses the NICE CXone REST API endpoints for agents and skills.
  • The programming language covered is Python, utilizing the requests library for HTTP interactions.

Prerequisites

  • OAuth Client Type: Service Account (Client Credentials Grant).
  • Required Scopes:
    • agent:read (to retrieve agent IDs)
    • agent:write (to update agent details including skills)
    • skill:read (to retrieve skill IDs if not known)
  • SDK/API Version: NICE CXone Admin API (REST). No specific SDK is required as we are using raw HTTP requests for clarity and flexibility, but the logic applies to any HTTP client.
  • Language/Runtime Requirements: Python 3.8+
  • External Dependencies:
    • requests (for HTTP calls)
    • python-dotenv (for secure credential management)

Install dependencies:

pip install requests python-dotenv

Authentication Setup

NICE CXone uses OAuth 2.0 Client Credentials Grant for service-to-service authentication. You must obtain an access token before making any API calls. The token is valid for a limited time (typically 1 hour) and must be refreshed.

Create a .env file in your project root with the following variables:

CXONE_TENANT_DOMAIN=your-tenant-domain.com
CXONE_CLIENT_ID=your-client-id
CXONE_CLIENT_SECRET=your-client-secret

Create a helper function to handle authentication and token caching:

import os
import requests
from dotenv import load_dotenv
import time

load_dotenv()

CXONE_BASE_URL = f"https://{os.getenv('CXONE_TENANT_DOMAIN')}"
CLIENT_ID = os.getenv('CXONE_CLIENT_ID')
CLIENT_SECRET = os.getenv('CXONE_CLIENT_SECRET')

# Simple in-memory token cache with expiration
_token_cache = {
    'token': None,
    'expires_at': 0
}

def get_access_token():
    """
    Retrieves an OAuth2 access token from CXone.
    Uses simple in-memory caching to avoid unnecessary requests.
    """
    current_time = time.time()
    
    # Return cached token if still valid (subtract 60s for safety buffer)
    if _token_cache['token'] and current_time < _token_cache['expires_at'] - 60:
        return _token_cache['token']

    token_url = f"{CXONE_BASE_URL}/oauth/token"
    payload = {
        'grant_type': 'client_credentials',
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET,
        'scope': 'agent:read agent:write skill:read'
    }

    response = requests.post(token_url, data=payload)
    
    if response.status_code != 200:
        raise Exception(f"Failed to obtain access token: {response.text}")

    data = response.json()
    access_token = data['access_token']
    expires_in = data['expires_in']
    
    # Update cache
    _token_cache['token'] = access_token
    _token_cache['expires_at'] = current_time + expires_in
    
    return access_token

Implementation

Step 1: Retrieve Agent IDs and Current Skill Proficiencies

Before updating, you need to identify which agents to update and their current skill proficiencies. NICE CXone does not have a single “bulk update skills” endpoint. Instead, you must update the agent resource (PUT /api/v2/users/{userId}) with the new skill proficiencies included in the request body.

First, fetch the list of agents. We will filter for agents with a specific status or skill if needed, but for this example, we will fetch all active agents.

import json

def get_agents(page_size=100):
    """
    Retrieves a paginated list of agents.
    Returns a list of agent dictionaries.
    """
    access_token = get_access_token()
    headers = {
        'Authorization': f'Bearer {access_token}',
        'Content-Type': 'application/json'
    }
    
    agents = []
    page = 1
    
    while True:
        url = f"{CXONE_BASE_URL}/api/v2/users"
        params = {
            'pageSize': page_size,
            'pageNumber': page,
            'expansion': 'skills'  # Expand skills to see current proficiencies
        }
        
        response = requests.get(url, headers=headers, params=params)
        
        if response.status_code != 200:
            raise Exception(f"Failed to retrieve agents: {response.text}")
        
        data = response.json()
        entities = data.get('entities', [])
        
        if not entities:
            break
            
        agents.extend(entities)
        
        # Check if more pages exist
        if data.get('hasMore', False):
            page += 1
        else:
            break
            
    return agents

Step 2: Define Skill Updates and Prepare Payloads

You need to define which skills to update and what the new proficiency level should be. Proficiency is typically represented as an integer (e.g., 0-100 or 1-5 depending on your tenant configuration). For this tutorial, we assume a 0-100 scale.

We will create a mapping of agent IDs to the skills they should have. This mapping could come from a CSV, a database, or a hardcoded list.

# Example: Define the skills to update
# In a real scenario, this might be loaded from a CSV or database
SKILL_UPDATES = {
    # Format: agent_id: { skill_id: proficiency_value }
    # Note: You must replace these IDs with actual IDs from your tenant
    "agent-id-1": {
        "skill-id-1": 85,
        "skill-id-2": 90
    },
    "agent-id-2": {
        "skill-id-1": 70
    }
}

def get_skill_ids():
    """
    Retrieves a list of skills to map IDs to names if needed.
    """
    access_token = get_access_token()
    headers = {
        'Authorization': f'Bearer {access_token}',
        'Content-Type': 'application/json'
    }
    
    url = f"{CXONE_BASE_URL}/api/v2/skills"
    params = {
        'pageSize': 100
    }
    
    response = requests.get(url, headers=headers, params=params)
    
    if response.status_code != 200:
        raise Exception(f"Failed to retrieve skills: {response.text}")
    
    data = response.json()
    return data.get('entities', [])

Step 3: Bulk Update Agent Skill Proficiencies

Now, iterate through the agents and update their skill proficiencies. Since there is no single bulk endpoint, we will make individual PUT requests for each agent. To avoid rate limiting, we will implement a small delay between requests and handle errors gracefully.

import time
from requests.exceptions import RequestException

def update_agent_skills(agent_id, skill_updates, current_agent_data):
    """
    Updates the skill proficiencies for a single agent.
    
    Args:
        agent_id (str): The ID of the agent to update.
        skill_updates (dict): A dictionary of skill_id: proficiency_value.
        current_agent_data (dict): The current agent data retrieved from the API.
    """
    access_token = get_access_token()
    headers = {
        'Authorization': f'Bearer {access_token}',
        'Content-Type': 'application/json'
    }
    
    url = f"{CXONE_BASE_URL}/api/v2/users/{agent_id}"
    
    # Prepare the request body
    # We must include all required fields for the agent resource
    # Only the skill proficiencies will be changed; other fields remain as-is
    
    # Deep copy to avoid modifying the original data
    update_payload = json.loads(json.dumps(current_agent_data))
    
    # Ensure the 'skills' key exists
    if 'skills' not in update_payload:
        update_payload['skills'] = []
    
    # Update or add skills
    for skill_id, proficiency in skill_updates.items():
        # Check if skill already exists in the agent's skills list
        found = False
        for skill in update_payload['skills']:
            if skill['id'] == skill_id:
                skill['proficiency'] = proficiency
                found = True
                break
        
        if not found:
            # Add new skill proficiency
            update_payload['skills'].append({
                'id': skill_id,
                'proficiency': proficiency
            })
    
    # Remove fields that are not allowed in the update payload
    # The CXone API is sensitive to extra fields
    allowed_fields = [
        'id', 'name', 'email', 'skills', 'status', 'routingStatus', 
        'type', 'divisionId', 'version', 'dateCreated', 'dateUpdated'
    ]
    
    filtered_payload = {k: v for k, v in update_payload.items() if k in allowed_fields}
    
    # Make the PUT request
    try:
        response = requests.put(url, headers=headers, json=filtered_payload)
        
        if response.status_code == 200:
            print(f"Successfully updated skills for agent {agent_id}")
        elif response.status_code == 429:
            print(f"Rate limited for agent {agent_id}. Retrying in 1 second.")
            time.sleep(1)
            # Recursive retry with backoff could be implemented here
        else:
            print(f"Failed to update agent {agent_id}: {response.status_code} - {response.text}")
            
    except RequestException as e:
        print(f"Network error updating agent {agent_id}: {e}")

def bulk_update_agents():
    """
    Main function to orchestrate the bulk update.
    """
    print("Retrieving agents...")
    agents = get_agents()
    
    print(f"Retrieved {len(agents)} agents.")
    
    # Create a map of agent_id to agent_data for quick lookup
    agent_map = {agent['id']: agent for agent in agents}
    
    print("Starting bulk update...")
    for agent_id, skill_updates in SKILL_UPDATES.items():
        if agent_id in agent_map:
            current_agent_data = agent_map[agent_id]
            update_agent_skills(agent_id, skill_updates, current_agent_data)
            # Add a small delay to avoid rate limiting
            time.sleep(0.5)
        else:
            print(f"Agent ID {agent_id} not found in retrieved agents.")

if __name__ == "__main__":
    bulk_update_agents()

Complete Working Example

Below is the complete, copy-pasteable script. Ensure you have set up the .env file with your credentials and updated the SKILL_UPDATES dictionary with actual agent and skill IDs from your tenant.

import os
import json
import time
import requests
from dotenv import load_dotenv
from requests.exceptions import RequestException

load_dotenv()

# Configuration
CXONE_BASE_URL = f"https://{os.getenv('CXONE_TENANT_DOMAIN')}"
CLIENT_ID = os.getenv('CXONE_CLIENT_ID')
CLIENT_SECRET = os.getenv('CXONE_CLIENT_SECRET')

# Token Cache
_token_cache = {
    'token': None,
    'expires_at': 0
}

# Skill Updates Configuration
# Replace with actual Agent IDs and Skill IDs from your CXone tenant
SKILL_UPDATES = {
    "agent-id-1": {
        "skill-id-1": 85,
        "skill-id-2": 90
    },
    "agent-id-2": {
        "skill-id-1": 70
    }
}

def get_access_token():
    """
    Retrieves an OAuth2 access token from CXone.
    Uses simple in-memory caching to avoid unnecessary requests.
    """
    current_time = time.time()
    
    if _token_cache['token'] and current_time < _token_cache['expires_at'] - 60:
        return _token_cache['token']

    token_url = f"{CXONE_BASE_URL}/oauth/token"
    payload = {
        'grant_type': 'client_credentials',
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET,
        'scope': 'agent:read agent:write skill:read'
    }

    response = requests.post(token_url, data=payload)
    
    if response.status_code != 200:
        raise Exception(f"Failed to obtain access token: {response.text}")

    data = response.json()
    access_token = data['access_token']
    expires_in = data['expires_in']
    
    _token_cache['token'] = access_token
    _token_cache['expires_at'] = current_time + expires_in
    
    return access_token

def get_agents(page_size=100):
    """
    Retrieves a paginated list of agents.
    Returns a list of agent dictionaries.
    """
    access_token = get_access_token()
    headers = {
        'Authorization': f'Bearer {access_token}',
        'Content-Type': 'application/json'
    }
    
    agents = []
    page = 1
    
    while True:
        url = f"{CXONE_BASE_URL}/api/v2/users"
        params = {
            'pageSize': page_size,
            'pageNumber': page,
            'expansion': 'skills'
        }
        
        response = requests.get(url, headers=headers, params=params)
        
        if response.status_code != 200:
            raise Exception(f"Failed to retrieve agents: {response.text}")
        
        data = response.json()
        entities = data.get('entities', [])
        
        if not entities:
            break
            
        agents.extend(entities)
        
        if data.get('hasMore', False):
            page += 1
        else:
            break
            
    return agents

def update_agent_skills(agent_id, skill_updates, current_agent_data):
    """
    Updates the skill proficiencies for a single agent.
    """
    access_token = get_access_token()
    headers = {
        'Authorization': f'Bearer {access_token}',
        'Content-Type': 'application/json'
    }
    
    url = f"{CXONE_BASE_URL}/api/v2/users/{agent_id}"
    
    update_payload = json.loads(json.dumps(current_agent_data))
    
    if 'skills' not in update_payload:
        update_payload['skills'] = []
    
    for skill_id, proficiency in skill_updates.items():
        found = False
        for skill in update_payload['skills']:
            if skill['id'] == skill_id:
                skill['proficiency'] = proficiency
                found = True
                break
        
        if not found:
            update_payload['skills'].append({
                'id': skill_id,
                'proficiency': proficiency
            })
    
    allowed_fields = [
        'id', 'name', 'email', 'skills', 'status', 'routingStatus', 
        'type', 'divisionId', 'version', 'dateCreated', 'dateUpdated'
    ]
    
    filtered_payload = {k: v for k, v in update_payload.items() if k in allowed_fields}
    
    try:
        response = requests.put(url, headers=headers, json=filtered_payload)
        
        if response.status_code == 200:
            print(f"Successfully updated skills for agent {agent_id}")
        elif response.status_code == 429:
            print(f"Rate limited for agent {agent_id}. Retrying in 1 second.")
            time.sleep(1)
            update_agent_skills(agent_id, skill_updates, current_agent_data) # Retry
        else:
            print(f"Failed to update agent {agent_id}: {response.status_code} - {response.text}")
            
    except RequestException as e:
        print(f"Network error updating agent {agent_id}: {e}")

def bulk_update_agents():
    """
    Main function to orchestrate the bulk update.
    """
    print("Retrieving agents...")
    agents = get_agents()
    
    print(f"Retrieved {len(agents)} agents.")
    
    agent_map = {agent['id']: agent for agent in agents}
    
    print("Starting bulk update...")
    for agent_id, skill_updates in SKILL_UPDATES.items():
        if agent_id in agent_map:
            current_agent_data = agent_map[agent_id]
            update_agent_skills(agent_id, skill_updates, current_agent_data)
            time.sleep(0.5)
        else:
            print(f"Agent ID {agent_id} not found in retrieved agents.")

if __name__ == "__main__":
    bulk_update_agents()

Common Errors & Debugging

Error: 401 Unauthorized

Cause: The access token is invalid, expired, or missing.

Fix:

  • Ensure your .env file has the correct CXONE_CLIENT_ID and CXONE_CLIENT_SECRET.
  • Check that the CXONE_TENANT_DOMAIN is correct (e.g., nicecxone.com or your custom domain).
  • Verify that the token cache logic is working. If the token expires during a long bulk update, the script will fail. The provided code refreshes the token if it is older than 60 seconds.

Error: 403 Forbidden

Cause: The OAuth client does not have the required scopes, or the service account does not have permission to update agents.

Fix:

  • Ensure the OAuth client has the agent:write scope.
  • Verify that the service account associated with the OAuth client has the necessary role permissions in the CXone admin console. The role must allow “Edit Agent” permissions.

Error: 429 Too Many Requests

Cause: You are making too many requests in a short period. CXone enforces rate limits.

Fix:

  • The provided code includes a time.sleep(0.5) between updates. If you still encounter 429 errors, increase the delay.
  • Implement exponential backoff for retries. The provided code has a basic retry for 429 responses.

Error: 400 Bad Request

Cause: The request body contains invalid data or missing required fields.

Fix:

  • Ensure the filtered_payload contains all required fields for the PUT /api/v2/users/{userId} endpoint.
  • Verify that skill IDs are valid and exist in your tenant.
  • Check that proficiency values are within the allowed range (typically 0-100).

Official References