Implementing Branching Logic in NICE CXone Studio Using ASSIGN and IF Actions

StarAdmin · May 15, 2026, 9:00am

Implementing Branching Logic in NICE CXone Studio Using ASSIGN and IF Actions

What You Will Build

A CXone Studio workflow that evaluates incoming call data, assigns a dynamic variable based on caller ID, and routes the interaction to different queues using conditional branching.
This tutorial uses the NICE CXone Studio REST API to programmatically create and deploy a workflow configuration.
The implementation is covered in Python using the requests library and the NICE CXone Python SDK.

Prerequisites

OAuth Client: A NICE CXone Studio Client ID and Client Secret with the studio:workflow:write and studio:workflow:read scopes.
SDK Version: nice-cxone-python-sdk version 100+ (or equivalent HTTP client).
Language/Runtime: Python 3.8+.
External Dependencies: pip install nice-cxone-python-sdk requests.
Environment: Access to a NICE CXone Studio environment with permissions to create and deploy workflows.

Authentication Setup

NICE CXone uses OAuth 2.0 for authentication. You must obtain an access token before making any API calls. The following code demonstrates how to retrieve a token using the client credentials flow.

import requests
import os
import json

# Configuration
CLIENT_ID = os.getenv("CXONE_CLIENT_ID")
CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET")
BASE_URL = "https://api.mypurecloud.com" # Replace with your specific region if different, e.g., api.us-east-1.mypurecloud.com

def get_access_token():
    """
    Retrieves an OAuth 2.0 access token from NICE CXone.
    Returns:
        str: The access token.
    Raises:
        Exception: If the token request fails.
    """
    url = f"{BASE_URL}/oauth/token"
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    data = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "scope": "studio:workflow:write studio:workflow:read"
    }

    response = requests.post(url, headers=headers, data=data)
    
    if response.status_code != 200:
        raise Exception(f"Failed to obtain access token. Status: {response.status_code}, Response: {response.text}")
    
    token_data = response.json()
    return token_data["access_token"]

# Example usage
# token = get_access_token()
# print(f"Token: {token[:10]}...")

Implementation

Step 1: Define the Workflow Structure

NICE CXone Studio workflows are defined as JSON documents. The core components for branching logic are the ASSIGN action (to set variables) and the IF action (to evaluate conditions).

The workflow structure consists of:

Triggers: Define when the workflow starts (e.g., inbound call).
Actions: Sequence of operations.
Conditions: Logic for branching.

We will create a workflow that:

Assigns the CallerID to a variable named caller_number.
Checks if caller_number starts with “1800”.
If true, routes to the “Sales” queue.
If false, routes to the “Support” queue.

def build_workflow_definition():
    """
    Constructs the JSON payload for the CXone Studio workflow.
    Returns:
        dict: The workflow definition dictionary.
    """
    workflow = {
        "name": "BranchingLogicExample",
        "description": "Uses ASSIGN and IF to route calls based on CallerID",
        "type": "IVR",
        "triggers": [
            {
                "type": "inboundCall",
                "id": "trigger_01",
                "properties": {
                    "phoneNumber": "+15551234567" # Replace with your test number
                }
            }
        ],
        "actions": [
            {
                "id": "assign_01",
                "type": "ASSIGN",
                "name": "Set Caller Number",
                "properties": {
                    "variableName": "caller_number",
                    "value": "${CallerID}" # Built-in variable for inbound CallerID
                },
                "next": "if_01"
            },
            {
                "id": "if_01",
                "type": "IF",
                "name": "Check for Toll-Free Number",
                "properties": {
                    "condition": {
                        "type": "string",
                        "operator": "startsWith",
                        "left": "${caller_number}",
                        "right": "1800"
                    }
                },
                "trueNext": "route_01",
                "falseNext": "route_02"
            },
            {
                "id": "route_01",
                "type": "ROUTE",
                "name": "Route to Sales",
                "properties": {
                    "queueId": "sales_queue_id_here" # Replace with actual Queue ID
                },
                "next": None # End of workflow
            },
            {
                "id": "route_02",
                "type": "ROUTE",
                "name": "Route to Support",
                "properties": {
                    "queueId": "support_queue_id_here" # Replace with actual Queue ID
                },
                "next": None # End of workflow
            }
        ]
    }
    return workflow

Step 2: Create the Workflow via API

Once the definition is built, you must POST it to the Studio API. The endpoint is /api/v2/studio/workflows.

