How to implement branching logic with ASSIGN and IF in CXone Studio
What You Will Build
- You will build a CXone Studio Snippet that uses the
ASSIGNaction to store dynamic data and theIFaction to route conversation flow based on that data. - You will use the NICE CXone Studio REST API to create and deploy this snippet programmatically.
- You will use Python with the
requestslibrary to interact with the CXone API.
Prerequisites
- A valid NICE CXone tenant with Studio access.
- An API Key or OAuth 2.0 Client Credentials setup with the
studio:snippet:writeandstudio:snippet:readscopes. - Python 3.8+ installed.
- The
requestslibrary installed (pip install requests).
Authentication Setup
NICE CXone uses OAuth 2.0 for API authentication. For programmatic access to Studio resources, the Client Credentials flow is the standard approach. You must obtain an access token before making any requests to the Studio API.
The following Python function handles the token acquisition. It caches the token in memory for the duration of the script execution to avoid unnecessary re-authentication calls.
import requests
import json
import time
from typing import Optional
# Configuration - Replace these with your actual CXone tenant details
CXONE_TENANT = "https://your-tenant-name.cxone.com"
CLIENT_ID = "your-client-id"
CLIENT_SECRET = "your-client-secret"
# Scopes required for Studio Snippet operations
SCOPES = "studio:snippet:write studio:snippet:read"
class CXoneClient:
def __init__(self, tenant: str, client_id: str, client_secret: str):
self.tenant = tenant
self.client_id = client_id
self.client_secret = client_secret
self.access_token: Optional[str] = None
self.token_expiry: float = 0
def _get_auth_url(self) -> str:
return f"{self.tenant}/api/v2/oauth/token"
def _refresh_token_if_needed(self) -> None:
if self.access_token and time.time() < self.token_expiry:
return
payload = {
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret,
"scope": SCOPES
}
try:
response = requests.post(self._get_auth_url(), data=payload)
response.raise_for_status()
data = response.json()
self.access_token = data["access_token"]
self.token_expiry = time.time() + data["expires_in"] - 60 # Buffer for expiry
except requests.exceptions.HTTPError as e:
if e.response.status_code == 401:
raise Exception("Authentication failed: Invalid Client ID or Secret.") from e
raise e
def get_headers(self) -> dict:
self._refresh_token_if_needed()
return {
"Authorization": f"Bearer {self.access_token}",
"Content-Type": "application/json",
"Accept": "application/json"
}
# Initialize the client
client = CXoneClient(CXONE_TENANT, CLIENT_ID, CLIENT_SECRET)
Implementation
Studio Snippets are defined as JSON documents that describe the flow of actions. To implement branching logic, you must understand how the ASSIGN action populates context variables and how the IF action evaluates conditions against those variables or system attributes.
Step 1: Define the ASSIGN Action
The ASSIGN action is used to set the value of a variable. In Studio, variables can be global (available across the entire interaction) or local (scoped to the current block). For branching logic, we typically assign a value that represents the user’s choice or a calculated result.
In the Studio API, an action is an object within the steps array. The ASSIGN action requires a name (the variable name) and a value (the data to store).
def create_assign_action(variable_name: str, value: str) -> dict:
"""
Creates a Studio ASSIGN action configuration.
Args:
variable_name: The name of the variable to set (e.g., "user_choice").
value: The value to assign. This can be a static string or a reference
to an input variable like "${input.text}".
Returns:
A dictionary representing the ASSIGN action.
"""
assign_action = {
"type": "ASSIGN",
"name": f"Set_{variable_name.replace(' ', '_')}",
"parameters": {
"variable": variable_name,
"value": value,
"scope": "global" # Options: "global", "local"
}
}
return assign_action
# Example: Assigning a static value for testing
assign_step = create_assign_action("test_choice", "premium")
print(json.dumps(assign_step, indent=2))
Expected Output:
{
"type": "ASSIGN",
"name": "Set_test_choice",
"parameters": {
"variable": "test_choice",
"value": "premium",
"scope": "global"
}
}
Error Handling Note: If the variable name contains special characters or spaces, the Studio API may reject the snippet during validation. Always use alphanumeric characters and underscores for variable names.
Step 2: Define the IF Action with Branching Logic
The IF action evaluates a condition and directs the flow to one of two branches: true or false. The condition is defined using a simple expression language.
Common comparison operators include:
==: Equal to!=: Not equal to>,<,>=,<=: Numeric comparisons
The IF action must define trueStep and falseStep. These are references to the id of other steps in the snippet.
def create_if_action(condition: str, true_step_id: str, false_step_id: str) -> dict:
"""
Creates a Studio IF action configuration.
Args:
condition: The expression to evaluate (e.g., "${test_choice} == 'premium'").
true_step_id: The ID of the step to execute if the condition is true.
false_step_id: The ID of the step to execute if the condition is false.
Returns:
A dictionary representing the IF action.
"""
if_action = {
"type": "IF",
"name": "Check_User_Choice",
"parameters": {
"condition": condition,
"trueStep": true_step_id,
"falseStep": false_step_id
}
}
return if_action
# Example: Checking if the choice is "premium"
# Note: The step IDs must match the IDs assigned in the final snippet structure
if_step = create_if_action("${test_choice} == 'premium'", "step_premium", "step_standard")
print(json.dumps(if_step, indent=2))
Expected Output:
{
"type": "IF",
"name": "Check_User_Choice",
"parameters": {
"condition": "${test_choice} == 'premium'",
"trueStep": "step_premium",
"falseStep": "step_standard"
}
}
Critical Parameter Explanation:
condition: The string inside${...}must reference a variable that has already been assigned or is available in the context (e.g.,${input.text},${agent.name}).trueStep/falseStep: These are not names, but unique identifiers for steps within the same snippet. They must correspond to theidfield of the target steps.
Step 3: Construct the Complete Snippet Structure
A Studio Snippet is a JSON object containing metadata and a list of steps. Each step has a unique id, a type, and parameters. The first step must be an INPUT or ASSIGN type, and the last step must be an OUTPUT or END type.
To implement branching, you need:
- An
ASSIGNstep to set the variable. - An
IFstep to evaluate the variable. - Two subsequent steps (one for
true, one forfalse) to handle the branches. - An
ENDstep to terminate the flow.
def build_branching_snippet() -> dict:
"""
Constructs a complete Studio Snippet JSON structure with branching logic.
"""
snippet = {
"name": "Branching_Logic_Demo",
"description": "Demonstrates ASSIGN and IF actions for branching.",
"version": 1,
"steps": [
{
"id": "step_assign",
"type": "ASSIGN",
"name": "Set_Choice",
"parameters": {
"variable": "user_choice",
"value": "premium", # In production, this might be "${input.text}"
"scope": "global"
},
"nextStep": "step_check"
},
{
"id": "step_check",
"type": "IF",
"name": "Check_Choice",
"parameters": {
"condition": "${user_choice} == 'premium'",
"trueStep": "step_premium",
"falseStep": "step_standard"
}
},
{
"id": "step_premium",
"type": "OUTPUT",
"name": "Premium_Response",
"parameters": {
"value": "You have selected the Premium plan.",
"type": "text"
},
"nextStep": "step_end"
},
{
"id": "step_standard",
"type": "OUTPUT",
"name": "Standard_Response",
"parameters": {
"value": "You have selected the Standard plan.",
"type": "text"
},
"nextStep": "step_end"
},
{
"id": "step_end",
"type": "END",
"name": "End_Flow",
"parameters": {}
}
]
}
return snippet
snippet_json = build_branching_snippet()
Complete Working Example
The following Python script combines authentication, snippet construction, and API submission. It creates the snippet, uploads it to CXone Studio, and verifies the upload.
import requests
import json
import sys
# Reuse CXoneClient class from Authentication Setup section
def create_snippet(client: CXoneClient, snippet_data: dict) -> dict:
"""
Creates a new Studio Snippet via the CXone API.
Args:
client: An initialized CXoneClient instance.
snippet_data: The JSON structure of the snippet.
Returns:
The response JSON containing the snippet ID.
"""
endpoint = f"{client.tenant}/api/v2/studio/snippets"
headers = client.get_headers()
try:
response = requests.post(endpoint, headers=headers, json=snippet_data)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as e:
print(f"Error creating snippet: {e.response.status_code} - {e.response.text}")
if e.response.status_code == 400:
print("Bad Request: Check the snippet JSON structure for syntax errors.")
elif e.response.status_code == 401:
print("Unauthorized: Check your API credentials.")
sys.exit(1)
def get_snippet(client: CXoneClient, snippet_id: str) -> dict:
"""
Retrieves a Studio Snippet by ID to verify creation.
Args:
client: An initialized CXoneClient instance.
snippet_id: The ID of the snippet to retrieve.
Returns:
The full snippet JSON.
"""
endpoint = f"{client.tenant}/api/v2/studio/snippets/{snippet_id}"
headers = client.get_headers()
try:
response = requests.get(endpoint, headers=headers)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as e:
print(f"Error retrieving snippet: {e.response.status_code} - {e.response.text}")
sys.exit(1)
def main():
# 1. Build the snippet
print("Building branching logic snippet...")
snippet = build_branching_snippet()
# 2. Create the snippet in CXone
print("Uploading snippet to CXone Studio...")
result = create_snippet(client, snippet)
snippet_id = result.get("id")
if not snippet_id:
print("Failed to retrieve Snippet ID from response.")
sys.exit(1)
print(f"Snippet created successfully with ID: {snippet_id}")
# 3. Verify the snippet
print("Verifying snippet content...")
retrieved_snippet = get_snippet(client, snippet_id)
# Validate that the IF action exists
steps = retrieved_snippet.get("steps", [])
if_step_exists = any(step.get("type") == "IF" for step in steps)
if if_step_exists:
print("SUCCESS: Branching logic (IF action) confirmed in snippet.")
else:
print("ERROR: IF action not found in retrieved snippet.")
sys.exit(1)
if __name__ == "__main__":
main()
Common Errors & Debugging
Error: 400 Bad Request - Invalid Step Reference
What causes it:
The IF action references a trueStep or falseStep ID that does not exist in the steps array, or the referenced step is not reachable (e.g., circular reference or missing nextStep chain).
How to fix it:
Ensure that every id referenced in trueStep or falseStep matches an id defined in the steps array exactly. Case sensitivity matters.
Code Fix:
# Incorrect
"parameters": {
"trueStep": "Premium_Response" # This is the 'name', not the 'id'
}
# Correct
"parameters": {
"trueStep": "step_premium" # This is the 'id'
}
Error: 400 Bad Request - Variable Not Found
What causes it:
The IF condition references a variable that has not been assigned yet in the flow, or the variable scope is incorrect (e.g., trying to access a local variable from a different branch).
How to fix it:
Ensure the ASSIGN action occurs before the IF action in the execution flow. Verify that the variable name in the ASSIGN parameters.variable matches the reference in the IF parameters.condition.
Code Fix:
# Ensure the ASSIGN step has a nextStep pointing to the IF step
{
"id": "step_assign",
"type": "ASSIGN",
"nextStep": "step_check" # Critical link
},
{
"id": "step_check",
"type": "IF",
"parameters": {
"condition": "${user_choice} == 'premium'" # Must match variable name above
}
}
Error: 429 Too Many Requests
What causes it:
CXone API enforces rate limits. Rapidly creating or updating snippets can trigger this.
How to fix it:
Implement exponential backoff in your retry logic.
Code Fix:
import time
def post_with_retry(client, url, data, retries=3):
for attempt in range(retries):
response = requests.post(url, headers=client.get_headers(), json=data)
if response.status_code == 429:
wait_time = 2 ** attempt
print(f"Rate limited. Waiting {wait_time} seconds...")
time.sleep(wait_time)
continue
return response
raise Exception("Max retries exceeded")