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

What You Will Build

• This tutorial builds a CXone Studio flow that dynamically routes inbound calls based on caller ID patterns and custom data attributes.
• It utilizes the CXone Studio Visual Flow API and the Studio Snippet language to demonstrate programmatic flow construction.
• The implementation covers Python for API interactions and CXone Studio Snippet syntax for in-flow logic.

Prerequisites

• OAuth Client Type: Internal Client or Client Credentials Grant.
• Required Scopes: studio:flow:read, studio:flow:write, studio:flow:execute.
• SDK Version: NICE CXone Python SDK @nice-dcv/sdk (latest stable) or direct REST API usage.
• Language/Runtime: Python 3.9+ with requests and json libraries.
• External Dependencies: nice-dcv-sdk (optional, but recommended for type safety) or raw requests.
• CXone Environment: A valid CXone tenant with Studio access enabled.

Authentication Setup

CXone uses OAuth 2.0 for API authentication. For programmatic flow construction, the Client Credentials Grant is the standard approach. You must obtain an access token before issuing any Studio API requests.

The following Python snippet demonstrates how to acquire and cache a token. In production, implement a refresh mechanism or a token cache to avoid unnecessary network calls.

import requests
import json
from typing import Optional

class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str, region: str = "us"):
        self.client_id = client_id
        self.client_secret = client_secret
        # Map region to OAuth endpoint
        oauth_endpoints = {
            "us": "https://api.mynicecx.com/oauth2/token",
            "eu": "https://api.eu.mynicecx.com/oauth2/token",
            "au": "https://api.ap.mynicecx.com/oauth2/token"
        }
        self.token_url = oauth_endpoints.get(region, oauth_endpoints["us"])
        self.access_token: Optional[str] = None
        self.headers: dict = {}

    def get_access_token(self) -> str:
        """
        Retrieves an OAuth2 access token using Client Credentials Grant.
        Returns the token string.
        """
        if self.access_token:
            return self.access_token

        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        try:
            response = requests.post(
                self.token_url,
                data=payload,
                headers={"Content-Type": "application/x-www-form-urlencoded"}
            )
            response.raise_for_status()
            token_data = response.json()
            self.access_token = token_data["access_token"]
            self.headers = {
                "Authorization": f"Bearer {self.access_token}",
                "Content-Type": "application/json"
            }
            return self.access_token
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise Exception("Authentication failed: Invalid client ID or secret.")
            elif response.status_code == 429:
                raise Exception("Rate limit exceeded. Retry after delay.")
            else:
                raise Exception(f"Failed to get token: {e}")
        except requests.exceptions.ConnectionError:
            raise Exception("Network error connecting to CXone OAuth endpoint.")

# Usage
auth = CXoneAuth(client_id="your_client_id", client_secret="your_client_secret")
token = auth.get_access_token()

Implementation

Step 1: Constructing the ASSIGN Action

The ASSIGN action in CXone Studio is used to set variables, modify caller data, or prepare data for downstream actions. In the Studio JSON schema, this is represented as a node with type: "assign".

When building flows via API, you must define the assigns array. Each object in this array contains a variable name and a value. The value can be a literal string, a number, or a Snippet expression (enclosed in {{ }}).

The following code creates a minimal Studio flow definition containing a single ASSIGN action that sets a custom variable call_priority based on a static value.

import requests

def create_assign_node(variable_name: str, value: str) -> dict:
    """
    Generates a JSON-serializable dictionary for a Studio ASSIGN node.
    
    Args:
        variable_name: The name of the CXone variable to set (e.g., 'call_priority').
        value: The value to assign. Can be a literal or a Snippet expression.
    
    Returns:
        A dictionary representing the Studio node.
    """
    assign_node = {
        "type": "assign",
        "name": f"Set {variable_name}",
        "assigns": [
            {
                "variable": variable_name,
                "value": value
            }
        ]
    }
    return assign_node

# Example: Setting a literal value
literal_assign = create_assign_node("call_priority", "High")

# Example: Setting a value using Snippet syntax (Caller ID)
# Note: In Studio UI, this is {{ caller.id }}. In API JSON, it is often passed as a string literal that Studio evaluates.
snippet_assign = create_assign_node("caller_number", "{{ caller.id }}")

print(json.dumps(literal_assign, indent=2))

Expected Response Structure (Node Definition):

{
  "type": "assign",
  "name": "Set call_priority",
  "assigns": [
    {
      "variable": "call_priority",
      "value": "High"
    }
  ]
}

Error Handling:
If the variable name contains invalid characters (e.g., spaces without proper quoting in Snippet context, or reserved keywords), the Studio API may return a 400 Bad Request when you attempt to save the flow. Always validate variable names against CXone naming conventions (alphanumeric and underscores only).

