CXone Admin API — Bulk Update Agent Skill Proficiencies via REST

What You Will Build

• This tutorial demonstrates how to update skill proficiencies for multiple agents simultaneously using the NICE CXone Admin REST API.
• The solution uses the PATCH /api/v2/agents endpoint with a batch update payload to modify skill assignments efficiently.
• The programming language covered is Python 3.10+ using the httpx library for robust HTTP handling.

Prerequisites

• OAuth Client Type: Service Account or Web Server (Client Credentials Grant or Authorization Code Grant).
• Required Scopes:
  • admin:agent:write (Required for updating agent profiles)
  • admin:agent:read (Required for retrieving current agent data if needed for comparison)
• SDK/API Version: CXone Admin API v2.
• Language/Runtime: Python 3.10 or higher.
• External Dependencies:
  • httpx: For asynchronous HTTP requests with native support for retries and timeouts.
  • pydantic: For data validation and type safety (optional but recommended for production).

Install dependencies via pip:

pip install httpx pydantic

Authentication Setup

NICE CXone uses OAuth 2.0 for authentication. For server-to-server integrations, the Client Credentials Grant flow is the standard approach. This flow exchanges a client ID and secret for an access token.

The following code demonstrates how to obtain an access token and handle the expiration logic.

import httpx
from typing import Optional
import time

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

    async def get_access_token(self) -> str:
        """
        Retrieves a new access token if the current one is expired or missing.
        """
        if self.access_token and time.time() < self.token_expiry:
            return self.access_token

        async with httpx.AsyncClient() as client:
            response = await client.post(
                self.token_url,
                data={
                    "grant_type": "client_credentials",
                    "client_id": self.client_id,
                    "client_secret": self.client_secret
                },
                headers={
                    "Content-Type": "application/x-www-form-urlencoded"
                }
            )

            if response.status_code != 200:
                raise Exception(f"Failed to obtain access token: {response.text}")

            token_data = response.json()
            self.access_token = token_data["access_token"]
            # Set expiry slightly before actual expiry to avoid race conditions
            self.token_expiry = time.time() + token_data["expires_in"] - 10

            return self.access_token

    async def get_headers(self) -> dict:
        """
        Returns the standard headers required for CXone API calls.
        """
        token = await self.get_access_token()
        return {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }

Key Authentication Details:

• The realm parameter is critical. It determines the base URL (e.g., na1, eu1, ap1).
• The access_token expires typically after 1 hour. The get_access_token method checks the timestamp before making a new request, preventing unnecessary network calls.

Implementation

Step 1: Define the Skill Proficiency Payload

The CXone Agent entity contains a skills array. Each skill object in this array defines the relationship between an agent and a specific skill. To update proficiencies, you must provide the correct structure for the skills array within the agent update payload.

The PATCH operation on /api/v2/agents supports partial updates. However, when updating the skills array, the API expects the complete desired state of skills for that agent, not just the delta. Therefore, you must include all existing skills you wish to retain, along with the updated or new skills.

Skill Object Structure:

{
  "skill": {
    "id": "skill-uuid-here"
  },
  "proficiency": 85,
  "available": true
}

• id: The UUID of the skill.
• proficiency: An integer between 0 and 100. This value influences routing weight in skills-based routing.
• available: A boolean indicating if the agent is available for this skill.

Step 2: Construct the Batch Update Payload

To bulk-update agents, you do not call the endpoint once per agent. Instead, you can send multiple PATCH requests in parallel or use a loop with concurrency controls. The CXone API does not have a single “bulk update all agents” endpoint that accepts a list of agents in one JSON body. Instead, “bulk” in this context means automating individual PATCH calls efficiently.

We will use httpx.AsyncClient to handle concurrent requests. This is crucial because updating 1,000 agents sequentially could take minutes. With concurrency, it can be done in seconds.

Important: CXone enforces rate limits. The default rate limit for Admin APIs is typically 100 requests per minute per tenant, but this can vary. We must implement exponential backoff for 429 Too Many Requests responses.

Step 3: Implement Concurrent Updates with Retry Logic

The following class handles the core logic: fetching agents (optional, if you already have IDs), constructing the payloads, and executing the updates concurrently.

