Writing a CXone Studio Script That Performs a Database Lookup via DBConnector

What You Will Build

• A CXone Studio workflow script that queries an external database using the DBConnector action and routes execution based on the returned record set.
• The CXone Studio API (/api/v2/scripts) and Python httpx for programmatic script deployment and validation.
• Python 3.9+ with httpx and pydantic for type-safe payload construction and HTTP client management.

Prerequisites

• CXone OAuth2 client credentials with studio:scripts:write and studio:scripts:read scopes.
• CXone API v2 (current stable release).
• Python 3.9+, httpx>=0.24.0, pydantic>=2.0.0.
• A preconfigured DBConnector instance in CXone Admin (requires a valid connectionId).
• Network access to https://{{domain}}.api.cxone.com.

Authentication Setup

CXone uses standard OAuth2 client credentials flow for server-to-server API access. The token endpoint is /api/v2/oauth/token. You must cache the access token and handle expiration. The following implementation includes token caching, expiration tracking, and automatic retry logic for HTTP 429 rate-limit responses.

import time
import httpx
from typing import Optional

CXONE_BASE_URL = "https://myorg.api.cxone.com"
OAUTH_TOKEN_URL = f"{CXONE_BASE_URL}/api/v2/oauth/token"

class CXoneAuthClient:
    def __init__(self, client_id: str, client_secret: str, scopes: list[str]):
        self.client_id = client_id
        self.client_secret = client_secret
        self.scopes = scopes
        self._token: Optional[str] = None
        self._expires_at: float = 0.0
        self.http = httpx.Client(base_url=CXONE_BASE_URL, timeout=30.0)

    def _get_token(self) -> str:
        if self._token and time.time() < self._expires_at:
            return self._token

        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": " ".join(self.scopes)
        }

        response = self.http.post(OAUTH_TOKEN_URL, data=payload)
        response.raise_for_status()
        data = response.json()
        
        self._token = data["access_token"]
        self._expires_at = time.time() + data["expires_in"] - 30  # 30s buffer
        return self._token

    def get_auth_headers(self) -> dict:
        return {
            "Authorization": f"Bearer {self._get_token()}",
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

    def request_with_retry(self, method: str, path: str, **kwargs) -> httpx.Response:
        max_retries = 3
        for attempt in range(max_retries):
            response = self.http.request(method, path, headers=self.get_auth_headers(), **kwargs)
            
            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 2 ** attempt))
                print(f"Rate limited. Retrying in {retry_after}s (attempt {attempt + 1}/{max_retries})")
                time.sleep(retry_after)
                continue
                
            return response
            
        raise httpx.HTTPStatusError("Max retries exceeded for 429", request=response.request, response=response)

The request_with_retry method handles 429 responses by reading the Retry-After header or falling back to exponential backoff. This prevents cascading rate-limit failures during bulk script deployments.

Implementation

Step 1: Define the Studio Script JSON Structure

CXone Studio scripts are authored as JSON documents containing blocks, variables, and transitions. The DBConnector action requires a connectionId, a SQL query string, parameter bindings, and output variable mapping. The following payload defines a script that accepts a customer identifier, queries an external database, and branches based on the result.

