Triggering a CXone Outbound Call via the Personal Connection API

What You Will Build

• A Python script that authenticates with the NICE CXone platform and triggers an outbound call using the Personal Connection API.
• This tutorial utilizes the CXone REST API for Outbound Campaigns and Connections, specifically the POST /api/v2/outbound/connections endpoint.
• The implementation is written in Python 3.9+ using the requests library for HTTP communication.

Prerequisites

• OAuth Client Type: A Service Account (Client Credentials) or User OAuth client configured in the NICE CXone Admin Portal.
• Required Scopes:
  • outbound:connection:create (Required to initiate the call)
  • outbound:campaign:read (Optional but recommended to validate campaign existence)
  • interaction:create (Often required depending on how the contact is created in the interaction center)
• API Version: CXone API v2.
• Language/Runtime: Python 3.9 or higher.
• External Dependencies:
  • requests: For handling HTTP requests.
  • python-dotenv (Optional): For managing environment variables securely.

Install the dependencies via pip:

pip install requests python-dotenv

Authentication Setup

NICE CXone uses OAuth 2.0 for authentication. For server-to-server integrations, the Client Credentials Grant flow is the standard approach. This flow exchanges your client ID and secret for an access token that is valid for a specific duration (typically 3600 seconds).

You must retrieve your Client ID and Client Secret from the CXone Admin Portal under Administration > Security > OAuth Clients. You also need the Realm ID (or Subdomain) which is part of your CXone URL (e.g., niceincontact.com or a custom domain).

Token Retrieval Code

The following function handles the OAuth handshake. It caches the token to avoid unnecessary requests and includes error handling for invalid credentials (401) or expired tokens.

import requests
import time
import os
from typing import Optional, Dict

class CXoneAuth:
    def __init__(self, realm_id: str, client_id: str, client_secret: str):
        self.realm_id = realm_id
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = f"https://{realm_id}.niceincontact.com"
        self.token_url = f"{self.base_url}/api/v2/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: Optional[float] = None

    def get_access_token(self) -> str:
        """
        Retrieves a valid OAuth access token.
        Uses cached token if it has not expired.
        """
        # Check if we have a valid cached token
        if self.access_token and self.token_expiry and time.time() < self.token_expiry:
            return self.access_token

        # Prepare the token request body
        # The grant_type must be 'client_credentials' for service accounts
        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"]
            
            # Calculate expiry time (expires_in is in seconds)
            # Subtract 60 seconds to ensure we refresh before actual expiration
            self.token_expiry = time.time() + (token_data["expires_in"] - 60)
            
            return self.access_token

        except requests.exceptions.HTTPError as http_err:
            if response.status_code == 401:
                raise Exception("Authentication failed: Invalid Client ID or Secret.") from http_err
            elif response.status_code == 400:
                raise Exception("Bad Request: Check the format of the grant_type or payload.") from http_err
            else:
                raise Exception(f"HTTP Error during token retrieval: {http_err}") from http_err
        except requests.exceptions.RequestException as req_err:
            raise Exception(f"Network error during token retrieval: {req_err}") from req_err

Implementation

Step 1: Define the Connection Payload

The Personal Connection API allows you to trigger a call without enrolling a contact in a standard campaign. This is useful for one-off calls, triggered workflows, or external application integrations.

The core endpoint is POST /api/v2/outbound/connections. The request body must contain specific fields to identify the caller, the callee, and the context of the call.

Critical Fields:

• campaignId: Even for “personal” connections, CXone often requires a campaign ID to route the call to the correct pool of agents or IVR. If you do not have a standard campaign, you may need to create a “dummy” campaign in the admin console dedicated to API-triggered calls.
• contact: The phone number being called. Must be in E.164 format (e.g., +14155552671).
• phoneNumber: The outbound phone number (ANI) that will appear on the recipient’s caller ID. This number must be provisioned in your CXone account.
• userData: Optional JSON payload that can be passed to the IVR or agent screen pop. This is how you pass context (like order numbers) into the call flow.

Step 2: Constructing the API Request

We will create a class CXoneOutbound that handles the logic for triggering the call. This class will use the CXoneAuth class from the previous step.

