Generate Long-Lived API Tokens for CI/CD Pipelines

How-To-Articles
1

Generate Long-Lived API Tokens for CI/CD Pipelines

What You Will Build

  • Create a script that authenticates to Genesys Cloud or NICE CXone using client credentials to obtain an access token.
  • Implement robust token caching and automatic refresh logic to ensure the token remains valid throughout long-running CI/CD jobs.
  • Use Python and JavaScript to demonstrate production-ready implementations that handle rate limits and scope requirements.

Prerequisites

Genesys Cloud

  • OAuth Client Type: Service Account (Client Credentials Grant).
  • Required Scopes: Minimum conversation:call:view or user:profile:read depending on your pipeline tasks. For full administrative access, admin:all is often used but violates least-privilege principles.
  • SDK Version: genesys-cloud-sdk-python v10+ or @genesyscloud/api-client v6+.
  • Dependencies: requests, python-dotenv, cachetools.

NICE CXone

  • OAuth Client Type: Service Account (Client Credentials Grant).
  • Required Scopes: read:users, write:users, or specific resource scopes like read:interactions.
  • SDK Version: nice-cxone-sdk (Node.js) or direct REST via axios.
  • Dependencies: axios, dotenv, node-cache.

General

  • A CI/CD environment variable store for CLIENT_ID, CLIENT_SECRET, and SUB_DOMAIN (Genesys) or TENANT_ID (CXone).
  • Python 3.8+ or Node.js 16+.

Authentication Setup

CI/CD pipelines require non-interactive authentication. The standard password grant is unsuitable because it requires a username and password, which are subject to MFA and expiration policies. The Client Credentials Grant is the correct flow for service-to-service communication.

This flow exchanges a client_id and client_secret for a short-lived access token (typically 1 hour for Genesys, 1 hour for CXone). Because CI/CD jobs can run for hours, the token will expire mid-execution. Therefore, the implementation must cache the token and request a new one only when necessary.

Genesys Cloud OAuth Endpoint

  • URL: https://api.mypurecloud.com/oauth/token
  • Method: POST
  • Content-Type: application/x-www-form-urlencoded

NICE CXone OAuth Endpoint

  • URL: https://api.cxone.com/oauth/token
  • Method: POST
  • Content-Type: application/x-www-form-urlencoded

Implementation

Step 1: Create a Token Cache with Expiration Logic

The first step is to build a wrapper that manages the lifecycle of the token. We do not want to hit the OAuth endpoint for every single API call. We also do not want to crash when the token expires.

We will use a simple in-memory cache with a TTL (Time-To-Live). To be safe, we will invalidate the cache 60 seconds before the actual token expiration reported by the OAuth server.

Python Implementation (Genesys Cloud)

import requests
import time
from typing import Optional
import os

