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

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

What You Will Build

  • You will build a CXone Studio flow that evaluates an incoming caller’s account balance and routes them to different queue groups based on specific thresholds.
  • This tutorial uses the NICE CXone Studio API to programmatically deploy a flow containing ASSIGN and IF actions.
  • The implementation is covered in Python using the requests library for direct API interaction, as the CXone Studio API is primarily REST-based and does not have a dedicated high-level SDK for flow topology definition.

Prerequisites

  • OAuth Client: A CXone OAuth client with the studio:flows:write and studio:flows:read scopes.
  • CXone Tenant: Access to a CXone tenant with Studio enabled.
  • Runtime: Python 3.8+ with the requests library installed (pip install requests).
  • Knowledge: Basic understanding of CXone Flow JSON structure and how actionId references work within the flow graph.

Authentication Setup

CXone uses OAuth 2.0 for API authentication. You must obtain a bearer token before making any requests to the Studio API. The following Python code demonstrates how to retrieve and cache a token.

import requests
import json
import time
from typing import Optional

class CXoneAuth:
    def __init__(self, realm: str, client_id: str, client_secret: str):
        self.realm = realm
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = f"https://{realm}.niceincontact.com/oauth2/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0

    def get_token(self) -> str:
        """
        Retrieves an OAuth access token.
        Implements basic caching to avoid unnecessary requests.
        """
        if self.access_token and time.time() < self.token_expiry:
            return self.access_token

        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }
        data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        try:
            response = requests.post(self.token_url, headers=headers, data=data)
            response.raise_for_status()
            token_data = response.json()
            
            self.access_token = token_data["access_token"]
            # Set expiry slightly earlier than actual to prevent edge-case expiration
            self.token_expiry = time.time() + (token_data["expires_in"] - 10)
            
            return self.access_token
        except requests.exceptions.RequestException as e:
            raise Exception(f"Failed to obtain OAuth token: {e}")

    def get_headers(self) -> dict:
        """
        Returns headers required for CXone API requests.
        """
        return {
            "Authorization": f"Bearer {self.get_token()}",
            "Content-Type": "application/json"
        }

Implementation

Step 1: Defining the Flow Structure with ASSIGN Actions

In CXone Studio, a flow is a directed graph of nodes. To implement branching logic, you first need to store data. The ASSIGN action allows you to set variables that can be evaluated later.

We will create a flow that assigns the caller’s account balance (simulated here for demonstration) to a variable named accountBalance.

Key Concepts:

  • type: Must be assign for assignment actions.
  • name: The display name of the action in the Studio UI.
  • assignments: An array of objects defining the variable name and value.
  • variableName: The name of the variable to set.
  • value: The expression or literal value to assign.
def create_assign_action(action_id: str, var_name: str, var_value: str) -> dict:
    """
    Creates a JSON representation of an ASSIGN action.
    
    Args:
        action_id: Unique identifier for this node in the flow graph.
        var_name: The name of the variable to create/update.
        var_value: The value to assign. Can be a string, number, or expression.
    
    Returns:
        A dictionary representing the ASSIGN action node.
    """
    return {
        "id": action_id,
        "type": "assign",
        "name": f"Assign {var_name}",
        "assignments": [
            {
                "variableName": var_name,
                "value": var_value,
                "dataType": "number" # Explicitly define type for safety
            }
        ]
    }

Step 2: Implementing Branching Logic with IF Actions

The IF action evaluates a condition and routes the flow to different child nodes based on the result. Unlike traditional programming languages, CXone Studio uses a specific JSON structure for conditions.

Key Concepts:

  • type: Must be if.
  • condition: The logical expression to evaluate.
  • children: An object mapping outcomes (true, false) to the actionId of the next node.
  • else: Optional. If not specified, the flow ends when the condition is false.

The condition syntax supports standard operators (=, !=, >, <, >=, <=) and logical operators (AND, OR).

def create_if_action(action_id: str, condition_expr: str, true_action_id: str, false_action_id: Optional[str] = None) -> dict:
    """
    Creates a JSON representation of an IF action.
    
    Args:
        action_id: Unique identifier for this node.
        condition_expr: The condition string (e.g., "{{accountBalance}} > 1000").
        true_action_id: The actionId to route to if condition is true.
        false_action_id: The actionId to route to if condition is false.
    
    Returns:
        A dictionary representing the IF action node.
    """
    children = {
        "true": {
            "actionId": true_action_id
        }
    }
    
    if false_action_id:
        children["false"] = {
            "actionId": false_action_id
        }
    else:
        # If no false path, explicitly state the flow ends or goes to a default
        # In Studio, omitting 'false' usually implies end of branch, 
        # but for API completeness, we often point to an 'End' node.
        children["false"] = {
            "actionId": "end_node" # Assuming an end node exists
        }

    return {
        "id": action_id,
        "type": "if",
        "name": f"If {condition_expr}",
        "condition": condition_expr,
        "children": children
    }

Step 3: Assembling the Complete Flow Graph

A CXone flow requires a root node and a complete graph definition. We must ensure every node has a unique id and that all actionId references in children or nextActionId point to existing nodes.

We will build a flow that:

  1. Starts with a Start node.
  2. Assigns a balance value.
  3. Checks if the balance is greater than 1000.
  4. Routes to a QueuePremium node if true.
  5. Routes to a QueueStandard node if false.
  6. Ends the flow.
