Processing NICE CXone Web Messaging Bot Handoffs via REST API with Python

What You Will Build

This tutorial builds a Python service that programmatically triggers NICE CXone web messaging bot handoffs to human agent queues. It uses the NICE CXone Bot API and Conversations API to manage handoff payloads, validate eligibility, preserve context, and synchronize events. All code examples use Python 3.9+ with the requests library and production-grade error handling.

Prerequisites

• OAuth 2.0 Client Credentials flow with scopes: bot:manage, conversations:read, conversations:write, users:read, analytics:read
• CXone API version: v2 (standard REST interface)
• Python 3.9+ runtime
• External dependencies: requests, pydantic, python-dotenv
• A configured CXone environment with at least one active queue and a bot flow capable of triggering handoff events

Authentication Setup

CXone uses standard OAuth 2.0 Client Credentials for server-to-server communication. The authentication handler caches tokens and refreshes them before expiration to prevent unnecessary network calls.

import requests
import time
from typing import Optional
from requests.auth import HTTPBasicAuth

class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str, base_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = base_url.rstrip("/")
        self.token_url = f"{self.base_url}/api/v2/oauth2/token"
        self._token: Optional[str] = None
        self._expires_at: float = 0.0

    def get_token(self) -> str:
        if self._token and time.time() < self._expires_at - 60:
            return self._token
        
        payload = {
            "grant_type": "client_credentials",
            "scope": "bot:manage conversations:read conversations:write users:read analytics:read"
        }
        auth = HTTPBasicAuth(self.client_id, self.client_secret)
        response = requests.post(self.token_url, data=payload, auth=auth)
        
        if response.status_code == 401:
            raise RuntimeError("Invalid client credentials or misconfigured OAuth application.")
        if response.status_code == 403:
            raise RuntimeError("OAuth application lacks required scopes or is disabled.")
            
        response.raise_for_status()
        data = response.json()
        self._token = data["access_token"]
        self._expires_at = time.time() + data["expires_in"]
        return self._token

Implementation

Step 1: Construct Handoff Payloads with Queue Assignments and Context Transfer

The handoff payload must specify the target queue, transfer context, and preservation flags. CXone expects a JSON body with targetType, targetId, context, message, and preserveHistory.

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

class HandoffPayload(BaseModel):
    target_type: str = Field(default="queue", alias="targetType")
    target_id: str = Field(alias="targetId")
    context: Dict[str, Any]
    message: str
    preserve_history: bool = Field(default=True, alias="preserveHistory")
    conversation_id: str = Field(alias="conversationId")

    class Config:
        populate_by_name = True

Required scope: bot:manage
The payload structure ensures the receiving agent queue inherits the bot session metadata. The context object carries custom key-value pairs that CXone merges into the conversation transcript.

Step 2: Validate Handoff Eligibility Based on Confidence and Intent

Bot handoffs should only trigger when the bot confidence score falls below a threshold or when the user intent matches predefined escalation categories. This validation prevents unnecessary agent interruptions.

MIN_CONFIDENCE_THRESHOLD = 0.75
ALLOWED_ESCALATION_INTENTS = {"billing_dispute", "technical_failure", "complaint", "sales_escalation"}

def validate_handoff_eligibility(confidence_score: float, detected_intent: str) -> tuple[bool, str]:
    if confidence_score >= MIN_CONFIDENCE_THRESHOLD:
        return False, "Bot confidence exceeds escalation threshold."
    
    if detected_intent not in ALLOWED_ESCALATION_INTENTS:
        return False, f"Intent '{detected_intent}' is not configured for human handoff."
        
    return True, "Eligible for handoff."

Required scope: bot:manage (validation occurs server-side before API invocation)
The function returns a boolean and a diagnostic message. Integration layers should log the diagnostic message for audit trails.

Step 3: Preserve Conversation History and Execute Seamless Transition

Preserving history requires fetching the full conversation thread before initiating the handoff. The Conversations API supports pagination via nextPageToken. We fetch all interactions, attach them to the context, and execute the handoff.

