Bulk-Update NICE CXone Agent Skill Proficiencies via REST

Bulk-Update NICE CXone Agent Skill Proficiencies via REST

What You Will Build

  • A Python script that retrieves a list of agents, modifies their skill proficiencies for specific languages and technical domains, and pushes those updates back to the NICE CXone platform.
  • This tutorial uses the NICE CXone Admin API (/api/v2/users and /api/v2/users/{userId}) and the REST client pattern.
  • The implementation is written in Python 3.9+ using requests and httpx.

Prerequisites

OAuth Client and Scopes

You must have a NICE CXone OAuth Client ID and Secret with the following scopes assigned:

  • admin:agent:read — To fetch user details and current skill sets.
  • admin:agent:update — To modify user profiles and skill proficiencies.
  • admin:skill:read — To resolve skill IDs if you are mapping by name.

SDK/API Version

  • API Version: v2 (The v1 API is deprecated for user management).
  • Base URL: https://api-us-2.nicecxone.com/api/v2 (Adjust for your region: api-eu-1, api-au-1, etc.).

Dependencies

Install the required Python packages:

pip install requests httpx

Authentication Setup

NICE CXone uses OAuth 2.0 Client Credentials flow for server-to-server API access. You must obtain a bearer token before making any API calls.

Token Retrieval Code

import httpx
import time
from typing import Optional

class CxoneAuth:
    def __init__(self, client_id: str, client_secret: str, region: str = "us-2"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.region = region
        self.token_url = f"https://auth-{region}.nicecxone.com/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0

    def get_token(self) -> str:
        """
        Retrieves an OAuth token. Handles caching if the token is still valid.
        """
        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
        }

        with httpx.Client() as client:
            response = client.post(self.token_url, headers=headers, data=data)
            response.raise_for_status()

            token_data = response.json()
            self.access_token = token_data["access_token"]
            # Expires in is in seconds; add to current time
            self.token_expiry = time.time() + token_data["expires_in"] - 60 # Buffer 60s

        return self.access_token

    def get_headers(self) -> dict:
        return {
            "Authorization": f"Bearer {self.get_token()}",
            "Accept": "application/json",
            "Content-Type": "application/json"
        }

Implementation

Step 1: Identify Skills by Name

NICE CXone APIs require Skill IDs, not names. If your bulk update input is based on skill names (e.g., “English”, “Billing”), you must resolve these to IDs first.

Endpoint: GET /api/v2/skills
Scope: admin:skill:read

import httpx
from typing import Dict, List

class CxoneSkillManager:
    def __init__(self, auth: CxoneAuth):
        self.auth = auth
        self.base_url = f"https://api-{auth.region}.nicecxone.com/api/v2"

    def get_skill_id_by_name(self, skill_name: str) -> Optional[str]:
        """
        Searches the skills list to find the ID for a given skill name.
        Note: This endpoint supports pagination. For simplicity, we fetch the first page.
        In production, iterate through 'nextPageUri' if skills exceed 100.
        """
        headers = self.auth.get_headers()
        params = {
            "skillName": skill_name,
            "pageSize": 1
        }

        with httpx.Client() as client:
            try:
                response = client.get(
                    f"{self.base_url}/skills",
                    headers=headers,
                    params=params
                )
                response.raise_for_status()
                data = response.json()
                
                if data.get("items") and len(data["items"]) > 0:
                    return data["items"][0]["id"]
            except httpx.HTTPStatusError as e:
                print(f"Error fetching skill '{skill_name}': {e.response.text}")
            return None

    def resolve_skill_ids(self, skill_names: List[str]) -> Dict[str, str]:
        """
        Maps a list of skill names to their IDs.
        Returns { skill_name: skill_id }
        """
        mapping = {}
        for name in skill_names:
            sid = self.get_skill_id_by_name(name)
            if sid:
                mapping[name] = sid
            else:
                print(f"Warning: Skill '{name}' not found.")
        return mapping

Step 2: Fetch Agents and Current Proficiencies

You cannot blindly overwrite an agent’s skill set without knowing their existing proficiencies unless you intend to wipe all other skills. The standard pattern is: Read → Modify → Write.

Endpoint: GET /api/v2/users
Scope: admin:agent:read

import httpx
from typing import List, Dict, Any

