How to Trigger a CXone Outbound Call via the Personal Connection API

How to Trigger a CXone Outbound Call via the Personal Connection API

What You Will Build

  • You will build a script that initiates an outbound voice call to a specific phone number using the NICE CXone Personal Connection API.
  • This tutorial uses the CXone REST API (/api/v2/personal-connections/outbound-calls) to create and execute the call.
  • The implementation is provided in Python using the requests library.

Prerequisites

  • OAuth Client: A CXone OAuth Application with the personal-connections scope.
  • API Version: CXone API v2.
  • Language/ Runtime: Python 3.8+.
  • Dependencies: requests (install via pip install requests).
  • Configuration: You must have a valid client_id, client_secret, and the base URL for your CXone environment (e.g., https://us-east-1.api.nice.com).

Authentication Setup

CXone uses standard OAuth 2.0 Client Credentials flow for server-to-server API access. You must obtain an access token before making any API calls. The token expires after 3600 seconds (1 hour), so production code should implement caching or refresh logic.

Step 1: Obtain Access Token

The following function authenticates against the CXone Identity Provider.

import requests
import json
import time
from typing import Optional

# Configuration - Replace with your actual CXone environment details
CXONE_BASE_URL = "https://us-east-1.api.nice.com"  # Example: us-east-1, eu-west-1, etc.
CLIENT_ID = "your_client_id"
CLIENT_SECRET = "your_client_secret"

class CXoneAuth:
    def __init__(self, base_url: str, client_id: str, client_secret: str):
        self.base_url = base_url.rstrip('/')
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = f"{self.base_url}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0.0

    def get_access_token(self) -> str:
        """
        Retrieves an OAuth2 access token.
        Implements basic caching to avoid requesting a new token on every call.
        """
        # Check if we have a valid token
        if self.access_token and time.time() < self.token_expiry:
            return self.access_token

        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }

        try:
            response = requests.post(self.token_url, data=payload, headers=headers)
            response.raise_for_status()
            
            token_data = response.json()
            self.access_token = token_data["access_token"]
            # Set expiry to slightly before actual expiry to handle clock skew
            self.token_expiry = time.time() + token_data["expires_in"] - 60
            
            return self.access_token

        except requests.exceptions.HTTPError as e:
            print(f"Authentication failed: {e.response.status_code} - {e.response.text}")
            raise
        except requests.exceptions.RequestException as e:
            print(f"Network error during authentication: {e}")
            raise

OAuth Scope Requirement: The client application must have the personal-connections scope granted in the CXone Admin Console. Without this scope, the API will return a 403 Forbidden error.

Implementation

Step 2: Construct the Outbound Call Payload

The Personal Connection API allows you to trigger calls directly. The core endpoint is POST /api/v2/personal-connections/outbound-calls.

You must define:

  1. To: The recipient’s phone number (E.164 format recommended).
  2. From: The caller ID number associated with your CXone instance.
  3. Type: The type of connection (e.g., voice).
  4. Context: Optional metadata or IVR prompts if you are integrating with a flow. For a simple direct call, this can be minimal or empty depending on your use case.

Required Parameters

  • to: String. The destination phone number.
  • from: String. The source phone number (must be verified in CXone).
  • type: String. Must be voice for standard phone calls.

Optional but Recommended Parameters

  • context: Object. Can include prompt to play a message before connecting, or flowId to route into a specific CXone flow.
  • callbackUrl: String. URL to receive webhook events for call status (answered, missed, failed).

Step 3: Execute the Outbound Call

The following class wraps the API call logic, including error handling and retry logic for rate limits (429).

