Trigger NICE CXone Outbound Calls via the Personal Connection API

What You Will Build

• This tutorial demonstrates how to programmatically initiate an outbound call from a specific agent using the NICE CXone Personal Connection API.
• The code utilizes the NICE CXone REST API endpoint /api/2.0/omnichannel/outbound/personal-connection/calls.
• The implementation is provided in Python using the requests library for HTTP communication and JSON handling.

Prerequisites

• OAuth Client Type: A CXone application with the Agent or Admin role assigned. The token must be generated using the client_credentials grant type.
• Required Scopes: Outbound:PersonalConnection:Create and Outbound:PersonalConnection:Read.
• SDK/API Version: CXone API v2.0 (REST). No specific SDK is required; direct HTTP calls are used for clarity and flexibility.
• Language/Runtime: Python 3.8+.
• External Dependencies: requests (install via pip install requests).
• Environment Variables: You must have your CXone CLIENT_ID, CLIENT_SECRET, and TOKEN_URL (e.g., https://api-us-22.nice-incontact.com/oauth2/token) available.

Authentication Setup

Before making any outbound calls, you must obtain a valid OAuth 2.0 access token. NICE CXone uses standard OAuth 2.0 client credentials flow. This token is valid for 1 hour (3600 seconds). In production, you should implement token caching to avoid refreshing the token on every single call request.

The following Python function handles the authentication process. It sends a POST request to the CXone token endpoint.

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

class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str, token_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = token_url
        self.access_token: Optional[str] = None
        self.token_expiry: Optional[int] = None

    def get_access_token(self) -> str:
        """
        Retrieves a new OAuth2 access token from NICE CXone.
        Returns:
            str: The access token.
        Raises:
            requests.exceptions.RequestException: If the request fails.
        """
        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, timeout=10)
            response.raise_for_status()
            
            token_data = response.json()
            self.access_token = token_data["access_token"]
            self.token_expiry = token_data.get("expires_in", 3600)
            
            return self.access_token
            
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise Exception("Authentication failed: Invalid client_id or client_secret.") from e
            elif response.status_code == 403:
                raise Exception("Authentication failed: Client does not have permission to request tokens.") from e
            else:
                raise Exception(f"Authentication request failed with status {response.status_code}: {response.text}") from e
        except requests.exceptions.RequestException as e:
            raise Exception(f"Network error during authentication: {str(e)}") from e

    def get_valid_token(self) -> str:
        """
        Returns the current access token if valid, otherwise fetches a new one.
        Note: For simplicity in this tutorial, we always fetch a new token. 
        In production, check time.time() against self.token_expiry.
        """
        return self.get_access_token()

Implementation

Step 1: Constructing the Outbound Call Request

The core of the Personal Connection API is the request body. Unlike traditional outbound campaigns, Personal Connection allows an agent to dial directly from their interface or via API without enqueuing a campaign.

Key parameters in the JSON payload:

• userId: The unique identifier of the agent placing the call. This agent must be in a Ready or Available state.
• phoneNumber: The destination number to dial. It must include the country code.
• displayNumber: The number that appears on the recipient’s caller ID. This must be a valid, provisioned DID in your CXone tenant.

Here is the structure of the request body:

{
  "userId": "5f8a9b2c-1d2e-3f4a-5b6c-7d8e9f0a1b2c",
  "phoneNumber": "+14155552671",
  "displayNumber": "+18005551234"
}

Step 2: Executing the API Call

Now we combine authentication and the API call into a single class. This class handles the HTTP POST request to the Personal Connection endpoint.

The endpoint is: POST /api/2.0/omnichannel/outbound/personal-connection/calls

class CXonePersonalConnection:
    def __init__(self, base_url: str, auth: CXoneAuth):
        self.base_url = base_url.rstrip('/')
        self.auth = auth

    def trigger_outbound_call(self, user_id: str, phone_number: str, display_number: str) -> Dict:
        """
        Triggers an outbound call using the Personal Connection API.
        
        Args:
            user_id (str): The ID of the agent initiating the call.
            phone_number (str): The destination phone number (E.164 format).
            display_number (str): The caller ID number (E.164 format).
            
        Returns:
            Dict: The API response containing the call ID and status.
        """
        # 1. Get a valid token
        token = self.auth.get_valid_token()
        
        # 2. Define the endpoint
        endpoint = f"{self.base_url}/api/2.0/omnichannel/outbound/personal-connection/calls"
        
        # 3. Define headers
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {token}"
        }
        
        # 4. Define the payload
        payload = {
            "userId": user_id,
            "phoneNumber": phone_number,
            "displayNumber": display_number
        }
        
        try:
            # 5. Make the request
            response = requests.post(endpoint, json=payload, headers=headers, timeout=15)
            
            # 6. Handle response
            if response.status_code == 202:
                # 202 Accepted means the call is being processed
                return {
                    "success": True,
                    "status_code": response.status_code,
                    "data": response.json(),
                    "message": "Call request accepted."
                }
            elif response.status_code == 200:
                # 200 OK might also be returned depending on tenant configuration
                return {
                    "success": True,
                    "status_code": response.status_code,
                    "data": response.json(),
                    "message": "Call request successful."
                }
            else:
                # Handle errors
                error_data = response.json() if response.content else {}
                return {
                    "success": False,
                    "status_code": response.status_code,
                    "data": error_data,
                    "message": f"API Error: {error_data.get('message', 'Unknown error')}"
                }
                
        except requests.exceptions.Timeout:
            return {
                "success": False,
                "message": "Request timed out."
            }
        except requests.exceptions.RequestException as e:
            return {
                "success": False,
                "message": f"Network error: {str(e)}"
            }

