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

What You Will Build

• You will build a Python script that initiates an immediate outbound voice call to a specified destination using the NICE CXone Personal Connection API.
• This tutorial utilizes the NICE CXone REST API v3 for Personal Connection, specifically the POST /api/v3/pc/calls endpoint.
• The implementation is written in Python 3.9+ using the httpx library for asynchronous HTTP requests.

Prerequisites

• OAuth Client Type: You require a Service Account (Client Credentials Grant) or an Authorized User (Resource Owner Password Credentials Grant) with appropriate permissions. For automated outbound dialing, a Service Account is recommended for stability.
• Required Scopes: The OAuth token must include the scope pc:calls:create. If you need to monitor the call status subsequently, add pc:calls:read.
• SDK/API Version: This tutorial uses raw REST API calls via httpx to ensure transparency of the HTTP request. It targets the CXone API v3.
• Language/Runtime Requirements: Python 3.9 or higher.
• External Dependencies:
  • httpx: For async HTTP requests.
  • pydantic: For data validation (optional but recommended for robust code).

Install the dependencies:

pip install httpx pydantic

Authentication Setup

NICE CXone uses OAuth 2.0 for authentication. You must obtain an access token before making any API calls. The token is valid for one hour and must be refreshed.

The following code defines a helper class to handle the OAuth flow. It supports the Client Credentials grant type, which is standard for server-to-server integrations like outbound dialing.

import httpx
import json
from typing import Optional, Dict, Any

class CXoneAuth:
    def __init__(self, env: str, client_id: str, client_secret: str):
        """
        Initialize the CXone Authentication handler.
        
        Args:
            env: The CXone environment (e.g., 'us-22', 'eu-1', 'ap-1').
            client_id: Your OAuth Client ID.
            client_secret: Your OAuth Client Secret.
        """
        self.env = env
        self.client_id = client_id
        self.client_secret = client_secret
        
        # Determine the base URL based on the environment
        if env == 'prod':
            self.base_url = "https://api.nicecxone.com"
        else:
            self.base_url = f"https://{env}.api.nicecxone.com"
            
        self.token_url = f"{self.base_url}/oauth/token"
        self.access_token: Optional[str] = None
        self.refresh_token: Optional[str] = None

    async def get_access_token(self) -> str:
        """
        Fetches a new access token using Client Credentials flow.
        
        Returns:
            str: The OAuth access token.
        
        Raises:
            httpx.HTTPStatusError: If the authentication fails.
        """
        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }
        
        data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": "pc:calls:create pc:calls:read"
        }

        async with httpx.AsyncClient() as client:
            try:
                response = await client.post(
                    self.token_url, 
                    headers=headers, 
                    data=data
                )
                response.raise_for_status()
                token_data = response.json()
                self.access_token = token_data.get("access_token")
                return self.access_token
            except httpx.HTTPStatusError as e:
                print(f"Authentication failed with status {e.response.status_code}: {e.response.text}")
                raise

    def get_headers(self) -> Dict[str, str]:
        """
        Returns the headers required for CXone API requests.
        
        Returns:
            dict: Headers including Authorization and Content-Type.
        """
        if not self.access_token:
            raise ValueError("Access token not set. Call get_access_token() first.")
            
        return {
            "Authorization": f"Bearer {self.access_token}",
            "Content-Type": "application/json"
        }

Implementation

Step 1: Constructing the Call Payload

The Personal Connection API requires a specific JSON structure to initiate a call. The core component is the call object, which defines the type of call (voice, SMS, etc.) and the destination.

For an outbound voice call, you must specify:

1. type: Set to "voice".
2. to: The destination phone number in E.164 format (e.g., +14155552671).
3. from (Optional but recommended): The outbound caller ID number. This number must be provisioned in your CXone account.
4. callbackUrl (Optional): A webhook URL where CXone will post events about the call lifecycle (answered, disconnected, failed).

Here is the structure of the payload:

{
  "call": {
    "type": "voice",
    "to": "+14155552671",
    "from": "+18005551234",
    "callbackUrl": "https://your-server.com/webhook/cxone-calls"
  }
}

Important Note on from Number: If you omit the from field, CXone will use the default outbound number configured for the user or service account associated with the OAuth token. If no default is set, the call will fail with a 400 Bad Request error.

Step 2: Implementing the Outbound Call Logic

