Routing Cognigy Bot Intent Data to NICE CXone Dynamic Queues

What You Will Build

• This tutorial demonstrates how to capture intent confidence scores and entity data from a NICE Cognigy bot and use that data to dynamically route the conversation to the correct NICE CXone queue or skill group.
• It utilizes the NICE CXone REST API (specifically the Interaction and Routing APIs) and the Cognigy.AI SDK for Node.js.
• The implementation covers JavaScript/TypeScript for the Cognigy bot logic and Python for a backend middleware service that translates Cognigy payloads into CXone routing directives.

Prerequisites

• NICE CXone Account: An active tenant with access to the API.
• OAuth Credentials: A CXone OAuth client ID and secret with the routing:queue:write and interaction:conversation:write scopes.
• Cognigy.AI Project: An active project with a published bot.
• Node.js: Version 16 or higher for the Cognigy.AI SDK environment.
• Python: Version 3.9 or higher for the middleware service.
• Dependencies:
  • Node.js: @cognigy/sdk, axios
  • Python: requests, python-dotenv

Authentication Setup

NICE CXone uses OAuth 2.0 for API authentication. Because the Cognigy bot runs in a serverless environment or a separate container, it should not hold long-lived CXone tokens. Instead, a middleware service or a secure backend endpoint should handle token acquisition.

Below is a Python utility class to manage OAuth tokens. This class handles the initial grant and refreshes tokens before they expire to prevent 401 errors during high-volume routing.

import requests
import time
from typing import Optional
import os

class CxoneAuthManager:
    def __init__(self, client_id: str, client_secret: str, base_url: str = "https://api.us-gov-1.niceincontact.com"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = base_url
        self.token_endpoint = f"{base_url}/api/v2/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0
        self.refresh_threshold = 60  # Refresh 60 seconds before expiry

    def _get_token(self) -> dict:
        """Fetches a new OAuth token from CXone."""
        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }
        data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": "routing:queue:write interaction:conversation:write"
        }
        
        response = requests.post(self.token_endpoint, headers=headers, data=data)
        response.raise_for_status()
        return response.json()

    def get_valid_token(self) -> str:
        """Returns a valid access token, refreshing if necessary."""
        current_time = time.time()
        
        # Check if token exists and is not expired (accounting for threshold)
        if self.access_token and (current_time < self.token_expiry - self.refresh_threshold):
            return self.access_token

        # Fetch new token
        token_data = self._get_token()
        self.access_token = token_data["access_token"]
        # CXone tokens typically expire in 3600 seconds (1 hour)
        self.token_expiry = current_time + token_data.get("expires_in", 3600)
        
        return self.access_token

# Example Usage
# auth = CxoneAuthManager(os.getenv("CXONE_CLIENT_ID"), os.getenv("CXONE_CLIENT_SECRET"))
# token = auth.get_valid_token()

Implementation

Step 1: Extracting Intent Data in Cognigy.AI

The first step is to structure the data leaving the Cognigy bot. When the bot determines the user’s intent (e.g., “BillingQuestion”, “TechnicalSupport”), it must pass this information to the external routing system. We do not route directly from Cognigy to CXone queues in this architecture; instead, Cognigy sends a payload to a middleware endpoint, which then interacts with CXone.

In your Cognigy.AI project, create a new Step called RouteToCXone. In the JavaScript code editor for this step, extract the intent and confidence score.

/**
 * Cognigy.AI Step: RouteToCXone
 * Extracts intent data and prepares payload for CXone middleware.
 */