class CxoneAgentManager:
    def __init__(self, auth: CxoneAuth):
        self.auth = auth
        self.base_url = f"https://api-{auth.region}.nicecxone.com/api/v2"

    def get_agents(self, query: str = "") -> List[Dict[str, Any]]:
        """
        Retrieves a list of users. Supports querying by name or email.
        Handles pagination automatically.
        """
        all_agents = []
        headers = self.auth.get_headers()
        params = {
            "pageSize": 100,
            "page": 1
        }
        
        if query:
            params["query"] = query

        with httpx.Client() as client:
            while True:
                response = client.get(
                    f"{self.base_url}/users",
                    headers=headers,
                    params=params
                )
                response.raise_for_status()
                data = response.json()

                all_agents.extend(data.get("items", []))
                
                # Check for next page
                if data.get("nextPageUri"):
                    # The nextPageUri is a full URL, but we need to keep headers
                    # It is safer to increment page number if the URI structure is consistent
                    # or parse the URI. For simplicity, we increment page.
                    params["page"] += 1
                else:
                    break

        return all_agents

Step 3: Update Agent Skill Proficiencies

The core logic. The user object in CXone has a skills array. Each item in this array contains:

  • skillId: The UUID of the skill.
  • proficiency: An integer (typically 0-100, or 0-5 depending on your skill configuration, but 0-100 is standard for most bulk operations).
  • status: Usually "ACTIVE".

Endpoint: PUT /api/v2/users/{userId}
Scope: admin:agent:update

Critical Note: When updating a user via PUT, you must send the entire user object back. If you omit fields like firstName, lastName, or other attributes, they may be reset to null or default values depending on the API version behavior. Always merge changes into the existing object.

import httpx
from typing import List, Dict, Any

class CxoneAgentManager:
    # ... (previous methods)

    def update_agent_skills(
        self, 
        agent: Dict[str, Any], 
        skill_updates: Dict[str, int]
    ) -> bool:
        """
        Updates an agent's skill proficiencies.
        
        Args:
            agent: The full user object retrieved from GET /users.
            skill_updates: A dict mapping { skill_id: proficiency_value }.
                          Example: { "abc-123": 80, "def-456": 60 }
        
        Returns:
            True if successful, False otherwise.
        """
        user_id = agent.get("id")
        if not user_id:
            return False

        # Deep copy to avoid mutating the original list reference if reused
        import copy
        updated_agent = copy.deepcopy(agent)

        # Ensure the skills list exists
        if "skills" not in updated_agent or updated_agent["skills"] is None:
            updated_agent["skills"] = []

        # Map existing skills by ID for quick lookup
        existing_skills_map = {s["skillId"]: s for s in updated_agent["skills"]}

        for skill_id, proficiency in skill_updates.items():
            if skill_id in existing_skills_map:
                # Update existing skill
                existing_skills_map[skill_id]["proficiency"] = proficiency
            else:
                # Add new skill
                new_skill = {
                    "skillId": skill_id,
                    "proficiency": proficiency,
                    "status": "ACTIVE" # Default status
                }
                updated_agent["skills"].append(new_skill)

        # Remove skills that are no longer in the update list?
        # Option A: Keep existing skills not in skill_updates (Safe)
        # Option B: Wipe all skills not in skill_updates (Destructive)
        # This implementation uses Option A (Merge).

        headers = self.auth.get_headers()
        
        with httpx.Client() as client:
            try:
                response = client.put(
                    f"{self.base_url}/users/{user_id}",
                    headers=headers,
                    json=updated_agent
                )
                
                # 204 No Content is standard for successful PUT updates in CXone
                if response.status_code == 204:
                    return True
                else:
                    print(f"Failed to update user {user_id}: {response.status_code} - {response.text}")
                    return False

            except httpx.HTTPStatusError as e:
                if e.response.status_code == 409:
                    print(f"Conflict updating user {user_id}. Another process may have modified this user.")
                elif e.response.status_code == 429:
                    print(f"Rate limit hit. Retry after backoff.")
                else:
                    print(f"HTTP Error {e.response.status_code}: {e.response.text}")
                return False

Complete Working Example

This script combines the above components. It reads a CSV-like structure (defined in memory for this example) containing Agent Email, Skill Name, and Desired Proficiency, then applies the updates.

import httpx
import time
import json
import sys
from typing import List, Dict, Any, Optional

# --- Configuration ---
CLIENT_ID = "your_client_id"
CLIENT_SECRET = "your_client_secret"
REGION = "us-2" # Change to eu-1, au-1, etc.

# --- Mock Input Data: List of updates to apply ---
# In production, load this from CSV/Excel/Database
BULK_UPDATES = [
    {
        "agent_email": "john.doe@company.com",
        "skill_name": "English",
        "proficiency": 90
    },
    {
        "agent_email": "john.doe@company.com",
        "skill_name": "Billing",
        "proficiency": 70
    },
    {
        "agent_email": "jane.smith@company.com",
        "skill_name": "Spanish",
        "proficiency": 85
    }
]