class CXoneOutbound:
    def __init__(self, realm_id: str, client_id: str, client_secret: str):
        self.auth = CXoneAuth(realm_id, client_id, client_secret)
        self.base_url = f"https://{realm_id}.niceincontact.com"
        self.connections_endpoint = f"{self.base_url}/api/v2/outbound/connections"

    def trigger_personal_connection(
        self, 
        campaign_id: str, 
        phone_number: str, 
        aninumber: str,
        user_data: Optional[Dict] = None
    ) -> Dict:
        """
        Triggers an outbound call using the Personal Connection API.
        
        Args:
            campaign_id (str): The ID of the campaign to use for routing.
            phone_number (str): The recipient's phone number in E.164 format.
            aninumber (str): The outbound caller ID number (must be provisioned).
            user_data (dict, optional): Contextual data to pass to the IVR/Agent.
            
        Returns:
            dict: The API response containing the connection ID and status.
        """
        access_token = self.auth.get_access_token()
        
        # Construct the headers
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {access_token}"
        }

        # Construct the request body
        # The 'contact' object defines who is being called
        # The 'phoneNumber' at the root level defines the ANI (Caller ID)
        payload = {
            "campaignId": campaign_id,
            "contact": {
                "phoneNumber": phone_number
            },
            "phoneNumber": aninumber,
            "userData": user_data or {}
        }

        try:
            # Execute the POST request
            response = requests.post(
                self.connections_endpoint,
                json=payload,
                headers=headers
            )
            
            # Raise an exception for HTTP errors
            response.raise_for_status()
            
            return response.json()

        except requests.exceptions.HTTPError as http_err:
            self._handle_http_error(response, http_err)
        except requests.exceptions.RequestException as req_err:
            raise Exception(f"Network error during call trigger: {req_err}") from req_err

    def _handle_http_error(self, response: requests.Response, error: Exception):
        """
        Specific error handling for CXone API responses.
        """
        status_code = response.status_code
        try:
            error_detail = response.json()
        except ValueError:
            error_detail = response.text

        if status_code == 400:
            # Common issues: Invalid phone number format, missing campaign ID, 
            # or ANI number not provisioned.
            raise Exception(f"Bad Request (400): {error_detail}") from error
        elif status_code == 403:
            # Missing scopes or insufficient permissions on the campaign.
            raise Exception(f"Forbidden (403): Check OAuth scopes. Detail: {error_detail}") from error
        elif status_code == 429:
            # Rate limited. Implement retry logic in production.
            raise Exception(f"Rate Limited (429): Too many requests. Wait before retrying.") from error
        elif status_code == 500:
            raise Exception(f"Server Error (500): CXone platform error. Detail: {error_detail}") from error
        else:
            raise Exception(f"HTTP Error {status_code}: {error_detail}") from error

Step 3: Processing Results and Validation

When the API call succeeds (HTTP 200 or 201), the response contains a connectionId. This ID is crucial for tracking the call status later. You can use this ID to query the connection status via GET /api/v2/outbound/connections/{connectionId}.

A successful response typically looks like this:

{
  "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "campaignId": "12345678-90ab-cdef-1234-567890abcdef",
  "contact": {
    "phoneNumber": "+14155552671"
  },
  "status": "PENDING",
  "createdTime": "2023-10-27T10:00:00.000Z",
  "userData": {
    "order_id": "ORD-998877"
  }
}

Complete Working Example

The following script combines the authentication and outbound logic into a runnable module. It uses environment variables for secure credential management.

Setup:

1. Create a .env file in the same directory as the script.
2. Add your CXone credentials:

   CXONE_REALM_ID=your-realm-id
   CXONE_CLIENT_ID=your-client-id
   CXONE_CLIENT_SECRET=your-client-secret
   CXONE_CAMPAIGN_ID=your-campaign-id
   CXONE_ANI_NUMBER=+12345678901
   

Code:

import os
import sys
import json
import requests
import time
from typing import Optional, Dict

# Load environment variables
try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    print("Warning: python-dotenv not installed. Ensure environment variables are set.")

class CXoneAuth:
    def __init__(self, realm_id: str, client_id: str, client_secret: str):
        self.realm_id = realm_id
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = f"https://{realm_id}.niceincontact.com"
        self.token_url = f"{self.base_url}/api/v2/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: Optional[float] = None

    def get_access_token(self) -> str:
        if self.access_token and self.token_expiry 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"]
            self.token_expiry = time.time() + (token_data["expires_in"] - 60)
            
            return self.access_token

        except requests.exceptions.HTTPError as http_err:
            raise Exception(f"Authentication failed: {http_err}") from http_err
        except requests.exceptions.RequestException as req_err:
            raise Exception(f"Network error during token retrieval: {req_err}") from req_err