def fetch_full_conversation_history(auth: CXoneAuth, conversation_id: str) -> list[Dict[str, Any]]:
    headers = {"Authorization": f"Bearer {auth.get_token()}"}
    interactions = []
    page_token = None
    
    while True:
        params = {"expand": "interactions", "pageSize": 50}
        if page_token:
            params["pageToken"] = page_token
            
        url = f"{auth.base_url}/api/v2/conversations/{conversation_id}"
        response = requests.get(url, headers=headers, params=params)
        response.raise_for_status()
        
        data = response.json()
        interactions.extend(data.get("interactions", []))
        
        page_token = data.get("nextPageToken")
        if not page_token:
            break
            
    return interactions

def execute_handoff(auth: CXoneAuth, payload: HandoffPayload) -> Dict[str, Any]:
    headers = {
        "Authorization": f"Bearer {auth.get_token()}",
        "Content-Type": "application/json"
    }
    url = f"{auth.base_url}/api/v2/bot/conversations/{payload.conversation_id}/handoff"
    
    # HTTP Request Cycle
    # Method: POST
    # Path: /api/v2/bot/conversations/{conversationId}/handoff
    # Headers: Authorization: Bearer <token>, Content-Type: application/json
    # Body: {"targetType":"queue","targetId":"q-123","context":{"orderId":"ORD-99"},"message":"Transferring to billing","preserveHistory":true,"conversationId":"c-456"}
    
    response = requests.post(url, headers=headers, json=payload.dict(by_alias=True))
    
    # Realistic Response Body
    # {
    #   "conversationId": "c-456",
    #   "handoffId": "h-789",
    #   "status": "transferring",
    #   "targetType": "queue",
    #   "targetId": "q-123",
    #   "timestamp": "2024-05-15T10:30:00Z"
    # }
    
    if response.status_code == 404:
        raise RuntimeError(f"Conversation {payload.conversation_id} not found.")
    if response.status_code == 422:
        raise RuntimeError(f"Invalid handoff payload: {response.text}")
        
    response.raise_for_status()
    return response.json()

Required scopes: conversations:read, bot:manage
The pagination loop ensures no message is dropped during context transfer. The handoff endpoint returns a handoffId that tracks the transition state.

Step 4: Handle Failures, Fallback Responses, and Retry Logic

Network throttling and transient 5xx errors require exponential backoff. The retry wrapper catches 429 and 5xx responses, applies delay, and falls back to a bot response if all attempts fail.

import time
from requests.exceptions import HTTPError

def execute_handoff_with_retry(auth: CXoneAuth, payload: HandoffPayload, max_retries: int = 3) -> Dict[str, Any]:
    for attempt in range(max_retries):
        try:
            return execute_handoff(auth, payload)
        except HTTPError as e:
            status_code = e.response.status_code if e.response else 500
            
            if status_code == 429:
                retry_after = int(e.response.headers.get("Retry-After", 2 ** attempt))
                time.sleep(retry_after)
                continue
                
            if 500 <= status_code < 600:
                time.sleep(2 ** attempt)
                continue
                
            raise
            
    # Fallback bot response when all retries are exhausted
    fallback_payload = {
        "targetType": "bot",
        "targetId": "fallback_queue",
        "context": {"error": "handoff_failed", "retry_count": max_retries},
        "message": "An agent will contact you shortly. Please remain on the line.",
        "preserveHistory": True,
        "conversationId": payload.conversation_id
    }
    return execute_handoff(auth, HandoffPayload(**fallback_payload))

Required scopes: bot:manage
The retry logic respects Retry-After headers for 429 responses. If the queue is unavailable after three attempts, the system routes the conversation to a fallback bot flow that queues a callback request.

Step 5: Synchronize CRM, Track Metrics, and Log Audit Trails

External CRM synchronization enriches the agent workspace with customer data. Latency tracking and resolution metrics feed operational dashboards. Audit logging captures every handoff decision for quality analysis.

import json
import logging
import time
from datetime import datetime, timezone

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

class HandoffMetrics:
    def __init__(self):
        self.latencies: list[float] = []
        self.success_count: int = 0
        self.failure_count: int = 0

    def record_latency(self, duration: float):
        self.latencies.append(duration)
        
    def get_resolution_rate(self) -> float:
        total = self.success_count + self.failure_count
        return (self.success_count / total) * 100 if total > 0 else 0.0

metrics = HandoffMetrics()

def sync_with_crm(conversation_id: str, context: Dict[str, Any]) -> bool:
    crm_url = "https://api.example-crm.com/v1/contacts/sync"
    headers = {"Authorization": "Bearer CRM_API_KEY", "Content-Type": "application/json"}
    payload = {
        "externalId": conversation_id,
        "attributes": context,
        "eventType": "BOT_HANDOFF"
    }
    response = requests.post(crm_url, headers=headers, json=payload)
    return response.status_code == 200