class CxoneBulkSkillUpdater:
    def __init__(self, client_id: str, client_secret: str, region: str):
        self.auth = CxoneAuth(client_id, client_secret, region)
        self.skill_mgr = CxoneSkillManager(self.auth)
        self.agent_mgr = CxoneAgentManager(self.auth)
        self.base_url = f"https://api-{region}.nicecxone.com/api/v2"

    def run(self):
        print("1. Resolving Skill Names to IDs...")
        # Extract unique skill names
        unique_skills = list(set([u["skill_name"] for u in BULK_UPDATES]))
        skill_map = self.skill_mgr.resolve_skill_ids(unique_skills)
        
        if not skill_map:
            print("No skills found. Aborting.")
            return

        print("2. Fetching Agents...")
        # We need to find agents by email. 
        # CXone /users endpoint does not support direct email lookup in query param easily 
        # without fetching all or using a specific search. 
        # For this example, we will fetch all active users (or a large subset) 
        # and build a local lookup map.
        
        # Optimization: If you know specific User IDs, skip this step and fetch by ID.
        all_agents = self.agent_mgr.get_agents()
        
        # Build email -> agent mapping
        agent_by_email = {agent["email"]: agent for agent in all_agents}
        
        print(f"Found {len(all_agents)} agents. Processing {len(BULK_UPDATES)} updates...")

        # Group updates by agent to minimize API calls
        # Structure: { email: { skill_id: proficiency } }
        updates_by_agent: Dict[str, Dict[str, int]] = {}
        
        for update in BULK_UPDATES:
            email = update["agent_email"]
            skill_name = update["skill_name"]
            proficiency = update["proficiency"]
            
            skill_id = skill_map.get(skill_name)
            if not skill_id:
                print(f"Skipping update for {email}: Skill '{skill_name}' not found.")
                continue
            
            if email not in agent_by_email:
                print(f"Skipping update: Agent with email '{email}' not found.")
                continue

            if email not in updates_by_agent:
                updates_by_agent[email] = {}
            
            updates_by_agent[email][skill_id] = proficiency

        # Apply Updates
        success_count = 0
        fail_count = 0
        
        for email, skill_updates in updates_by_agent.items():
            agent_obj = agent_by_email[email]
            print(f"Updating skills for {email}...")
            
            # Rate Limiting: CXone typically allows ~100 req/s but varies.
            # Adding a small delay to be safe in bulk operations.
            time.sleep(0.1)

            if self.agent_mgr.update_agent_skills(agent_obj, skill_updates):
                success_count += 1
            else:
                fail_count += 1

        print(f"\nCompleted. Success: {success_count}, Failed: {fail_count}")

if __name__ == "__main__":
    if CLIENT_ID == "your_client_id":
        print("Error: Please configure CLIENT_ID and CLIENT_SECRET in the script.")
        sys.exit(1)
        
    updater = CxoneBulkSkillUpdater(CLIENT_ID, CLIENT_SECRET, REGION)
    updater.run()

Common Errors & Debugging

Error: 403 Forbidden

Cause: The OAuth client lacks the admin:agent:update scope.
Fix: Log into the NICE CXone Admin UI, navigate to Administration > Security > OAuth Clients, edit your client, and ensure admin:agent:update is checked.

Error: 400 Bad Request - “Invalid Skill ID”

Cause: The skillId provided does not exist in the tenant, or it is a soft-deleted skill.
Fix: Verify the skill ID exists by calling GET /api/v2/skills/{skillId}. Ensure the skill is active.

Error: 409 Conflict

Cause: Another process (or admin user) modified the agent’s profile between your GET and PUT calls. CXone uses optimistic locking.
Fix: Implement a retry mechanism with exponential backoff. Re-fetch the user object (GET /users/{id}) immediately before retrying the PUT.

def update_with_retry(self, agent, skill_updates, max_retries=3):
    for attempt in range(max_retries):
        success = self.agent_mgr.update_agent_skills(agent, skill_updates)
        if success:
            return True
        # If failed, re-fetch the latest version
        print(f"Conflict detected. Re-fetching user data (Attempt {attempt + 1})...")
        # Re-fetch logic here
        # agent = self.agent_mgr.get_user_by_id(agent["id"])
        time.sleep(2 ** attempt)
    return False

Error: 429 Too Many Requests

Cause: You are exceeding the rate limit for the users endpoint.
Fix: Add a delay between requests. The time.sleep(0.1) in the complete example is a conservative starting point. Monitor the Retry-After header in 429 responses if available.

Official References