Bulk-Update Agent Skill Proficiencies via NICE CXone REST API

What You Will Build

• A script that retrieves a list of agents and updates their skill proficiency levels in bulk using the NICE CXone Administration API.
• This tutorial utilizes the NICE CXone admin REST API endpoints for users and skills.
• The primary programming language covered is Python 3.9+ using the requests library.

Prerequisites

• OAuth Client Type: Service Account (Client Credentials Grant) is recommended for backend automation scripts.
• Required Scopes:
  • admin:user:read (to list agents)
  • admin:user:write (to update agent profiles)
  • admin:skill:read (to fetch skill IDs if not known)
  • admin:site:read (often required for context in multi-site deployments)
• SDK Version: No official SDK is required for this specific task; raw REST calls via requests provide the most control over bulk operations and error handling.
• Language/Runtime: Python 3.9 or higher.
• External Dependencies:
  • requests: For HTTP communication.
  • python-dotenv: For secure environment variable management.
  • tenacity: For robust retry logic on rate limits.

Install dependencies:

pip install requests python-dotenv tenacity

Authentication Setup

NICE CXone uses OAuth 2.0 for authentication. For server-to-server integration, the Client Credentials flow is the standard. You must obtain an access token before making any API calls. The token expires after a short duration (typically 1 hour), so your application should handle token refresh or re-authentication.

The following Python class handles authentication and provides a method to get a valid token.

import os
import requests
from dotenv import load_dotenv
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

load_dotenv()

class CXoneAuth:
    def __init__(self, domain: str, client_id: str, client_secret: str):
        self.domain = domain
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = f"https://{domain}/api/auth/oauth2/token"
        self.access_token = None
        self.token_expiry = 0

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type(requests.exceptions.RequestException)
    )
    def get_token(self) -> str:
        """
        Retrieves an OAuth2 access token using client credentials.
        Caches the token if it is not expired.
        """
        import time
        
        # Simple cache check. In production, use a proper cache with expiration.
        if self.access_token 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
        }

        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 expiry to avoid edge cases
            self.token_expiry = time.time() + token_data.get("expires_in", 3600) - 60
            
            return self.access_token

        except requests.exceptions.HTTPError as e:
            print(f"Authentication failed: {e.response.status_code} - {e.response.text}")
            raise

# Usage initialization
auth = CXoneAuth(
    domain=os.getenv("CXONE_DOMAIN"),
    client_id=os.getenv("CXONE_CLIENT_ID"),
    client_secret=os.getenv("CXONE_CLIENT_SECRET")
)

Implementation

Step 1: Retrieve Target Skills and Agents

Before updating proficiencies, you must identify the skillId for the skills you intend to update and the userId for the agents. CXone IDs are UUIDs. Hardcoding IDs is fragile; querying them dynamically is safer.

First, we fetch the skills. We assume you know the skill name (e.g., “Spanish - Fluent”).

import requests

def get_skill_id_by_name(auth: CXoneAuth, skill_name: str) -> str:
    """
    Fetches the UUID of a skill by its name.
    Endpoint: GET /api/v2/admin/skills
    Scope: admin:skill:read
    """
    url = f"https://{auth.domain}/api/v2/admin/skills"
    token = auth.get_token()
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    try:
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        skills = response.json().get("skills", [])
        
        for skill in skills:
            if skill.get("name").lower() == skill_name.lower():
                return skill.get("skillId")
                
        return None

    except requests.exceptions.HTTPError as e:
        print(f"Error fetching skills: {e.response.status_code}")
        raise

# Example usage
spanish_skill_id = get_skill_id_by_name(auth, "Spanish - Fluent")
if not spanish_skill_id:
    raise ValueError("Skill 'Spanish - Fluent' not found.")

Next, we retrieve the list of agents. In CXone, agents are represented as users with a specific userType (usually agent). We will filter for active agents.

def get_active_agents(auth: CXoneAuth, limit: int = 100) -> list:
    """
    Fetches a list of active agents.
    Endpoint: GET /api/v2/admin/users
    Scope: admin:user:read
    """
    url = f"https://{auth.domain}/api/v2/admin/users"
    token = auth.get_token()
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    params = {
        "userType": "agent",
        "limit": limit
    }

    try:
        response = requests.get(url, headers=headers, params=params)
        response.raise_for_status()
        users = response.json().get("users", [])
        return users
    except requests.exceptions.HTTPError as e:
        print(f"Error fetching agents: {e.response.status_code}")
        raise

