Triggering NICE CXone Outbound Calls via the Personal Connection API

What You Will Build

• A Python script that programmatically initiates an outbound voice call to a specific telephone number using the NICE CXone Personal Connection API.
• The integration uses the CXone REST API for campaign execution and the Personal Connection endpoint for immediate, ad-hoc dialing.
• The tutorial covers Python 3.9+ using the requests library for HTTP communication.

Prerequisites

• OAuth Client Type: A CXone OAuth client with Service Account or Resource Owner credentials.
• Required Scopes:
  • personal_connection:write (Required to initiate calls)
  • personal_connection:read (Optional, useful for verifying call status later)
  • agent:read (If you need to look up agent IDs dynamically, though hardcoding is fine for this tutorial)
• CXone Environment: Access to a CXone tenant with the Personal Connection feature enabled. Ensure the “Personal Connection” feature flag is active in your tenant settings.
• Python Runtime: Python 3.9 or higher.
• Dependencies:

  pip install requests
  

Authentication Setup

CXone APIs use OAuth 2.0 for authentication. You must obtain an access token before making any API calls. The token expires after a specific duration (typically 1 hour), so production systems should implement token caching and refresh logic. For this tutorial, we will implement a simple token fetcher.

Replace CLIENT_ID, CLIENT_SECRET, and TENANT_DOMAIN with your actual CXone credentials.

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

