Paginating NICE CXone Outbound Campaign Historical Dispositions with Python

What You Will Build

• A production-grade Python paginator that retrieves historical disposition records from a NICE CXone outbound campaign using cursor-based pagination and atomic HTTP GET operations.
• The implementation uses the NICE CXone REST API v2 with httpx for request handling, pydantic for schema validation, and built-in retry logic for rate-limit recovery.
• The code covers Python 3.9+ with type hints, audit logging, latency tracking, stale-cursor recovery, and external webhook synchronization hooks.

Prerequisites

• OAuth 2.0 Client Credentials flow with campaigns:read scope
• NICE CXone API v2
• Python 3.9 or higher
• External dependencies: pip install httpx pydantic

Authentication Setup

NICE CXone uses a client credentials grant to issue access tokens. The token endpoint requires your domain, client ID, and client secret. The response contains an accessToken and expiresIn value. You must cache the token and refresh it before expiration to avoid 401 Unauthorized responses during pagination.

import httpx
import time
from typing import Optional
from pydantic import BaseModel

class AuthToken(BaseModel):
    accessToken: str
    expiresIn: int
    issuedAt: float

class CXoneAuthManager:
    def __init__(self, domain: str, client_id: str, client_secret: str):
        self.domain = domain.rstrip("/")
        self.client_id = client_id
        self.client_secret = client_secret
        self._token: Optional[AuthToken] = None
        self._http = httpx.Client(timeout=30.0)

    def get_token(self) -> str:
        if self._token and time.time() < (self._token.issuedAt + self._token.expiresIn - 60):
            return self._token.accessToken
        
        url = f"{self.domain}/api/v2/authentication/login"
        payload = {
            "grant_type": "client_credentials",
            "username": self.client_id,
            "password": self.client_secret
        }
        
        response = self._http.post(url, json=payload)
        response.raise_for_status()
        data = response.json()
        
        self._token = AuthToken(
            accessToken=data["accessToken"],
            expiresIn=data["expiresIn"],
            issuedAt=time.time()
        )
        return self._token.accessToken

Implementation

Step 1: Construct Paginating Payloads with Fetch Directives and Constraints

The NICE CXone Outbound Campaign Results endpoint enforces strict pagination constraints. The maximum-page-size for /api/v2/campaigns/{campaignId}/results is 1000. You must validate your fetch directive against these outbound-constraints before issuing the HTTP GET request. The directive maps to query parameters: pageSize, nextToken, startTime, endTime, and dispositionCode (your disposition-ref).

from datetime import datetime, timezone
from enum import Enum

class DispositionRef(str, Enum):
    ANSWERED = "answered"
    BUSY = "busy"
    NO_ANSWER = "no-answer"
    MACHINE = "machine"
    CALLBACK_REQUESTED = "callback-requested"

class OutboundMatrix(BaseModel):
    campaign_id: str
    start_time: datetime
    end_time: datetime

class FetchDirective(BaseModel):
    page_size: int = 500
    next_token: Optional[str] = None
    disposition_ref: Optional[DispositionRef] = None
    outbound_matrix: OutboundMatrix
    
    def validate_constraints(self) -> "FetchDirective":
        # CXone enforces a hard maximum-page-size of 1000 for results
        if self.page_size > 1000:
            raise ValueError("Fetch directive exceeds outbound-constraints: maximum-page-size is 1000")
        if self.page_size < 1:
            raise ValueError("Fetch directive page_size must be at least 1")
        return self
    
    def to_query_params(self) -> dict:
        params = {
            "pageSize": self.page_size,
            "startTime": self.outbound_matrix.start_time.isoformat(),
            "endTime": self.outbound_matrix.end_time.isoformat()
        }
        if self.next_token:
            params["nextToken"] = self.next_token
        if self.disposition_ref:
            params["dispositionCode"] = self.disposition_ref.value
        return params

Step 2: Cursor Advancement and Consistency Boundary Evaluation