class GenesysTokenManager:
    def __init__(self, client_id: str, client_secret: str, sub_domain: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.sub_domain = sub_domain
        self.token_url = f"https://api.{sub_domain}.mypurecloud.com/oauth/token"
        
        # Cache state
        self.access_token: Optional[str] = None
        self.expires_at: float = 0.0
        self.token_buffer_seconds = 60  # Refresh 60s before actual expiry

    def _is_token_valid(self) -> bool:
        """Check if the current token is still valid considering the buffer."""
        return self.access_token is not None and time.time() < (self.expires_at - self.token_buffer_seconds)

    def get_token(self) -> str:
        """
        Returns a valid access token.
        If the current token is expired or invalid, it fetches a new one.
        """
        if self._is_token_valid():
            return self.access_token

        # Token is expired or not set, fetch new one
        self._refresh_token()
        return self.access_token

    def _refresh_token(self):
        """
        Performs the client credentials grant to obtain a new token.
        """
        payload = {
            'grant_type': 'client_credentials',
            'client_id': self.client_id,
            'client_secret': self.client_secret
        }

        try:
            response = requests.post(self.token_url, data=payload, timeout=30)
            response.raise_for_status()
            data = response.json()

            self.access_token = data['access_token']
            # expires_in is in seconds, convert to absolute timestamp
            self.expires_at = time.time() + data['expires_in']
            
            print(f"Token refreshed. Expires in {data['expires_in']} seconds.")

        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise Exception("Invalid Client ID or Secret. Check your environment variables.") from e
            elif response.status_code == 429:
                raise Exception("Rate limited on OAuth endpoint. Wait before retrying.") from e
            else:
                raise Exception(f"OAuth failed with status {response.status_code}: {response.text}") from e
        except requests.exceptions.RequestException as e:
            raise Exception(f"Network error during token refresh: {str(e)}") from e

JavaScript Implementation (NICE CXone)

const axios = require('axios');

class CXoneTokenManager {
    constructor(clientId, clientSecret) {
        this.clientId = clientId;
        this.clientSecret = clientSecret;
        this.tokenUrl = 'https://api.cxone.com/oauth/token';
        
        this.accessToken = null;
        this.expiresAt = 0;
        this.tokenBufferSeconds = 60; // Refresh 60s before actual expiry
    }

    _isTokenValid() {
        return this.accessToken !== null && Date.now() < (this.expiresAt - (this.tokenBufferSeconds * 1000));
    }

    async getToken() {
        if (this._isTokenValid()) {
            return this.accessToken;
        }

        await this._refreshToken();
        return this.accessToken;
    }

    async _refreshToken() {
        const payload = new URLSearchParams();
        payload.append('grant_type', 'client_credentials');
        payload.append('client_id', this.clientId);
        payload.append('client_secret', this.clientSecret);

        try {
            const response = await axios.post(this.tokenUrl, payload, {
                headers: {
                    'Content-Type': 'application/x-www-form-urlencoded'
                },
                timeout: 30000
            });

            this.accessToken = response.data.access_token;
            // expires_in is in seconds, convert to ms timestamp
            this.expiresAt = Date.now() + (response.data.expires_in * 1000);
            
            console.log(`Token refreshed. Expires in ${response.data.expires_in} seconds.`);

        } catch (error) {
            if (error.response) {
                if (error.response.status === 401) {
                    throw new Error("Invalid Client ID or Secret.");
                } else if (error.response.status === 429) {
                    throw new Error("Rate limited on OAuth endpoint.");
                } else {
                    throw new Error(`OAuth failed: ${error.response.status} - ${error.response.data}`);
                }
            } else {
                throw new Error(`Network error during token refresh: ${error.message}`);
            }
        }
    }
}

module.exports = CXoneTokenManager;

Step 2: Integrate Token Manager with API Calls

Now that we have a token manager, we must integrate it into the actual API calls. The key is to call get_token() immediately before making any REST request or SDK call. This ensures that if the job has been running for 45 minutes, the token is refreshed automatically before the 50-minute mark.

Genesys Cloud: Using the Python SDK with Custom Auth

The Genesys Cloud Python SDK allows you to inject a custom authentication provider. This is cleaner than manually adding headers to every request.

from purecloud_platform_client import Configuration, PureCloudAuthProvider, PureCloudPlatformClientV2
import os

class CustomTokenAuth(PureCloudAuthProvider):
    def __init__(self, token_manager: GenesysTokenManager):
        self.token_manager = token_manager

    def get_access_token(self) -> str:
        # This method is called by the SDK whenever it needs a token
        return self.token_manager.get_token()

def create_genesys_client():
    client_id = os.getenv('GENESYS_CLIENT_ID')
    client_secret = os.getenv('GENESYS_CLIENT_SECRET')
    sub_domain = os.getenv('GENESYS_SUB_DOMAIN')

    if not all([client_id, client_secret, sub_domain]):
        raise ValueError("Missing environment variables for Genesys Cloud.")

    # 1. Initialize Token Manager
    token_mgr = GenesysTokenManager(client_id, client_secret, sub_domain)

    # 2. Create Configuration with Custom Auth
    config = Configuration()
    config.host = f"https://api.{sub_domain}.mypurecloud.com"
    
    # Inject our custom auth provider
    config.auth_provider = CustomTokenAuth(token_mgr)

    # 3. Initialize the Platform Client
    client = PureCloudPlatformClientV2(config)
    return client

NICE CXone: Using Axios with Interceptors

For CXone, we will use Axios interceptors to automatically attach the token and handle 401 errors if they occur unexpectedly (e.g., token revoked server-side).

const axios = require('axios');
const CXoneTokenManager = require('./CXoneTokenManager');
require('dotenv').config();

const CXoneApiClient = class {
    constructor() {
        this.clientId = process.env.CXONE_CLIENT_ID;
        this.clientSecret = process.env.CXONE_CLIENT_SECRET;
        this.tokenManager = new CXoneTokenManager(this.clientId, this.clientSecret);
        
        // Create a persistent Axios instance
        this.api = axios.create({
            baseURL: 'https://api.cxone.com',
            headers: {
                'Content-Type': 'application/json'
            }
        });

        this._setupInterceptors();
    }

    _setupInterceptors() {
        // Request Interceptor: Attach Token
        this.api.interceptors.request.use(async (config) => {
            const token = await this.tokenManager.getToken();
            config.headers['Authorization'] = `Bearer ${token}`;
            return config;
        }, (error) => {
            return Promise.reject(error);
        });

        // Response Interceptor: Handle 401 (Unauthorized)
        this.api.interceptors.response.use(
            (response) => response,
            async (error) => {
                const originalRequest = error.config;
                
                // If 401 and we haven't already retried
                if (error.response.status === 401 && !originalRequest._retry) {
                    originalRequest._retry = true;
                    try {
                        // Force a refresh
                        await this.tokenManager._refreshToken();
                        const newToken = await this.tokenManager.getToken();
                        originalRequest.headers['Authorization'] = `Bearer ${newToken}`;
                        return this.api(originalRequest);
                    } catch (refreshError) {
                        return Promise.reject(refreshError);
                    }
                }
                
                return Promise.reject(error);
            }
        );
    }

    // Example Method: Get User Profile
    async getUser(userId) {
        const response = await this.api.get(`/users/${userId}`);
        return response.data;
    }
};

module.exports = CXoneApiClient;

Step 3: Handling Pagination and Large Data Sets

CI/CD pipelines often export large datasets (e.g., all users, all interactions). These endpoints are paginated. A common mistake is to request the token once at the start of the script and then loop through pages for 20 minutes. The token will expire, and subsequent pages will return 401 errors.

By using the token manager from Step 1 and Step 2, every page request automatically checks the token validity.

Genesys Cloud: Paginated User Export

def export_all_users(client):
    """
    Exports all users from Genesys Cloud.
    Demonstrates automatic token refresh across pages.
    """
    users_api = client.users
    all_users = []
    page_size = 100
    page_number = 1
    total_records = 0

    print("Starting user export...")

    while True:
        try:
            # The SDK calls get_access_token() internally before making this request
            response = users_api.post_users_search(
                body={
                    "pageSize": page_size,
                    "pageNumber": page_number,
                    "query": "active:true" # Example filter
                }
            )

            # Update total records for loop control
            if page_number == 1:
                total_records = response.total

            all_users.extend(response.entities)
            print(f"Fetched page {page_number}. Total users so far: {len(all_users)}")

            # Check if there are more pages
            if page_number * page_size >= total_records:
                break
            
            page_number += 1

        except Exception as e:
            # If the error is related to auth, the TokenManager should have handled it.
            # If not, it might be a 429 or 5xx.
            print(f"Error fetching page {page_number}: {e}")
            raise

    print(f"Export complete. Total users: {len(all_users)}")
    return all_users

NICE CXone: Paginated Interaction Query

async function exportAllInteractions(apiClient) {
    const allInteractions = [];
    let pageNumber = 1;
    const pageSize = 100;
    let hasNextPage = true;

    console.log("Starting interaction export...");

    while (hasNextPage) {
        try {
            // The interceptor ensures a valid token is attached
            const response = await apiClient.api.post('/interactions/search', {
                pageSize: pageSize,
                pageNumber: pageNumber,
                query: {
                    type: 'interaction',
                    filters: [] // Add filters as needed
                }
            });

            const { data, pagination } = response.data;
            allInteractions.push(...data);

            console.log(`Fetched page ${pageNumber}. Total interactions so far: ${allInteractions.length}`);

            // Check pagination
            if (pageNumber >= pagination.totalPages) {
                hasNextPage = false;
            } else {
                pageNumber++;
            }

        } catch (error) {
            console.error(`Error fetching page ${pageNumber}:`, error.message);
            throw error;
        }
    }

    console.log(`Export complete. Total interactions: ${allInteractions.length}`);
    return allInteractions;
}

Complete Working Example

Below is a complete, runnable Python script for Genesys Cloud that sets up the environment, initializes the token manager, and performs a sample API call.

File: genesys_ci_cd_demo.py

import os
import sys
import requests
import time
from typing import Optional

# --- Configuration ---
# In a real CI/CD pipeline, these come from environment variables
CLIENT_ID = os.getenv('GENESYS_CLIENT_ID')
CLIENT_SECRET = os.getenv('GENESYS_CLIENT_SECRET')
SUB_DOMAIN = os.getenv('GENESYS_SUB_DOMAIN')

if not all([CLIENT_ID, CLIENT_SECRET, SUB_DOMAIN]):
    print("Error: Missing environment variables.")
    print("Please set GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET, and GENESYS_SUB_DOMAIN.")
    sys.exit(1)

# --- Token Manager ---
class GenesysTokenManager:
    def __init__(self, client_id: str, client_secret: str, sub_domain: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.sub_domain = sub_domain
        self.token_url = f"https://api.{sub_domain}.mypurecloud.com/oauth/token"
        self.access_token: Optional[str] = None
        self.expires_at: float = 0.0
        self.token_buffer_seconds = 60

    def _is_token_valid(self) -> bool:
        return self.access_token is not None and time.time() < (self.expires_at - self.token_buffer_seconds)

    def get_token(self) -> str:
        if self._is_token_valid():
            return self.access_token
        self._refresh_token()
        return self.access_token

    def _refresh_token(self):
        payload = {
            'grant_type': 'client_credentials',
            'client_id': self.client_id,
            'client_secret': self.client_secret
        }
        try:
            response = requests.post(self.token_url, data=payload, timeout=30)
            response.raise_for_status()
            data = response.json()
            self.access_token = data['access_token']
            self.expires_at = time.time() + data['expires_in']
            print(f"[Auth] Token refreshed. Expires in {data['expires_in']}s.")
        except requests.exceptions.HTTPError as e:
            raise Exception(f"OAuth Error {response.status_code}: {response.text}") from e

# --- API Caller ---
class GenesysApiClient:
    def __init__(self, token_manager: GenesysTokenManager, sub_domain: str):
        self.token_manager = token_manager
        self.base_url = f"https://api.{sub_domain}.mypurecloud.com"

    def get_user(self, user_id: str):
        """Fetches a specific user by ID."""
        token = self.token_manager.get_token()
        headers = {
            'Authorization': f'Bearer {token}',
            'Content-Type': 'application/json'
        }
        
        url = f"{self.base_url}/api/v2/users/{user_id}"
        response = requests.get(url, headers=headers, timeout=30)
        response.raise_for_status()
        return response.json()

# --- Main Execution ---
def main():
    print(f"Initializing CI/CD script for Sub-domain: {SUB_DOMAIN}")
    
    # 1. Setup
    token_mgr = GenesysTokenManager(CLIENT_ID, CLIENT_SECRET, SUB_DOMAIN)
    client = GenesysApiClient(token_mgr, SUB_DOMAIN)

    # 2. Simulate a long-running job
    # We will fetch the same user 3 times with a delay to demonstrate token validity checks
    # In a real scenario, this would be different API calls
    
    # Note: You need a valid user ID from your org. 
    # For this demo, we will try to get the first user via search to get an ID.
    
    try:
        # Get a user ID first (using the token manager internally)
        token = token_mgr.get_token()
        headers = {'Authorization': f'Bearer {token}', 'Content-Type': 'application/json'}
        search_url = f"{client.base_url}/api/v2/users/search"
        search_body = {"query": "active:true", "pageSize": 1}
        
        search_resp = requests.post(search_url, headers=headers, json=search_body)
        search_resp.raise_for_status()
        user_data = search_resp.json()
        
        if not user_data['entities']:
            print("No active users found.")
            return

        target_user_id = user_data['entities'][0]['id']
        print(f"Target User ID: {target_user_id}")

        # Simulate work
        for i in range(1, 4):
            print(f"\n--- Iteration {i} ---")
            print("Checking token validity...")
            user_profile = client.get_user(target_user_id)
            print(f"Successfully fetched user: {user_profile['name']}")
            
            # Add a delay to simulate processing time
            # In a real CI/CD, this might be 10-30 minutes
            if i < 3:
                print("Simulating long task... (sleeping 5s)")
                time.sleep(5)

    except Exception as e:
        print(f"Critical Error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

Cause: The client_id or client_secret is incorrect, or the service account has been disabled.
Fix: Verify the credentials in your CI/CD environment variables. Ensure the Service Account is enabled in the Genesys Cloud Admin Center or CXone Admin Console.
Code Check:

if response.status_code == 401:
    print("Credentials are invalid. Check CLIENT_ID and CLIENT_SECRET.")

Error: 403 Forbidden

Cause: The Service Account lacks the required OAuth scopes for the specific API endpoint.
Fix:

  1. Identify the endpoint (e.g., /api/v2/users).
  2. Check the documentation for required scopes (e.g., user:profile:read).
  3. Edit the Service Account in the Admin Console and add the missing scopes.
    Note: Scope changes can take up to 15 minutes to propagate.

Error: 429 Too Many Requests

Cause: You are hitting the OAuth endpoint too frequently. This happens if you request a new token for every single API call instead of caching it.
Fix: Ensure your TokenManager implements caching with a TTL. Do not call post_oauth_token more than once per hour (or once per expires_in duration minus buffer).
Code Check:

# Correct: Check cache first
if self._is_token_valid():
    return self.access_token
# Only refresh if expired
self._refresh_token()

Error: Token Expired Mid-Request

Cause: The expires_in value was not correctly converted to an absolute timestamp, or the server clock is skewed.
Fix: Use time.time() (Python) or Date.now() (JS) to calculate the expiration absolute timestamp. Always subtract a buffer (e.g., 60 seconds) to account for network latency and clock drift.

Official References