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

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

What You Will Build

  • This tutorial builds a script that programmatically initiates an outbound voice call to a specified number using the NICE CXone Personal Connection API.
  • The solution uses the CXone REST API v2 endpoints for Personal Connections to create a call intent and execute the dial.
  • The primary implementation language is Python, with supplementary examples in JavaScript/TypeScript for Node.js environments.

Prerequisites

  • OAuth Client: A CXone API client configured with client_credentials flow.
  • Required Scopes: personal_connections:write, personal_connections:read.
  • CXone Tenant: A valid CXone tenant ID and a user context (typically an agent or supervisor with permissions to make outbound calls).
  • SDK Version: @nice-dcv/sdk (NPM) or raw HTTP requests via requests (Python).
  • Runtime: Python 3.9+ or Node.js 18+.
  • Dependencies:
    • Python: pip install requests python-dotenv
    • Node.js: npm install axios dotenv

Authentication Setup

NICE CXone uses OAuth 2.0 for API authentication. Before triggering a call, you must obtain an access token. The Personal Connection API requires the token to be associated with a specific user context. While standard client_credentials grants machine access, Personal Connections often require a user-bound token to simulate the action of an agent making a call.

For this tutorial, we assume a standard client_credentials flow where the client ID and secret are exchanged for a token. If your tenant requires user impersonation, you must use the urn:nice:cxone:context:user grant type or include the x-nice-impersonate-user header.

Python Authentication Helper

import requests
import os
from typing import Optional

CXONE_API_BASE = "https://api.nicecxone.com"
TOKEN_ENDPOINT = f"{CXONE_API_BASE}/api/v2/oauth/token"

def get_access_token(client_id: str, client_secret: str) -> str:
    """
    Retrieves an OAuth2 access token from CXone.
    
    Args:
        client_id: Your CXone API Client ID.
        client_secret: Your CXone API Client Secret.
        
    Returns:
        The access token string.
        
    Raises:
        requests.exceptions.HTTPError: If authentication fails.
    """
    headers = {
        "Content-Type": "application/x-www-form-urlencoded",
        "Authorization": f"Basic {base64_b64encode(f'{client_id}:{client_secret}'.encode()).decode()}"
    }
    
    payload = {
        "grant_type": "client_credentials",
        "scope": "personal_connections:write personal_connections:read"
    }
    
    try:
        response = requests.post(TOKEN_ENDPOINT, headers=headers, data=payload)
        response.raise_for_status()
        return response.json()["access_token"]
    except requests.exceptions.HTTPError as e:
        print(f"Authentication failed: {e.response.text}")
        raise

# Helper for base64 encoding in basic auth header
from base64 import b64encode

def base64_b64encode(data: bytes) -> bytes:
    return b64encode(data)

JavaScript Authentication Helper

const axios = require('axios');

const CXONE_API_BASE = 'https://api.nicecxone.com';
const TOKEN_ENDPOINT = `${CXONE_API_BASE}/api/v2/oauth/token`;

async function getAccessToken(clientId, clientSecret) {
    const auth = Buffer.from(`${clientId}:${clientSecret}`).toString('base64');
    
    try {
        const response = await axios.post(
            TOKEN_ENDPOINT,
            new URLSearchParams({
                grant_type: 'client_credentials',
                scope: 'personal_connections:write personal_connections:read'
            }).toString(),
            {
                headers: {
                    'Authorization': `Basic ${auth}`,
                    'Content-Type': 'application/x-www-form-urlencoded'
                }
            }
        );
        return response.data.access_token;
    } catch (error) {
        console.error('Authentication failed:', error.response?.data || error.message);
        throw error;
    }
}

Implementation

The Personal Connection API is designed to allow users (agents, supervisors, or admin users) to make calls directly from the CXone interface or via API as if they were clicking “Call” in the desktop client. The workflow involves two main steps:

  1. Create a Personal Connection: This establishes the intent to call a specific number.
  2. Execute the Call: This triggers the actual telephony leg.

Note: In many CXone configurations, these are combined into a single POST request to the /personal-connections endpoint with an action parameter, or handled via the specific /calls sub-resource depending on the exact API version and tenant configuration. The standard approach for “Personal Connection” specifically is using the personal-connections resource.

Step 1: Construct the Call Payload

The API expects a JSON body defining the target number, the type of connection, and the user context.

Key Fields:

  • to: The destination phone number in E.164 format (e.g., +14155552671).
  • type: Usually CALL for voice.
  • from: Optional. The outbound caller ID. If omitted, the tenant default or user default is used.