We will create a class PersonalConnectionClient that handles the API interaction. This class will use the CXoneAuth helper to get headers and send the POST request to the /api/v3/pc/calls endpoint.

import asyncio
from datetime import datetime

class PersonalConnectionClient:
    def __init__(self, auth: CXoneAuth):
        self.auth = auth
        self.base_url = f"{auth.base_url}/api/v3"
        self.endpoint = f"{self.base_url}/pc/calls"

    async def make_outbound_call(self, to_number: str, from_number: Optional[str] = None, callback_url: Optional[str] = None) -> Dict[str, Any]:
        """
        Initiates an outbound voice call.
        
        Args:
            to_number: The destination phone number in E.164 format.
            from_number: The outbound caller ID (optional).
            callback_url: URL to receive call events (optional).
            
        Returns:
            dict: The response from CXone containing the call ID and status.
        """
        # Construct the payload
        call_payload = {
            "call": {
                "type": "voice",
                "to": to_number
            }
        }
        
        # Add optional fields if provided
        if from_number:
            call_payload["call"]["from"] = from_number
            
        if callback_url:
            call_payload["call"]["callbackUrl"] = callback_url

        # Get authenticated headers
        headers = self.auth.get_headers()

        async with httpx.AsyncClient() as client:
            try:
                print(f"Initiating call to {to_number}...")
                response = await client.post(
                    self.endpoint,
                    headers=headers,
                    json=call_payload
                )
                
                # Check for success
                response.raise_for_status()
                result = response.json()
                print(f"Call initiated successfully. Call ID: {result.get('id')}")
                return result
                
            except httpx.HTTPStatusError as e:
                error_detail = e.response.text
                print(f"Failed to initiate call. Status: {e.response.status_code}")
                print(f"Error Detail: {error_detail}")
                raise
            except Exception as e:
                print(f"An unexpected error occurred: {str(e)}")
                raise

Step 3: Processing Results and Handling Asynchronous Nature

When you trigger a call via the Personal Connection API, the response is immediate. However, the call itself is asynchronous. The API returns a 201 Created status with a JSON body containing the id of the call.

Expected Response:

{
  "id": "call-uuid-12345-67890",
  "state": "queued",
  "type": "voice",
  "to": "+14155552671",
  "from": "+18005551234",
  "createdTime": "2023-10-27T10:00:00.000Z"
}

The state field will initially be "queued" or "ringing". To track the final disposition (answered, no-answer, busy), you should either:

1. Use the callbackUrl provided in the payload to receive webhooks.
2. Poll the GET /api/v3/pc/calls/{callId} endpoint.

For this tutorial, we will focus on the initiation. If you need to poll, you can extend the PersonalConnectionClient with a get_call_status method.

Complete Working Example

Below is the complete, runnable Python script. It combines authentication, payload construction, and the API call.

Instructions:

1. Save this code as cxone_outbound.py.
2. Install dependencies: pip install httpx.
3. Replace the placeholder values in the main function with your actual CXone credentials.

import httpx
import asyncio
import sys
from typing import Optional, Dict, Any

# --- Authentication Module ---

class CXoneAuth:
    def __init__(self, env: str, client_id: str, client_secret: str):
        self.env = env
        self.client_id = client_id
        self.client_secret = client_secret
        
        if env == 'prod':
            self.base_url = "https://api.nicecxone.com"
        else:
            self.base_url = f"https://{env}.api.nicecxone.com"
            
        self.token_url = f"{self.base_url}/oauth/token"
        self.access_token: Optional[str] = None

    async def get_access_token(self) -> str:
        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }
        
        data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": "pc:calls:create pc:calls:read"
        }

        async with httpx.AsyncClient() as client:
            try:
                response = await client.post(
                    self.token_url, 
                    headers=headers, 
                    data=data
                )
                response.raise_for_status()
                token_data = response.json()
                self.access_token = token_data.get("access_token")
                return self.access_token
            except httpx.HTTPStatusError as e:
                print(f"Authentication failed with status {e.response.status_code}: {e.response.text}")
                raise

    def get_headers(self) -> Dict[str, str]:
        if not self.access_token:
            raise ValueError("Access token not set. Call get_access_token() first.")
        return {
            "Authorization": f"Bearer {self.access_token}",
            "Content-Type": "application/json"
        }

# --- Personal Connection Client Module ---