import asyncio
import httpx
from typing import List, Dict, Any, Optional
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class AgentSkillUpdater:
    def __init__(self, auth: CXoneAuth, realm: str, max_concurrency: int = 10):
        self.auth = auth
        self.realm = realm
        self.base_url = f"https://{realm}.cxone.com/api/v2"
        self.max_concurrency = max_concurrency
        self.semaphore = asyncio.Semaphore(max_concurrency)

    async def update_agent_skills(
        self,
        agent_id: str,
        skills_payload: List[Dict[str, Any]]
    ) -> Dict[str, Any]:
        """
        Updates the skills for a single agent.
        
        Args:
            agent_id: The UUID of the agent.
            skills_payload: A list of skill objects to set as the agent's complete skill set.
        
        Returns:
            The response JSON from the API.
        """
        url = f"{self.base_url}/agents/{agent_id}"
        headers = await self.auth.get_headers()

        # The payload must include only the fields being updated
        payload = {
            "skills": skills_payload
        }

        async with httpx.AsyncClient(timeout=30.0) as client:
            retries = 3
            for attempt in range(retries):
                try:
                    async with self.semaphore:
                        response = await client.patch(
                            url,
                            json=payload,
                            headers=headers
                        )

                    if response.status_code == 200:
                        logger.info(f"Successfully updated agent {agent_id}")
                        return response.json()
                    elif response.status_code == 429:
                        # Rate limited. Wait and retry.
                        wait_time = 2 ** attempt
                        logger.warning(f"Rate limited for agent {agent_id}. Retrying in {wait_time}s...")
                        await asyncio.sleep(wait_time)
                        continue
                    elif response.status_code == 404:
                        logger.error(f"Agent {agent_id} not found.")
                        return {"error": "Agent not found"}
                    elif response.status_code == 400:
                        logger.error(f"Bad request for agent {agent_id}: {response.text}")
                        return {"error": "Bad request", "details": response.text}
                    else:
                        logger.error(f"Unexpected error for agent {agent_id}: {response.status_code} {response.text}")
                        return {"error": response.text}

                except httpx.RequestError as e:
                    logger.error(f"Network error for agent {agent_id}: {e}")
                    await asyncio.sleep(1)
                    continue

            return {"error": "Max retries exceeded"}

    async def bulk_update_agents(
        self,
        updates: List[Dict[str, Any]]
    ) -> List[Dict[str, Any]]:
        """
        Executes concurrent updates for a list of agents.
        
        Args:
            updates: A list of dictionaries, each containing:
                - 'agent_id': str
                - 'skills': List[Dict]
        
        Returns:
            A list of response dictionaries.
        """
        tasks = []
        for update in updates:
            task = asyncio.create_task(
                self.update_agent_skills(
                    agent_id=update["agent_id"],
                    skills_payload=update["skills"]
                )
            )
            tasks.append(task)

        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Process results
        processed_results = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                logger.error(f"Task for agent {updates[i]['agent_id']} failed with exception: {result}")
                processed_results.append({"error": str(result)})
            else:
                processed_results.append(result)
        
        return processed_results

Explanation of the Code:

1. asyncio.Semaphore: This limits the number of concurrent HTTP requests to max_concurrency. Without this, you might open thousands of connections simultaneously, triggering rate limits or exhausting local OS resources.
2. Retry Logic: The loop inside update_agent_skills handles 429 errors by waiting exponentially longer between attempts. This is essential for bulk operations.
3. Payload Structure: The PATCH request only sends the skills field. This ensures that other agent attributes (like name, email, or default queues) remain untouched.

Complete Working Example

The following script ties everything together. It assumes you have a list of agent IDs and the desired skill proficiencies.

import asyncio
import os
from cxone_auth import CXoneAuth  # Assuming the auth class is in cxone_auth.py
from agent_skill_updater import AgentSkillUpdater

async def main():
    # Configuration
    CLIENT_ID = os.getenv("CXONE_CLIENT_ID")
    CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET")
    REALM = os.getenv("CXONE_REALM", "na1")

    if not CLIENT_ID or not CLIENT_SECRET:
        raise ValueError("Missing CXONE_CLIENT_ID or CXONE_CLIENT_SECRET environment variables.")

    # Initialize Authentication
    auth = CXoneAuth(
        client_id=CLIENT_ID,
        client_secret=CLIENT_SECRET,
        realm=REALM
    )

    # Initialize Updater
    updater = AgentSkillUpdater(auth=auth, realm=REALM, max_concurrency=10)

    # Define Updates
    # Scenario: Update proficiency for 'Skill A' and 'Skill B' for two agents.
    # Note: In a real scenario, you would likely fetch the current skills first 
    # to ensure you are not deleting existing skills you want to keep.
    
    skill_a_id = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
    skill_b_id = "b2c3d4e5-f6a7-8901-bcde-f12345678901"

    updates = [
        {
            "agent_id": "agent-uuid-1",
            "skills": [
                {
                    "skill": {"id": skill_a_id},
                    "proficiency": 90,
                    "available": True
                },
                {
                    "skill": {"id": skill_b_id},
                    "proficiency": 75,
                    "available": True
                }
            ]
        },
        {
            "agent_id": "agent-uuid-2",
            "skills": [
                {
                    "skill": {"id": skill_a_id},
                    "proficiency": 100,
                    "available": True
                },
                {
                    "skill": {"id": skill_b_id},
                    "proficiency": 50,
                    "available": False
                }
            ]
        }
    ]

    print("Starting bulk update...")
    results = await updater.bulk_update_agents(updates)
    
    print("\nResults:")
    for i, result in enumerate(results):
        if "error" in result:
            print(f"Agent {updates[i]['agent_id']}: FAILED - {result}")
        else:
            print(f"Agent {updates[i]['agent_id']}: SUCCESS")

if __name__ == "__main__":
    asyncio.run(main())

Common Errors & Debugging

Error: 400 Bad Request - “Invalid skill ID”

Cause: The skill.id provided in the payload does not exist in the CXone tenant, or the format is incorrect.
Fix: Verify the skill UUIDs. You can list all skills using GET /api/v2/skills to find the correct IDs. Ensure the UUIDs are valid strings.

Error: 400 Bad Request - “Proficiency out of range”

Cause: The proficiency field must be an integer between 0 and 100.
Fix: Validate your input data before sending. Ensure no floating-point numbers are passed.

Error: 403 Forbidden

Cause: The OAuth token lacks the admin:agent:write scope.
Fix: Re-authenticate with the correct scopes. Check your client application settings in the CXone Admin Portal under Platform > OAuth Applications.

Error: 429 Too Many Requests

Cause: You exceeded the tenant’s rate limit for API calls.
Fix: The provided code includes exponential backoff. If you still see 429s, reduce the max_concurrency parameter in AgentSkillUpdater. Start with 5 and increase gradually.

Error: 409 Conflict

Cause: Another process is currently updating the same agent. CXone uses optimistic locking.
Fix: Implement a retry mechanism with a random jitter. The current retry logic handles this implicitly by retrying after a delay.

Official References

• NICE CXone Admin API Documentation - Agents
• NICE CXone Developer Portal
• OAuth 2.0 Client Credentials Grant