import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class CXonePersonalConnection:
    def __init__(self, auth: CXoneAuth):
        self.auth = auth
        self.api_base = f"{auth.base_url}/api/v2"

    def trigger_outbound_call(
        self, 
        to_number: str, 
        from_number: str, 
        callback_url: Optional[str] = None
    ) -> dict:
        """
        Triggers an outbound voice call via the Personal Connection API.
        
        Args:
            to_number: The recipient's phone number (E.164 format).
            from_number: The caller ID number (must be owned by the account).
            callback_url: Optional URL to receive call status webhooks.
            
        Returns:
            The JSON response from the API containing the connection ID.
        """
        url = f"{self.api_base}/personal-connections/outbound-calls"
        
        # Construct the payload
        payload = {
            "to": to_number,
            "from": from_number,
            "type": "voice"
        }

        if callback_url:
            payload["callbackUrl"] = callback_url

        headers = {
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

        access_token = self.auth.get_access_token()
        headers["Authorization"] = f"Bearer {access_token}"

        max_retries = 3
        retry_count = 0

        while retry_count < max_retries:
            try:
                logger.info(f"Initiating call from {from_number} to {to_number}")
                response = requests.post(url, json=payload, headers=headers, timeout=30)
                
                # Handle Rate Limiting (429)
                if response.status_code == 429:
                    retry_after = int(response.headers.get("Retry-After", 5))
                    logger.warning(f"Rate limited. Retrying in {retry_after} seconds...")
                    time.sleep(retry_after)
                    retry_count += 1
                    continue
                
                # Handle Success
                if response.status_code == 201:
                    result = response.json()
                    logger.info(f"Call triggered successfully. Connection ID: {result.get('id')}")
                    return result
                
                # Handle Other Errors
                response.raise_for_status()

            except requests.exceptions.HTTPError as e:
                logger.error(f"HTTP Error: {e.response.status_code} - {e.response.text}")
                raise
            except requests.exceptions.Timeout:
                logger.error("Request timed out.")
                raise
            except requests.exceptions.RequestException as e:
                logger.error(f"Request failed: {e}")
                raise

        logger.error("Max retries exceeded for outbound call.")
        raise Exception("Failed to trigger outbound call after retries.")

Step 4: Processing Results and Webhooks

When the call is successfully initiated, the API returns a 201 Created status with a JSON body containing the id of the personal connection.

Realistic Response Body:

{
  "id": "a1b2c3d4-e5f6-7890-g1h2-i3j4k5l6m7n8",
  "type": "voice",
  "to": "+15551234567",
  "from": "+15559876543",
  "status": "ringing",
  "callbackUrl": "https://your-server.com/webhooks/cxone-call-status"
}

Webhook Handling:
If you provided a callbackUrl, CXone will send POST requests to that URL with updates on the call lifecycle. The payload typically includes:

  • id: The connection ID.
  • status: ringing, answered, missed, failed, completed.
  • startTime: Timestamp when the call started.
  • endTime: Timestamp when the call ended (if completed).
  • duration: Duration in seconds.

You must implement an endpoint to receive these webhooks to track the actual outcome of the call. The API response only confirms the initiation of the call, not the answer.

Complete Working Example

This is a full, copy-pasteable script. Replace the placeholder credentials and phone numbers before running.

import requests
import json
import time
from typing import Optional

# --- Configuration ---
CXONE_BASE_URL = "https://us-east-1.api.nice.com"  # Update to your region
CLIENT_ID = "YOUR_CLIENT_ID"
CLIENT_SECRET = "YOUR_CLIENT_SECRET"

# --- Authentication Module ---
class CXoneAuth:
    def __init__(self, base_url: str, client_id: str, client_secret: str):
        self.base_url = base_url.rstrip('/')
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = f"{self.base_url}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0.0

    def get_access_token(self) -> str:
        if self.access_token and time.time() < self.token_expiry:
            return self.access_token

        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }

        response = requests.post(self.token_url, data=payload, headers=headers)
        response.raise_for_status()
        
        token_data = response.json()
        self.access_token = token_data["access_token"]
        self.token_expiry = time.time() + token_data["expires_in"] - 60
        
        return self.access_token