module.exports = async ({ session, utils }) => {
    try {
        // 1. Retrieve the resolved intent from the session
        const intent = session.getVariable("intent");
        const confidence = session.getVariable("confidence");
        
        // 2. Validate that an intent was successfully recognized
        if (!intent || confidence < 0.7) {
            // Fallback logic: send to general queue
            session.setVariable("routingQueue", "GeneralSupport");
            session.setVariable("routingReason", "LowConfidenceFallback");
            return;
        }

        // 3. Map Cognigy Intents to CXone Queue Names/Skills
        // This mapping should ideally be in a configuration file or database
        const queueMapping = {
            "BillingQuestion": "Billing_Tier1",
            "TechnicalSupport": "Tech_Support_L2",
            "SalesInquiry": "Sales_NewBusiness",
            "AccountManagement": "Account_Mgmt"
        };

        const targetQueue = queueMapping[intent] || "GeneralSupport";

        // 4. Set session variables for the webhook payload
        session.setVariable("cxoneTargetQueue", targetQueue);
        session.setVariable("cxoneIntent", intent);
        session.setVariable("cxoneConfidence", confidence);

        // 5. Prepare the payload for the HTTP request step
        // We will use a subsequent HTTP Step to send this data
        const payload = {
            sessionId: session.id,
            userId: session.user.id,
            targetQueue: targetQueue,
            intent: intent,
            confidence: confidence,
            timestamp: new Date().toISOString()
        };

        session.setVariable("routingPayload", JSON.stringify(payload));

    } catch (error) {
        console.error("Error preparing routing data:", error);
        session.setVariable("cxoneTargetQueue", "ErrorHandling");
    }
};

After this step, add an HTTP Request step in Cognigy.AI. Configure it to POST to your middleware endpoint (e.g., https://your-middleware.com/route). Set the body to {{routingPayload}} and Content-Type to application/json.

Step 2: Building the Middleware Router

The middleware receives the payload from Cognigy and translates it into a CXone API call. This service acts as the bridge, handling authentication and error retries.

Create a Python FastAPI application (or Flask/Express) to handle the incoming webhook.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import requests
import logging
import os

# Import the AuthManager from Step 1
from cxone_auth import CxoneAuthManager

app = FastAPI()
auth_manager = CxoneAuthManager(
    os.getenv("CXONE_CLIENT_ID"),
    os.getenv("CXONE_CLIENT_SECRET")
)

class RoutingPayload(BaseModel):
    sessionId: str
    userId: str
    targetQueue: str
    intent: str
    confidence: float
    timestamp: str

@app.post("/route")
def route_interaction(payload: RoutingPayload):
    """
    Receives intent data from Cognigy and updates the CXone interaction 
    to route to the specified queue.
    """
    try:
        # 1. Authenticate with CXone
        token = auth_manager.get_valid_token()
        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }

        # 2. Identify the Interaction in CXone
        # In a real scenario, you might link the Cognigy Session ID to 
        # a CXone Interaction ID via a database or header passed during IVR handoff.
        # For this example, we assume the Cognigy bot is embedded in a CXone 
        # Web Chat or Voice channel where the Interaction ID is known or 
        # we are creating a new interaction context.
        
        # NOTE: If this is a voice call, the Interaction ID is passed from the IVR.
        # If this is Web Chat, the Interaction ID is generated by CXone.
        # We will use a placeholder Interaction ID here. 
        # In production, retrieve this from a session store or request header.
        interaction_id = payload.userId # Simplified for example
        
        # 3. Prepare the Routing Directive
        # We use the Interaction API to update the routing target.
        # Endpoint: PATCH /api/v2/interactions/conversations/{interactionId}
        
        # Construct the update payload
        # We are updating the 'routing' object to specify the queue
        update_payload = {
            "routing": {
                "queueId": payload.targetQueue, 
                # Note: In production, you should map Queue Name to Queue ID 
                # by first querying GET /api/v2/routing/queues/name/{name}
                "skillRequirements": [] 
            },
            "metadata": {
                "intent": payload.intent,
                "confidence": payload.confidence
            }
        }

        # 4. Execute the API Call
        cxone_base = "https://api.us-gov-1.niceincontact.com" # Adjust for your region
        url = f"{cxone_base}/api/v2/interactions/conversations/{interaction_id}"
        
        response = requests.patch(url, json=update_payload, headers=headers)
        
        if response.status_code == 200:
            logging.info(f"Successfully routed session {payload.sessionId} to {payload.targetQueue}")
            return {"status": "success", "queue": payload.targetQueue}
        elif response.status_code == 404:
            raise HTTPException(status_code=404, detail="Interaction not found in CXone")
        elif response.status_code == 429:
            # Handle Rate Limiting
            logging.warning("CXone API Rate Limited. Retry logic should be implemented.")
            raise HTTPException(status_code=503, detail="Service temporarily unavailable due to rate limits")
        else:
            raise HTTPException(status_code=response.status_code, detail=response.text)

    except Exception as e:
        logging.error(f"Routing failed for session {payload.sessionId}: {str(e)}")
        raise HTTPException(status_code=500, detail="Internal Routing Error")