class CXoneOutbound:
    def __init__(self, realm_id: str, client_id: str, client_secret: str):
        self.auth = CXoneAuth(realm_id, client_id, client_secret)
        self.base_url = f"https://{realm_id}.niceincontact.com"
        self.connections_endpoint = f"{self.base_url}/api/v2/outbound/connections"

    def trigger_personal_connection(
        self, 
        campaign_id: str, 
        phone_number: str, 
        aninumber: str,
        user_data: Optional[Dict] = None
    ) -> Dict:
        access_token = self.auth.get_access_token()
        
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {access_token}"
        }

        payload = {
            "campaignId": campaign_id,
            "contact": {
                "phoneNumber": phone_number
            },
            "phoneNumber": aninumber,
            "userData": user_data or {}
        }

        try:
            response = requests.post(
                self.connections_endpoint,
                json=payload,
                headers=headers
            )
            
            response.raise_for_status()
            return response.json()

        except requests.exceptions.HTTPError as http_err:
            self._handle_http_error(response, http_err)
        except requests.exceptions.RequestException as req_err:
            raise Exception(f"Network error during call trigger: {req_err}") from req_err

    def _handle_http_error(self, response: requests.Response, error: Exception):
        status_code = response.status_code
        try:
            error_detail = response.json()
        except ValueError:
            error_detail = response.text

        if status_code == 400:
            raise Exception(f"Bad Request (400): {error_detail}") from error
        elif status_code == 403:
            raise Exception(f"Forbidden (403): Check OAuth scopes. Detail: {error_detail}") from error
        elif status_code == 429:
            raise Exception(f"Rate Limited (429): Too many requests.") from error
        elif status_code == 500:
            raise Exception(f"Server Error (500): {error_detail}") from error
        else:
            raise Exception(f"HTTP Error {status_code}: {error_detail}") from error

def main():
    # Retrieve credentials from environment variables
    realm_id = os.getenv("CXONE_REALM_ID")
    client_id = os.getenv("CXONE_CLIENT_ID")
    client_secret = os.getenv("CXONE_CLIENT_SECRET")
    campaign_id = os.getenv("CXONE_CAMPAIGN_ID")
    ani_number = os.getenv("CXONE_ANI_NUMBER")

    if not all([realm_id, client_id, client_secret, campaign_id, ani_number]):
        print("Error: Missing required environment variables.")
        print("Please set CXONE_REALM_ID, CXONE_CLIENT_ID, CXONE_CLIENT_SECRET, CXONE_CAMPAIGN_ID, and CXONE_ANI_NUMBER")
        sys.exit(1)

    # Initialize the outbound client
    outbound_client = CXoneOutbound(realm_id, client_id, client_secret)

    # Define the recipient and context
    recipient_number = "+14155552671"  # Replace with actual test number
    context_data = {
        "order_id": "ORD-12345",
        "customer_name": "John Doe",
        "purpose": "Order Confirmation"
    }

    try:
        print(f"Initiating call to {recipient_number}...")
        result = outbound_client.trigger_personal_connection(
            campaign_id=campaign_id,
            phone_number=recipient_number,
            aninumber=ani_number,
            user_data=context_data
        )
        
        print("Call triggered successfully!")
        print(json.dumps(result, indent=2))
        
        # Extract connection ID for tracking
        connection_id = result.get("id")
        if connection_id:
            print(f"\nConnection ID: {connection_id}")
            print("Use this ID to track call status via the API.")

    except Exception as e:
        print(f"Failed to trigger call: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 400 Bad Request - “Invalid Phone Number”

• Cause: The phone number provided in contact.phoneNumber or phoneNumber (ANI) is not in strict E.164 format.
• Fix: Ensure the number starts with a + followed by the country code and number, with no spaces or dashes. Example: +14155552671.
• Code Check: Validate input using a library like phonenumbers before sending to the API.

Error: 403 Forbidden - “Insufficient Scope”

• Cause: The OAuth client used to generate the token does not have the outbound:connection:create scope.
• Fix: Go to CXone Admin Portal > Security > OAuth Clients. Edit your client and add the outbound:connection:create scope. Re-generate the token.

Error: 400 Bad Request - “Campaign Not Found” or “Campaign Disabled”

• Cause: The campaignId provided is invalid, does not exist, or the campaign is in a paused/disabled state.
• Fix: Verify the Campaign ID in the CXone Admin Portal. Ensure the campaign is Active. For Personal Connections, the campaign does not need to have contacts enrolled, but it must be active and configured to accept API triggers.

Error: 400 Bad Request - “ANI Number Not Provisioned”

• Cause: The phoneNumber (ANI) used in the request is not registered in your CXone account as an outbound number.
• Fix: Ensure the number is purchased and activated in the CXone Telephony/Number Management section.

Error: 429 Too Many Requests

• Cause: You have exceeded the rate limit for outbound connection triggers.
• Fix: Implement exponential backoff in your retry logic. Do not simply retry immediately.

Official References

• NICE CXone API Reference: Outbound Connections
• NICE CXone API Reference: OAuth Authentication
• NICE CXone Developer Hub