Step 2: Implementing the IF Action for Branching

The IF action evaluates a boolean condition. If true, the flow proceeds to the true port; otherwise, it proceeds to the false port. In the Studio JSON schema, an IF node must define conditions, truePort, and falsePort.

The condition is typically defined using Snippet syntax. For example, {{ caller.id == "5551234567" }}.

Crucially, when defining an IF node via API, you must ensure that the ports referenced in truePort and falsePort actually exist in the flow definition. You cannot reference a port on a node that has not been defined or connected.

def create_if_node(condition_snippet: str, true_port_name: str, false_port_name: str) -> dict:
    """
    Generates a JSON-serializable dictionary for a Studio IF node.
    
    Args:
        condition_snippet: The Snippet expression to evaluate (e.g., '{{ caller.id == "5551234567" }}').
        true_port_name: The name of the port to connect if the condition is true.
        false_port_name: The name of the port to connect if the condition is false.
    
    Returns:
        A dictionary representing the Studio IF node.
    """
    if_node = {
        "type": "if",
        "name": "Check Caller ID",
        "conditions": [
            {
                "snippet": condition_snippet
            }
        ],
        "truePort": true_port_name,
        "falsePort": false_port_name
    }
    return if_node

# Example: Check if caller ID matches a specific VIP number
vip_condition = "{{ caller.id == '+15551234567' }}"
vip_if_node = create_if_node(vip_condition, "vip_route", "standard_route")

print(json.dumps(vip_if_node, indent=2))

Expected Response Structure (Node Definition):

{
  "type": "if",
  "name": "Check Caller ID",
  "conditions": [
    {
      "snippet": "{{ caller.id == '+15551234567' }}"
    }
  ],
  "truePort": "vip_route",
  "falsePort": "standard_route"
}

Edge Cases:

• Null Values: If caller.id is null (e.g., anonymous call), the Snippet evaluation may return false or throw an error depending on strictness settings. Use {{ caller.id is not null and caller.id == '+15551234567' }} for robustness.
• Data Types: Ensure the data types match. Comparing a string "123" to a number 123 may yield unexpected results in Snippet engine. Use {{ caller.id == "+15551234567" }} (string comparison) rather than {{ caller.id == 15551234567 }}.

Step 3: Connecting Nodes and Publishing the Flow

Creating individual nodes is insufficient. You must assemble them into a valid flow structure with a start node, connections, and an end node. Then, you must POST this structure to the CXone Studio API.

The following code constructs a complete flow: Start → Assign → IF → End (True) / End (False).

def build_complete_flow(auth: CXoneAuth, flow_name: str, flow_id: Optional[str] = None) -> dict:
    """
    Builds and publishes a complete CXone Studio flow.
    
    Args:
        auth: An authenticated CXoneAuth instance.
        flow_name: The name of the flow.
        flow_id: Optional ID for updating an existing flow. If None, creates a new flow.
    
    Returns:
        The API response containing the flow ID and status.
    """
    # 1. Define Nodes
    start_node = {
        "type": "start",
        "name": "Start",
        "outPort": "assign_priority"
    }

    assign_node = create_assign_node("call_priority", "Standard")

    if_node = create_if_node(
        condition_snippet="{{ caller.id == '+15551234567' }}",
        true_port_name="vip_end",
        false_port_name="standard_end"
    )

    vip_end_node = {
        "type": "end",
        "name": "VIP Route End",
        "id": "vip_end" # Explicit ID for port reference
    }

    standard_end_node = {
        "type": "end",
        "name": "Standard Route End",
        "id": "standard_end"
    }

    # 2. Assemble Flow Structure
    flow_definition = {
        "name": flow_name,
        "description": "Dynamic branching based on Caller ID",
        "type": "studio",
        "nodes": [
            start_node,
            assign_node,
            if_node,
            vip_end_node,
            standard_end_node
        ],
        "connections": [
            {
                "fromNode": "start",
                "fromPort": "outPort",
                "toNode": assign_node["name"],
                "toPort": "in"
            },
            {
                "fromNode": assign_node["name"],
                "fromPort": "outPort",
                "toNode": if_node["name"],
                "toPort": "in"
            },
            {
                "fromNode": if_node["name"],
                "fromPort": "truePort",
                "toNode": vip_end_node["id"],
                "toPort": "in"
            },
            {
                "fromNode": if_node["name"],
                "fromPort": "falsePort",
                "toNode": standard_end_node["id"],
                "toPort": "in"
            }
        ]
    }

    # 3. API Endpoint
    base_url = "https://api.mynicecx.com/api/v2/studio/flows"
    endpoint = f"{base_url}/{flow_id}" if flow_id else base_url
    method = "PUT" if flow_id else "POST"

    try:
        response = requests.request(
            method,
            endpoint,
            json=flow_definition,
            headers=auth.headers
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        print(f"API Error: {e}")
        print(f"Response Body: {response.text}")
        raise

Complete Working Example:

import requests
import json
from typing import Optional

class CXoneAuth:
    def __init__(self, client_id: str, client_secret: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = "https://api.mynicecx.com/oauth2/token"
        self.access_token: Optional[str] = None
        self.headers: dict = {}

    def get_access_token(self) -> str:
        if self.access_token:
            return self.access_token
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }
        response = requests.post(
            self.token_url,
            data=payload,
            headers={"Content-Type": "application/x-www-form-urlencoded"}
        )
        response.raise_for_status()
        self.access_token = response.json()["access_token"]
        self.headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Content-Type": "application/json"
        }
        return self.access_token