Step 2: Trigger the Outbound Call

We will use the POST /api/v2/personal-connections endpoint. This endpoint creates the connection and immediately attempts to dial if the action is set to CREATE (default behavior for new connections).

Python Implementation

import requests
import json
from typing import Dict, Any

def trigger_outbound_call(access_token: str, tenant_id: str, target_number: str, from_number: Optional[str] = None) -> Dict[str, Any]:
    """
    Triggers an outbound call using the CXone Personal Connection API.
    
    Args:
        access_token: Valid OAuth2 access token.
        tenant_id: Your CXone Tenant ID.
        target_number: E.164 format phone number to call.
        from_number: Optional E.164 format caller ID.
        
    Returns:
        The API response JSON containing the connection ID and status.
    """
    endpoint = f"{CXONE_API_BASE}/api/v2/personal-connections"
    
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "x-nice-tenant": tenant_id
    }
    
    # Construct the payload
    payload = {
        "to": target_number,
        "type": "CALL",
        "direction": "OUTBOUND"
    }
    
    if from_number:
        payload["from"] = from_number
        
    try:
        # The Personal Connection API is asynchronous in nature for some actions, 
        # but CREATE usually returns the connection object immediately.
        response = requests.post(endpoint, headers=headers, json=payload)
        
        # Handle 429 Rate Limiting
        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 1))
            print(f"Rate limited. Retrying after {retry_after} seconds...")
            import time
            time.sleep(retry_after)
            response = requests.post(endpoint, headers=headers, json=payload)
            
        response.raise_for_status()
        return response.json()
        
    except requests.exceptions.HTTPError as e:
        print(f"API Error: {e.response.status_code} - {e.response.text}")
        raise
    except Exception as e:
        print(f"Unexpected error: {str(e)}")
        raise

JavaScript Implementation

async function triggerOutboundCall(accessToken, tenantId, targetNumber, fromNumber = null) {
    const endpoint = `${CXONE_API_BASE}/api/v2/personal-connections`;
    
    const headers = {
        'Authorization': `Bearer ${accessToken}`,
        'Content-Type': 'application/json',
        'x-nice-tenant': tenantId
    };
    
    const payload = {
        to: targetNumber,
        type: 'CALL',
        direction: 'OUTBOUND'
    };
    
    if (fromNumber) {
        payload.from = fromNumber;
    }
    
    try {
        const response = await axios.post(endpoint, payload, { headers });
        return response.data;
    } catch (error) {
        if (error.response?.status === 429) {
            const retryAfter = error.response.headers['retry-after'] || 1;
            console.log(`Rate limited. Retrying after ${retryAfter} seconds...`);
            await new Promise(resolve => setTimeout(resolve, retryAfter * 1000));
            return await triggerOutboundCall(accessToken, tenantId, targetNumber, fromNumber);
        }
        console.error('API Error:', error.response?.data || error.message);
        throw error;
    }
}

Step 3: Processing Results

The response from POST /api/v2/personal-connections returns a PersonalConnection object. Crucially, it contains a id which you can use to track the call status later.

Expected Response Structure:

{
  "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "to": "+14155552671",
  "from": "+18005551234",
  "type": "CALL",
  "direction": "OUTBOUND",
  "status": "INITIATED",
  "createdTime": "2023-10-27T10:00:00.000Z",
  "modifiedTime": "2023-10-27T10:00:00.000Z"
}

The status field may vary (INITIATED, RINGING, ANSWERED, FAILED, COMPLETED). For real-time monitoring of the call state, you should poll the GET /api/v2/personal-connections/{id} endpoint or subscribe to CXone Event Streams (if available in your plan) for personal_connection events.

Complete Working Example

Below is a complete, runnable Python script. Save this as trigger_call.py. Ensure you have created a .env file with your credentials.

.env file:

CXONE_CLIENT_ID=your_client_id_here
CXONE_CLIENT_SECRET=your_client_secret_here
CXONE_TENANT_ID=your_tenant_id_here
CXONE_TARGET_NUMBER=+14155552671
CXONE_FROM_NUMBER=+18005551234

trigger_call.py:

import os
import requests
import time
import sys
from base64 import b64encode
from dotenv import load_dotenv
from typing import Optional, Dict, Any

# Load environment variables
load_dotenv()

CXONE_API_BASE = "https://api.nicecxone.com"
TOKEN_ENDPOINT = f"{CXONE_API_BASE}/api/v2/oauth/token"