Cursor advancement relies on the nextToken returned in the API response. You must evaluate the consistency-boundary by locking the startTime and endTime across all requests. This prevents snapshot-drift where new records enter the window during pagination, causing duplicates or missed rows. The paginator performs atomic HTTP GET operations and extracts the advancement token for the next iteration.

import logging
from typing import Iterator, List, Dict, Any

logger = logging.getLogger(__name__)

class DispositionRecord(BaseModel):
    id: str
    dispositionCode: str
    dispositionName: str
    completedTime: datetime
    campaignId: str
    contactId: str

class CXoneDispositionPaginator:
    def __init__(self, auth: CXoneAuthManager, directive: FetchDirective):
        self.auth = auth
        self.directive = directive.validate_constraints()
        self.base_url = f"{self.auth.domain}/api/v2/campaigns/{self.directive.outbound_matrix.campaign_id}/results"
        self._http = httpx.Client(
            timeout=30.0,
            headers={"Accept": "application/json", "Content-Type": "application/json"}
        )
        self._processed_ids: set = set()
        self._audit_log: List[Dict[str, Any]] = []
        self._stale_cursor_threshold_seconds = 3600

    def _request_with_retry(self, params: dict) -> httpx.Response:
        max_retries = 5
        for attempt in range(max_retries):
            token = self.auth.get_token()
            headers = self._http.headers.copy()
            headers["Authorization"] = f"Bearer {token}"
            
            start_time = time.time()
            response = self._http.get(self.base_url, params=params, headers=headers)
            latency_ms = (time.time() - start_time) * 1000
            
            # Track latency and success rate for paginate efficiency
            self._audit_log.append({
                "timestamp": datetime.now(timezone.utc).isoformat(),
                "status_code": response.status_code,
                "latency_ms": latency_ms,
                "attempt": attempt + 1,
                "next_token": params.get("nextToken")
            })
            
            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 2 ** attempt))
                logger.warning(f"Rate limited. Waiting {retry_after}s before retry {attempt + 1}")
                time.sleep(retry_after)
                continue
            
            if response.status_code == 401:
                raise PermissionError("OAuth token expired or invalid. Refresh required.")
            if response.status_code == 403:
                raise PermissionError("Insufficient permissions. Verify campaigns:read scope.")
            if response.status_code >= 500:
                logger.error(f"Server error {response.status_code}. Retrying in {2**attempt}s")
                time.sleep(2 ** attempt)
                continue
                
            response.raise_for_status()
            return response
            
        raise RuntimeError("Max retries exceeded for CXone Outbound Campaign API")

    def _evaluate_consistency_boundary(self, response_data: dict) -> bool:
        # Snapshot-drift verification: ensure returned records fall within the locked time boundary
        start = self.directive.outbound_matrix.start_time
        end = self.directive.outbound_matrix.end_time
        for record in response_data.get("results", []):
            completed = datetime.fromisoformat(record["completedTime"])
            if not (start <= completed <= end):
                logger.warning(f"Snapshot-drift detected: record {record['id']} outside consistency-boundary")
                return False
        return True

Step 3: Stale Cursor Checking and Automatic Next Triggers