def create_assign_node(variable_name: str, value: str) -> dict:
    return {
        "type": "assign",
        "name": f"Set {variable_name}",
        "assigns": [{"variable": variable_name, "value": value}]
    }

def create_if_node(condition_snippet: str, true_port_name: str, false_port_name: str) -> dict:
    return {
        "type": "if",
        "name": "Check Caller ID",
        "conditions": [{"snippet": condition_snippet}],
        "truePort": true_port_name,
        "falsePort": false_port_name
    }

def publish_branching_flow(client_id: str, client_secret: str, flow_name: str):
    auth = CXoneAuth(client_id, client_secret)
    auth.get_access_token()

    start_node = {"type": "start", "name": "Start", "outPort": "assign_priority"}
    assign_node = create_assign_node("call_priority", "Standard")
    if_node = create_if_node(
        condition_snippet="{{ caller.id == '+15551234567' }}",
        true_port_name="vip_end",
        false_port_name="standard_end"
    )
    vip_end_node = {"type": "end", "name": "VIP Route End", "id": "vip_end"}
    standard_end_node = {"type": "end", "name": "Standard Route End", "id": "standard_end"}

    flow_definition = {
        "name": flow_name,
        "description": "Dynamic branching based on Caller ID",
        "type": "studio",
        "nodes": [start_node, assign_node, if_node, vip_end_node, standard_end_node],
        "connections": [
            {"fromNode": "start", "fromPort": "outPort", "toNode": assign_node["name"], "toPort": "in"},
            {"fromNode": assign_node["name"], "fromPort": "outPort", "toNode": if_node["name"], "toPort": "in"},
            {"fromNode": if_node["name"], "fromPort": "truePort", "toNode": vip_end_node["id"], "toPort": "in"},
            {"fromNode": if_node["name"], "fromPort": "falsePort", "toNode": standard_end_node["id"], "toPort": "in"}
        ]
    }

    endpoint = "https://api.mynicecx.com/api/v2/studio/flows"
    response = requests.post(endpoint, json=flow_definition, headers=auth.headers)
    
    if response.status_code == 200 or response.status_code == 201:
        print(f"Flow published successfully. ID: {response.json().get('id')}")
    else:
        print(f"Failed to publish flow. Status: {response.status_code}")
        print(response.text)

if __name__ == "__main__":
    # Replace with your actual credentials
    CLIENT_ID = "your_client_id"
    CLIENT_SECRET = "your_client_secret"
    publish_branching_flow(CLIENT_ID, CLIENT_SECRET, "Caller ID Branching Flow")

Common Errors & Debugging

Error: 400 Bad Request - Invalid Node Configuration

• Cause: The IF node references a truePort or falsePort that does not match any node id or name in the nodes array. Alternatively, the connections array has a mismatched fromPort or toPort.
• Fix: Verify that every port referenced in connections exists in the nodes array. Ensure that IF node ports (truePort, falsePort) point to the id of the target node.
• Code Fix: Check the id field in your target nodes. If you omit id, CXone generates one, but for API construction, explicit id fields are safer for reference.

Error: 422 Unprocessable Entity - Snippet Syntax Error

• Cause: The Snippet expression in the IF condition is malformed. For example, missing quotes around string literals or incorrect variable names.
• Fix: Validate the Snippet syntax. Ensure strings are quoted: {{ caller.id == "123" }}. Ensure variables exist: {{ custom.my_var == "value" }}.
• Debugging: Use the CXone Studio UI “Validate Flow” button if you can import the JSON, or check the API response body for specific Snippet compilation errors.

Error: 403 Forbidden - Insufficient Permissions

• Cause: The OAuth client lacks the studio:flow:write scope.
• Fix: Update your OAuth client configuration in the CXone Admin Portal to include the required Studio scopes.

Official References

• CXone Studio API Documentation
• CXone Studio Snippet Guide
• OAuth 2.0 Client Credentials Grant