def create_workflow(token: str, workflow_def: dict):
    """
    Creates a new workflow in NICE CXone Studio.
    
    Args:
        token (str): The OAuth access token.
        workflow_def (dict): The workflow definition dictionary.
        
    Returns:
        dict: The response containing the workflow ID.
    """
    url = f"{BASE_URL}/api/v2/studio/workflows"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    # Validate JSON structure before sending
    try:
        json.dumps(workflow_def)
    except TypeError as e:
        raise Exception(f"Invalid workflow definition structure: {e}")

    response = requests.post(url, headers=headers, json=workflow_def)
    
    if response.status_code == 201:
        return response.json()
    elif response.status_code == 400:
        raise Exception(f"Bad Request: {response.json()}")
    elif response.status_code == 401:
        raise Exception("Unauthorized. Check your token.")
    elif response.status_code == 403:
        raise Exception("Forbidden. Check your scopes.")
    else:
        raise Exception(f"Failed to create workflow. Status: {response.status_code}, Response: {response.text}")

# Example usage:
# token = get_access_token()
# workflow_def = build_workflow_definition()
# result = create_workflow(token, workflow_def)
# print(f"Workflow ID: {result['id']}")

Step 3: Deploy the Workflow

Creating a workflow does not make it active. You must deploy it to the production environment. Deployment is asynchronous. You must poll the deployment status until it completes or fails.

import time

def deploy_workflow(token: str, workflow_id: str):
    """
    Deploys a workflow and polls for completion.
    
    Args:
        token (str): The OAuth access token.
        workflow_id (str): The ID of the workflow to deploy.
        
    Returns:
        dict: The final deployment status.
    """
    deploy_url = f"{BASE_URL}/api/v2/studio/workflows/{workflow_id}/deploy"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    # Initiate deployment
    response = requests.post(deploy_url, headers=headers, json={})
    
    if response.status_code != 202:
        raise Exception(f"Deployment initiation failed. Status: {response.status_code}, Response: {response.text}")
    
    deploy_id = response.json()["id"]
    print(f"Deployment initiated with ID: {deploy_id}")
    
    # Poll for status
    status_url = f"{BASE_URL}/api/v2/studio/workflows/{workflow_id}/deployments/{deploy_id}"
    
    for _ in range(30): # Poll up to 30 times
        time.sleep(5) # Wait 5 seconds between polls
        
        status_response = requests.get(status_url, headers=headers)
        
        if status_response.status_code != 200:
            raise Exception(f"Failed to get deployment status. Status: {status_response.status_code}")
        
        status_data = status_response.json()
        status = status_data["status"]
        
        print(f"Current Deployment Status: {status}")
        
        if status == "SUCCESS":
            return status_data
        elif status == "FAILED":
            raise Exception(f"Deployment failed: {status_data.get('error', 'Unknown error')}")
        elif status in ["IN_PROGRESS", "QUEUED"]:
            continue
        else:
            raise Exception(f"Unexpected deployment status: {status}")
            
    raise Exception("Deployment timed out.")

# Example usage:
# workflow_id = result["id"]
# deploy_result = deploy_workflow(token, workflow_id)
# print("Deployment successful!")

Complete Working Example

The following script combines all steps into a single executable module. Replace the placeholder values for CLIENT_ID, CLIENT_SECRET, and queue IDs with your actual environment values.

import requests
import os
import json
import time

# Configuration
CLIENT_ID = os.getenv("CXONE_CLIENT_ID", "your_client_id_here")
CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET", "your_client_secret_here")
BASE_URL = "https://api.mypurecloud.com" # Adjust for your region
SALES_QUEUE_ID = "your_sales_queue_id_here"
SUPPORT_QUEUE_ID = "your_support_queue_id_here"
TEST_PHONE_NUMBER = "+15551234567"

def get_access_token():
    url = f"{BASE_URL}/oauth/token"
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    data = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "scope": "studio:workflow:write studio:workflow:read"
    }
    response = requests.post(url, headers=headers, data=data)
    if response.status_code != 200:
        raise Exception(f"Auth failed: {response.text}")
    return response.json()["access_token"]

def build_workflow_definition():
    return {
        "name": "BranchingLogicExample",
        "description": "Routes calls based on CallerID prefix",
        "type": "IVR",
        "triggers": [
            {
                "type": "inboundCall",
                "id": "trigger_01",
                "properties": {"phoneNumber": TEST_PHONE_NUMBER}
            }
        ],
        "actions": [
            {
                "id": "assign_01",
                "type": "ASSIGN",
                "name": "Set Caller Number",
                "properties": {
                    "variableName": "caller_number",
                    "value": "${CallerID}"
                },
                "next": "if_01"
            },
            {
                "id": "if_01",
                "type": "IF",
                "name": "Check for Toll-Free",
                "properties": {
                    "condition": {
                        "type": "string",
                        "operator": "startsWith",
                        "left": "${caller_number}",
                        "right": "1800"
                    }
                },
                "trueNext": "route_sales",
                "falseNext": "route_support"
            },
            {
                "id": "route_sales",
                "type": "ROUTE",
                "name": "Route to Sales",
                "properties": {"queueId": SALES_QUEUE_ID},
                "next": None
            },
            {
                "id": "route_support",
                "type": "ROUTE",
                "name": "Route to Support",
                "properties": {"queueId": SUPPORT_QUEUE_ID},
                "next": None
            }
        ]
    }