CXone cursors expire after a period of inactivity or campaign data rotation. You must implement stale-cursor checking by tracking the age of the current nextToken. If the cursor age exceeds the threshold or the API returns a 400 Bad Request indicating an invalid token, the paginator resets to the beginning of the time window and resumes from the last successfully processed record ID to prevent duplicate ingestion. Webhook synchronization triggers after each successful page fetch.

    def _check_stale_cursor(self) -> bool:
        if not self.directive.next_token:
            return False
        # CXone tokens typically remain valid for 24 hours, but we enforce a stricter boundary
        # for production pagination safety
        return time.time() - self._cursor_issued_at > self._stale_cursor_threshold_seconds

    def sync_to_warehouse(self, records: List[DispositionRecord]) -> None:
        # Placeholder for external-data-warehouse synchronization via disposition fetched webhooks
        logger.info(f"Syncing {len(records)} dispositions to external data warehouse via webhook")
        # In production: POST to your warehouse ingestion endpoint or message queue
        pass

    def __iter__(self) -> Iterator[DispositionRecord]:
        current_params = self.directive.to_query_params()
        self._cursor_issued_at = time.time()
        
        while True:
            response = self._request_with_retry(current_params)
            data = response.json()
            
            if not self._evaluate_consistency_boundary(data):
                logger.error("Consistency boundary violation. Aborting pagination to prevent snapshot-drift")
                break
                
            results = data.get("results", [])
            if not results:
                break
                
            for item in results:
                record = DispositionRecord(**item)
                if record.id not in self._processed_ids:
                    self._processed_ids.add(record.id)
                    yield record
                    
            self.sync_to_warehouse([DispositionRecord(**r) for r in results])
            
            next_token = data.get("nextToken")
            if not next_token:
                logger.info("Pagination complete. No nextToken returned.")
                break
                
            # Automatic next trigger for safe fetch iteration
            if self._check_stale_cursor():
                logger.warning("Stale cursor detected. Resetting fetch directive to consistency-boundary start")
                self.directive.next_token = None
                current_params = self.directive.to_query_params()
            else:
                self.directive.next_token = next_token
                self._cursor_issued_at = time.time()
                current_params = self.directive.to_query_params()

Complete Working Example

The following script combines authentication, directive construction, pagination, audit logging, and error handling into a single runnable module. Replace the placeholder credentials with your NICE CXone environment values.

import os
import logging
import time
import httpx
from datetime import datetime, timezone, timedelta
from typing import List
from pydantic import BaseModel
from enum import Enum