{
  "name": "CustomerDBLookup",
  "description": "Performs a database lookup via DBConnector and routes based on result count",
  "version": "1.0.0",
  "entryBlock": "Start",
  "variables": [
    {"name": "input.customerId", "type": "String", "value": ""},
    {"name": "db.result", "type": "Object", "value": null},
    {"name": "db.recordCount", "type": "Number", "value": 0}
  ],
  "blocks": [
    {
      "name": "Start",
      "type": "Start",
      "properties": {},
      "transitions": {"next": "DBLookup"}
    },
    {
      "name": "DBLookup",
      "type": "DBConnector",
      "properties": {
        "connectionId": "conn_a1b2c3d4-e5f6-7890-abcd-ef1234567890",
        "query": "SELECT customer_id, first_name, loyalty_tier FROM customers WHERE customer_id = ?",
        "parameters": ["{{input.customerId}}"],
        "outputVariable": "db.result",
        "timeout": 5000
      },
      "transitions": {
        "success": "CheckRecordCount",
        "error": "HandleDBError"
      }
    },
    {
      "name": "CheckRecordCount",
      "type": "Condition",
      "properties": {
        "conditions": [
          {"left": "{{db.result.length}}", "operator": "GREATER_THAN", "right": "0"}
        ]
      },
      "transitions": {
        "true": "ProcessCustomer",
        "false": "NoRecordFound"
      }
    },
    {
      "name": "ProcessCustomer",
      "type": "SetVariable",
      "properties": {
        "variableName": "db.recordCount",
        "value": "{{db.result.length}}"
      },
      "transitions": {"next": "EndSuccess"}
    },
    {
      "name": "NoRecordFound",
      "type": "SetVariable",
      "properties": {
        "variableName": "db.recordCount",
        "value": "0"
      },
      "transitions": {"next": "EndSuccess"}
    },
    {
      "name": "HandleDBError",
      "type": "SetVariable",
      "properties": {
        "variableName": "error.message",
        "value": "{{error.description}}"
      },
      "transitions": {"next": "EndError"}
    },
    {
      "name": "EndSuccess",
      "type": "End",
      "properties": {"result": "SUCCESS"},
      "transitions": {}
    },
    {
      "name": "EndError",
      "type": "End",
      "properties": {"result": "ERROR"},
      "transitions": {}
    }
  ]
}

The parameters array uses CXone expression syntax ({{input.customerId}}) to bind runtime values to the SQL query. The outputVariable captures the result set as a JSON array. The timeout property prevents blocking the Studio runtime during slow database queries.

Step 2: Configure DBConnector Parameters and Transitions

The DBConnector action validates the SQL syntax against the registered database schema during deployment. You must ensure the connectionId matches a DBConnector instance configured in CXone Admin. Parameter binding uses zero-indexed placeholders (?) in the query string. Each placeholder maps to a corresponding index in the parameters array.

When the DBConnector executes, CXone evaluates the transitions object. The success transition triggers when the query returns without SQL errors, regardless of row count. The error transition triggers on connection failures, timeout violations, or syntax errors. Always define both transitions to prevent orphaned execution paths.

Step 3: Deploy the Script via API with Validation

The CXone Studio API accepts script definitions via POST /api/v2/scripts. The endpoint performs structural validation before persistence. You must include the Content-Type: application/json header and pass the script payload in the request body. The following Python code demonstrates the deployment call with comprehensive error handling.

import httpx
from typing import Any, Dict

