How to use ASSIGN and IF actions in CXone Studio to implement branching logic

What You Will Build

• A CXone Studio flow that captures an inbound voice call, assigns the caller’s input to a variable, and branches to different IVR menus based on that value.
• This tutorial uses the NICE CXone Studio API (REST) and the Python requests library to programmatically create and update a Studio Flow.
• The programming language covered is Python 3.9+.

Prerequisites

• OAuth Client: A CXone Developer Account with a registered OAuth 2.0 Client Application. The client must have the studio:flows:write and studio:flows:read scopes.
• API Version: CXone Studio API (v2).
• Language/Runtime: Python 3.9 or higher.
• External Dependencies:
  • requests (for HTTP calls)
  • python-dotenv (for managing environment variables securely)

Install the dependencies via pip:

pip install requests python-dotenv

Authentication Setup

CXone uses standard OAuth 2.0 Client Credentials flow for server-to-server API access. You must obtain a bearer token before making any Studio API calls.

Create a .env file in your project root with the following variables:

# .env
CXONE_CLIENT_ID=your_client_id
CXONE_CLIENT_SECRET=your_client_secret
CXONE_TENANT_URL=https://your-tenant.nicecxone.com

Create a helper module auth.py to handle token retrieval and caching. This prevents unnecessary token requests on every API call.

import os
import time
import requests
from dotenv import load_dotenv

load_dotenv()

CLIENT_ID = os.getenv("CXONE_CLIENT_ID")
CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET")
TENANT_URL = os.getenv("CXONE_TENANT_URL")

# Cache for the token to avoid re-authenticating within the token's lifetime
_token_cache = {
    "access_token": None,
    "expires_in": 0,
    "issued_at": 0
}

def get_access_token() -> str:
    """
    Retrieves an OAuth2 access token from CXone.
    Uses a simple cache to reuse the token until it expires.
    """
    current_time = time.time()
    
    # If we have a valid token, return it
    if (_token_cache["access_token"] and 
        (current_time - _token_cache["issued_at"] < _token_cache["expires_in"] - 60)):
        return _token_cache["access_token"]

    # Token is expired or missing; fetch a new one
    token_url = f"{TENANT_URL}/oauth2/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET
    }

    try:
        response = requests.post(token_url, data=payload)
        response.raise_for_status()
        data = response.json()
        
        # Update cache
        _token_cache["access_token"] = data["access_token"]
        _token_cache["expires_in"] = data["expires_in"]
        _token_cache["issued_at"] = current_time
        
        return data["access_token"]
    except requests.exceptions.RequestException as e:
        raise RuntimeError(f"Failed to obtain access token: {e}")

Implementation

Step 1: Initialize the Studio Flow Structure

To use ASSIGN and IF actions, you must first create a valid Studio Flow JSON structure. CXone Studio flows are defined by a nodes array (the UI elements) and an edges array (the connections between them).

We will start by creating a minimal “Hello World” flow to get the flow_id, which is required for subsequent updates.

Create create_base_flow.py:

import requests
import json
from auth import get_access_token, TENANT_URL

def create_base_flow(name: str) -> str:
    """
    Creates a minimal Studio Flow and returns its ID.
    """
    token = get_access_token()
    endpoint = f"{TENANT_URL}/api/v2/studio/flows"
    
    # Minimal valid Studio Flow JSON structure
    # Note: Studio flows require a specific schema version and structure.
    flow_data = {
        "name": name,
        "description": "Base flow for ASSIGN/IF tutorial",
        "type": "voice",
        "nodes": [
            {
                "id": "start_node",
                "type": "start",
                "config": {}
            }
        ],
        "edges": []
    }

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }

    try:
        response = requests.post(endpoint, json=flow_data, headers=headers)
        response.raise_for_status()
        created_flow = response.json()
        print(f"Flow created successfully with ID: {created_flow['id']}")
        return created_flow["id"]
    except requests.exceptions.HTTPError as e:
        print(f"HTTP Error: {e.response.status_code} - {e.response.text}")
        raise
    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}")
        raise

if __name__ == "__main__":
    flow_id = create_base_flow("AssignIfTutorial_Flow")
    print(f"Store this ID for the next steps: {flow_id}")

Run this script to obtain the flow_id. You will need this ID for the remaining steps.

Step 2: Implement the ASSIGN Action

The ASSIGN action allows you to store data into a flow variable. In CXone Studio, this is typically represented by a setVariable or assign node type depending on the specific SDK version or API schema evolution. For the standard Voice API, we use the setVariable node.