Step 3: Resolving Queue Names to IDs

CXone APIs generally require IDs (UUIDs) rather than names for routing directives. The previous step used a simplified queueId field. In production, you must resolve the queue name to its ID.

Add a helper function to resolve queue IDs. This should be cached to avoid excessive API calls.

from functools import lru_cache
import requests

@lru_cache(maxsize=128)
def get_queue_id_by_name(queue_name: str, auth_token: str) -> str:
    """
    Resolves a CXone Queue Name to its UUID.
    Uses LRU cache to prevent repeated API calls for the same queue.
    """
    headers = {
        "Authorization": f"Bearer {auth_token}",
        "Content-Type": "application/json"
    }
    
    # CXone Endpoint: GET /api/v2/routing/queues/name/{name}
    # URL encode the queue name
    import urllib.parse
    encoded_name = urllib.parse.quote(queue_name, safe='')
    url = f"https://api.us-gov-1.niceincontact.com/api/v2/routing/queues/name/{encoded_name}"
    
    response = requests.get(url, headers=headers)
    
    if response.status_code == 200:
        return response.json().get("id")
    else:
        raise Exception(f"Failed to resolve queue ID for '{queue_name}'. Status: {response.status_code}")

# Update the /route endpoint to use this helper:
# queue_id = get_queue_id_by_name(payload.targetQueue, token)
# update_payload["routing"]["queueId"] = queue_id

Complete Working Example

Below is the complete Python middleware service. It includes the auth manager, queue resolution, and the FastAPI router.

import os
import time
import requests
import logging
import urllib.parse
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from functools import lru_cache

# Configure Logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# --- Configuration ---
CXONE_CLIENT_ID = os.getenv("CXONE_CLIENT_ID")
CXONE_CLIENT_SECRET = os.getenv("CXONE_CLIENT_SECRET")
CXONE_BASE_URL = os.getenv("CXONE_BASE_URL", "https://api.us-gov-1.niceincontact.com")

app = FastAPI()

# --- Authentication Module ---
class CxoneAuthManager:
    def __init__(self, client_id: str, client_secret: str, base_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = base_url
        self.token_endpoint = f"{base_url}/api/v2/oauth/token"
        self.access_token = None
        self.token_expiry = 0
        self.refresh_threshold = 60

    def _get_token(self) -> dict:
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": "routing:queue:write interaction:conversation:write routing:queue:read"
        }
        response = requests.post(self.token_endpoint, headers=headers, data=data)
        response.raise_for_status()
        return response.json()

    def get_valid_token(self) -> str:
        current_time = time.time()
        if self.access_token and (current_time < self.token_expiry - self.refresh_threshold):
            return self.access_token
        
        token_data = self._get_token()
        self.access_token = token_data["access_token"]
        self.token_expiry = current_time + token_data.get("expires_in", 3600)
        return self.access_token

auth_manager = CxoneAuthManager(CXONE_CLIENT_ID, CXONE_CLIENT_SECRET, CXONE_BASE_URL)