def get_access_token() -> str:
    """Retrieves an OAuth2 access token from CXone."""
    client_id = os.getenv("CXONE_CLIENT_ID")
    client_secret = os.getenv("CXONE_CLIENT_SECRET")
    
    if not client_id or not client_secret:
        raise ValueError("CXONE_CLIENT_ID and CXONE_CLIENT_SECRET must be set in .env")
    
    auth_string = f"{client_id}:{client_secret}"
    auth_header = "Basic " + b64encode(auth_string.encode()).decode()
    
    headers = {
        "Content-Type": "application/x-www-form-urlencoded",
        "Authorization": auth_header
    }
    
    payload = {
        "grant_type": "client_credentials",
        "scope": "personal_connections:write personal_connections:read"
    }
    
    try:
        response = requests.post(TOKEN_ENDPOINT, headers=headers, data=payload)
        response.raise_for_status()
        return response.json()["access_token"]
    except requests.exceptions.HTTPError as e:
        print(f"Authentication failed: {e.response.text}")
        sys.exit(1)

def trigger_outbound_call(access_token: str, target_number: str, from_number: Optional[str] = None) -> Dict[str, Any]:
    """Triggers an outbound call using the CXone Personal Connection API."""
    tenant_id = os.getenv("CXONE_TENANT_ID")
    if not tenant_id:
        raise ValueError("CXONE_TENANT_ID must be set in .env")
        
    endpoint = f"{CXONE_API_BASE}/api/v2/personal-connections"
    
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "x-nice-tenant": tenant_id
    }
    
    payload = {
        "to": target_number,
        "type": "CALL",
        "direction": "OUTBOUND"
    }
    
    if from_number:
        payload["from"] = from_number
        
    max_retries = 3
    for attempt in range(max_retries):
        try:
            response = requests.post(endpoint, headers=headers, json=payload)
            
            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 1))
                print(f"Rate limited (429). Retrying after {retry_after} seconds... (Attempt {attempt + 1}/{max_retries})")
                time.sleep(retry_after)
                continue
                
            response.raise_for_status()
            return response.json()
            
        except requests.exceptions.HTTPError as e:
            if response.status_code in [401, 403]:
                print(f"Authentication/Authorization Error: {e.response.text}")
                sys.exit(1)
            elif response.status_code == 422:
                print(f"Validation Error: {e.response.text}")
                sys.exit(1)
            else:
                print(f"API Error: {e.response.status_code} - {e.response.text}")
                sys.exit(1)
                
    print("Max retries exceeded due to rate limiting.")
    sys.exit(1)

def main():
    target_number = os.getenv("CXONE_TARGET_NUMBER")
    from_number = os.getenv("CXONE_FROM_NUMBER")
    
    if not target_number:
        print("CXONE_TARGET_NUMBER must be set in .env")
        sys.exit(1)
        
    print(f"Initiating call to {target_number}...")
    
    try:
        token = get_access_token()
        result = trigger_outbound_call(token, target_number, from_number)
        
        print("Call triggered successfully!")
        print(f"Connection ID: {result.get('id')}")
        print(f"Status: {result.get('status')}")
        print(f"Response: {json.dumps(result, indent=2)}")
        
    except Exception as e:
        print(f"Failed to trigger call: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

  • Cause: The access token is expired, invalid, or missing the required scopes.
  • Fix: Ensure the token was generated recently. Verify that the scope parameter in the token request includes personal_connections:write. Check that the Client ID and Secret match the tenant.

Error: 403 Forbidden

  • Cause: The OAuth client does not have permission to use the Personal Connection API, or the tenant ID is incorrect.
  • Fix: Verify the x-nice-tenant header matches the tenant associated with the API client. Check the API Client settings in the CXone Admin Portal to ensure the “Personal Connections” permissions are granted.

Error: 400 Bad Request

  • Cause: Invalid phone number format or missing required fields.
  • Fix: Ensure the to and from numbers are in strict E.164 format (e.g., +1 prefix). Ensure the type is CALL and direction is OUTBOUND.

Error: 429 Too Many Requests

  • Cause: Exceeding the API rate limits. Personal Connection APIs often have lower rate limits than other endpoints to prevent abuse.
  • Fix: Implement exponential backoff. The code example above includes a basic retry mechanism with Retry-After header parsing.

Error: 422 Unprocessable Entity

  • Cause: The target number is invalid, blocked, or the caller ID (from) is not authorized for outbound dialing in the tenant configuration.
  • Fix: Verify that the from number is a valid, provisioned DID in your CXone tenant. Check if the target number is on a blocklist.

Official References