Configuring NICE CXone ICM Dial Plan Rules via API with Python

What You Will Build

• This tutorial builds a Python module that constructs, validates, and applies NICE CXone ICM dial plan rules with time scheduling, holiday calendar integration, and conflict detection.
• The implementation uses the NICE CXone ICM REST API endpoints under /icm/api/v1/ and the nice-cxone-python SDK for configuration management.
• The code is written in Python 3.9+ with type hints, requests for explicit HTTP cycles, and pydantic for schema validation.

Prerequisites

• OAuth client type: Confidential client (Client Credentials flow). Required scopes: icm:plan:read, icm:plan:write
• SDK version: nice-cxone-python>=3.0.0
• Language/runtime: Python 3.9 or higher
• External dependencies: requests>=2.31.0, pydantic>=2.0.0, python-dotenv>=1.0.0

Authentication Setup

NICE CXone uses OAuth 2.0 Client Credentials flow. The following class handles token acquisition, caching, and automatic refresh before expiration.

import os
import time
import requests
from typing import Optional

class CxoneOAuth:
    def __init__(self, tenant: str, client_id: str, client_secret: str):
        self.tenant = tenant
        self.client_id = client_id
        self.client_secret = client_secret
        self.access_token: Optional[str] = None
        self.expires_at: float = 0.0

    def get_access_token(self) -> str:
        if time.time() < self.expires_at - 60:
            return self.access_token

        token_url = f"https://{self.tenant}.nicecv.com/oauth/token"
        payload = {"grant_type": "client_credentials"}
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        auth = requests.auth.HTTPBasicAuth(self.client_id, self.client_secret)

        response = requests.post(token_url, data=payload, headers=headers, auth=auth, timeout=10)
        response.raise_for_status()

        token_data = response.json()
        self.access_token = token_data["access_token"]
        self.expires_at = time.time() + token_data["expires_in"]
        return self.access_token

Implementation

Step 1: Initialize Client and Fetch Target Plan

The ICM API requires you to retrieve the existing plan before modification to preserve versioning and prevent overwriting concurrent changes. The GET /icm/api/v1/plans/{planId} endpoint returns the full plan structure.

import requests
import logging
from typing import Dict, Any

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

class IcmPlanFetcher:
    def __init__(self, oauth: CxoneOAuth):
        self.oauth = oauth
        self.base_url = f"https://{oauth.tenant}.nicecv.com/icm/api/v1"

    def fetch_plan(self, plan_id: str) -> Dict[str, Any]:
        url = f"{self.base_url}/plans/{plan_id}"
        headers = {
            "Authorization": f"Bearer {self.oauth.get_access_token()}",
            "Accept": "application/json",
            "Content-Type": "application/json"
        }
        # OAuth Scope: icm:plan:read

        response = requests.get(url, headers=headers, timeout=15)
        if response.status_code == 404:
            raise ValueError(f"Plan {plan_id} not found in ICM engine.")
        response.raise_for_status()
        return response.json()

Step 2: Construct Rule Matrix Payload with Scheduling and Holiday References

Dial plan rules require a structured matrix containing time windows, holiday calendar references, target routing, and sequence ordering. The payload must match the ICM engine schema exactly.

from datetime import datetime
from pydantic import BaseModel, validator
from typing import List, Optional

class TimeWindow(BaseModel):
    type: str
    dayOfWeek: Optional[str] = None
    startTime: str
    endTime: str

    @validator("startTime", "endTime")
    def validate_time_format(cls, v: str) -> str:
        try:
            datetime.strptime(v, "%H:%M")
        except ValueError:
            raise ValueError("Time must be in HH:MM format (24-hour).")
        return v

class Rule(BaseModel):
    id: str
    name: str
    sequence: int
    enabled: bool = True
    timeWindows: List[TimeWindow]
    holidayCalendarId: Optional[str] = None
    targets: List[Dict[str, Any]]

class PlanPayload(BaseModel):
    id: str
    name: str
    rules: List[Rule]
    description: Optional[str] = None

Step 3: Validate Schema, Rule Limits, Precedence, and Conflict Detection

The ICM engine enforces strict constraints. You must validate maximum rule counts, unique sequence ordering, and overlapping time windows before submission. This step prevents 400 Bad Request failures during the atomic PUT operation.