def deploy_studio_script(auth_client: CXoneAuthClient, script_payload: Dict[str, Any]) -> Dict[str, Any]:
    path = "/api/v2/scripts"
    
    request_headers = {
        "Authorization": f"Bearer {auth_client._get_token()}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }
    
    response = auth_client.request_with_retry("POST", path, json=script_payload, headers=request_headers)
    
    if response.status_code == 201:
        return response.json()
    elif response.status_code == 400:
        error_body = response.json()
        raise ValueError(f"Validation failed: {error_body.get('description', 'Unknown validation error')}")
    elif response.status_code == 409:
        raise ConflictError(f"Script name already exists: {script_payload['name']}")
    else:
        response.raise_for_status()

class ConflictError(Exception):
    pass

HTTP Request Cycle Example

POST /api/v2/scripts HTTP/1.1
Host: myorg.api.cxone.com
Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...
Content-Type: application/json
Accept: application/json

{
  "name": "CustomerDBLookup",
  "description": "Performs a database lookup via DBConnector and routes based on result count",
  "version": "1.0.0",
  "entryBlock": "Start",
  "variables": [...],
  "blocks": [...]
}

HTTP Response Cycle Example (Success)

HTTP/1.1 201 Created
Content-Type: application/json
Location: /api/v2/scripts/script_x9y8z7w6-v5u4-3210-abcd-ef1234567890
Date: Mon, 15 Oct 2024 14:32:00 GMT

{
  "id": "script_x9y8z7w6-v5u4-3210-abcd-ef1234567890",
  "name": "CustomerDBLookup",
  "version": "1.0.0",
  "status": "VALID",
  "createdTimestamp": 1729000320000,
  "modifiedTimestamp": 1729000320000
}

The API returns 201 Created with a Location header pointing to the resource URI. The status field indicates VALID when the JSON structure passes schema validation. If the DBConnector connectionId references a disabled connection, the API returns 400 with a detailed validation error.

Step 4: List Scripts with Pagination

The CXone Studio API supports cursor-based pagination for collection endpoints. You must handle the nextPageToken to retrieve all scripts in large environments. The following implementation demonstrates pagination with automatic continuation.

def list_all_scripts(auth_client: CXoneAuthClient, page_size: int = 50) -> list[Dict[str, Any]]:
    all_scripts = []
    next_token: Optional[str] = None
    
    while True:
        params = {"pageSize": page_size}
        if next_token:
            params["nextPageToken"] = next_token
            
        response = auth_client.request_with_retry("GET", "/api/v2/scripts", params=params)
        response.raise_for_status()
        
        data = response.json()
        all_scripts.extend(data.get("entities", []))
        
        next_token = data.get("nextPageToken")
        if not next_token:
            break
            
    return all_scripts

The nextPageToken field is omitted from the response when the final page is reached. This pattern prevents infinite loops and ensures complete dataset retrieval without manual offset calculation.

Complete Working Example

The following module combines authentication, script definition, deployment, and listing into a single executable workflow. Replace the credential placeholders before execution.

import time
import httpx
from typing import Optional, Any, Dict

CXONE_BASE_URL = "https://myorg.api.cxone.com"
OAUTH_TOKEN_URL = f"{CXONE_BASE_URL}/api/v2/oauth/token"

class CXoneAuthClient:
    def __init__(self, client_id: str, client_secret: str, scopes: list[str]):
        self.client_id = client_id
        self.client_secret = client_secret
        self.scopes = scopes
        self._token: Optional[str] = None
        self._expires_at: float = 0.0
        self.http = httpx.Client(base_url=CXONE_BASE_URL, timeout=30.0)

    def _get_token(self) -> str:
        if self._token and time.time() < self._expires_at:
            return self._token
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": " ".join(self.scopes)
        }
        response = self.http.post(OAUTH_TOKEN_URL, data=payload)
        response.raise_for_status()
        data = response.json()
        self._token = data["access_token"]
        self._expires_at = time.time() + data["expires_in"] - 30
        return self._token

    def request_with_retry(self, method: str, path: str, **kwargs) -> httpx.Response:
        max_retries = 3
        for attempt in range(max_retries):
            headers = kwargs.pop("headers", {})
            headers["Authorization"] = f"Bearer {self._get_token()}"
            response = self.http.request(method, path, headers=headers, **kwargs)
            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 2 ** attempt))
                time.sleep(retry_after)
                continue
            return response
        raise httpx.HTTPStatusError("Max retries exceeded", request=response.request, response=response)

def deploy_studio_script(auth_client: CXoneAuthClient, script_payload: Dict[str, Any]) -> Dict[str, Any]:
    path = "/api/v2/scripts"
    response = auth_client.request_with_retry("POST", path, json=script_payload, headers={"Accept": "application/json"})
    if response.status_code == 201:
        return response.json()
    elif response.status_code == 400:
        raise ValueError(f"Validation failed: {response.json().get('description')}")
    else:
        response.raise_for_status()