We will add an ASSIGN node that sets a variable named user_choice to a default value of "1". This simulates capturing a DTMF press.

Create update_flow_assign.py:

import requests
import json
from auth import get_access_token, TENANT_URL

def update_flow_with_assign(flow_id: str) -> None:
    """
    Updates the flow to include an ASSIGN node that sets 'user_choice' to '1'.
    """
    token = get_access_token()
    endpoint = f"{TENANT_URL}/api/v2/studio/flows/{flow_id}"
    
    # Fetch the current flow to ensure we are updating the latest version
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }
    
    try:
        get_response = requests.get(endpoint, headers=headers)
        get_response.raise_for_status()
        current_flow = get_response.json()
    except requests.exceptions.RequestException as e:
        raise RuntimeError(f"Failed to fetch flow: {e}")

    # Define the ASSIGN node
    assign_node_id = "assign_user_choice"
    assign_node = {
        "id": assign_node_id,
        "type": "setVariable",
        "config": {
            "variableName": "user_choice",
            "value": "1",  # Default value
            "type": "string"
        },
        "position": {
            "x": 0,
            "y": 100
        }
    }

    # Add the node to the existing nodes list
    current_flow["nodes"].append(assign_node)

    # Update the edges to connect Start -> Assign
    # Remove existing edges from start_node if any
    current_flow["edges"] = [
        edge for edge in current_flow["edges"] 
        if edge["sourceNode"] != "start_node"
    ]
    
    # Add new edge
    current_flow["edges"].append({
        "sourceNode": "start_node",
        "targetNode": assign_node_id,
        "sourcePort": "default",
        "targetPort": "default"
    })

    # Push the updated flow back to CXone
    try:
        put_response = requests.put(endpoint, json=current_flow, headers=headers)
        put_response.raise_for_status()
        print("Flow updated successfully with ASSIGN node.")
    except requests.exceptions.HTTPError as e:
        print(f"HTTP Error: {e.response.status_code} - {e.response.text}")
        raise
    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}")
        raise

if __name__ == "__main__":
    # Replace with the flow_id from Step 1
    FLOW_ID = "your_flow_id_here" 
    if FLOW_ID == "your_flow_id_here":
        print("Please replace FLOW_ID with the actual ID from Step 1")
    else:
        update_flow_with_assign(FLOW_ID)

Key Technical Detail: The config object for setVariable requires variableName, value, and type. The type must match the data type of the value. If you are capturing DTMF, it is always a string. If you are calculating a numeric score, use integer or decimal.

Step 3: Implement the IF Action for Branching Logic

The IF action evaluates a condition and routes the flow to different branches based on the result. In CXone Studio, this is typically a conditional node. It requires a condition expression and multiple output ports (e.g., true, false).

We will add an IF node that checks if user_choice equals "1". If true, it routes to a “Sales Menu” node. If false, it routes to a “Support Menu” node.

Create update_flow_if.py:

import requests
import json
from auth import get_access_token, TENANT_URL

def update_flow_with_if(flow_id: str) -> None:
    """
    Updates the flow to include an IF node that branches based on 'user_choice'.
    """
    token = get_access_token()
    endpoint = f"{TENANT_URL}/api/v2/studio/flows/{flow_id}"
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }
    
    try:
        get_response = requests.get(endpoint, headers=headers)
        get_response.raise_for_status()
        current_flow = get_response.json()
    except requests.exceptions.RequestException as e:
        raise RuntimeError(f"Failed to fetch flow: {e}")

    # Define the IF (Conditional) node
    if_node_id = "check_user_choice"
    if_node = {
        "id": if_node_id,
        "type": "conditional",
        "config": {
            "condition": "{{user_choice}} == '1'",  // CXone Expression Syntax
            "trueLabel": "Go to Sales",
            "falseLabel": "Go to Support"
        },
        "position": {
            "x": 0,
            "y": 200
        }
    }

    # Define Target Nodes for Branching (Placeholders for this example)
    sales_node_id = "sales_menu"
    support_node_id = "support_menu"
    
    sales_node = {
        "id": sales_node_id,
        "type": "play",
        "config": {
            "text": "Thank you for choosing Sales. Please hold.",
            "voice": "default"
        },
        "position": {"x": -200, "y": 300}
    }

    support_node = {
        "id": support_node_id,
        "type": "play",
        "config": {
            "text": "Thank you for choosing Support. Please hold.",
            "voice": "default"
        },
        "position": {"x": 200, "y": 300}
    }

    # Add nodes to flow
    current_flow["nodes"].append(if_node)
    current_flow["nodes"].append(sales_node)
    current_flow["nodes"].append(support_node)

    # Update Edges
    # 1. Connect Assign -> IF
    current_flow["edges"] = [
        edge for edge in current_flow["edges"] 
        if edge["sourceNode"] != "assign_user_choice"
    ]
    current_flow["edges"].append({
        "sourceNode": "assign_user_choice",
        "targetNode": if_node_id,
        "sourcePort": "default",
        "targetPort": "default"
    })

    # 2. Connect IF -> Sales (True Branch)
    current_flow["edges"].append({
        "sourceNode": if_node_id,
        "targetNode": sales_node_id,
        "sourcePort": "true",
        "targetPort": "default"
    })

    # 3. Connect IF -> Support (False Branch)
    current_flow["edges"].append({
        "sourceNode": if_node_id,
        "targetNode": support_node_id,
        "sourcePort": "false",
        "targetPort": "default"
    })

    # Push the updated flow back to CXone
    try:
        put_response = requests.put(endpoint, json=current_flow, headers=headers)
        put_response.raise_for_status()
        print("Flow updated successfully with IF branching logic.")
    except requests.exceptions.HTTPError as e:
        print(f"HTTP Error: {e.response.status_code} - {e.response.text}")
        raise
    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}")
        raise