Step 3: Processing Results and Edge Cases

When the API returns a 202 Accepted, the response body typically contains the callId. This ID is crucial for tracking the call’s progress if you integrate with the Call Details API later.

Edge Case: Agent Not Ready
If the userId provided is not in a Ready state (e.g., they are on a break or in a call), the API will return a 400 Bad Request or 409 Conflict. You must ensure the agent’s state is validated before triggering the call if you want a seamless experience.

Edge Case: Invalid Display Number
If the displayNumber is not provisioned in your CXone tenant, the call will fail. Always use a DID that is active and assigned to the specific site or global pool accessible by the agent.

Edge Case: Rate Limiting
NICE CXone enforces rate limits. If you receive a 429 Too Many Requests, you must implement exponential backoff. The response headers will include Retry-After.

Complete Working Example

Below is the full, runnable Python script. Replace the placeholder values with your actual CXone credentials.

import os
import sys
import json
import requests

# --- Configuration ---
# In production, use environment variables or a secure secrets manager
CXONE_CLIENT_ID = os.getenv("CXONE_CLIENT_ID", "YOUR_CLIENT_ID")
CXONE_CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET", "YOUR_CLIENT_SECRET")
CXONE_TOKEN_URL = os.getenv("CXONE_TOKEN_URL", "https://api-us-22.nice-incontact.com/oauth2/token")
CXONE_BASE_URL = os.getenv("CXONE_BASE_URL", "https://api-us-22.nice-incontact.com")

# Target Agent and Phone Numbers
AGENT_USER_ID = os.getenv("AGENT_USER_ID", "5f8a9b2c-1d2e-3f4a-5b6c-7d8e9f0a1b2c")
DESTINATION_NUMBER = os.getenv("DESTINATION_NUMBER", "+14155552671")
CALLER_ID_NUMBER = os.getenv("CALLER_ID_NUMBER", "+18005551234")

class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str, token_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = token_url

    def get_access_token(self) -> str:
        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, timeout=10)
            response.raise_for_status()
            return response.json()["access_token"]
        except requests.exceptions.HTTPError as e:
            raise Exception(f"Auth Failed: {e.response.text}") from e
        except requests.exceptions.RequestException as e:
            raise Exception(f"Network Error: {str(e)}") from e

class CXonePersonalConnection:
    def __init__(self, base_url: str, auth: CXoneAuth):
        self.base_url = base_url.rstrip('/')
        self.auth = auth

    def trigger_outbound_call(self, user_id: str, phone_number: str, display_number: str):
        token = self.auth.get_access_token()
        endpoint = f"{self.base_url}/api/2.0/omnichannel/outbound/personal-connection/calls"
        
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {token}"
        }
        
        payload = {
            "userId": user_id,
            "phoneNumber": phone_number,
            "displayNumber": display_number
        }
        
        print(f"Attempting to trigger call for Agent {user_id}...")
        print(f"Payload: {json.dumps(payload, indent=2)}")
        
        try:
            response = requests.post(endpoint, json=payload, headers=headers, timeout=15)
            
            if response.status_code in [200, 202]:
                result = response.json()
                print("Call Triggered Successfully!")
                print(f"Response: {json.dumps(result, indent=2)}")
                return result
            else:
                print(f"Failed to trigger call. Status: {response.status_code}")
                print(f"Error Details: {response.text}")
                return None
                
        except Exception as e:
            print(f"An error occurred: {str(e)}")
            return None

def main():
    if CXONE_CLIENT_ID == "YOUR_CLIENT_ID":
        print("Error: Please set CXONE_CLIENT_ID and CXONE_CLIENT_SECRET environment variables.")
        sys.exit(1)

    auth = CXoneAuth(CXONE_CLIENT_ID, CXONE_CLIENT_SECRET, CXONE_TOKEN_URL)
    pc_api = CXonePersonalConnection(CXONE_BASE_URL, auth)
    
    result = pc_api.trigger_outbound_call(
        user_id=AGENT_USER_ID,
        phone_number=DESTINATION_NUMBER,
        display_number=CALLER_ID_NUMBER
    )
    
    if result:
        print("\nProcess completed.")
    else:
        print("\nProcess failed.")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token is invalid, expired, or missing.
• Fix: Verify that CXONE_CLIENT_ID and CXONE_CLIENT_SECRET are correct. Ensure the token URL is correct for your region (e.g., api-us-22, api-eu-01). Check that the client has the Outbound:PersonalConnection:Create scope.

Error: 403 Forbidden

• Cause: The OAuth client does not have the necessary permissions, or the agent ID provided does not belong to the tenant.
• Fix: Check the application permissions in the CXone Admin Portal. Ensure the userId is a valid agent within the same tenant as the OAuth client.

Error: 400 Bad Request

• Cause: Invalid input parameters. Common reasons include:
  • phoneNumber or displayNumber is not in E.164 format.
  • displayNumber is not provisioned in the tenant.
  • userId is not a valid agent ID.
• Fix: Validate phone numbers using a library like phonenumbers. Ensure the DID is active in the CXone Admin Portal under Voice > Numbers.

Error: 409 Conflict

• Cause: The agent is not in a Ready state.
• Fix: Check the agent’s current state using the /api/2.0/users/{userId}/state endpoint. Wait for the agent to become available before triggering the call.

Error: 429 Too Many Requests

• Cause: You have exceeded the rate limit for the Personal Connection API.
• Fix: Implement retry logic with exponential backoff. Check the Retry-After header in the response.

Official References

• NICE CXone Personal Connection API Documentation
• NICE CXone OAuth 2.0 Authentication
• NICE CXone User States API