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
ASSIGNandIFactions. - The implementation is covered in Python using the
requestslibrary 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:writeandstudio:flows:readscopes. - CXone Tenant: Access to a CXone tenant with Studio enabled.
- Runtime: Python 3.8+ with the
requestslibrary installed (pip install requests). - Knowledge: Basic understanding of CXone Flow JSON structure and how
actionIdreferences 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 beassignfor 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 beif.condition: The logical expression to evaluate.children: An object mapping outcomes (true,false) to theactionIdof 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:
- Starts with a
Startnode. - Assigns a balance value.
- Checks if the balance is greater than 1000.
- Routes to a
QueuePremiumnode if true. - Routes to a
QueueStandardnode if false. - 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
actionIdthat does not exist in thenodesarray. - The
startnode does not have anextActionId. - Circular references in the graph (e.g., Node A → Node B → Node A).
- Missing
dataTypeinASSIGNactions.
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'"