if __name__ == "__main__":
    FLOW_ID = "your_flow_id_here" 
    if FLOW_ID == "your_flow_id_here":
        print("Please replace FLOW_ID with the actual ID from Step 1")
    else:
        update_flow_with_if(FLOW_ID)

Critical Syntax Note: The condition "{{user_choice}} == '1'" uses CXone’s expression language. The double curly braces {{ }} indicate variable interpolation. The comparison operator == checks for equality. Ensure the data types match. If user_choice was set as an integer, you would compare {{user_choice}} == 1.

Step 4: Deploying the Flow

Updating the flow JSON does not automatically deploy it to production. You must use the Deploy API.

Create deploy_flow.py:

import requests
from auth import get_access_token, TENANT_URL

def deploy_flow(flow_id: str) -> None:
    """
    Deploys the specified flow to production.
    """
    token = get_access_token()
    endpoint = f"{TENANT_URL}/api/v2/studio/flows/{flow_id}/deploy"
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }

    # Deploy request body is typically empty or contains optional metadata
    payload = {}

    try:
        response = requests.post(endpoint, json=payload, headers=headers)
        response.raise_for_status()
        print(f"Flow {flow_id} deployed successfully.")
        print(response.json())
    except requests.exceptions.HTTPError as e:
        print(f"HTTP Error: {e.response.status_code} - {e.response.text}")
        raise
    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}")
        raise

if __name__ == "__main__":
    FLOW_ID = "your_flow_id_here" 
    if FLOW_ID == "your_flow_id_here":
        print("Please replace FLOW_ID with the actual ID from Step 1")
    else:
        deploy_flow(FLOW_ID)

Complete Working Example

Combine the logic into a single script build_assign_if_flow.py for ease of testing. This script creates a flow, adds the ASSIGN and IF nodes, and deploys it.

import os
import time
import requests
import json
from dotenv import load_dotenv

load_dotenv()

CLIENT_ID = os.getenv("CXONE_CLIENT_ID")
CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET")
TENANT_URL = os.getenv("CXONE_TENANT_URL")

_token_cache = {"access_token": None, "expires_in": 0, "issued_at": 0}

def get_access_token() -> str:
    current_time = time.time()
    if (_token_cache["access_token"] and 
        (current_time - _token_cache["issued_at"] < _token_cache["expires_in"] - 60)):
        return _token_cache["access_token"]

    token_url = f"{TENANT_URL}/oauth2/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET
    }

    try:
        response = requests.post(token_url, data=payload)
        response.raise_for_status()
        data = response.json()
        _token_cache["access_token"] = data["access_token"]
        _token_cache["expires_in"] = data["expires_in"]
        _token_cache["issued_at"] = current_time
        return data["access_token"]
    except requests.exceptions.RequestException as e:
        raise RuntimeError(f"Failed to obtain access token: {e}")