class PersonalConnectionClient:
    def __init__(self, auth: CXoneAuth):
        self.auth = auth
        self.base_url = f"{auth.base_url}/api/v3"
        self.endpoint = f"{self.base_url}/pc/calls"

    async def make_outbound_call(self, to_number: str, from_number: Optional[str] = None, callback_url: Optional[str] = None) -> Dict[str, Any]:
        call_payload = {
            "call": {
                "type": "voice",
                "to": to_number
            }
        }
        
        if from_number:
            call_payload["call"]["from"] = from_number
            
        if callback_url:
            call_payload["call"]["callbackUrl"] = callback_url

        headers = self.auth.get_headers()

        async with httpx.AsyncClient() as client:
            try:
                print(f"Initiating call to {to_number}...")
                response = await client.post(
                    self.endpoint,
                    headers=headers,
                    json=call_payload
                )
                
                response.raise_for_status()
                result = response.json()
                print(f"Call initiated successfully.")
                print(f"Call ID: {result.get('id')}")
                print(f"State: {result.get('state')}")
                return result
                
            except httpx.HTTPStatusError as e:
                error_detail = e.response.text
                print(f"Failed to initiate call. Status: {e.response.status_code}")
                print(f"Error Detail: {error_detail}")
                raise
            except Exception as e:
                print(f"An unexpected error occurred: {str(e)}")
                raise

# --- Main Execution ---

async def main():
    # Configuration
    CXONE_ENV = "us-22"          # Replace with your environment (e.g., us-22, eu-1)
    CLIENT_ID = "your_client_id" # Replace with your Client ID
    CLIENT_SECRET = "your_client_secret" # Replace with your Client Secret
    
    # Call Details
    TO_NUMBER = "+14155552671"   # Replace with destination number (E.164)
    FROM_NUMBER = "+18005551234" # Replace with your provisioned outbound number (E.164)
    CALLBACK_URL = "https://webhook.site/your-unique-url" # Optional: Replace with your webhook URL

    try:
        # Step 1: Initialize Authentication
        auth = CXoneAuth(CXONE_ENV, CLIENT_ID, CLIENT_SECRET)
        await auth.get_access_token()
        
        # Step 2: Initialize Client
        pc_client = PersonalConnectionClient(auth)
        
        # Step 3: Make the Call
        await pc_client.make_outbound_call(
            to_number=TO_NUMBER,
            from_number=FROM_NUMBER,
            callback_url=CALLBACK_URL
        )
        
    except Exception as e:
        print(f"Execution failed: {e}")
        sys.exit(1)

if __name__ == "__main__":
    asyncio.run(main())

Common Errors & Debugging

Error: 401 Unauthorized

Cause: The OAuth token is invalid, expired, or missing.
Fix:

1. Ensure your client_id and client_secret are correct.
2. Verify that the token was successfully retrieved in the authentication step.
3. Check that the token has not expired. The example code fetches a fresh token on every run. In a long-running application, implement token caching and refresh logic.

Error: 403 Forbidden

Cause: The OAuth client does not have the required scopes.
Fix:

1. Go to the CXone Admin Console.
2. Navigate to Administration > Security > OAuth Clients.
3. Edit your client.
4. Ensure the scope pc:calls:create is checked.
5. Regenerate the token.

Error: 400 Bad Request - “Invalid phone number format”

Cause: The to or from number is not in E.164 format.
Fix:

1. Ensure the number starts with a + followed by the country code.
2. Example: +14155552671 is correct. 14155552671 or (415) 555-2671 is incorrect.
3. Use a library like phonenumbers in Python to validate and format numbers before sending them to the API.

Error: 400 Bad Request - “Outbound number not provisioned”

Cause: The from number provided is not associated with your CXone account or is not enabled for outbound calling.
Fix:

1. Log in to the CXone Admin Console.
2. Navigate to Administration > Communications > Numbers.
3. Verify that the number exists and is active.
4. Ensure that the number is assigned to the user or service account making the API call, or that it is designated as a default outbound number.

Error: 429 Too Many Requests

Cause: You have exceeded the rate limit for the Personal Connection API.
Fix:

1. Implement exponential backoff in your retry logic.
2. Check the Retry-After header in the response to determine how long to wait.
3. Contact NICE Support to review your rate limits if you have high-volume requirements.

Official References

• NICE CXone Personal Connection API Documentation
• NICE CXone OAuth 2.0 Documentation
• httpx Documentation