# --- Queue Resolution Module ---
@lru_cache(maxsize=128)
def get_queue_id_by_name(queue_name: str, auth_token: str) -> str:
    headers = {"Authorization": f"Bearer {auth_token}"}
    encoded_name = urllib.parse.quote(queue_name, safe='')
    url = f"{CXONE_BASE_URL}/api/v2/routing/queues/name/{encoded_name}"
    
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        return response.json().get("id")
    else:
        raise Exception(f"Queue '{queue_name}' not found. Status: {response.status_code}")

# --- Data Models ---
class RoutingPayload(BaseModel):
    sessionId: str
    userId: str
    targetQueue: str
    intent: str
    confidence: float
    timestamp: str

# --- API Endpoints ---
@app.post("/route")
def route_interaction(payload: RoutingPayload):
    try:
        token = auth_manager.get_valid_token()
        
        # 1. Resolve Queue Name to ID
        try:
            queue_id = get_queue_id_by_name(payload.targetQueue, token)
        except Exception as e:
            logger.error(f"Queue resolution failed: {e}")
            raise HTTPException(status_code=400, detail=f"Invalid Queue Name: {payload.targetQueue}")

        # 2. Prepare Routing Update
        # Note: In a real Voice scenario, you would have the Interaction ID from the IVR.
        # Here we assume the userId is linked to the Interaction ID for demonstration.
        interaction_id = payload.userId 
        
        update_payload = {
            "routing": {
                "queueId": queue_id
            },
            "metadata": {
                "cognigyIntent": payload.intent,
                "confidenceScore": payload.confidence
            }
        }

        # 3. Send to CXone
        url = f"{CXONE_BASE_URL}/api/v2/interactions/conversations/{interaction_id}"
        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }

        response = requests.patch(url, json=update_payload, headers=headers)

        if response.status_code == 200:
            logger.info(f"Routed {payload.sessionId} to Queue ID: {queue_id}")
            return {"status": "success", "queueId": queue_id}
        elif response.status_code == 404:
            raise HTTPException(status_code=404, detail=f"Interaction {interaction_id} not found")
        elif response.status_code == 429:
            raise HTTPException(status_code=503, detail="CXone API Rate Limited")
        else:
            raise HTTPException(status_code=response.status_code, detail=response.text)

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Unexpected error: {str(e)}")
        raise HTTPException(status_code=500, detail="Internal Server Error")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Common Errors & Debugging

Error: 401 Unauthorized

Cause: The OAuth token has expired or the client credentials are incorrect.
Fix: Ensure the CxoneAuthManager is correctly fetching a new token. Check that the client_id and client_secret in your environment variables match the CXone Admin Console. Verify that the scope routing:queue:write is included in the token request.

Error: 404 Interaction Not Found

Cause: The interactionId passed to the PATCH endpoint does not exist in CXone.
Fix: In a Voice scenario, ensure the IVR is correctly passing the CXone Interaction ID to the Cognigy webhook. In a Web Chat scenario, ensure the Chat SDK has initialized the interaction before the bot attempts to route. Log the interactionId to verify it is a valid UUID.

Error: 403 Forbidden

Cause: The OAuth token lacks the necessary scope.
Fix: Check the token response body. Ensure the scope parameter in the OAuth request includes interaction:conversation:write. If using a restricted OAuth client, verify that the client has access to the specific routing queues.

Error: 429 Too Many Requests

Cause: The middleware is sending requests faster than CXone allows (rate limiting).
Fix: Implement exponential backoff in the CxoneAuthManager or the /route endpoint. For high-volume bots, batch routing updates or use a message queue (like RabbitMQ or AWS SQS) to throttle requests to the CXone API.

Official References

• NICE CXone Interaction API Documentation
• NICE CXone Routing API Documentation
• NICE CXone OAuth 2.0 Guide
• Cognigy.AI SDK Documentation