def build_and_deploy_flow(name: str) -> str:
    token = get_access_token()
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }

    # 1. Create Base Flow
    print("1. Creating base flow...")
    create_endpoint = f"{TENANT_URL}/api/v2/studio/flows"
    base_flow = {
        "name": name,
        "description": "Automated Assign/IF Flow",
        "type": "voice",
        "nodes": [{"id": "start_node", "type": "start", "config": {}}],
        "edges": []
    }
    create_resp = requests.post(create_endpoint, json=base_flow, headers=headers)
    create_resp.raise_for_status()
    flow_id = create_resp.json()["id"]
    print(f"   Flow ID: {flow_id}")

    # 2. Update with ASSIGN and IF
    print("2. Updating flow with ASSIGN and IF logic...")
    update_endpoint = f"{TENANT_URL}/api/v2/studio/flows/{flow_id}"
    
    # Fetch current to ensure version consistency
    get_resp = requests.get(update_endpoint, headers=headers)
    get_resp.raise_for_status()
    current_flow = get_resp.json()

    # Define Nodes
    assign_node = {
        "id": "assign_choice",
        "type": "setVariable",
        "config": {"variableName": "user_choice", "value": "1", "type": "string"},
        "position": {"x": 0, "y": 100}
    }
    
    if_node = {
        "id": "branch_logic",
        "type": "conditional",
        "config": {
            "condition": "{{user_choice}} == '1'",
            "trueLabel": "Sales",
            "falseLabel": "Support"
        },
        "position": {"x": 0, "y": 200}
    }

    sales_node = {
        "id": "sales_play",
        "type": "play",
        "config": {"text": "You selected Sales.", "voice": "default"},
        "position": {"x": -150, "y": 300}
    }

    support_node = {
        "id": "support_play",
        "type": "play",
        "config": {"text": "You selected Support.", "voice": "default"},
        "position": {"x": 150, "y": 300}
    }

    # Add Nodes
    current_flow["nodes"].extend([assign_node, if_node, sales_node, support_node])

    # Define Edges
    edges = [
        {"sourceNode": "start_node", "targetNode": "assign_choice", "sourcePort": "default", "targetPort": "default"},
        {"sourceNode": "assign_choice", "targetNode": "branch_logic", "sourcePort": "default", "targetPort": "default"},
        {"sourceNode": "branch_logic", "targetNode": "sales_play", "sourcePort": "true", "targetPort": "default"},
        {"sourceNode": "branch_logic", "targetNode": "support_play", "sourcePort": "false", "targetPort": "default"}
    ]
    current_flow["edges"] = edges

    # Update Flow
    put_resp = requests.put(update_endpoint, json=current_flow, headers=headers)
    put_resp.raise_for_status()
    print("   Flow structure updated.")

    # 3. Deploy
    print("3. Deploying flow...")
    deploy_endpoint = f"{TENANT_URL}/api/v2/studio/flows/{flow_id}/deploy"
    deploy_resp = requests.post(deploy_endpoint, json={}, headers=headers)
    deploy_resp.raise_for_status()
    print("   Flow deployed to production.")

    return flow_id

if __name__ == "__main__":
    try:
        flow_id = build_and_deploy_flow("DevAdv_AssignIf_Example")
        print(f"\nSuccess! Flow ID: {flow_id}")
    except Exception as e:
        print(f"\nError: {e}")

Common Errors & Debugging

Error: 400 Bad Request - Invalid Flow JSON

Cause: The Studio API is strict about the JSON schema. Common issues include missing position objects, incorrect sourcePort/targetPort names, or circular references in edges.
Fix: Validate that every node has an id, type, config, and position. Ensure that every edge connects a valid source port to a valid target port. For conditional nodes, the source ports are true and false. For most other nodes, it is default.

Error: 403 Forbidden - Insufficient Scopes

Cause: The OAuth token does not have the studio:flows:write scope.
Fix: Check your OAuth Client configuration in the CXone Admin Console. Ensure the scope studio:flows:write is granted. Re-generate the token.

Error: 409 Conflict - Flow Already Exists

Cause: Attempting to create a flow with a name that already exists in the tenant, or trying to update a flow with a stale version ID (if the API uses optimistic locking).
Fix: For creation, use a unique name. For updates, always fetch the latest flow JSON before modifying it, as shown in Step 2 and Step 3.

Error: Syntax Error in Condition

Cause: The condition string in the IF node is malformed. For example, using {{user_choice}} = 1 instead of ==, or mixing types ({{user_choice}} == 1 when user_choice is a string).
Fix: Use the CXone Expression Language syntax. Strings must be quoted: '1'. Numbers are not quoted: 1. Use == for equality. Use && and || for boolean logic.

Official References

• NICE CXone Studio API Documentation
• OAuth 2.0 Client Credentials Flow
• Studio Flow JSON Schema Reference