Synchronizing NICE CXone SCIM Users with Azure AD Using Python

What You Will Build

This script polls Microsoft Graph API for incremental user property changes, maps Active Directory groups to NICE CXone roles using a bidirectional lookup table, constructs SCIM PATCH requests with minimal payloads to reduce API load, resolves attribute conflicts by prioritizing a configurable source of truth, and maintains a SQLite state store to track synchronization progress and recover from partial failures. The implementation uses the Microsoft Graph REST API and the NICE CXone SCIM 2.0 API. The code is written in Python 3.10+.

Prerequisites

• Azure AD Application Registration with delegated or application permissions for User.Read.All
• NICE CXone OAuth 2.0 Client Credentials grant configured with scim:users:readwrite scope
• Python 3.10 or higher
• External dependencies: httpx, tenacity, pydantic
• Installation command: pip install httpx tenacity pydantic

Authentication Setup

Both platforms require OAuth 2.0 client credentials flow. You must cache tokens and handle expiration before making API calls. The following function handles token retrieval and automatic refresh using httpx.

import httpx
import time
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from pydantic import BaseModel
import threading

class TokenCache:
    def __init__(self):
        self._tokens = {}
        self._lock = threading.Lock()

    def get(self, provider: str) -> dict | None:
        with self._lock:
            entry = self._tokens.get(provider)
            if entry and time.time() < entry["expires_at"]:
                return entry["token"]
            return None

    def set(self, provider: str, token: str, expires_in: int):
        with self._lock:
            self._tokens[provider] = {
                "token": token,
                "expires_at": time.time() + expires_in - 60  # Refresh 60s early
            }

token_cache = TokenCache()

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type(httpx.HTTPStatusError)
)
def fetch_graph_token(tenant_id: str, client_id: str, client_secret: str) -> str:
    """Fetches Microsoft Graph API access token."""
    url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
        "scope": "https://graph.microsoft.com/.default"
    }
    response = httpx.post(url, data=payload)
    response.raise_for_status()
    data = response.json()
    token_cache.set("graph", data["access_token"], data["expires_in"])
    return data["access_token"]

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type(httpx.HTTPStatusError)
)
def fetch_cxone_token(subdomain: str, client_id: str, client_secret: str) -> str:
    """Fetches NICE CXone access token. Required scope: scim:users:readwrite"""
    url = f"https://{subdomain}.api.nicecxone.com/oauth/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret
    }
    response = httpx.post(url, data=payload)
    response.raise_for_status()
    data = response.json()
    token_cache.set("cxone", data["access_token"], data["expires_in"])
    return data["access_token"]

Implementation

Step 1: Poll Graph API for User Property Changes

Microsoft Graph provides an incremental delta query endpoint. You must include the ConsistencyLevel: eventual header and paginate using the @odata.deltaLink response field. This endpoint returns only changed or deleted users since the last query.

import httpx
from typing import Generator, Dict, Any

def poll_graph_delta_changes(token: str, delta_link: str | None = None) -> Generator[Dict[str, Any], None, None]:
    """Polls Graph API for user delta changes. Handles pagination via deltaLink."""
    url = delta_link or "https://graph.microsoft.com/v1.0/users/delta"
    headers = {
        "Authorization": f"Bearer {token}",
        "ConsistencyLevel": "eventual",
        "Content-Type": "application/json"
    }

    while url:
        response = httpx.get(url, headers=headers)
        response.raise_for_status()
        data = response.json()

        # Realistic response structure:
        # {
        #   "@odata.context": "https://graph.microsoft.com/v1.0/$metadata#users",
        #   "@odata.deltaLink": "https://graph.microsoft.com/v1.0/users/delta?$deltatoken=...",
        #   "value": [ { "id": "obj-id-1", "displayName": "Jane Doe", "userPrincipalName": "jane@contoso.com", "deleted": false } ]
        # }

        for user in data.get("value", []):
            yield user

        url = data.get("@odata.deltaLink")
        if url:
            # Store delta link for state persistence (handled in Step 4)
            print(f"Next delta link: {url}")

Step 2: Map AD Groups to CXone Roles Using a Bidirectional Lookup Table