def build_branching_flow() -> dict:
    """
    Constructs the full JSON payload for a branching flow.
    """
    # Define Node IDs
    start_node_id = "start_node"
    assign_node_id = "assign_balance"
    if_node_id = "check_balance"
    queue_premium_id = "queue_premium"
    queue_standard_id = "queue_standard"
    end_node_id = "end_node"

    # 1. Start Node
    start_node = {
        "id": start_node_id,
        "type": "start",
        "name": "Start",
        "nextActionId": assign_node_id
    }

    # 2. Assign Action: Set accountBalance to a simulated value
    # In a real scenario, this might read from a CRM integration or IVR input
    assign_node = create_assign_action(assign_node_id, "accountBalance", "1500")

    # 3. If Action: Check if balance > 1000
    # Note: The syntax {{variableName}} is required for variable interpolation in conditions
    if_node = create_if_action(
        if_node_id, 
        "{{accountBalance}} > 1000", 
        true_action_id=queue_premium_id, 
        false_action_id=queue_standard_id
    )

    # 4. Queue Actions (Simplified for demonstration)
    # In production, these would be 'queue' type actions with specific queue IDs
    queue_premium_node = {
        "id": queue_premium_id,
        "type": "queue",
        "name": "Queue Premium",
        "queueId": "premium_queue_id_123", # Replace with real Queue ID
        "nextActionId": end_node_id
    }

    queue_standard_node = {
        "id": queue_standard_id,
        "type": "queue",
        "name": "Queue Standard",
        "queueId": "standard_queue_id_456", # Replace with real Queue ID
        "nextActionId": end_node_id
    }

    # 5. End Node
    end_node = {
        "id": end_node_id,
        "type": "end",
        "name": "End Flow"
    }

    # Assemble the full flow definition
    flow_definition = {
        "name": "Dynamic Branching Flow",
        "description": "Routes calls based on account balance using ASSIGN and IF actions.",
        "version": "1.0",
        "nodes": [
            start_node,
            assign_node,
            if_node,
            queue_premium_node,
            queue_standard_node,
            end_node
        ]
    }

    return flow_definition

Complete Working Example

The following script combines authentication, flow construction, and API submission. It creates a new flow in your CXone tenant.

Required Scopes: studio:flows:write

import requests
import json
import sys

# Configuration
REALM = "your-realm"
CLIENT_ID = "your-client-id"
CLIENT_SECRET = "your-client-secret"

# Initialize Auth
auth = CXoneAuth(REALM, CLIENT_ID, CLIENT_SECRET)

def deploy_flow(flow_data: dict) -> None:
    """
    Deploys the flow to CXone Studio.
    """
    api_url = f"https://{REALM}.niceincontact.com/api/v2/studio/flows"
    
    headers = auth.get_headers()
    
    try:
        response = requests.post(api_url, headers=headers, json=flow_data)
        
        if response.status_code == 200 or response.status_code == 201:
            result = response.json()
            print(f"Success: Flow created/updated.")
            print(f"Flow ID: {result.get('id')}")
            print(f"Flow Name: {result.get('name')}")
            print(f"Version: {result.get('version')}")
            print(f"API Response: {json.dumps(result, indent=2)}")
        else:
            print(f"Error: HTTP {response.status_code}")
            print(f"Response: {response.text}")
            sys.exit(1)
            
    except requests.exceptions.RequestException as e:
        print(f"Network Error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    # Build the flow
    flow_payload = build_branching_flow()
    
    # Debug: Print the JSON payload
    print("Deploying Flow Payload:")
    print(json.dumps(flow_payload, indent=2))
    print("-" * 50)
    
    # Deploy
    deploy_flow(flow_payload)

Common Errors & Debugging

Error: 400 Bad Request - Invalid Flow Structure

Cause: The flow graph is not valid. Common issues include:

  • A node references an actionId that does not exist in the nodes array.
  • The start node does not have a nextActionId.
  • Circular references in the graph (e.g., Node A → Node B → Node A).
  • Missing dataType in ASSIGN actions.

Fix: Validate the graph structure before sending. Ensure every actionId in children or nextActionId matches a unique id in the nodes list.

def validate_flow_graph(nodes: list) -> bool:
    """
    Basic validation to check for missing references.
    """
    node_ids = {node["id"] for node in nodes}
    
    for node in nodes:
        # Check nextActionId
        if "nextActionId" in node:
            if node["nextActionId"] not in node_ids:
                print(f"Error: Node {node['id']} references nextActionId {node['nextActionId']} which does not exist.")
                return False
        
        # Check children (for IF, SWITCH, etc.)
        if "children" in node:
            for outcome, child in node["children"].items():
                if child["actionId"] not in node_ids:
                    print(f"Error: Node {node['id']} references actionId {child['actionId']} in children which does not exist.")
                    return False
                    
    return True

Error: 401 Unauthorized - Invalid Token

Cause: The OAuth token is expired or invalid.

Fix: Ensure the CXoneAuth class is refreshing the token correctly. Check that the client ID and secret are correct and that the client has the studio:flows:write scope.

Error: 403 Forbidden - Insufficient Permissions

Cause: The OAuth client lacks the required scopes.

Fix: Go to the CXone Admin Console > Integrations > OAuth Clients. Select your client and ensure studio:flows:write is checked.

Error: Condition Syntax Error

Cause: The condition string in the IF action is malformed.

Fix: Ensure variable names are wrapped in double curly braces {{variableName}}. Ensure operators are valid (>, <, =, !=, >=, <=). Ensure logical operators (AND, OR) are uppercase.

Example of correct syntax:
"{{accountBalance}} > 1000 AND {{customerType}} = 'Premium'"

Official References