def main():
    auth = CXoneAuthClient(
        client_id="your_client_id",
        client_secret="your_client_secret",
        scopes=["studio:scripts:write", "studio:scripts:read"]
    )

    script_definition = {
        "name": "CustomerDBLookup",
        "description": "Performs a database lookup via DBConnector and routes based on result count",
        "version": "1.0.0",
        "entryBlock": "Start",
        "variables": [
            {"name": "input.customerId", "type": "String", "value": ""},
            {"name": "db.result", "type": "Object", "value": None},
            {"name": "db.recordCount", "type": "Number", "value": 0}
        ],
        "blocks": [
            {"name": "Start", "type": "Start", "properties": {}, "transitions": {"next": "DBLookup"}},
            {
                "name": "DBLookup",
                "type": "DBConnector",
                "properties": {
                    "connectionId": "conn_a1b2c3d4-e5f6-7890-abcd-ef1234567890",
                    "query": "SELECT customer_id, first_name, loyalty_tier FROM customers WHERE customer_id = ?",
                    "parameters": ["{{input.customerId}}"],
                    "outputVariable": "db.result",
                    "timeout": 5000
                },
                "transitions": {"success": "CheckRecordCount", "error": "HandleDBError"}
            },
            {
                "name": "CheckRecordCount",
                "type": "Condition",
                "properties": {"conditions": [{"left": "{{db.result.length}}", "operator": "GREATER_THAN", "right": "0"}]},
                "transitions": {"true": "ProcessCustomer", "false": "NoRecordFound"}
            },
            {"name": "ProcessCustomer", "type": "SetVariable", "properties": {"variableName": "db.recordCount", "value": "{{db.result.length}}"}, "transitions": {"next": "EndSuccess"}},
            {"name": "NoRecordFound", "type": "SetVariable", "properties": {"variableName": "db.recordCount", "value": "0"}, "transitions": {"next": "EndSuccess"}},
            {"name": "HandleDBError", "type": "SetVariable", "properties": {"variableName": "error.message", "value": "{{error.description}}"}, "transitions": {"next": "EndError"}},
            {"name": "EndSuccess", "type": "End", "properties": {"result": "SUCCESS"}, "transitions": {}},
            {"name": "EndError", "type": "End", "properties": {"result": "ERROR"}, "transitions": {}}
        ]
    }

    try:
        result = deploy_studio_script(auth, script_definition)
        print(f"Script deployed successfully: {result['id']}")
    except Exception as e:
        print(f"Deployment failed: {e}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: HTTP 400 Bad Request

• Cause: The script JSON violates the CXone Studio schema. Common triggers include missing transitions, invalid connectionId, or mismatched parameter counts in the DBConnector query and parameters arrays.
• Fix: Validate the JSON structure against the Studio schema. Ensure every block defines all required transitions. Verify the DBConnector connection exists and is enabled in Admin.
• Code showing the fix:

if response.status_code == 400:
    errors = response.json().get("validationErrors", [])
    for err in errors:
        print(f"Block {err.get('blockName')}: {err.get('message')}")

Error: HTTP 401 Unauthorized / 403 Forbidden

• Cause: The OAuth token expired, or the client credentials lack studio:scripts:write scope.
• Fix: Regenerate the token using the _get_token method. Verify the OAuth client configuration in CXone Admin includes the required scopes.
• Code showing the fix:

# Token refresh is automatic in _get_token()
# Verify scopes during initialization
auth = CXoneAuthClient(client_id, client_secret, scopes=["studio:scripts:write"])

Error: HTTP 429 Too Many Requests

• Cause: Exceeded CXone API rate limits. Studio script deployments count against the organization-wide API quota.
• Fix: Implement exponential backoff. The request_with_retry method handles this automatically by reading Retry-After and sleeping before the next attempt.
• Code showing the fix:

# Already implemented in CXoneAuthClient.request_with_retry
# Returns after successful retry or raises HTTPStatusError on max retries

Error: DBConnector Runtime Timeout

• Cause: The database query exceeds the timeout property value (default 5000ms). Complex joins or unindexed columns cause slow execution.
• Fix: Increase the timeout property to 10000ms or optimize the SQL query. Add database indexes on filtered columns.
• Code showing the fix:

"properties": {
  "connectionId": "conn_xxx",
  "query": "SELECT ...",
  "parameters": ["{{input.customerId}}"],
  "outputVariable": "db.result",
  "timeout": 10000
}

Official References

• CXone Studio Scripts API Reference
• CXone DBConnector Action Documentation
• CXone OAuth2 Authentication Guide
• CXone API Pagination & Rate Limiting