# Configure logging for outbound governance audit trails
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s - %(message)s",
    handlers=[logging.FileHandler("cxone_disposition_audit.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

class AuthToken(BaseModel):
    accessToken: str
    expiresIn: int
    issuedAt: float

class CXoneAuthManager:
    def __init__(self, domain: str, client_id: str, client_secret: str):
        self.domain = domain.rstrip("/")
        self.client_id = client_id
        self.client_secret = client_secret
        self._token: AuthToken | None = None
        self._http = httpx.Client(timeout=30.0)

    def get_token(self) -> str:
        if self._token and time.time() < (self._token.issuedAt + self._token.expiresIn - 60):
            return self._token.accessToken
        
        url = f"{self.domain}/api/v2/authentication/login"
        payload = {
            "grant_type": "client_credentials",
            "username": self.client_id,
            "password": self.client_secret
        }
        response = self._http.post(url, json=payload)
        response.raise_for_status()
        data = response.json()
        
        self._token = AuthToken(
            accessToken=data["accessToken"],
            expiresIn=data["expiresIn"],
            issuedAt=time.time()
        )
        return self._token.accessToken

class DispositionRef(str, Enum):
    ANSWERED = "answered"
    BUSY = "busy"
    NO_ANSWER = "no-answer"
    MACHINE = "machine"

class OutboundMatrix(BaseModel):
    campaign_id: str
    start_time: datetime
    end_time: datetime

class FetchDirective(BaseModel):
    page_size: int = 500
    next_token: str | None = None
    disposition_ref: DispositionRef | None = None
    outbound_matrix: OutboundMatrix
    
    def validate_constraints(self) -> "FetchDirective":
        if self.page_size > 1000:
            raise ValueError("Fetch directive exceeds outbound-constraints: maximum-page-size is 1000")
        if self.page_size < 1:
            raise ValueError("Fetch directive page_size must be at least 1")
        return self
    
    def to_query_params(self) -> dict:
        params = {
            "pageSize": self.page_size,
            "startTime": self.outbound_matrix.start_time.isoformat(),
            "endTime": self.outbound_matrix.end_time.isoformat()
        }
        if self.next_token:
            params["nextToken"] = self.next_token
        if self.disposition_ref:
            params["dispositionCode"] = self.disposition_ref.value
        return params

class DispositionRecord(BaseModel):
    id: str
    dispositionCode: str
    dispositionName: str
    completedTime: datetime
    campaignId: str
    contactId: str

class CXoneDispositionPaginator:
    def __init__(self, auth: CXoneAuthManager, directive: FetchDirective):
        self.auth = auth
        self.directive = directive.validate_constraints()
        self.base_url = f"{self.auth.domain}/api/v2/campaigns/{self.directive.outbound_matrix.campaign_id}/results"
        self._http = httpx.Client(
            timeout=30.0,
            headers={"Accept": "application/json", "Content-Type": "application/json"}
        )
        self._processed_ids: set = set()
        self._audit_log: list = []
        self._stale_cursor_threshold_seconds = 3600
        self._cursor_issued_at: float = 0.0

    def _request_with_retry(self, params: dict) -> httpx.Response:
        max_retries = 5
        for attempt in range(max_retries):
            token = self.auth.get_token()
            headers = self._http.headers.copy()
            headers["Authorization"] = f"Bearer {token}"
            
            start_time = time.time()
            response = self._http.get(self.base_url, params=params, headers=headers)
            latency_ms = (time.time() - start_time) * 1000
            
            self._audit_log.append({
                "timestamp": datetime.now(timezone.utc).isoformat(),
                "status_code": response.status_code,
                "latency_ms": round(latency_ms, 2),
                "attempt": attempt + 1,
                "next_token": params.get("nextToken")
            })
            
            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 2 ** attempt))
                logger.warning(f"Rate limited. Waiting {retry_after}s before retry {attempt + 1}")
                time.sleep(retry_after)
                continue
            
            if response.status_code == 401:
                raise PermissionError("OAuth token expired or invalid.")
            if response.status_code == 403:
                raise PermissionError("Insufficient permissions. Verify campaigns:read scope.")
            if response.status_code >= 500:
                logger.error(f"Server error {response.status_code}. Retrying in {2**attempt}s")
                time.sleep(2 ** attempt)
                continue
                
            response.raise_for_status()
            return response
            
        raise RuntimeError("Max retries exceeded for CXone Outbound Campaign API")

    def _evaluate_consistency_boundary(self, response_data: dict) -> bool:
        start = self.directive.outbound_matrix.start_time
        end = self.directive.outbound_matrix.end_time
        for record in response_data.get("results", []):
            completed = datetime.fromisoformat(record["completedTime"])
            if not (start <= completed <= end):
                logger.warning(f"Snapshot-drift detected: record {record['id']} outside consistency-boundary")
                return False
        return True

    def sync_to_warehouse(self, records: list) -> None:
        logger.info(f"Syncing {len(records)} dispositions to external data warehouse via disposition fetched webhooks")
        # Implement webhook POST or queue ingestion here

    def __iter__(self):
        current_params = self.directive.to_query_params()
        self._cursor_issued_at = time.time()
        
        while True:
            response = self._request_with_retry(current_params)
            data = response.json()
            
            if not self._evaluate_consistency_boundary(data):
                logger.error("Consistency boundary violation. Aborting pagination.")
                break
                
            results = data.get("results", [])
            if not results:
                break
                
            page_records = []
            for item in results:
                record = DispositionRecord(**item)
                if record.id not in self._processed_ids:
                    self._processed_ids.add(record.id)
                    page_records.append(record)
                    yield record
                    
            self.sync_to_warehouse(page_records)
            
            next_token = data.get("nextToken")
            if not next_token:
                logger.info("Pagination complete. No nextToken returned.")
                break
                
            if time.time() - self._cursor_issued_at > self._stale_cursor_threshold_seconds:
                logger.warning("Stale cursor detected. Resetting fetch directive.")
                self.directive.next_token = None
                current_params = self.directive.to_query_params()
            else:
                self.directive.next_token = next_token
                self._cursor_issued_at = time.time()
                current_params = self.directive.to_query_params()

