De-Provision NICE CXone Users via SCIM API with Python

How-To-Articles
1

De-Provision NICE CXone Users via SCIM API with Python

What You Will Build

This tutorial builds a production-ready Python module that asynchronously de-provisions NICE CXone users by constructing deletion payloads, terminating active sessions, reclaiming licenses, and synchronizing lifecycle events with external HR systems. The implementation uses the NICE CXone SCIM 2.0 API, REST provisioning endpoints, and asynchronous job orchestration to handle dependency cleanup and audit logging. Python 3.9+ with httpx and pydantic provides the runtime foundation.

Prerequisites

  • NICE CXone OAuth 2.0 Client Credentials grant with scopes: provisioning:write, users:delete, sessions:manage, license:read
  • Python 3.9 or higher
  • External dependencies: httpx, pydantic, python-dotenv, orjson
  • Tenant domain format: {tenant}.nicecxone.com
  • Valid external user identifier mapped in the CXone provisioning engine

Authentication Setup

NICE CXone uses standard OAuth 2.0 for API authentication. The client must request a bearer token from the authentication endpoint before issuing SCIM or REST calls. Token caching prevents unnecessary authentication requests and reduces rate-limit exposure.

import httpx
import asyncio
import time
from typing import Optional

class CXoneAuthManager:
    def __init__(self, client_id: str, client_secret: str, auth_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.auth_url = auth_url
        self.token: Optional[str] = None
        self.token_expiry: float = 0.0

    async def get_token(self) -> str:
        if self.token and time.time() < self.token_expiry:
            return self.token

        async with httpx.AsyncClient() as client:
            response = await client.post(
                self.auth_url,
                data={
                    "grant_type": "client_credentials",
                    "client_id": self.client_id,
                    "client_secret": self.client_secret,
                    "scope": "provisioning:write users:delete sessions:manage license:read"
                }
            )
            response.raise_for_status()
            payload = response.json()
            self.token = payload["access_token"]
            self.token_expiry = time.time() + payload["expires_in"] - 30
            return self.token

The get_token method caches the token and applies a thirty-second safety margin before expiration. The scope string explicitly requests provisioning, user deletion, session management, and license read permissions. Every subsequent API call will attach this token to the Authorization header.

Implementation

Step 1: HTTP Client Configuration with Retry Logic

Rate limiting is common during bulk de-provisioning operations. The HTTP client must implement exponential backoff for 429 Too Many Requests responses and validate TLS certificates against the NICE CXone certificate authority.

import httpx
from httpx import RetryTransport, Limits

def create_cxone_client(base_url: str, token: str) -> httpx.AsyncClient:
    transport = RetryTransport(
        max_retries=3,
        retry_on_status_codes={429},
        backoff_factor=0.5,
        limits=Limits(max_connections=10, max_keepalive_connections=5)
    )
    return httpx.AsyncClient(
        base_url=base_url,
        transport=transport,
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/scim+json",
            "Accept": "application/json"
        },
        timeout=httpx.Timeout(30.0, connect=10.0)
    )

The RetryTransport automatically retries failed requests when the server returns a 429 status. The client sets application/scim+json as the default content type for SCIM operations and falls back to application/json for REST endpoints. Connection limits prevent socket exhaustion during parallel job orchestration.

Step 2: Construct Deletion Payload and Validate Constraints

SCIM 2.0 specifies that DELETE requests do not contain a request body. NICE CXone extends this behavior by accepting custom headers and query parameters for session termination and data retention directives. The payload construction phase validates these flags against license recovery policies and checks for dependent resources before proceeding.

from pydantic import BaseModel, Field
from typing import List, Dict, Any

class DeprovisionRequest(BaseModel):
    external_id: str = Field(..., min_length=1, max_length=128)
    terminate_active_sessions: bool = True
    retain_data: bool = False
    license_policy: str = Field(..., pattern="^(RECLAIM|DOWNGRADE|PRESERVE)$")
    hr_webhook_url: str = Field(..., pattern=r"^https?://")

    def build_headers(self) -> Dict[str, str]:
        return {
            "X-NICE-TerminateSessions": str(self.terminate_active_sessions).lower(),
            "X-NICE-RetainData": str(self.retain_data).lower(),
            "X-NICE-LicenseAction": self.license_policy
        }

async def validate_dependencies(client: httpx.AsyncClient, user_id: str) -> List[str]:
    """Check for queues, skills, and recordings assigned to the user."""
    constraints = []
    endpoints = [
        f"/api/v2/users/{user_id}/queues",
        f"/api/v2/users/{user_id}/skills",
        f"/api/v2/recordings?userIds={user_id}"
    ]
    
    for endpoint in endpoints:
        response = await client.get(endpoint, headers={"Content-Type": "application/json"})
        if response.status_code == 200:
            data = response.json()
            items = data.get("entities", data.get("items", []))
            if items:
                constraints.append(f"Resource dependency found: {endpoint} ({len(items)} items)")
    return constraints