def log_audit_trail(handoff_id: str, conversation_id: str, status: str, context: Dict[str, Any]):
    audit_entry = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "handoffId": handoff_id,
        "conversationId": conversation_id,
        "status": status,
        "context": context,
        "latency_ms": metrics.latencies[-1] * 1000 if metrics.latencies else 0
    }
    logger.info(json.dumps(audit_entry))

Required scopes: analytics:read, users:read
The sync_with_crm function pushes handoff context to an external system. The HandoffMetrics class tracks latency and success rates. The audit logger writes structured JSON entries that integrate with SIEM or quality assurance platforms.

Step 6: Expose Handoff Simulator for Workflow Testing

The simulator mocks CXone responses to validate payload construction, eligibility logic, and retry behavior without consuming production API quotas.

from unittest.mock import patch, MagicMock
from requests.models import Response

class HandoffSimulator:
    def __init__(self, auth: CXoneAuth):
        self.auth = auth
        self.mock_responses: list[Dict[str, Any]] = []
        
    def simulate_handoff(self, payload: HandoffPayload, force_failure: bool = False) -> Dict[str, Any]:
        start_time = time.time()
        
        with patch("requests.post") as mock_post:
            response_obj = Response()
            response_obj.status_code = 503 if force_failure else 200
            response_obj._content = json.dumps({
                "conversationId": payload.conversation_id,
                "handoffId": "sim-h-001",
                "status": "transferring",
                "targetType": payload.target_type,
                "targetId": payload.target_id,
                "timestamp": datetime.now(timezone.utc).isoformat()
            }).encode()
            mock_post.return_value = response_obj
            
            result = execute_handoff_with_retry(self.auth, payload)
            duration = time.time() - start_time
            
            metrics.record_latency(duration)
            metrics.success_count += 1 if not force_failure else 0
            metrics.failure_count += 1 if force_failure else 0
            
            log_audit_trail(
                handoff_id=result.get("handoffId", "unknown"),
                conversation_id=payload.conversation_id,
                status="success" if not force_failure else "fallback_triggered",
                context=payload.context
            )
            
            return result

Required scopes: None (simulator runs offline)
The simulator patches requests.post to return deterministic responses. Developers run it against unit tests to verify payload serialization, retry timing, and audit log formatting before deploying to production environments.

Complete Working Example

import os
import sys
import json
import time
import logging
import requests
from typing import Dict, Any, Optional
from datetime import datetime, timezone
from requests.auth import HTTPBasicAuth
from pydantic import BaseModel, Field
from unittest.mock import patch, MagicMock
from requests.models import Response
from requests.exceptions import HTTPError

# --- Configuration ---
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("cxone_handoff")

MIN_CONFIDENCE_THRESHOLD = 0.75
ALLOWED_ESCALATION_INTENTS = {"billing_dispute", "technical_failure", "complaint", "sales_escalation"}

# --- Authentication ---
class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str, base_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = base_url.rstrip("/")
        self.token_url = f"{self.base_url}/api/v2/oauth2/token"
        self._token: Optional[str] = None
        self._expires_at: float = 0.0

    def get_token(self) -> str:
        if self._token and time.time() < self._expires_at - 60:
            return self._token
        payload = {
            "grant_type": "client_credentials",
            "scope": "bot:manage conversations:read conversations:write users:read analytics:read"
        }
        auth = HTTPBasicAuth(self.client_id, self.client_secret)
        response = requests.post(self.token_url, data=payload, auth=auth)
        if response.status_code == 401:
            raise RuntimeError("Invalid client credentials or misconfigured OAuth application.")
        if response.status_code == 403:
            raise RuntimeError("OAuth application lacks required scopes or is disabled.")
        response.raise_for_status()
        data = response.json()
        self._token = data["access_token"]
        self._expires_at = time.time() + data["expires_in"]
        return self._token

# --- Payload & Validation ---
class HandoffPayload(BaseModel):
    target_type: str = Field(default="queue", alias="targetType")
    target_id: str = Field(alias="targetId")
    context: Dict[str, Any]
    message: str
    preserve_history: bool = Field(default=True, alias="preserveHistory")
    conversation_id: str = Field(alias="conversationId")

    class Config:
        populate_by_name = True