def create_workflow(token, workflow_def):
    url = f"{BASE_URL}/api/v2/studio/workflows"
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    response = requests.post(url, headers=headers, json=workflow_def)
    if response.status_code == 201:
        return response.json()
    raise Exception(f"Create failed: {response.status_code} - {response.text}")

def deploy_workflow(token, workflow_id):
    deploy_url = f"{BASE_URL}/api/v2/studio/workflows/{workflow_id}/deploy"
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    response = requests.post(deploy_url, headers=headers, json={})
    if response.status_code != 202:
        raise Exception(f"Deploy init failed: {response.status_code} - {response.text}")
    
    deploy_id = response.json()["id"]
    status_url = f"{BASE_URL}/api/v2/studio/workflows/{workflow_id}/deployments/{deploy_id}"
    
    for _ in range(30):
        time.sleep(5)
        status_resp = requests.get(status_url, headers=headers)
        if status_resp.status_code != 200:
            raise Exception(f"Status check failed: {status_resp.status_code}")
        
        status_data = status_resp.json()
        status = status_data["status"]
        print(f"Status: {status}")
        
        if status == "SUCCESS":
            return status_data
        elif status == "FAILED":
            raise Exception(f"Deploy failed: {status_data.get('error', 'Unknown')}")
            
    raise Exception("Deploy timed out")

def main():
    try:
        print("1. Authenticating...")
        token = get_access_token()
        
        print("2. Building workflow definition...")
        workflow_def = build_workflow_definition()
        
        print("3. Creating workflow...")
        result = create_workflow(token, workflow_def)
        workflow_id = result["id"]
        print(f"Workflow created with ID: {workflow_id}")
        
        print("4. Deploying workflow...")
        deploy_result = deploy_workflow(token, workflow_id)
        print("Deployment successful!")
        
    except Exception as e:
        print(f"Error: {e}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 400 Bad Request - Invalid Workflow Structure

What causes it: The JSON payload does not conform to the Studio schema. Common issues include missing next fields in actions, incorrect type values, or invalid operator names in conditions.
How to fix it: Validate the JSON against the Studio Workflow API Schema. Ensure every action except the last one has a next ID pointing to a valid action ID.

Code showing the fix:

# Ensure 'next' is defined for non-terminal actions
"actions": [
    {
        "id": "assign_01",
        "type": "ASSIGN",
        "properties": {...},
        "next": "if_01" # Must match an existing action ID
    }
]

Error: 401 Unauthorized

What causes it: The access token is expired, invalid, or missing the required scopes.
How to fix it: Re-authenticate using the get_access_token function. Ensure the client credentials have studio:workflow:write scope.

Code showing the fix:

# Add error handling for token refresh
if response.status_code == 401:
    print("Token expired. Re-authenticating...")
    token = get_access_token()

Error: 403 Forbidden

What causes it: The OAuth client does not have permission to modify workflows in the target environment.
How to fix it: Contact your NICE CXone administrator to grant the necessary permissions to the client ID.

Error: Deployment Failed - “Queue ID not found”

What causes it: The queueId specified in the ROUTE action does not exist or is not accessible.
How to fix it: Verify the Queue IDs using the CXone API.
```
curl -X GET "https://api.mypurecloud.com/api/v2/routing/queues" \
     -H "Authorization: Bearer $TOKEN"
```
Use the actual id from the response in your workflow definition.

Implementing Branching Logic in NICE CXone Studio Using ASSIGN and IF Actions

Implementing Branching Logic in NICE CXone Studio Using ASSIGN and IF Actions

What You Will Build

Prerequisites

Authentication Setup

Implementation

Step 1: Define the Workflow Structure

Step 2: Create the Workflow via API

Step 3: Deploy the Workflow

Complete Working Example

Common Errors & Debugging

Error: 400 Bad Request - Invalid Workflow Structure

Error: 401 Unauthorized

Error: 403 Forbidden

Error: Deployment Failed - “Queue ID not found”

Official References