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

What You Will Build

• One sentence: You will build a Studio flow that captures user input, stores it in a variable, and routes the conversation based on that value.
• One sentence: This uses the NICE CXone Studio API to programmatically create and publish a flow with logic nodes.
• One sentence: The tutorial covers Python using the requests library to interact with the CXone REST API.

Prerequisites

• OAuth client type: Confidential Client (Client Credentials Grant).
• Required scopes: flow:flow:create, flow:flow:publish, flow:flow:read.
• SDK/API version: CXone Studio API v2 (beta/stable endpoints vary; this tutorial uses the standard /api/v2/flow/ paths).
• Language/runtime requirements: Python 3.8+.
• External dependencies: requests (install via pip install requests).

Authentication Setup

NICE CXone uses OAuth 2.0 for authentication. You must obtain an access token before calling the Studio API. The token is valid for one hour. For production systems, implement a caching mechanism to reuse the token until expiration.

The following Python function handles the Client Credentials flow. It requires your client_id, client_id_secret, and domain (e.g., api.mynicecxone.com).

import requests
import json
import time

class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str, domain: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.domain = domain
        self.access_token = None
        self.token_expiry = 0

    def get_token(self) -> str:
        """
        Retrieves an OAuth2 access token from CXone.
        Implements basic caching to avoid unnecessary refreshes.
        """
        # Check if we have a valid token
        if self.access_token and time.time() < self.token_expiry:
            return self.access_token

        url = f"https://{self.domain}/oauth/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(url, headers=headers, data=data)
            response.raise_for_status()
            
            token_data = response.json()
            self.access_token = token_data["access_token"]
            # Set expiry to slightly before actual expiry to handle clock skew
            self.token_expiry = time.time() + (token_data["expires_in"] - 60)
            
            return self.access_token

        except requests.exceptions.HTTPError as e:
            print(f"Authentication failed: {e.response.status_code} - {e.response.text}")
            raise
        except requests.exceptions.RequestException as e:
            print(f"Network error during authentication: {e}")
            raise

# Usage Example:
# auth = CXoneAuth("your_client_id", "your_client_secret", "api.mynicecxone.com")
# token = auth.get_token()

Implementation

Step 1: Define the Flow Structure with ASSIGN

In CXone Studio, a flow is a directed graph of nodes. To implement branching logic, you first need data to branch on. The ASSIGN action sets a variable that subsequent nodes can reference.

We will create a flow that:

1. Starts with a START node.
2. Uses an ASSIGN node to set a variable named user_choice to a specific value (simulating user input for this example).
3. Uses an IF node to check the value of user_choice.

First, we define the base flow JSON structure. The Studio API expects a specific schema.

import requests

def create_base_flow(auth: CXoneAuth, flow_name: str) -> str:
    """
    Creates a new empty flow in CXone Studio.
    Returns the flow ID.
    """
    url = f"https://{auth.domain}/api/v2/flow"
    headers = {
        "Authorization": f"Bearer {auth.get_token()}",
        "Content-Type": "application/json"
    }
    
    # Minimal flow payload
    payload = {
        "name": flow_name,
        "description": "Flow created via API for branching logic tutorial",
        "type": "IVR", # Can be IVR, Chat, Email, etc.
        "language": "en-us",
        "nodes": [],
        "edges": []
    }

    try:
        response = requests.post(url, headers=headers, json=payload)
        response.raise_for_status()
        
        flow_data = response.json()
        print(f"Flow created successfully with ID: {flow_data['id']}")
        return flow_data["id"]

    except requests.exceptions.HTTPError as e:
        print(f"Failed to create flow: {e.response.status_code} - {e.response.text}")
        raise

Step 2: Add ASSIGN and IF Nodes

Now we populate the flow with the logic nodes. The Studio API uses a specific node type for assignments (ASSIGN) and conditional checks (IF).

Key Concepts:

• ASSIGN Node: Sets a variable. You must define the variable name and the value expression.
• IF Node: Evaluates a condition. It has multiple outgoing edges (True/False or multiple cases).
• Edges: Connect nodes. Each edge has a sourceNodeId, targetNodeId, and a label (for IF nodes, this corresponds to the condition outcome).

We will add:

1. An ASSIGN node that sets user_choice to "sales".
2. An IF node that checks if user_choice equals "sales".
3. Two END nodes: one for the “Sales” path (True) and one for the “Other” path (False).