Active Directory groups do not map directly to CXone roles. You must maintain a bidirectional configuration table that resolves memberOf IDs from Graph to CXone role identifiers. The following function performs the mapping and handles missing mappings gracefully.

from typing import List

# Configuration: Bidirectional lookup table
GROUP_ROLE_MAPPING = {
    "a1b2c3d4-e5f6-7890-abcd-ef1234567890": "cxone_role_agent",
    "b2c3d4e5-f6a7-8901-bcde-f12345678901": "cxone_role_supervisor",
    "cxone_role_agent": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
    "cxone_role_supervisor": "b2c3d4e5-f6a7-8901-bcde-f12345678901"
}

def map_ad_groups_to_cxone_roles(ad_group_ids: List[str]) -> List[str]:
    """Maps Azure AD group IDs to CXone role identifiers using bidirectional lookup."""
    cxone_roles = []
    for group_id in ad_group_ids:
        role_id = GROUP_ROLE_MAPPING.get(group_id)
        if role_id:
            cxone_roles.append(role_id)
        else:
            # Log unmapped groups in production. Skip to avoid sync failure.
            print(f"Warning: No CXone role mapping found for AD group {group_id}")
    return cxone_roles

Step 3: Resolve Attribute Conflicts and Construct Minimal SCIM PATCH Payloads

SCIM 2.0 PATCH operations require an Operations array. You must compare incoming Graph data against the current CXone state or your configured source of truth. The following function builds minimal payloads by only including changed fields, respecting a priority configuration.

import json
from typing import Dict, Any, Optional

# Configuration: Source of truth per attribute
ATTRIBUTE_SOURCE_OF_TRUTH = {
    "emails": "azure_ad",
    "name": "azure_ad",
    "active": "azure_ad",
    "roles": "cxone"  # Roles are managed internally in CXone, not overridden by AD
}

def resolve_and_build_scim_patch(
    graph_user: Dict[str, Any],
    cxone_user: Optional[Dict[str, Any]],
    external_id: str
) -> Optional[Dict[str, Any]]:
    """
    Resolves attribute conflicts based on configuration and constructs a minimal SCIM PATCH payload.
    Returns None if no changes are required.
    """
    operations = []

    # Resolve emails
    if ATTRIBUTE_SOURCE_OF_TRUTH.get("emails") == "azure_ad":
        new_email = graph_user.get("userPrincipalName")
        current_email = cxone_user.get("emails", [{}])[0].get("value") if cxone_user else None
        if new_email and new_email != current_email:
            operations.append({
                "op": "replace",
                "path": "emails[0].value",
                "value": new_email
            })

    # Resolve display name
    if ATTRIBUTE_SOURCE_OF_TRUTH.get("name") == "azure_ad":
        new_name = graph_user.get("displayName")
        current_name = cxone_user.get("name", {}).get("formatted") if cxone_user else None
        if new_name and new_name != current_name:
            operations.append({
                "op": "replace",
                "path": "name.formatted",
                "value": new_name
            })

    # Resolve active status
    if ATTRIBUTE_SOURCE_OF_TRUTH.get("active") == "azure_ad":
        # Graph uses accountEnabled, CXone uses active
        new_active = graph_user.get("accountEnabled", True)
        current_active = cxone_user.get("active", True) if cxone_user else True
        if new_active != current_active:
            operations.append({
                "op": "replace",
                "path": "active",
                "value": new_active
            })

    if not operations:
        return None

    # Realistic SCIM PATCH request payload:
    # POST/PATCH headers: Content-Type: application/scim+json
    # Body:
    # {
    #   "Operations": [
    #     { "op": "replace", "path": "emails[0].value", "value": "jane@contoso.com" },
    #     { "op": "replace", "path": "name.formatted", "value": "Jane Doe" }
    #   ]
    # }

    return {
        "Operations": operations
    }

Step 4: Maintain a State Store to Track Sync Progress and Handle Partial Failures

You must persist the @odata.deltaLink, processed user IDs, and failure logs to ensure idempotency and recovery. SQLite provides file-based locking and transaction safety. The following functions handle state initialization, success logging, and failure tracking.

import sqlite3
import time
from datetime import datetime, timezone