def validate_handoff_eligibility(confidence_score: float, detected_intent: str) -> tuple[bool, str]:
    if confidence_score >= MIN_CONFIDENCE_THRESHOLD:
        return False, "Bot confidence exceeds escalation threshold."
    if detected_intent not in ALLOWED_ESCALATION_INTENTS:
        return False, f"Intent '{detected_intent}' is not configured for human handoff."
    return True, "Eligible for handoff."

# --- History & Execution ---
def fetch_full_conversation_history(auth: CXoneAuth, conversation_id: str) -> list[Dict[str, Any]]:
    headers = {"Authorization": f"Bearer {auth.get_token()}"}
    interactions = []
    page_token = None
    while True:
        params = {"expand": "interactions", "pageSize": 50}
        if page_token:
            params["pageToken"] = page_token
        url = f"{auth.base_url}/api/v2/conversations/{conversation_id}"
        response = requests.get(url, headers=headers, params=params)
        response.raise_for_status()
        data = response.json()
        interactions.extend(data.get("interactions", []))
        page_token = data.get("nextPageToken")
        if not page_token:
            break
    return interactions

def execute_handoff(auth: CXoneAuth, payload: HandoffPayload) -> Dict[str, Any]:
    headers = {
        "Authorization": f"Bearer {auth.get_token()}",
        "Content-Type": "application/json"
    }
    url = f"{auth.base_url}/api/v2/bot/conversations/{payload.conversation_id}/handoff"
    response = requests.post(url, headers=headers, json=payload.dict(by_alias=True))
    if response.status_code == 404:
        raise RuntimeError(f"Conversation {payload.conversation_id} not found.")
    if response.status_code == 422:
        raise RuntimeError(f"Invalid handoff payload: {response.text}")
    response.raise_for_status()
    return response.json()

def execute_handoff_with_retry(auth: CXoneAuth, payload: HandoffPayload, max_retries: int = 3) -> Dict[str, Any]:
    for attempt in range(max_retries):
        try:
            return execute_handoff(auth, payload)
        except HTTPError as e:
            status_code = e.response.status_code if e.response else 500
            if status_code == 429:
                retry_after = int(e.response.headers.get("Retry-After", 2 ** attempt))
                time.sleep(retry_after)
                continue
            if 500 <= status_code < 600:
                time.sleep(2 ** attempt)
                continue
            raise
    fallback_payload = {
        "targetType": "bot",
        "targetId": "fallback_queue",
        "context": {"error": "handoff_failed", "retry_count": max_retries},
        "message": "An agent will contact you shortly. Please remain on the line.",
        "preserveHistory": True,
        "conversationId": payload.conversation_id
    }
    return execute_handoff(auth, HandoffPayload(**fallback_payload))

# --- Metrics, CRM, Audit ---
class HandoffMetrics:
    def __init__(self):
        self.latencies: list[float] = []
        self.success_count: int = 0
        self.failure_count: int = 0
    def record_latency(self, duration: float):
        self.latencies.append(duration)
    def get_resolution_rate(self) -> float:
        total = self.success_count + self.failure_count
        return (self.success_count / total) * 100 if total > 0 else 0.0

metrics = HandoffMetrics()

def sync_with_crm(conversation_id: str, context: Dict[str, Any]) -> bool:
    crm_url = "https://api.example-crm.com/v1/contacts/sync"
    headers = {"Authorization": "Bearer CRM_API_KEY", "Content-Type": "application/json"}
    payload = {"externalId": conversation_id, "attributes": context, "eventType": "BOT_HANDOFF"}
    response = requests.post(crm_url, headers=headers, json=payload)
    return response.status_code == 200

def log_audit_trail(handoff_id: str, conversation_id: str, status: str, context: Dict[str, Any]):
    audit_entry = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "handoffId": handoff_id,
        "conversationId": conversation_id,
        "status": status,
        "context": context,
        "latency_ms": metrics.latencies[-1] * 1000 if metrics.latencies else 0
    }
    logger.info(json.dumps(audit_entry))