class PlanValidator:
    MAX_RULES = 200

    @staticmethod
    def validate_plan(payload: PlanPayload) -> List[str]:
        errors: List[str] = []

        # Rule count limit
        if len(payload.rules) > PlanValidator.MAX_RULES:
            errors.append(f"Rule count {len(payload.rules)} exceeds maximum limit of {PlanValidator.MAX_RULES}.")

        # Precedence and sequence validation
        sequences = [r.sequence for r in payload.rules]
        if len(sequences) != len(set(sequences)):
            errors.append("Duplicate sequence values detected. Each rule must have a unique sequence number.")
        if sequences != sorted(sequences):
            errors.append("Rule sequences must be ordered sequentially starting from 1.")

        # Conflict detection: overlapping time windows with identical targets
        for i, rule_a in enumerate(payload.rules):
            for rule_b in payload.rules[i + 1:]:
                if rule_a.targets == rule_b.targets:
                    if PlanValidator._check_time_overlap(rule_a.timeWindows, rule_b.timeWindows):
                        errors.append(f"Conflict: Rules '{rule_a.name}' and '{rule_b.name}' have overlapping time windows and identical targets.")

        return errors

    @staticmethod
    def _check_time_overlap(windows_a: List[TimeWindow], windows_b: List[TimeWindow]) -> bool:
        for wa in windows_a:
            for wb in windows_b:
                if wa.dayOfWeek == wb.dayOfWeek:
                    if wa.startTime < wb.endTime and wb.startTime < wa.endTime:
                        return True
        return False

Step 4: Execute Atomic PUT and Trigger Rule Compilation

NICE CXone requires an atomic PUT /icm/api/v1/plans/{planId} to replace the entire plan, followed by a POST /icm/api/v1/plans/{planId}/apply to trigger the ICM engine compilation. This two-step process ensures deterministic routing and prevents partial configurations.

import time
from typing import Dict, Any

class IcmPlanApplicator:
    def __init__(self, oauth: CxoneOAuth):
        self.oauth = oauth
        self.base_url = f"https://{oauth.tenant}.nicecv.com/icm/api/v1"

    def apply_plan(self, plan_id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
        headers = {
            "Authorization": f"Bearer {self.oauth.get_access_token()}",
            "Accept": "application/json",
            "Content-Type": "application/json",
            "x-nice-icm-plan-version": payload.get("version", "1")
        }
        # OAuth Scope: icm:plan:write

        # Atomic PUT to persist plan structure
        put_url = f"{self.base_url}/plans/{plan_id}"
        put_start = time.time()
        put_response = requests.put(put_url, json=payload, headers=headers, timeout=20)
        put_latency = time.time() - put_start

        if put_response.status_code == 409:
            raise RuntimeError("Plan version conflict. Another process modified the plan. Fetch latest version and retry.")
        put_response.raise_for_status()

        # Trigger compilation via apply directive
        apply_url = f"{self.base_url}/plans/{plan_id}/apply"
        apply_start = time.time()
        apply_response = requests.post(apply_url, headers=headers, timeout=20)
        apply_latency = time.time() - apply_start

        apply_response.raise_for_status()

        return {
            "put_status": put_response.status_code,
            "apply_status": apply_response.status_code,
            "put_latency_ms": round(put_latency * 1000, 2),
            "apply_latency_ms": round(apply_latency * 1000, 2),
            "apply_result": apply_response.json()
        }

Step 5: Webhook Synchronization, Latency Tracking, and Audit Logging

External scheduling systems require synchronization when a plan is configured. NICE CXone emits icm.plan.configured webhooks. The following handler parses the webhook payload, logs the event for ICM governance, and tracks configuration efficiency metrics.

import json
from datetime import datetime, timezone

class IcmAuditTracker:
    def __init__(self):
        self.audit_log: List[Dict[str, Any]] = []
        self.success_count = 0
        self.total_attempts = 0

    def log_configuration(self, plan_id: str, result: Dict[str, Any], payload_hash: str) -> None:
        self.total_attempts += 1
        success = result.get("apply_status") == 200
        if success:
            self.success_count += 1

        audit_entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "plan_id": plan_id,
            "payload_hash": payload_hash,
            "put_latency_ms": result.get("put_latency_ms"),
            "apply_latency_ms": result.get("apply_latency_ms"),
            "success": success,
            "apply_result": result.get("apply_result")
        }
        self.audit_log.append(audit_entry)
        logging.info(f"Audit: Plan {plan_id} configured. Success={success}, Latency={result.get('apply_latency_ms')}ms")

    def get_efficiency_metrics(self) -> Dict[str, float]:
        if self.total_attempts == 0:
            return {"success_rate": 0.0, "avg_latency_ms": 0.0}
        latencies = [entry["apply_latency_ms"] for entry in self.audit_log if entry["success"]]
        avg_latency = sum(latencies) / len(latencies) if latencies else 0.0
        return {
            "success_rate": round(self.success_count / self.total_attempts * 100, 2),
            "avg_latency_ms": round(avg_latency, 2)
        }

def handle_webhook_sync(event_payload: Dict[str, Any]) -> None:
    """Processes icm.plan.configured webhook for external system alignment."""
    event_type = event_payload.get("type")
    if event_type != "icm.plan.configured":
        return

    plan_id = event_payload.get("data", {}).get("planId")
    logging.info(f"Webhook sync triggered for plan {plan_id}. External scheduling alignment initiated.")
    # Insert external system API call here (e.g., update CRM routing rules, sync shift schedules)