def add_branching_logic(auth: CXoneAuth, flow_id: str) -> None:
    """
    Adds ASSIGN and IF nodes to the existing flow.
    """
    url = f"https://{auth.domain}/api/v2/flow/{flow_id}"
    headers = {
        "Authorization": f"Bearer {auth.get_token()}",
        "Content-Type": "application/json"
    }

    # 1. Define Nodes
    nodes = [
        {
            "id": "start_node",
            "type": "START",
            "position": {"x": 0, "y": 0}
        },
        {
            "id": "assign_choice",
            "type": "ASSIGN",
            "position": {"x": 0, "y": 200},
            "properties": {
                "variables": [
                    {
                        "name": "user_choice",
                        "value": {"type": "literal", "value": "sales"} 
                        # In a real scenario, this might be {"type": "input", "name": "user_input"}
                    }
                ]
            }
        },
        {
            "id": "check_choice",
            "type": "IF",
            "position": {"x": 0, "y": 400},
            "properties": {
                "condition": {
                    "type": "equals",
                    "left": {"type": "variable", "name": "user_choice"},
                    "right": {"type": "literal", "value": "sales"}
                }
            }
        },
        {
            "id": "end_sales",
            "type": "END",
            "position": {"x": -200, "y": 600},
            "properties": {
                "result": "transferred_to_sales"
            }
        },
        {
            "id": "end_other",
            "type": "END",
            "position": {"x": 200, "y": 600},
            "properties": {
                "result": "transferred_to_general"
            }
        }
    ]

    # 2. Define Edges
    edges = [
        {
            "id": "edge_start_to_assign",
            "sourceNodeId": "start_node",
            "targetNodeId": "assign_choice"
        },
        {
            "id": "edge_assign_to_if",
            "sourceNodeId": "assign_choice",
            "targetNodeId": "check_choice"
        },
        {
            "id": "edge_if_to_sales",
            "sourceNodeId": "check_choice",
            "targetNodeId": "end_sales",
            "label": "true" # The condition evaluated to true
        },
        {
            "id": "edge_if_to_other",
            "sourceNodeId": "check_choice",
            "targetNodeId": "end_other",
            "label": "false" # The condition evaluated to false
        }
    ]

    # 3. Prepare Payload
    payload = {
        "nodes": nodes,
        "edges": edges
    }

    try:
        response = requests.put(url, headers=headers, json=payload)
        response.raise_for_status()
        print("Branching logic added successfully.")
    except requests.exceptions.HTTPError as e:
        print(f"Failed to update flow: {e.response.status_code} - {e.response.text}")
        raise

Step 3: Publish the Flow

A flow in the “Draft” state cannot be executed by the CXone runtime. You must publish it to make it live. Publishing validates the flow structure. If there are broken edges or invalid node configurations, the API will return an error.

def publish_flow(auth: CXoneAuth, flow_id: str) -> None:
    """
    Publishes the flow to make it executable.
    """
    url = f"https://{auth.domain}/api/v2/flow/{flow_id}/publish"
    headers = {
        "Authorization": f"Bearer {auth.get_token()}",
        "Content-Type": "application/json"
    }

    try:
        response = requests.post(url, headers=headers)
        response.raise_for_status()
        
        result = response.json()
        print(f"Flow published successfully. Version: {result.get('version', 'N/A')}")
        
    except requests.exceptions.HTTPError as e:
        # Common error: 400 Bad Request with validation details
        print(f"Publish failed: {e.response.status_code}")
        print(f"Details: {e.response.text}")
        raise
    except requests.exceptions.RequestException as e:
        print(f"Network error during publish: {e}")
        raise

Complete Working Example

The following script combines all steps into a single runnable module. It creates a flow, adds the branching logic, and publishes it.

import requests
import time
import json

class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str, domain: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.domain = domain
        self.access_token = None
        self.token_expiry = 0

    def get_token(self) -> str:
        if self.access_token and time.time() < self.token_expiry:
            return self.access_token

        url = f"https://{self.domain}/oauth/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(url, headers=headers, data=data)
            response.raise_for_status()
            token_data = response.json()
            self.access_token = token_data["access_token"]
            self.token_expiry = time.time() + (token_data["expires_in"] - 60)
            return self.access_token
        except requests.exceptions.HTTPError as e:
            print(f"Authentication failed: {e.response.status_code} - {e.response.text}")
            raise