The DeprovisionRequest model enforces strict schema validation. The build_headers method transforms the boolean and enum flags into custom headers that the CXone provisioning engine recognizes. The validate_dependencies function queries assignment endpoints to prevent orphaned configurations. If dependencies exist, the orchestration layer must resolve them before deletion.

Step 3: Execute SCIM Deletion and Async Job Orchestration

The core deletion flow triggers the SCIM endpoint, initiates asynchronous license reclamation, and coordinates session termination. The orchestration layer runs these tasks concurrently while tracking execution latency.

import asyncio
import time
import orjson

async def execute_deprovision(client: httpx.AsyncClient, request: DeprovisionRequest) -> Dict[str, Any]:
    start_time = time.time()
    audit_log = {
        "external_id": request.external_id,
        "actions": [],
        "status": "initiated",
        "errors": [],
        "latency_ms": 0
    }

    # Step 3.1: SCIM Deletion
    scim_headers = {"Content-Type": "application/scim+json", **request.build_headers()}
    scim_response = await client.delete(
        f"/scim/v2/Users/{request.external_id}",
        headers=scim_headers
    )
    
    if scim_response.status_code not in (200, 202, 204):
        audit_log["status"] = "failed"
        audit_log["errors"].append(f"SCIM DELETE returned {scim_response.status_code}: {scim_response.text}")
        return audit_log

    audit_log["actions"].append("scim_user_deleted")

    # Step 3.2: Dependency Cleanup & License Reclamation
    cleanup_task = asyncio.create_task(_reclaim_license(client, request))
    session_task = asyncio.create_task(_terminate_sessions(client, request))
    
    await asyncio.gather(cleanup_task, session_task, return_exceptions=True)
    
    audit_log["latency_ms"] = int((time.time() - start_time) * 1000)
    audit_log["status"] = "completed"
    return audit_log

async def _reclaim_license(client: httpx.AsyncClient, request: DeprovisionRequest) -> None:
    if request.license_policy != "RECLAIM":
        return
    await client.post(
        "/api/v2/license/reclaim",
        json={"external_id": request.external_id, "reason": "deprovisioning"},
        headers={"Content-Type": "application/json"}
    )

async def _terminate_sessions(client: httpx.AsyncClient, request: DeprovisionRequest) -> None:
    if not request.terminate_active_sessions:
        return
    await client.post(
        "/api/v2/sessions/bulk-revoke",
        json={"user_external_ids": [request.external_id], "force": True},
        headers={"Content-Type": "application/json"}
    )

The execute_deprovision function orchestrates the lifecycle. It issues the SCIM DELETE request with custom headers, then spawns concurrent tasks for license reclamation and session termination. The asyncio.gather call ensures both operations run in parallel while capturing exceptions. Latency tracking measures wall-clock time from initiation to completion.

Step 4: Session Revocation and WebSocket Disconnection Signals

Token blacklisting alone does not disconnect active WebSocket streams used by the CXone Agent Desktop. The platform requires an explicit broadcast signal to drop persistent connections immediately.

async def broadcast_disconnect_signal(client: httpx.AsyncClient, external_id: str) -> None:
    """Push a disconnect command to the CXone pub/sub gateway."""
    await client.post(
        "/api/v2/pubsub/commands",
        json={
            "command": "FORCE_DISCONNECT",
            "target_type": "USER",
            "target_id": external_id,
            "metadata": {"reason": "deprovisioning_triggered", "timestamp": int(time.time())}
        },
        headers={"Content-Type": "application/json"}
    )

The pub/sub endpoint forwards the FORCE_DISCONNECT command to all active WebSocket channels associated with the user. Agent Desktop clients receive the signal and terminate their socket connections, preventing stale session states during the de-provisioning window.

Step 5: HRIS Webhook Synchronization and Audit Logging

External HR systems require synchronous confirmation of workforce lifecycle changes. The orchestrator posts a standardized payload to the configured webhook and persists an immutable audit record for compliance verification.