Complete Working Example

The following script combines all components into a runnable dial plan configurator. Replace the environment variables with your NICE CXone tenant credentials.

import os
import hashlib
import json
from dotenv import load_dotenv

load_dotenv()

def generate_payload_hash(payload: Dict[str, Any]) -> str:
    return hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()

def main():
    tenant = os.getenv("CXONE_TENANT")
    client_id = os.getenv("CXONE_CLIENT_ID")
    client_secret = os.getenv("CXONE_CLIENT_SECRET")
    plan_id = os.getenv("CXONE_PLAN_ID")

    if not all([tenant, client_id, client_secret, plan_id]):
        raise EnvironmentError("Missing required environment variables.")

    oauth = CxoneOAuth(tenant, client_id, client_secret)
    fetcher = IcmPlanFetcher(oauth)
    applicator = IcmPlanApplicator(oauth)
    auditor = IcmAuditTracker()

    # Step 1: Fetch existing plan
    existing_plan = fetcher.fetch_plan(plan_id)

    # Step 2: Construct new rule matrix
    new_rules = [
        Rule(
            id="rule-001",
            name="Business Hours Primary",
            sequence=1,
            enabled=True,
            timeWindows=[
                TimeWindow(type="weekday", dayOfWeek="MONDAY", startTime="09:00", endTime="17:00"),
                TimeWindow(type="weekday", dayOfWeek="TUESDAY", startTime="09:00", endTime="17:00")
            ],
            holidayCalendarId="cal-us-federal-2024",
            targets=[{"id": "target-sales-queue", "weight": 100}]
        ),
        Rule(
            id="rule-002",
            name="After Hours Overflow",
            sequence=2,
            enabled=True,
            timeWindows=[
                TimeWindow(type="all", startTime="00:00", endTime="23:59")
            ],
            holidayCalendarId=None,
            targets=[{"id": "target-voicemail", "weight": 100}]
        )
    ]

    plan_payload = PlanPayload(
        id=plan_id,
        name="Automated Sales Dial Plan",
        rules=new_rules,
        description="Managed via ICM API automation pipeline"
    )

    # Step 3: Validate
    validation_errors = PlanValidator.validate_plan(plan_payload)
    if validation_errors:
        logging.error(f"Validation failed: {validation_errors}")
        return

    # Prepare JSON payload for API
    api_payload = plan_payload.dict()
    payload_hash = generate_payload_hash(api_payload)

    # Step 4: Apply
    try:
        result = applicator.apply_plan(plan_id, api_payload)
        auditor.log_configuration(plan_id, result, payload_hash)
        metrics = auditor.get_efficiency_metrics()
        logging.info(f"Configuration complete. Success Rate: {metrics['success_rate']}%, Avg Latency: {metrics['avg_latency_ms']}ms")
    except requests.exceptions.HTTPError as e:
        logging.error(f"API Error during apply: {e.response.status_code} - {e.response.text}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 400 Bad Request

• What causes it: The payload violates ICM engine schema constraints, contains invalid time formats, or references a non-existent holiday calendar ID.
• How to fix it: Verify startTime and endTime use HH:MM format. Ensure holidayCalendarId matches an active calendar in your tenant. Run the PlanValidator locally before submission.
• Code showing the fix: The TimeWindow Pydantic model enforces %H:%M validation. The PlanValidator checks sequence ordering and rule count limits before the HTTP request is sent.

Error: 409 Conflict

• What causes it: The x-nice-icm-plan-version header does not match the current server version, indicating concurrent modification.
• How to fix it: Implement an optimistic locking retry loop. Fetch the latest plan version, merge your rule changes, increment the version header, and retry the PUT.
• Code showing the fix: The IcmPlanApplicator raises a RuntimeError on 409. Wrap the apply_plan call in a retry decorator that re-fetches via IcmPlanFetcher before retrying.

Error: 429 Too Many Requests

• What causes it: Exceeding the ICM API rate limit (typically 50 requests per minute per client).
• How to fix it: Implement exponential backoff with jitter. The requests library does not include this by default, so you must add it to the HTTP call layer.
• Code showing the fix: Add a retry loop around requests.put and requests.post that sleeps for 2 ** attempt + random.uniform(0, 1) seconds when response.status_code == 429.

Error: 500 Internal Server Error

• What causes it: ICM engine compilation failure due to ambiguous routing rules or target capacity limits.
• How to fix it: Review the apply_result payload for compilationErrors. Check that target queues exist and have assigned agents. Ensure no two rules route identical caller patterns to conflicting targets during overlapping windows.

Official References

• NICE CXone ICM API Reference
• nice-cxone-python SDK Repository
• OAuth 2.0 Client Credentials Flow Documentation
• ICM Plan and Rule Schema Specification