def create_and_publish_branching_flow(auth: CXoneAuth, flow_name: str) -> str:
    """
    Main orchestration function to create, configure, and publish a flow.
    """
    
    # Step 1: Create Base Flow
    print(f"Creating flow: {flow_name}")
    create_url = f"https://{auth.domain}/api/v2/flow"
    create_headers = {
        "Authorization": f"Bearer {auth.get_token()}",
        "Content-Type": "application/json"
    }
    create_payload = {
        "name": flow_name,
        "description": "Branching logic example",
        "type": "IVR",
        "language": "en-us",
        "nodes": [],
        "edges": []
    }

    try:
        response = requests.post(create_url, headers=create_headers, json=create_payload)
        response.raise_for_status()
        flow_id = response.json()["id"]
        print(f"Flow created with ID: {flow_id}")
    except requests.exceptions.HTTPError as e:
        print(f"Failed to create flow: {e.response.text}")
        return None

    # Step 2: Add Nodes and Edges
    print("Adding branching logic nodes...")
    update_url = f"https://{auth.domain}/api/v2/flow/{flow_id}"
    
    nodes = [
        {
            "id": "start_node",
            "type": "START",
            "position": {"x": 0, "y": 0}
        },
        {
            "id": "assign_choice",
            "type": "ASSIGN",
            "position": {"x": 0, "y": 200},
            "properties": {
                "variables": [
                    {
                        "name": "user_choice",
                        "value": {"type": "literal", "value": "sales"}
                    }
                ]
            }
        },
        {
            "id": "check_choice",
            "type": "IF",
            "position": {"x": 0, "y": 400},
            "properties": {
                "condition": {
                    "type": "equals",
                    "left": {"type": "variable", "name": "user_choice"},
                    "right": {"type": "literal", "value": "sales"}
                }
            }
        },
        {
            "id": "end_sales",
            "type": "END",
            "position": {"x": -200, "y": 600},
            "properties": {"result": "sales_queue"}
        },
        {
            "id": "end_other",
            "type": "END",
            "position": {"x": 200, "y": 600},
            "properties": {"result": "general_queue"}
        }
    ]

    edges = [
        {"id": "e1", "sourceNodeId": "start_node", "targetNodeId": "assign_choice"},
        {"id": "e2", "sourceNodeId": "assign_choice", "targetNodeId": "check_choice"},
        {"id": "e3", "sourceNodeId": "check_choice", "targetNodeId": "end_sales", "label": "true"},
        {"id": "e4", "sourceNodeId": "check_choice", "targetNodeId": "end_other", "label": "false"}
    ]

    update_payload = {"nodes": nodes, "edges": edges}

    try:
        response = requests.put(update_url, headers=create_headers, json=update_payload)
        response.raise_for_status()
        print("Nodes and edges updated.")
    except requests.exceptions.HTTPError as e:
        print(f"Failed to update flow: {e.response.text}")
        return None

    # Step 3: Publish Flow
    print("Publishing flow...")
    publish_url = f"https://{auth.domain}/api/v2/flow/{flow_id}/publish"
    
    try:
        response = requests.post(publish_url, headers=create_headers)
        response.raise_for_status()
        print("Flow published successfully.")
        return flow_id
    except requests.exceptions.HTTPError as e:
        print(f"Failed to publish flow: {e.response.text}")
        return None

if __name__ == "__main__":
    # Replace with your actual credentials
    CLIENT_ID = "your_client_id"
    CLIENT_SECRET = "your_client_secret"
    DOMAIN = "api.mynicecxone.com"
    
    if CLIENT_ID == "your_client_id":
        raise ValueError("Please update CLIENT_ID and CLIENT_SECRET in the script.")

    auth = CXoneAuth(CLIENT_ID, CLIENT_SECRET, DOMAIN)
    create_and_publish_branching_flow(auth, "API_Branching_Logic_Test")

Common Errors & Debugging

Error: 400 Bad Request - Invalid Flow Structure

What causes it:
The Studio API validates the flow graph on publish. Common causes include:

• A node referenced in an edge does not exist in the nodes array.
• An edge is missing a label when connecting from an IF node.
• The IF condition syntax is invalid (e.g., missing left or right operands).

How to fix it:
Check the response body for specific validation messages. Ensure every IF node has outgoing edges for all possible outcomes (typically true and false for binary conditions).

// Example Error Response
{
    "message": "Flow validation failed",
    "details": [
        "Node 'check_choice' has an outgoing edge with label 'true' but the condition does not support this label."
    ]
}

Code Fix:
Verify that the label in the edge matches the condition type. For equals, true and false are standard.

Error: 401 Unauthorized

What causes it:
The OAuth token is expired or invalid.

How to fix it:
Ensure your CXoneAuth class is caching the token and refreshing it when time.time() > self.token_expiry. If you are making rapid API calls, ensure you are not hitting the rate limit during token refresh.

Error: 429 Too Many Requests

What causes it:
CXone APIs have rate limits. Creating flows is a write operation and may have stricter limits than read operations.

How to fix it:
Implement exponential backoff in your request function.

import time

def make_request_with_retry(url, method, headers, payload=None, max_retries=3):
    for attempt in range(max_retries):
        try:
            if method == "POST":
                response = requests.post(url, headers=headers, json=payload)
            elif method == "PUT":
                response = requests.put(url, headers=headers, json=payload)
            
            if response.status_code == 429:
                wait_time = 2 ** attempt
                print(f"Rate limited. Waiting {wait_time} seconds...")
                time.sleep(wait_time)
                continue
            
            response.raise_for_status()
            return response
        except requests.exceptions.RequestException as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(1)
    raise Exception("Max retries exceeded")

Official References

• NICE CXone Studio API Documentation
• OAuth 2.0 Authentication Guide
• Flow Node Types Reference