Segmenting NICE CXone Outbound Contact Lists Dynamically with Python

What You Will Build

• A Python script that extracts historical interaction data, calculates recency and frequency scores, creates segmented contact lists, and launches predictive campaigns with dynamic abandonment thresholds.
• This implementation uses the NICE CXone REST APIs for Interactions, Lists, and Campaigns.
• The tutorial covers Python 3.9+ with the requests library and standard type hints.

Prerequisites

• OAuth 2.0 Client Credentials grant configured in CXone with the following scopes: interactions:read, lists:read, lists:write, campaigns:read, campaigns:write
• CXone API version 1 (v1)
• Python 3.9 or higher
• External dependencies: requests>=2.28.0, urllib3>=1.26.0
• Valid CXone tenant URL (e.g., https://api.mynicecx.com or tenant-specific endpoint)

Authentication Setup

CXone uses standard OAuth 2.0 Client Credentials flow. The token expires after one hour and must be refreshed before expiration. The code below implements a session-based client with automatic retry logic for rate limits and token caching.

import time
import logging
import requests
from typing import Dict, Optional
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter

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

class CXoneClient:
    def __init__(self, tenant_url: str, client_id: str, client_secret: str):
        self.tenant_url = tenant_url.rstrip("/")
        self.client_id = client_id
        self.client_secret = client_secret
        self.token: Optional[str] = None
        self.token_expiry: float = 0.0
        self.session = self._build_session()

    def _build_session(self) -> requests.Session:
        session = requests.Session()
        retry_strategy = Retry(
            total=3,
            backoff_factor=1.5,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["HEAD", "GET", "OPTIONS", "POST", "PUT", "PATCH"]
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("https://", adapter)
        session.mount("http://", adapter)
        return session

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

        url = f"{self.tenant_url}/oauth/token"
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": "interactions:read lists:read lists:write campaigns:read campaigns:write"
        }
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        response = self.session.post(url, data=payload, headers=headers)
        response.raise_for_status()
        data = response.json()
        self.token = data["access_token"]
        self.token_expiry = time.time() + data["expires_in"]
        return self.token

    def _make_request(self, method: str, path: str, **kwargs) -> requests.Response:
        headers = kwargs.pop("headers", {})
        headers["Authorization"] = f"Bearer {self._get_token()}"
        headers["Content-Type"] = "application/json"
        url = f"{self.tenant_url}{path}"
        return self.session.request(method, url, headers=headers, **kwargs)

The _build_session method attaches an HTTPAdapter with a Retry strategy. This automatically handles 429 Too Many Requests responses with exponential backoff, which prevents cascade failures during bulk list operations. The _get_token method caches the token and refreshes it only when expiration approaches within a five-minute window.

Implementation

Step 1: Query Interaction API for Historical Engagement Metrics

The CXone Interaction API requires a POST request to /api/v1/interactions/search. The request body accepts filter criteria and pagination tokens. You must specify the time window and interaction types to retrieve relevant outbound history.

from datetime import datetime, timedelta
from typing import List, Dict, Any

def fetch_interactions(client: CXoneClient, contact_ids: List[str], days_back: int = 90) -> List[Dict[str, Any]]:
    """
    Retrieves historical interactions for a set of contacts.
    Required scope: interactions:read
    """
    start_date = (datetime.utcnow() - timedelta(days=days_back)).isoformat() + "Z"
    end_date = datetime.utcnow().isoformat() + "Z"

    all_interactions: List[Dict[str, Any]] = []
    next_token: Optional[str] = None

    while True:
        payload = {
            "filters": {
                "contactIds": contact_ids,
                "startTime": start_date,
                "endTime": end_date,
                "interactionType": ["voice"]
            },
            "pageSize": 500,
            "pageToken": next_token
        }

        response = client._make_request("POST", "/api/v1/interactions/search", json=payload)
        if response.status_code == 429:
            logger.warning("Rate limited on interaction query. Retrying...")
            time.sleep(2)
            continue
        response.raise_for_status()

        data = response.json()
        results = data.get("data", [])
        all_interactions.extend(results)

        pagination = data.get("pagination", {})
        next_token = pagination.get("nextPageToken")
        if not next_token:
            break

    logger.info(f"Retrieved {len(all_interactions)} interactions for {len(contact_ids)} contacts.")
    return all_interactions

The API returns a data array containing interaction objects and a pagination object with nextPageToken. The loop continues until nextPageToken is null. Each interaction object contains id, startTime, contactId, and type. The 500 record page size balances throughput with memory usage.

Step 2: Apply Scoring Algorithm Based on Recency and Frequency Attributes

You must transform raw interaction timestamps into actionable scores. The algorithm calculates recency as days since the last interaction and frequency as the total interaction count within the window. A weighted score determines segment priority.

from collections import defaultdict

def calculate_segment_scores(interactions: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]:
    """
    Computes recency, frequency, and priority scores per contact.
    Returns a dictionary keyed by contactId.
    """
    contact_metrics: Dict[str, Dict[str, Any]] = defaultdict(lambda: {
        "frequency": 0,
        "last_interaction": None,
        "score": 0.0,
        "priority": "low"
    })

    now = datetime.utcnow()
    for interaction in interactions:
        contact_id = interaction.get("contactId")
        if not contact_id:
            continue

        start_time_str = interaction.get("startTime")
        if not start_time_str:
            continue

        start_time = datetime.fromisoformat(start_time_str.replace("Z", "+00:00"))
        days_since = (now - start_time).total_seconds() / 86400.0

        metrics = contact_metrics[contact_id]
        metrics["frequency"] += 1

        if metrics["last_interaction"] is None or start_time > metrics["last_interaction"]:
            metrics["last_interaction"] = start_time
            metrics["days_since"] = days_since

    for contact_id, metrics in contact_metrics.items():
        # Recency score: 0-50 points (lower days_since = higher score)
        recency_score = max(0, 50 - (metrics["days_since"] * 0.5))
        # Frequency score: 0-50 points (capped at 20 interactions)
        freq_score = min(metrics["frequency"], 20) * 2.5
        metrics["score"] = recency_score + freq_score

        if metrics["score"] >= 70:
            metrics["priority"] = "high"
        elif metrics["score"] >= 40:
            metrics["priority"] = "medium"
        else:
            metrics["priority"] = "low"

    return dict(contact_metrics)

The scoring formula weights recency and frequency equally at 50 points each. Recency decays linearly by 0.5 points per day. Frequency caps at 20 interactions to prevent outliers from dominating the score. The priority thresholds map directly to campaign abandonment thresholds in the next step.

Step 3: Generate Filtered Contact Subsets Using the List API

CXone requires explicit list creation before contact assignment. You will create three lists corresponding to the priority segments, then batch-add contacts to each list. The List API accepts up to 1,000 contacts per request, but 500 provides safer payload sizes.

def create_segment_lists(client: CXoneClient, prefix: str) -> Dict[str, str]:
    """
    Creates three outbound lists for high, medium, and low priority segments.
    Required scope: lists:write
    """
    list_ids = {}
    for priority in ["high", "medium", "low"]:
        payload = {
            "name": f"{prefix}_segment_{priority}",
            "description": f"Dynamic segment for {priority} priority contacts",
            "type": "outbound"
        }
        response = client._make_request("POST", "/api/v1/lists", json=payload)
        response.raise_for_status()
        list_ids[priority] = response.json()["id"]
        logger.info(f"Created list {list_ids[priority]} for {priority} priority.")
    return list_ids

def assign_contacts_to_lists(client: CXoneClient, list_ids: Dict[str, str], metrics: Dict[str, Dict[str, Any]]) -> None:
    """
    Batches contacts into their respective priority lists.
    Required scope: lists:write
    """
    batches: Dict[str, List[Dict[str, Any]]] = {k: [] for k in list_ids}

    for contact_id, data in metrics.items():
        contact_payload = {
            "contactId": contact_id,
            "data": {
                "priority": data["priority"],
                "score": data["score"]
            }
        }
        batches[data["priority"]].append(contact_payload)

    for priority, contacts in batches.items():
        list_id = list_ids[priority]
        for i in range(0, len(contacts), 500):
            batch = contacts[i:i+500]
            response = client._make_request("POST", f"/api/v1/lists/{list_id}/contacts", json=batch)
            if response.status_code == 429:
                logger.warning("Rate limited on list assignment. Backing off...")
                time.sleep(2)
                continue
            response.raise_for_status()
            logger.info(f"Assigned {len(batch)} contacts to {priority} list.")

The contactId field must match the identifier used in the interaction history. The data object stores custom attributes that CXone preserves for reporting. Batch processing prevents payload size errors and distributes load across the API gateway.

Step 4: Trigger Predictive Dialing Campaigns with Adjusted Abandonment Thresholds

Campaign creation requires explicit dial type configuration and abandonment thresholds. You will map segment priority to threshold values: high priority receives 3%, medium receives 5%, and low receives 8%. The campaign payload must include start time, list ID, and wrap-up configuration.

def launch_predictive_campaigns(client: CXoneClient, list_ids: Dict[str, str], prefix: str) -> Dict[str, str]:
    """
    Creates predictive campaigns with priority-based abandonment thresholds.
    Required scope: campaigns:write
    """
    threshold_map = {
        "high": 3.0,
        "medium": 5.0,
        "low": 8.0
    }

    campaign_ids = {}
    start_time = (datetime.utcnow() + timedelta(minutes=15)).isoformat() + "Z"
    end_time = (datetime.utcnow() + timedelta(hours=8)).isoformat() + "Z"

    for priority, list_id in list_ids.items():
        payload = {
            "name": f"{prefix}_campaign_{priority}",
            "dialType": "predictive",
            "listId": list_id,
            "abandonThreshold": threshold_map[priority],
            "startTime": start_time,
            "endTime": end_time,
            "wrapUpCode": "Campaign Complete",
            "status": "active",
            "predictiveDialingSettings": {
                "targetAnswerRate": 0.85,
                "maxCallsInProgress": 50,
                "agentAvailabilityFactor": 0.9
            }
        }

        response = client._make_request("POST", "/api/v1/campaigns", json=payload)
        if response.status_code == 429:
            logger.warning("Rate limited on campaign creation. Retrying...")
            time.sleep(2)
            continue
        response.raise_for_status()
        campaign_ids[priority] = response.json()["id"]
        logger.info(f"Launched campaign {campaign_ids[priority]} with {threshold_map[priority]}% abandon threshold.")

    return campaign_ids

The abandonThreshold field accepts a percentage value. Predictive dialing requires predictiveDialingSettings to define answer rate targets and concurrency limits. The status field set to active immediately queues the campaign for execution. CXone validates list population before activation, so list assignment must complete successfully.

Complete Working Example

The following script integrates all components into a single executable module. Replace the placeholder credentials before execution.

import time
import logging
import requests
from typing import Dict, List, Optional
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
from datetime import datetime, timedelta
from collections import defaultdict

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

class CXoneClient:
    def __init__(self, tenant_url: str, client_id: str, client_secret: str):
        self.tenant_url = tenant_url.rstrip("/")
        self.client_id = client_id
        self.client_secret = client_secret
        self.token: Optional[str] = None
        self.token_expiry: float = 0.0
        self.session = self._build_session()

    def _build_session(self) -> requests.Session:
        session = requests.Session()
        retry_strategy = Retry(
            total=3,
            backoff_factor=1.5,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["HEAD", "GET", "OPTIONS", "POST", "PUT", "PATCH"]
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("https://", adapter)
        session.mount("http://", adapter)
        return session

    def _get_token(self) -> str:
        if self.token and time.time() < self.token_expiry - 300:
            return self.token
        url = f"{self.tenant_url}/oauth/token"
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": "interactions:read lists:read lists:write campaigns:read campaigns:write"
        }
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        response = self.session.post(url, data=payload, headers=headers)
        response.raise_for_status()
        data = response.json()
        self.token = data["access_token"]
        self.token_expiry = time.time() + data["expires_in"]
        return self.token

    def _make_request(self, method: str, path: str, **kwargs) -> requests.Response:
        headers = kwargs.pop("headers", {})
        headers["Authorization"] = f"Bearer {self._get_token()}"
        headers["Content-Type"] = "application/json"
        url = f"{self.tenant_url}{path}"
        return self.session.request(method, url, headers=headers, **kwargs)

def fetch_interactions(client: CXoneClient, contact_ids: List[str], days_back: int = 90) -> List[Dict]:
    start_date = (datetime.utcnow() - timedelta(days=days_back)).isoformat() + "Z"
    end_date = datetime.utcnow().isoformat() + "Z"
    all_interactions = []
    next_token = None
    while True:
        payload = {
            "filters": {
                "contactIds": contact_ids,
                "startTime": start_date,
                "endTime": end_date,
                "interactionType": ["voice"]
            },
            "pageSize": 500,
            "pageToken": next_token
        }
        response = client._make_request("POST", "/api/v1/interactions/search", json=payload)
        if response.status_code == 429:
            logger.warning("Rate limited on interaction query. Retrying...")
            time.sleep(2)
            continue
        response.raise_for_status()
        data = response.json()
        all_interactions.extend(data.get("data", []))
        next_token = data.get("pagination", {}).get("nextPageToken")
        if not next_token:
            break
    return all_interactions

def calculate_segment_scores(interactions: List[Dict]) -> Dict[str, Dict]:
    contact_metrics = defaultdict(lambda: {"frequency": 0, "last_interaction": None, "score": 0.0, "priority": "low"})
    now = datetime.utcnow()
    for interaction in interactions:
        contact_id = interaction.get("contactId")
        start_time_str = interaction.get("startTime")
        if not contact_id or not start_time_str:
            continue
        start_time = datetime.fromisoformat(start_time_str.replace("Z", "+00:00"))
        days_since = (now - start_time).total_seconds() / 86400.0
        metrics = contact_metrics[contact_id]
        metrics["frequency"] += 1
        if metrics["last_interaction"] is None or start_time > metrics["last_interaction"]:
            metrics["last_interaction"] = start_time
            metrics["days_since"] = days_since
    for contact_id, metrics in contact_metrics.items():
        recency_score = max(0, 50 - (metrics["days_since"] * 0.5))
        freq_score = min(metrics["frequency"], 20) * 2.5
        metrics["score"] = recency_score + freq_score
        metrics["priority"] = "high" if metrics["score"] >= 70 else ("medium" if metrics["score"] >= 40 else "low")
    return dict(contact_metrics)

def create_segment_lists(client: CXoneClient, prefix: str) -> Dict[str, str]:
    list_ids = {}
    for priority in ["high", "medium", "low"]:
        payload = {"name": f"{prefix}_segment_{priority}", "description": f"Dynamic segment for {priority} priority contacts", "type": "outbound"}
        response = client._make_request("POST", "/api/v1/lists", json=payload)
        response.raise_for_status()
        list_ids[priority] = response.json()["id"]
    return list_ids

def assign_contacts_to_lists(client: CXoneClient, list_ids: Dict[str, str], metrics: Dict[str, Dict]) -> None:
    batches = {k: [] for k in list_ids}
    for contact_id, data in metrics.items():
        batches[data["priority"]].append({"contactId": contact_id, "data": {"priority": data["priority"], "score": data["score"]}})
    for priority, contacts in batches.items():
        list_id = list_ids[priority]
        for i in range(0, len(contacts), 500):
            batch = contacts[i:i+500]
            response = client._make_request("POST", f"/api/v1/lists/{list_id}/contacts", json=batch)
            if response.status_code == 429:
                time.sleep(2)
                continue
            response.raise_for_status()

def launch_predictive_campaigns(client: CXoneClient, list_ids: Dict[str, str], prefix: str) -> Dict[str, str]:
    threshold_map = {"high": 3.0, "medium": 5.0, "low": 8.0}
    campaign_ids = {}
    start_time = (datetime.utcnow() + timedelta(minutes=15)).isoformat() + "Z"
    end_time = (datetime.utcnow() + timedelta(hours=8)).isoformat() + "Z"
    for priority, list_id in list_ids.items():
        payload = {
            "name": f"{prefix}_campaign_{priority}", "dialType": "predictive", "listId": list_id,
            "abandonThreshold": threshold_map[priority], "startTime": start_time, "endTime": end_time,
            "wrapUpCode": "Campaign Complete", "status": "active",
            "predictiveDialingSettings": {"targetAnswerRate": 0.85, "maxCallsInProgress": 50, "agentAvailabilityFactor": 0.9}
        }
        response = client._make_request("POST", "/api/v1/campaigns", json=payload)
        if response.status_code == 429:
            time.sleep(2)
            continue
        response.raise_for_status()
        campaign_ids[priority] = response.json()["id"]
    return campaign_ids

if __name__ == "__main__":
    TENANT_URL = "https://api.mynicecx.com"
    CLIENT_ID = "YOUR_CLIENT_ID"
    CLIENT_SECRET = "YOUR_CLIENT_SECRET"
    CONTACT_IDS = ["contact_001", "contact_002", "contact_003"]
    PREFIX = "dynamic_outbound"

    client = CXoneClient(TENANT_URL, CLIENT_ID, CLIENT_SECRET)
    interactions = fetch_interactions(client, CONTACT_IDS, days_back=90)
    metrics = calculate_segment_scores(interactions)
    list_ids = create_segment_lists(client, PREFIX)
    assign_contacts_to_lists(client, list_ids, metrics)
    campaign_ids = launch_predictive_campaigns(client, list_ids, PREFIX)
    logger.info("Workflow complete. Campaign IDs: %s", campaign_ids)

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: Expired OAuth token or invalid client credentials.
• Fix: Verify the client ID and secret match the CXone OAuth application configuration. Ensure the token refresh logic runs before expiration. The provided client automatically refreshes tokens 300 seconds before expiry.
• Code Fix: Replace the hardcoded credentials with environment variables or a secrets manager. Never commit secrets to version control.

Error: 403 Forbidden

• Cause: Missing OAuth scopes or insufficient API permissions for the client application.
• Fix: Navigate to the CXone OAuth application settings and append interactions:read, lists:write, campaigns:write to the allowed scopes. Restart the script to trigger a fresh token request with the updated scope string.

Error: 429 Too Many Requests

• Cause: Exceeding tenant-level rate limits during bulk list assignments or interaction queries.
• Fix: The HTTPAdapter with Retry strategy handles automatic backoff. If persistent 429 errors occur, reduce batch sizes from 500 to 200 contacts per list assignment call. Add a fixed delay between campaign creation requests using time.sleep(1).

Error: 400 Bad Request on Campaign Creation

• Cause: Invalid abandonThreshold value or missing required predictive dialing parameters.
• Fix: Ensure abandonThreshold is a float between 1.0 and 10.0. Verify predictiveDialingSettings includes targetAnswerRate and maxCallsInProgress. CXone rejects campaigns that reference empty lists. Confirm list assignment completes before campaign creation.

Error: Pagination Token Exhaustion

• Cause: The Interaction API returns stale or corrupted nextPageToken values after network interruptions.
• Fix: Implement a maximum iteration counter to prevent infinite loops. Reset next_token to None and re-query from the start if the response lacks a data array. Log the total pages processed for audit trails.

Official References

• NICE CXone Interaction API Documentation
• NICE CXone List API Documentation
• NICE CXone Campaign API Documentation
• NICE CXone OAuth 2.0 Authentication Guide