agents = get_active_agents(auth, limit=50)
print(f"Retrieved {len(agents)} agents.")

Step 2: Construct the Update Payload

The CXone API does not have a single “bulk update proficiencies” endpoint. You must update each user individually via PUT /api/v2/admin/users/{userId}. However, to minimize API calls and avoid overwriting other user data, you must:

1. Fetch the current user profile (to get the current version number and existing skills).
2. Modify the skills array in the profile.
3. Send the updated profile back with the correct version number to prevent race conditions.

The version field is critical. CXone uses optimistic locking. If you submit a PUT request with an old version, the API will reject it with a 409 Conflict.

def update_agent_proficiency(auth: CXoneAuth, user_id: str, skill_id: str, proficiency_level: int, skill_name: str) -> bool:
    """
    Updates a single agent's proficiency for a specific skill.
    Endpoint: PUT /api/v2/admin/users/{userId}
    Scope: admin:user:write
    
    Args:
        user_id: The UUID of the agent.
        skill_id: The UUID of the skill.
        proficiency_level: 1-5 (1=Low, 5=High).
        skill_name: Name of the skill for logging purposes.
    
    Returns:
        True if successful, False otherwise.
    """
    token = auth.get_token()
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    # Step 2a: Get current user profile to preserve version and other data
    get_url = f"https://{auth.domain}/api/v2/admin/users/{user_id}"
    
    try:
        get_response = requests.get(get_url, headers=headers)
        get_response.raise_for_status()
        user_data = get_response.json()
        
    except requests.exceptions.HTTPError as e:
        print(f"Failed to fetch user {user_id}: {e.response.status_code}")
        return False

    # Step 2b: Modify the skills array
    # The user object contains a 'skills' array. Each item has 'skillId' and 'proficiency'
    skills = user_data.get("skills", [])
    skill_updated = False
    
    for skill in skills:
        if skill.get("skillId") == skill_id:
            skill["proficiency"] = proficiency_level
            skill_updated = True
            break
    
    if not skill_updated:
        # If the skill is not in the list, add it
        new_skill = {
            "skillId": skill_id,
            "proficiency": proficiency_level
        }
        skills.append(new_skill)
        user_data["skills"] = skills

    # Step 2c: Prepare the PUT payload
    # Important: We must keep the 'version' field from the GET response
    payload = {
        "version": user_data.get("version"),
        "userId": user_data.get("userId"),
        "userName": user_data.get("userName"),
        "firstName": user_data.get("firstName"),
        "lastName": user_data.get("lastName"),
        "email": user_data.get("email"),
        "userType": user_data.get("userType"),
        "skills": user_data.get("skills"),
        # Include other necessary fields if your tenant requires them
        # e.g., "groups", "roles", etc. Omitting them might default to empty depending on API behavior.
        # It is safer to only send fields you are changing + version + userId.
        # However, CXone PUT /users often requires the full object or at least the core identity fields.
    }

    # Step 2d: Send the update
    try:
        put_response = requests.put(get_url, headers=headers, json=payload)
        
        if put_response.status_code == 200:
            print(f"Successfully updated {user_data.get('userName')} for {skill_name}")
            return True
        elif put_response.status_code == 409:
            print(f"Conflict updating {user_id}. Version mismatch. User data may have changed.")
            return False
        else:
            print(f"Failed to update {user_id}: {put_response.status_code} - {put_response.text}")
            return False

    except requests.exceptions.RequestException as e:
        print(f"Network error updating {user_id}: {e}")
        return False

Step 3: Process Results and Handle Rate Limits

When updating many agents, you will hit rate limits. The CXone API returns 429 Too Many Requests with a Retry-After header. We will use the tenacity library to handle retries automatically.

We will also implement a small delay between requests to be polite to the API, even if we are not rate-limited.