class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str, tenant_domain: str):
        self.client_id = client_id
        self.client_secret = client_secret
        # The base URL for CXone APIs is typically https://{tenant}.niceincontact.com
        # However, OAuth tokens are fetched from the global auth endpoint or tenant-specific.
        # Most CXone tenants use the tenant-specific endpoint for token retrieval.
        self.auth_url = f"https://{tenant_domain}/oauth2/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0

    def get_token(self) -> str:
        """
        Retrieves an OAuth2 access token if not already cached or if expired.
        """
        current_time = time.time()
        
        # Check if we have a valid token
        if self.access_token and current_time < self.token_expiry:
            return self.access_token

        # Fetch new 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.auth_url, data=payload, headers=headers)
            response.raise_for_status()
            
            token_data = response.json()
            self.access_token = token_data["access_token"]
            
            # Set expiry with a small buffer (5 seconds) to avoid edge cases
            self.token_expiry = current_time + token_data["expires_in"] - 5
            
            return self.access_token

        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise Exception("Authentication failed: Invalid Client ID or Secret.") from e
            elif response.status_code == 403:
                raise Exception("Authentication forbidden: Client may not have permission.") from e
            else:
                raise Exception(f"HTTP Error during token fetch: {response.status_code}") from e
        except requests.exceptions.RequestException as e:
            raise Exception(f"Network error during token fetch: {e}") from e

    def get_headers(self) -> Dict[str, str]:
        """
        Returns standard headers including the Authorization bearer token.
        """
        token = self.get_token()
        return {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

Implementation

Step 1: Constructing the Personal Connection Request

The Personal Connection API allows you to place calls, send messages, or share files directly from your application. For voice calls, the endpoint is POST /api/v2/personal-connection/calls.

Unlike traditional outbound campaigns which require queue configuration and agent assignment, Personal Connection allows you to specify the caller ID (if permitted) and the recipient directly.

Key Parameters:

• to: The destination phone number in E.164 format (e.g., +12125551234).
• from: The caller ID number. This must be a number provisioned in your CXone tenant and allowed for outbound calling.
• mediaType: For voice calls, this is "voice".
• contextId: An optional string to correlate this call with your internal system records.

Here is the code structure to initiate the call:

import requests
from cxone_auth import CXoneAuth  # Assuming the auth class is in cxone_auth.py

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

    def place_outbound_call(self, to_number: str, from_number: str, context_id: str = "") -> Dict:
        """
        Initiates an outbound voice call using Personal Connection.
        
        Args:
            to_number: Destination number in E.164 format.
            from_number: Caller ID number (must be provisioned in CXone).
            context_id: Optional identifier for tracking.
            
        Returns:
            Dictionary containing the API response.
        """
        endpoint = f"{self.base_url}/personal-connection/calls"
        
        headers = self.auth.get_headers()
        
        payload = {
            "to": [
                {
                    "type": "phoneNumber",
                    "phoneNumber": to_number
                }
            ],
            "from": {
                "type": "phoneNumber",
                "phoneNumber": from_number
            },
            "mediaType": "voice",
            "contextId": context_id
        }

        try:
            response = requests.post(endpoint, json=payload, headers=headers)
            
            # Handle HTTP Errors
            if response.status_code == 201:
                print(f"Call initiated successfully. Context ID: {context_id}")
                return response.json()
            elif response.status_code == 400:
                print(f"Bad Request: Check phone number formats and provisioned numbers.")
                print(f"Response: {response.text}")
                return {}
            elif response.status_code == 401:
                print("Unauthorized: Token may be invalid or expired.")
                return {}
            elif response.status_code == 403:
                print("Forbidden: Check OAuth scopes. Requires 'personal_connection:write'.")
                return {}
            elif response.status_code == 429:
                print("Rate Limit Exceeded: Please wait and retry.")
                return {}
            else:
                print(f"Unexpected Error: {response.status_code}")
                print(f"Response: {response.text}")
                return {}

        except requests.exceptions.RequestException as e:
            print(f"Network error: {e}")
            return {}

Step 2: Handling Edge Cases and Validation

Before sending the request, it is critical to validate the phone numbers. CXone strictly enforces E.164 format. If you pass a number like (212) 555-1234, the API will return a 400 Bad Request.

Additionally, the from number must be a provisioned DID (Direct Inward Dialing) number in your CXone tenant. You cannot use a random number or a mobile number as the caller ID unless it has been explicitly whitelisted and provisioned in the CXone admin console under Telephony > Numbers.

Add this validation helper to your class:

import re

    @staticmethod
    def validate_e164(phone_number: str) -> bool:
        """
        Validates if a phone number is in E.164 format.
        E.164 format: +[country code][area code][local number]
        Max 15 digits.
        """
        # Regex for E.164: Starts with +, followed by 1-15 digits
        pattern = r"^\+[1-9]\d{1,14}$"
        return bool(re.match(pattern, phone_number))

    def place_outbound_call_with_validation(self, to_number: str, from_number: str, context_id: str = "") -> Dict:
        """
        Validates inputs before placing the call.
        """
        if not self.validate_e164(to_number):
            raise ValueError(f"Destination number '{to_number}' is not in valid E.164 format.")
        
        if not self.validate_e164(from_number):
            raise ValueError(f"Caller ID '{from_number}' is not in valid E.164 format.")

        return self.place_outbound_call(to_number, from_number, context_id)

Step 3: Retrieving Call Status (Optional but Recommended)

The POST request returns immediately. It does not wait for the call to connect or complete. It returns a callId or references the contextId. To know if the call was answered, you must poll the Personal Connection status endpoint or use the standard Call Detail Records (CDR) API.

For Personal Connection, you can query the status using the contextId if you provided one. However, the most robust way is to query the Call Records API using the from and to numbers and a time window.

Here is how to fetch recent call details to verify the outcome:

    def get_recent_call_status(self, to_number: str, from_number: str, start_time: str, end_time: str) -> Dict:
        """
        Queries the Call Records API to check the status of a recent call.
        
        Args:
            to_number: Destination number.
            from_number: Caller ID.
            start_time: ISO 8601 datetime string (e.g., '2023-10-27T10:00:00Z').
            end_time: ISO 8601 datetime string.
        """
        endpoint = f"{self.base_url}/analytics/conversations/details/query"
        
        headers = self.auth.get_headers()
        
        # Define the query body
        # We look for conversations involving these numbers within the time frame
        payload = {
            "dateRange": {
                "start": start_time,
                "end": end_time
            },
            "filter": {
                "and": [
                    {
                        "field": "to.phoneNumber",
                        "op": "equal",
                        "value": to_number
                    },
                    {
                        "field": "from.phoneNumber",
                        "op": "equal",
                        "value": from_number
                    }
                ]
            },
            "groupBy": [],
            "aggregations": [],
            "limit": 10,
            "offset": 0
        }

        try:
            response = requests.post(endpoint, json=payload, headers=headers)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            print(f"Error fetching call status: {e}")
            return {}

Complete Working Example

This script combines authentication, validation, call initiation, and status checking into a single executable module.

import sys
import time
from datetime import datetime, timezone, timedelta
import requests

# --- Authentication Module ---
class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str, tenant_domain: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.auth_url = f"https://{tenant_domain}/oauth2/token"
        self.access_token = None
        self.token_expiry = 0

    def get_token(self) -> str:
        current_time = time.time()
        if self.access_token and current_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.auth_url, data=payload, headers=headers)
            response.raise_for_status()
            token_data = response.json()
            self.access_token = token_data["access_token"]
            self.token_expiry = current_time + token_data["expires_in"] - 5
            return self.access_token
        except requests.exceptions.HTTPError as e:
            raise Exception(f"Auth Error: {response.status_code} - {response.text}") from e

    def get_headers(self) -> dict:
        return {
            "Authorization": f"Bearer {self.get_token()}",
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

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

    @staticmethod
    def validate_e164(phone_number: str) -> bool:
        import re
        pattern = r"^\+[1-9]\d{1,14}$"
        return bool(re.match(pattern, phone_number))

    def place_outbound_call(self, to_number: str, from_number: str, context_id: str = "") -> dict:
        if not self.validate_e164(to_number) or not self.validate_e164(from_number):
            raise ValueError("Phone numbers must be in E.164 format (e.g., +12125551234).")

        endpoint = f"{self.base_url}/personal-connection/calls"
        headers = self.auth.get_headers()
        
        payload = {
            "to": [{"type": "phoneNumber", "phoneNumber": to_number}],
            "from": {"type": "phoneNumber", "phoneNumber": from_number},
            "mediaType": "voice",
            "contextId": context_id
        }

        try:
            response = requests.post(endpoint, json=payload, headers=headers)
            if response.status_code == 201:
                print("SUCCESS: Call initiated.")
                return response.json()
            else:
                print(f"ERROR: {response.status_code} - {response.text}")
                return {}
        except Exception as e:
            print(f"Exception: {e}")
            return {}

    def check_call_status(self, to_number: str, from_number: str) -> dict:
        # Check calls from the last 5 minutes
        end_time = datetime.now(timezone.utc)
        start_time = end_time - timedelta(minutes=5)
        
        start_str = start_time.strftime("%Y-%m-%dT%H:%M:%SZ")
        end_str = end_time.strftime("%Y-%m-%dT%H:%M:%SZ")

        endpoint = f"{self.base_url}/analytics/conversations/details/query"
        headers = self.auth.get_headers()
        
        payload = {
            "dateRange": {"start": start_str, "end": end_str},
            "filter": {
                "and": [
                    {"field": "to.phoneNumber", "op": "equal", "value": to_number},
                    {"field": "from.phoneNumber", "op": "equal", "value": from_number}
                ]
            },
            "groupBy": [],
            "aggregations": [],
            "limit": 1,
            "offset": 0
        }

        try:
            response = requests.post(endpoint, json=payload, headers=headers)
            if response.status_code == 200:
                data = response.json()
                if data.get("results"):
                    conv = data["results"][0]
                    print(f"Call Status Found: {conv.get('mediaType')}")
                    print(f"Duration: {conv.get('duration', 'N/A')} seconds")
                    return conv
                else:
                    print("No recent calls found matching criteria.")
                    return {}
            else:
                print(f"Status Check Error: {response.status_code}")
                return {}
        except Exception as e:
            print(f"Status Check Exception: {e}")
            return {}

# --- Main Execution ---
if __name__ == "__main__":
    # Configuration
    CLIENT_ID = "YOUR_CLIENT_ID"
    CLIENT_SECRET = "YOUR_CLIENT_SECRET"
    TENANT_DOMAIN = "YOUR_TENANT.niceincontact.com"
    
    # Phone Numbers (E.164 Format)
    CALLER_ID = "+12125550100"  # Must be provisioned in CXone
    DESTINATION = "+12125550199"

    if CLIENT_ID == "YOUR_CLIENT_ID":
        print("Please update CLIENT_ID, CLIENT_SECRET, and TENANT_DOMAIN in the script.")
        sys.exit(1)

    try:
        # 1. Authenticate
        auth = CXoneAuth(CLIENT_ID, CLIENT_SECRET, TENANT_DOMAIN)
        
        # 2. Initialize Client
        pc_client = CXonePersonalConnection(auth, TENANT_DOMAIN)
        
        # 3. Place Call
        context_id = f"pc_call_{int(time.time())}"
        result = pc_client.place_outbound_call(DESTINATION, CALLER_ID, context_id)
        
        if result:
            print(f"Context ID: {context_id}")
            print("Waiting 10 seconds to check status...")
            time.sleep(10)
            
            # 4. Check Status
            pc_client.check_call_status(DESTINATION, CALLER_ID)

    except Exception as e:
        print(f"Fatal Error: {e}")

Common Errors & Debugging

Error: 400 Bad Request

Cause: The phone numbers are not in valid E.164 format, or the from number is not provisioned in your CXone tenant.
Fix:

1. Ensure numbers start with + followed by the country code (e.g., +1 for US/Canada).
2. Log into the CXone Admin Console. Navigate to Telephony > Numbers. Ensure the from number is present and has Outbound capability enabled.

Error: 403 Forbidden

Cause: The OAuth token lacks the required scope.
Fix:

1. Go to Security > OAuth Clients in the CXone Admin Console.
2. Select your client.
3. Edit the scopes. Ensure personal_connection:write is checked.
4. Save and regenerate the token.

Error: 401 Unauthorized

Cause: The access token is expired or invalid.
Fix:

1. Verify your CLIENT_ID and CLIENT_SECRET are correct.
2. Ensure your CXoneAuth class is correctly fetching a new token when the old one expires. In the example above, the get_token() method handles this automatically.

Error: Call Initiates but No Ringing

Cause: The call was successfully initiated by the API, but the telephony carrier blocked it, or the number is invalid.
Fix:

1. Use the check_call_status method in the code above.
2. Look at the disposition field in the response. Common dispositions include NO_ANSWER, BUSY, or FAILED.
3. If FAILED, check the failureReason field for carrier-specific error codes.

Official References

• NICE CXone Personal Connection API Documentation
• CXone OAuth 2.0 Guide
• CXone Analytics Query API