# --- Simulator ---
class HandoffSimulator:
    def __init__(self, auth: CXoneAuth):
        self.auth = auth
    def simulate_handoff(self, payload: HandoffPayload, force_failure: bool = False) -> Dict[str, Any]:
        start_time = time.time()
        with patch("requests.post") as mock_post:
            response_obj = Response()
            response_obj.status_code = 503 if force_failure else 200
            response_obj._content = json.dumps({
                "conversationId": payload.conversation_id,
                "handoffId": "sim-h-001",
                "status": "transferring",
                "targetType": payload.target_type,
                "targetId": payload.target_id,
                "timestamp": datetime.now(timezone.utc).isoformat()
            }).encode()
            mock_post.return_value = response_obj
            result = execute_handoff_with_retry(self.auth, payload)
            duration = time.time() - start_time
            metrics.record_latency(duration)
            metrics.success_count += 1 if not force_failure else 0
            metrics.failure_count += 1 if force_failure else 0
            log_audit_trail(
                handoff_id=result.get("handoffId", "unknown"),
                conversation_id=payload.conversation_id,
                status="success" if not force_failure else "fallback_triggered",
                context=payload.context
            )
            return result

# --- Entry Point ---
if __name__ == "__main__":
    CLIENT_ID = os.getenv("CXONE_CLIENT_ID")
    CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET")
    BASE_URL = os.getenv("CXONE_BASE_URL", "https://api-us-01.cxone.com")
    
    if not CLIENT_ID or not CLIENT_SECRET:
        logger.info("Credentials not provided. Running simulator.")
        auth = CXoneAuth("test_id", "test_secret", BASE_URL)
        sim = HandoffSimulator(auth)
        payload = HandoffPayload(
            targetId="q-billing-01",
            context={"orderId": "ORD-99281", "customerTier": "premium"},
            message="Transferring to billing specialist",
            conversationId="c-8812"
        )
        result = sim.simulate_handoff(payload, force_failure=False)
        logger.info(f"Simulator result: {json.dumps(result, indent=2)}")
    else:
        auth = CXoneAuth(CLIENT_ID, CLIENT_SECRET, BASE_URL)
        eligible, reason = validate_handoff_eligibility(0.62, "billing_dispute")
        if not eligible:
            logger.info(f"Handoff rejected: {reason}")
            sys.exit(0)
            
        payload = HandoffPayload(
            targetId="q-billing-01",
            context={"orderId": "ORD-99281", "customerTier": "premium"},
            message="Transferring to billing specialist",
            conversationId="c-8812"
        )
        result = execute_handoff_with_retry(auth, payload)
        sync_with_crm(payload.conversation_id, payload.context)
        log_audit_trail(result.get("handoffId"), payload.conversation_id, "completed", payload.context)
        logger.info(f"Production handoff completed: {json.dumps(result, indent=2)}")

Common Errors & Debugging

Error: 401 Unauthorized

• What causes it: Expired OAuth token, incorrect client credentials, or missing Authorization header.
• How to fix it: Verify the CXONE_CLIENT_ID and CXONE_CLIENT_SECRET environment variables. Ensure the token cache refreshes before expiration.
• Code showing the fix: The CXoneAuth.get_token() method checks time.time() < self._expires_at - 60 and automatically requests a new token when the window approaches.

Error: 403 Forbidden

• What causes it: The OAuth application lacks bot:manage or conversations:read scopes, or the tenant restricts API access to specific IP ranges.
• How to fix it: Navigate to the CXone Admin Console, open the OAuth application settings, and add the missing scopes. Whitelist your server IP if network restrictions are enforced.
• Code showing the fix: The authentication block explicitly raises RuntimeError("OAuth application lacks required scopes or is disabled.") on 403 responses to prevent silent failures.

Error: 429 Too Many Requests

• What causes it: Exceeding CXone rate limits (typically 100 requests per minute per tenant for bot endpoints).
• How to fix it: Implement exponential backoff and respect the Retry-After header.
• Code showing the fix: execute_handoff_with_retry catches 429, extracts Retry-After, sleeps for the specified duration, and resumes the loop.

Error: 422 Unprocessable Entity

• What causes it: Malformed handoff payload, invalid queue ID, or missing required fields like targetType.
• How to fix it: Validate the HandoffPayload against CXone schema requirements. Ensure targetId matches an active queue UUID.
• Code showing the fix: The execute_handoff function checks response.status_code == 422 and logs the exact validation error returned by CXone.

Official References

• NICE CXone Bot API Documentation
• NICE CXone Conversations API Documentation
• NICE CXone OAuth 2.0 Guide
• NICE CXone Rate Limiting & Error Codes