import time
from tenacity import retry, stop_after_attempt, wait_random_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(5),
    wait=wait_random_exponential(multiplier=1, max=60),
    retry=retry_if_exception_type(requests.exceptions.HTTPError),
    reraise=True
)
def safe_update_agent(auth: CXoneAuth, user_id: str, skill_id: str, proficiency: int, skill_name: str) -> bool:
    """
    Wrapper for update_agent_proficiency that handles retries on 429 errors.
    """
    # Check for 429 manually before calling the main function to allow custom retry logic if needed
    # However, tenacity handles the retry loop. We need to ensure the inner function raises the exception.
    
    # Re-implementing the core logic inside the retry decorator scope for simplicity
    token = auth.get_token()
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    get_url = f"https://{auth.domain}/api/v2/admin/users/{user_id}"
    
    # Fetch current data
    get_response = requests.get(get_url, headers=headers)
    if get_response.status_code == 429:
        retry_after = int(get_response.headers.get('Retry-After', 5))
        raise requests.exceptions.HTTPError(f"Rate limited. Retry after {retry_after}s", response=get_response)
    get_response.raise_for_status()
    
    user_data = get_response.json()
    
    # Modify skills
    skills = user_data.get("skills", [])
    skill_updated = False
    
    for skill in skills:
        if skill.get("skillId") == skill_id:
            skill["proficiency"] = proficiency
            skill_updated = True
            break
    
    if not skill_updated:
        skills.append({"skillId": skill_id, "proficiency": proficiency})
        user_data["skills"] = skills

    payload = {
        "version": user_data.get("version"),
        "userId": user_data.get("userId"),
        "userName": user_data.get("userName"),
        "firstName": user_data.get("firstName"),
        "lastName": user_data.get("lastName"),
        "email": user_data.get("email"),
        "userType": user_data.get("userType"),
        "skills": user_data.get("skills"),
    }

    # Update
    put_response = requests.put(get_url, headers=headers, json=payload)
    
    if put_response.status_code == 429:
        retry_after = int(put_response.headers.get('Retry-After', 5))
        raise requests.exceptions.HTTPError(f"Rate limited on PUT. Retry after {retry_after}s", response=put_response)
    
    if put_response.status_code == 409:
        # Version conflict. In a bulk job, we might want to log this and skip, or retry the GET.
        # For this tutorial, we treat 409 as a non-retryable error for this specific item.
        print(f"Conflict (409) for user {user_id}. Skipping.")
        return False

    put_response.raise_for_status()
    
    print(f"Updated: {user_data.get('userName')} ({user_id})")
    return True

# Bulk Execution Loop
def bulk_update_agents(auth: CXoneAuth, agents: list, skill_id: str, proficiency: int, skill_name: str):
    """
    Iterates through agents and updates proficiencies.
    """
    success_count = 0
    fail_count = 0
    
    for agent in agents:
        user_id = agent.get("userId")
        user_name = agent.get("userName")
        
        try:
            # Add a small delay between requests to stay within rate limits proactively
            time.sleep(0.5) 
            
            if safe_update_agent(auth, user_id, skill_id, proficiency, skill_name):
                success_count += 1
            else:
                fail_count += 1
                
        except Exception as e:
            print(f"Unexpected error updating {user_name}: {e}")
            fail_count += 1
            
    print(f"--- Bulk Update Complete ---")
    print(f"Success: {success_count}")
    print(f"Failed: {fail_count}")

# Run the bulk update
if spanish_skill_id:
    bulk_update_agents(auth, agents, spanish_skill_id, proficiency=5, skill_name="Spanish - Fluent")

Complete Working Example

Below is the consolidated, runnable script. Save this as bulk_update_skills.py.

import os
import time
import requests
from dotenv import load_dotenv
from tenacity import retry, stop_after_attempt, wait_random_exponential, retry_if_exception_type

load_dotenv()

class CXoneAuth:
    def __init__(self, domain: str, client_id: str, client_secret: str):
        self.domain = domain
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = f"https://{domain}/api/auth/oauth2/token"
        self.access_token = None
        self.token_expiry = 0

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_random_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type(requests.exceptions.RequestException)
    )
    def get_token(self) -> str:
        import time
        if self.access_token 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
        }

        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"]
        self.token_expiry = time.time() + token_data.get("expires_in", 3600) - 60
        return self.access_token

def get_skill_id_by_name(auth: CXoneAuth, skill_name: str) -> str:
    url = f"https://{auth.domain}/api/v2/admin/skills"
    token = auth.get_token()
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    skills = response.json().get("skills", [])
    
    for skill in skills:
        if skill.get("name").lower() == skill_name.lower():
            return skill.get("skillId")
    return None

def get_active_agents(auth: CXoneAuth, limit: int = 100) -> list:
    url = f"https://{auth.domain}/api/v2/admin/users"
    token = auth.get_token()
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    params = {"userType": "agent", "limit": limit}

    response = requests.get(url, headers=headers, params=params)
    response.raise_for_status()
    return response.json().get("users", [])