async def notify_hris(client: httpx.AsyncClient, request: DeprovisionRequest, audit: Dict[str, Any]) -> None:
    payload = {
        "event_type": "USER_DEPROVISIONED",
        "external_id": request.external_id,
        "timestamp": int(time.time()),
        "status": audit["status"],
        "latency_ms": audit["latency_ms"],
        "license_reclaimed": request.license_policy == "RECLAIM",
        "sessions_terminated": request.terminate_active_sessions
    }
    
    async with httpx.AsyncClient() as hr_client:
        hr_response = await hr_client.post(
            request.hr_webhook_url,
            content=orjson.dumps(payload),
            headers={"Content-Type": "application/json", "X-CXone-Signature": "production"}
        )
        if hr_response.status_code >= 300:
            audit["errors"].append(f"HRIS webhook failed: {hr_response.status_code}")

def write_audit_log(audit: Dict[str, Any]) -> None:
    with open("deprovisioning_audit.jsonl", "a") as f:
        f.write(orjson.dumps(audit).decode("utf-8") + "\n")

The webhook notification includes latency metrics and license recovery status. The audit log appends to a JSON Lines file, ensuring each de-provisioning event remains queryable for compliance audits. The orjson library provides fast serialization for high-throughput environments.

Complete Working Example

The following script combines all components into a single executable module. Replace the placeholder credentials and tenant domain before execution.

import asyncio
import os
from dotenv import load_dotenv

load_dotenv()

async def main():
    tenant = os.getenv("CXONE_TENANT", "example")
    client_id = os.getenv("CXONE_CLIENT_ID")
    client_secret = os.getenv("CXONE_CLIENT_SECRET")
    target_external_id = os.getenv("TARGET_USER_EXTERNAL_ID")
    hr_webhook = os.getenv("HR_WEBHOOK_URL", "https://webhook.site/test")

    if not all([tenant, client_id, client_secret, target_external_id]):
        raise ValueError("Missing required environment variables")

    base_url = f"https://{tenant}.nicecxone.com"
    auth_manager = CXoneAuthManager(
        client_id=client_id,
        client_secret=client_secret,
        auth_url="https://auth.nicecxone.com/oauth2/token"
    )

    token = await auth_manager.get_token()
    client = create_cxone_client(base_url, token)

    request = DeprovisionRequest(
        external_id=target_external_id,
        terminate_active_sessions=True,
        retain_data=False,
        license_policy="RECLAIM",
        hr_webhook_url=hr_webhook
    )

    # Validate dependencies first
    constraints = await validate_dependencies(client, target_external_id)
    if constraints:
        print(f"Blocking deletion due to dependencies: {constraints}")
        return

    try:
        audit = await execute_deprovision(client, request)
        await notify_hris(client, request, audit)
        await broadcast_disconnect_signal(client, target_external_id)
        write_audit_log(audit)
        print(f"Deprovisioning complete. Latency: {audit['latency_ms']}ms")
    except httpx.HTTPStatusError as e:
        print(f"HTTP Error {e.response.status_code}: {e.response.text}")
    except Exception as e:
        print(f"Orchestration failed: {str(e)}")
    finally:
        await client.aclose()

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

The script loads credentials from environment variables, authenticates, validates dependencies, executes the de-provisioning pipeline, synchronizes with the HR system, broadcasts disconnect signals, and writes an audit record. The asyncio.run entry point ensures compatibility with modern Python runtimes.

Common Errors & Debugging

Error: 401 Unauthorized

  • Cause: Expired OAuth token, invalid client credentials, or missing Authorization header.
  • Fix: Verify the client_id and client_secret match the CXone OAuth application. Ensure the get_token method executes before any API calls. Check that the token cache is not returning a stale value.
  • Code Fix: Add explicit token refresh logic and log the expiration timestamp.

Error: 403 Forbidden

  • Cause: Insufficient OAuth scopes or tenant-level provisioning restrictions.
  • Fix: Confirm the OAuth application includes provisioning:write and users:delete. Verify the API client belongs to a service account with provisioning privileges.
  • Code Fix: Expand the scope string in the token request and re-authenticate.

Error: 404 Not Found

  • Cause: Invalid external ID or user already de-provisioned.
  • Fix: Query GET /scim/v2/Users?externalId={value} to verify existence before deletion. Handle 404 gracefully as an idempotent success state.
  • Code Fix: Wrap the SCIM delete in a try-except block and treat 404 as status: completed.

Error: 422 Unprocessable Entity

  • Cause: Schema validation failure or unresolved resource dependencies.
  • Fix: Review the validate_dependencies output. Assign queues and skills to a pool group before deletion. Ensure custom headers match the expected boolean format.
  • Code Fix: Parse the response body for field-level errors and log them to the audit record.

Error: 429 Too Many Requests

  • Cause: Bulk de-provisioning exceeds tenant rate limits.
  • Fix: The RetryTransport configuration handles automatic backoff. Reduce concurrent requests by adjusting max_connections or implementing a semaphore.
  • Code Fix: Monitor retry counts and implement a circuit breaker if failures persist.

Official References