# --- Personal Connection Module ---
class CXonePersonalConnection:
    def __init__(self, auth: CXoneAuth):
        self.auth = auth
        self.api_base = f"{auth.base_url}/api/v2"

    def trigger_outbound_call(
        self, 
        to_number: str, 
        from_number: str, 
        callback_url: Optional[str] = None
    ) -> dict:
        url = f"{self.api_base}/personal-connections/outbound-calls"
        
        payload = {
            "to": to_number,
            "from": from_number,
            "type": "voice"
        }

        if callback_url:
            payload["callbackUrl"] = callback_url

        headers = {
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

        access_token = self.auth.get_access_token()
        headers["Authorization"] = f"Bearer {access_token}"

        max_retries = 3
        retry_count = 0

        while retry_count < max_retries:
            try:
                response = requests.post(url, json=payload, headers=headers, timeout=30)
                
                if response.status_code == 429:
                    retry_after = int(response.headers.get("Retry-After", 5))
                    print(f"Rate limited. Retrying in {retry_after} seconds...")
                    time.sleep(retry_after)
                    retry_count += 1
                    continue
                
                if response.status_code == 201:
                    result = response.json()
                    print(f"Call triggered successfully. Connection ID: {result.get('id')}")
                    return result
                
                response.raise_for_status()

            except requests.exceptions.HTTPError as e:
                print(f"HTTP Error: {e.response.status_code} - {e.response.text}")
                raise
            except requests.exceptions.RequestException as e:
                print(f"Request failed: {e}")
                raise

        print("Max retries exceeded.")
        raise Exception("Failed to trigger outbound call after retries.")

# --- Main Execution ---
if __name__ == "__main__":
    # 1. Initialize Authentication
    auth = CXoneAuth(
        base_url=CXONE_BASE_URL,
        client_id=CLIENT_ID,
        client_secret=CLIENT_SECRET
    )

    # 2. Initialize Personal Connection Client
    pc_client = CXonePersonalConnection(auth)

    # 3. Define Call Details
    TO_NUMBER = "+15551234567"       # Recipient
    FROM_NUMBER = "+15559876543"     # Caller ID (Must be verified in CXone)
    CALLBACK_URL = "https://your-server.com/webhooks/call-status" # Optional

    try:
        # 4. Trigger the Call
        result = pc_client.trigger_outbound_call(
            to_number=TO_NUMBER,
            from_number=FROM_NUMBER,
            callback_url=CALLBACK_URL
        )
        
        print("Final Result:")
        print(json.dumps(result, indent=2))

    except Exception as e:
        print(f"Failed to execute call: {e}")

Common Errors & Debugging

Error: 403 Forbidden

  • Cause: The OAuth client does not have the required scope.
  • Fix: Go to the CXone Admin Console → OAuth Applications → Edit your client. Ensure the personal-connections scope is checked. You must re-authorize the application or generate a new token after adding scopes.

Error: 400 Bad Request - “Invalid From Number”

  • Cause: The from number provided in the payload is not verified or owned by your CXone instance.
  • Fix: Navigate to CXone Admin Console → Settings → Phone Numbers. Ensure the from number is listed and verified. You cannot spoof arbitrary phone numbers via this API.

Error: 400 Bad Request - “Invalid To Number”

  • Cause: The to number is not in a valid E.164 format or contains invalid characters.
  • Fix: Ensure the phone number starts with a plus sign + followed by the country code (e.g., +1 for US/Canada). Do not include dashes, spaces, or parentheses.

Error: 429 Too Many Requests

  • Cause: You have exceeded the rate limit for the Personal Connection API.
  • Fix: Implement exponential backoff. The code example above includes basic retry logic for 429s. Check the Retry-After header in the response for the specific wait time recommended by the server.

Error: 500 Internal Server Error

  • Cause: A transient issue on the CXone platform.
  • Fix: Retry the request after a short delay (e.g., 1-5 seconds). If the error persists, check the CXone status page for ongoing outages.

Official References