@retry(
    stop=stop_after_attempt(5),
    wait=wait_random_exponential(multiplier=1, max=60),
    retry=retry_if_exception_type(requests.exceptions.HTTPError),
    reraise=True
)
def safe_update_agent(auth: CXoneAuth, user_id: str, skill_id: str, proficiency: int, skill_name: str) -> bool:
    token = auth.get_token()
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    get_url = f"https://{auth.domain}/api/v2/admin/users/{user_id}"
    
    # GET current user
    get_response = requests.get(get_url, headers=headers)
    if get_response.status_code == 429:
        raise requests.exceptions.HTTPError(f"Rate limited. Retry after {get_response.headers.get('Retry-After', 5)}s", response=get_response)
    get_response.raise_for_status()
    
    user_data = get_response.json()
    
    # Modify skills
    skills = user_data.get("skills", [])
    skill_updated = False
    
    for skill in skills:
        if skill.get("skillId") == skill_id:
            skill["proficiency"] = proficiency
            skill_updated = True
            break
    
    if not skill_updated:
        skills.append({"skillId": skill_id, "proficiency": proficiency})
        user_data["skills"] = skills

    # Payload construction
    payload = {
        "version": user_data.get("version"),
        "userId": user_data.get("userId"),
        "userName": user_data.get("userName"),
        "firstName": user_data.get("firstName"),
        "lastName": user_data.get("lastName"),
        "email": user_data.get("email"),
        "userType": user_data.get("userType"),
        "skills": user_data.get("skills"),
    }

    # PUT update
    put_response = requests.put(get_url, headers=headers, json=payload)
    
    if put_response.status_code == 429:
        raise requests.exceptions.HTTPError(f"Rate limited on PUT. Retry after {put_response.headers.get('Retry-After', 5)}s", response=put_response)
    
    if put_response.status_code == 409:
        print(f"Conflict (409) for user {user_id}. Skipping.")
        return False

    put_response.raise_for_status()
    print(f"Updated: {user_data.get('userName')} ({user_id})")
    return True

def main():
    auth = CXoneAuth(
        domain=os.getenv("CXONE_DOMAIN"),
        client_id=os.getenv("CXONE_CLIENT_ID"),
        client_secret=os.getenv("CXONE_CLIENT_SECRET")
    )

    skill_name = "Spanish - Fluent"
    target_proficiency = 5

    print(f"Fetching skill ID for: {skill_name}")
    skill_id = get_skill_id_by_name(auth, skill_name)
    if not skill_id:
        print(f"Error: Skill '{skill_name}' not found.")
        return

    print(f"Fetching active agents...")
    agents = get_active_agents(auth, limit=100)
    print(f"Found {len(agents)} agents.")

    print(f"Starting bulk update of {len(agents)} agents to proficiency {target_proficiency}...")
    
    success_count = 0
    fail_count = 0
    
    for agent in agents:
        user_id = agent.get("userId")
        try:
            time.sleep(0.5) # Polite delay
            if safe_update_agent(auth, user_id, skill_id, target_proficiency, skill_name):
                success_count += 1
            else:
                fail_count += 1
        except Exception as e:
            print(f"Failed to process agent {agent.get('userName')}: {e}")
            fail_count += 1

    print(f"--- Complete ---")
    print(f"Success: {success_count}, Failed: {fail_count}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 409 Conflict

• What causes it: The version field in your PUT payload does not match the current version of the user record in CXone. This happens if another process (or admin) updated the user between your GET and PUT calls.
• How to fix it: The provided code treats this as a skip. For critical updates, implement a retry loop that performs a fresh GET before the next PUT attempt. Ensure you are passing the version field exactly as returned by the GET request.

Error: 429 Too Many Requests

• What causes it: You have exceeded the rate limit for the admin/user endpoints. CXone enforces strict rate limits to protect platform stability.
• How to fix it: The code uses tenacity to retry with exponential backoff. Additionally, the time.sleep(0.5) in the main loop reduces the request rate proactively. If you still hit 429s, increase the sleep duration or reduce the limit in get_active_agents to process smaller batches.

Error: 403 Forbidden

• What causes it: The OAuth token does not have the required scopes (admin:user:write or admin:user:read).
• How to fix it: Verify your Service Account configuration in the CXone Admin UI. Ensure the Client ID and Secret match the Service Account that has been granted the necessary permissions.

Error: 400 Bad Request

• What causes it: The JSON payload is malformed or missing required fields. Specifically, omitting the version field or sending an invalid proficiency level (must be 1-5) will cause this.
• How to fix it: Ensure the proficiency integer is between 1 and 5. Verify that the userId and skillId are valid UUIDs.

Official References

• NICE CXone Admin API Documentation
• Users API Reference
• Skills API Reference
• OAuth 2.0 Authentication Guide