DB_PATH = "sync_state.db"

def init_state_store():
    """Creates SQLite table for tracking sync progress and delta links."""
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS sync_state (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id TEXT UNIQUE,
                last_synced_at TEXT,
                delta_link TEXT,
                status TEXT DEFAULT 'pending',
                error_message TEXT
            )
        """)
        conn.commit()

def update_state_success(user_id: str, delta_link: str):
    """Records successful synchronization and updates delta link."""
    timestamp = datetime.now(timezone.utc).isoformat()
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute(
            "UPDATE sync_state SET last_synced_at = ?, status = 'success', error_message = NULL WHERE user_id = ?",
            (timestamp, user_id)
        )
        conn.execute("UPDATE sync_state SET delta_link = ? WHERE id = (SELECT MIN(id) FROM sync_state)", (delta_link,))
        if conn.total_changes == 0:
            conn.execute(
                "INSERT INTO sync_state (user_id, last_synced_at, delta_link, status) VALUES (?, ?, ?, 'success')",
                (user_id, timestamp, delta_link)
            )
        conn.commit()

def update_state_failure(user_id: str, error: str):
    """Records partial failure for retry or manual review."""
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute(
            "UPDATE sync_state SET status = 'failed', error_message = ? WHERE user_id = ?",
            (error, user_id)
        )
        conn.commit()

def get_current_delta_link() -> str | None:
    """Retrieves the last processed delta link for resume capability."""
    with sqlite3.connect(DB_PATH) as conn:
        cursor = conn.execute("SELECT delta_link FROM sync_state ORDER BY id DESC LIMIT 1")
        row = cursor.fetchone()
        return row[0] if row else None

Complete Working Example

The following script combines all components into a runnable synchronization routine. It authenticates, polls Graph, resolves conflicts, applies SCIM PATCH operations, and updates the state store with retry logic for 429 rate limits.

import httpx
import sqlite3
import time
import json
from typing import Dict, Any, Optional
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

# Configuration constants
AZURE_TENANT_ID = "your-tenant-id"
AZURE_CLIENT_ID = "your-azure-client-id"
AZURE_CLIENT_SECRET = "your-azure-client-secret"

CXONE_SUBDOMAIN = "your-subdomain"
CXONE_CLIENT_ID = "your-cxone-client-id"
CXONE_CLIENT_SECRET = "your-cxone-client-secret"

DB_PATH = "sync_state.db"

# Reuse functions from previous steps (fetch_graph_token, fetch_cxone_token, 
# poll_graph_delta_changes, map_ad_groups_to_cxone_roles, 
# resolve_and_build_scim_patch, init_state_store, update_state_success, 
# update_state_failure, get_current_delta_link)

def fetch_cxone_user(token: str, external_id: str) -> Optional[Dict[str, Any]]:
    """Fetches current CXone user state for conflict resolution."""
    url = f"https://{CXONE_SUBDOMAIN}.api.nicecxone.com/scim/v2/Users/{external_id}"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/scim+json"
    }
    response = httpx.get(url, headers=headers)
    if response.status_code == 404:
        return None
    response.raise_for_status()
    return response.json()

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=2, min=4, max=60),
    retry=retry_if_exception_type(httpx.HTTPStatusError)
)
def apply_scim_patch(token: str, external_id: str, payload: Dict[str, Any]):
    """Sends minimal SCIM PATCH request to CXone. Handles 429 rate limiting."""
    url = f"https://{CXONE_SUBDOMAIN}.api.nicecxone.com/scim/v2/Users/{external_id}"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/scim+json"
    }
    # Realistic request:
    # PATCH /scim/v2/Users/obj-id-1 HTTP/1.1
    # Host: your-subdomain.api.nicecxone.com
    # Content-Type: application/scim+json
    # Authorization: Bearer <token>
    # Body: {"Operations": [{"op": "replace", "path": "emails[0].value", "value": "new@contoso.com"}]}
    response = httpx.patch(url, headers=headers, json=payload)
    response.raise_for_status()
    # Realistic response: 200 OK or 204 No Content with updated SCIM resource

def run_sync():
    """Main synchronization loop."""
    init_state_store()
    graph_token = fetch_graph_token(AZURE_TENANT_ID, AZURE_CLIENT_ID, AZURE_CLIENT_SECRET)
    cxone_token = fetch_cxone_token(CXONE_SUBDOMAIN, CXONE_CLIENT_ID, CXONE_CLIENT_SECRET)

    delta_link = get_current_delta_link()
    next_delta_link = None

    for graph_user in poll_graph_delta_changes(graph_token, delta_link):
        user_id = graph_user.get("id")
        if not user_id:
            continue

        # Handle deletions from Graph
        if graph_user.get("deleted"):
            print(f"User {user_id} deleted in Azure AD. Skipping CXone deactivation per config.")
            update_state_success(user_id, delta_link or "")
            continue

        # Fetch current CXone state for conflict resolution
        cxone_user = fetch_cxone_user(cxone_token, user_id)
        
        # Map groups if needed (roles are handled via config priority)
        _ = map_ad_groups_to_cxone_roles(graph_user.get("memberOf", []))

        # Build minimal PATCH payload
        patch_payload = resolve_and_build_scim_patch(graph_user, cxone_user, user_id)

        if patch_payload:
            try:
                apply_scim_patch(cxone_token, user_id, patch_payload)
                update_state_success(user_id, delta_link or "")
                print(f"Successfully synced user {user_id}")
            except httpx.HTTPStatusError as e:
                update_state_failure(user_id, str(e))
                print(f"Failed to sync user {user_id}: {e.response.status_code}")
        else:
            update_state_success(user_id, delta_link or "")
            print(f"No changes required for user {user_id}")

    # Update delta link in state store for next run
    if next_delta_link:
        with sqlite3.connect(DB_PATH) as conn:
            conn.execute("UPDATE sync_state SET delta_link = ? WHERE id = (SELECT MIN(id) FROM sync_state)", (next_delta_link,))
            conn.commit()

if __name__ == "__main__":
    run_sync()

Common Errors & Debugging

Error: 401 Unauthorized (Graph API)

• Cause: The User.Read.All scope is not granted by an Azure AD administrator, or the token has expired.
• Fix: Run fetch_graph_token() again. Verify that the app registration has User.Read.All under Application permissions and that admin consent is granted.
• Code showing the fix: The @retry decorator on fetch_graph_token automatically retries transient 401s during token issuance. Implement explicit token cache invalidation if you receive persistent 401s after cache expiry.

Error: 429 Too Many Requests (CXone SCIM API)

• Cause: CXone enforces rate limits per OAuth client. Bulk synchronization without backoff triggers cascading 429 responses.
• Fix: The apply_scim_patch function uses tenacity with exponential backoff. Ensure you do not parallelize PATCH requests beyond 10 concurrent connections per client ID.
• Code showing the fix: The @retry configuration on apply_scim_patch waits between 4 and 60 seconds on 429 responses. Add retry=retry_if_exception_type(httpx.HTTPStatusError) to catch rate limit headers.

Error: 400 Bad Request (SCIM Payload Format)

• Cause: The Content-Type header is not application/scim+json, or the Operations array contains invalid path syntax.
• Fix: Verify that path uses SCIM 2.0 filter syntax (e.g., emails[0].value, not emails.value). Ensure the request body is a valid JSON object with a top-level Operations key.
• Code showing the fix: The resolve_and_build_scim_patch function constructs paths explicitly. Replace path: "emails.value" with path: "emails[0].value" if CXone returns a 400.

Error: 404 Not Found (CXone User)

• Cause: The Graph user ID does not exist in CXone. SCIM PATCH requires an existing resource.
• Fix: Switch to SCIM PUT for creation or handle 404 by issuing a SCIM POST request instead.
• Code showing the fix: The fetch_cxone_user function returns None on 404. Extend resolve_and_build_scim_patch to return a POST payload when cxone_user is None and graph_user.get("accountEnabled") is true.

Official References

• Microsoft Graph API: Incremental Sync
• Microsoft Graph API: OAuth 2.0 Client Credentials Flow
• NICE CXone SCIM 2.0 API Documentation
• SCIM 2.0 Core Specification (RFC 7644)