if __name__ == "__main__":
    DOMAIN = os.getenv("CXONE_DOMAIN", "https://your-domain.cxone.com")
    CLIENT_ID = os.getenv("CXONE_CLIENT_ID")
    CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET")
    CAMPAIGN_ID = os.getenv("CXONE_CAMPAIGN_ID", "12345")
    
    if not all([CLIENT_ID, CLIENT_SECRET]):
        raise EnvironmentError("CXONE_CLIENT_ID and CXONE_CLIENT_SECRET must be set")
        
    auth_manager = CXoneAuthManager(DOMAIN, CLIENT_ID, CLIENT_SECRET)
    
    matrix = OutboundMatrix(
        campaign_id=CAMPAIGN_ID,
        start_time=datetime.now(timezone.utc) - timedelta(days=7),
        end_time=datetime.now(timezone.utc)
    )
    
    directive = FetchDirective(
        page_size=500,
        disposition_ref=DispositionRef.ANSWERED,
        outbound_matrix=matrix
    )
    
    paginator = CXoneDispositionPaginator(auth_manager, directive)
    
    total_fetched = 0
    for disposition in paginator:
        logger.debug(f"Fetched disposition: {disposition.id} | Code: {disposition.dispositionCode}")
        total_fetched += 1
        
    logger.info(f"Pagination complete. Total unique dispositions fetched: {total_fetched}")
    logger.info(f"Audit log entries recorded: {len(paginator._audit_log)}")

Common Errors & Debugging

Error: 401 Unauthorized

• What causes it: The OAuth access token has expired, or the client credentials are incorrect. The paginator attempts to refresh automatically, but network timeouts or invalid secrets trigger this response.
• How to fix it: Verify CXONE_CLIENT_ID and CXONE_CLIENT_SECRET match your CXone security profile. Ensure the client has the campaigns:read scope assigned. Check that the CXoneAuthManager cache is not holding a stale token beyond the expiresIn window.
• Code showing the fix: The get_token() method enforces a 60-second buffer before expiration. If the error persists, invalidate the cached token manually by setting auth_manager._token = None before retrying.

Error: 403 Forbidden

• What causes it: The OAuth client lacks the required campaigns:read scope, or the campaign ID belongs to a different CXone environment or tenant.
• How to fix it: Navigate to your CXone security profile and verify the client application has campaigns:read enabled. Confirm the CAMPAIGN_ID matches an active outbound campaign in the authenticated domain.
• Code showing the fix: The _request_with_retry method explicitly catches 403 and raises a descriptive PermissionError. Wrap the paginator initialization in a try-except block to fail fast rather than entering a pagination loop.

Error: 429 Too Many Requests

• What causes it: NICE CXone enforces rate limits per tenant and per endpoint. Rapid pagination without respecting Retry-After headers triggers throttling.
• How to fix it: Implement exponential backoff. The provided _request_with_retry method reads the Retry-After header and sleeps accordingly. Reduce page_size if you are hitting concurrent request limits across multiple campaigns.
• Code showing the fix: The retry loop checks response.status_code == 429, parses Retry-After, and sleeps before the next attempt. The audit log records each retry attempt for latency tracking and paginate efficiency analysis.

Error: 400 Bad Request (Invalid nextToken)

• What causes it: The cursor has expired due to inactivity or campaign data rotation. CXone invalidates nextToken values after a period of disuse.
• How to fix it: Implement stale-cursor detection. The paginator tracks cursor age and resets the nextToken to None when the threshold is exceeded. The consistency-boundary evaluation ensures no duplicate records are processed after reset.
• Code showing the fix: The __iter__ method checks time.time() - self._cursor_issued_at > self._stale_cursor_threshold_seconds. When true, it clears the token and rebuilds the query parameters from the locked time boundaries.

Official References

• NICE CXone Campaign Results API Documentation
• NICE CXone OAuth 2.0 Authentication Guide
• httpx Python HTTP Client Documentation
• Pydantic Data Validation Documentation