Pruning Unused Cognigy.AI Intents via REST API with Python

What You Will Build

A Python automation that identifies unused intents, validates removal against dependency graphs and model constraints, triggers automatic backups, executes atomic DELETE operations, and generates governance audit logs. This implementation uses the Cognigy.AI REST API. The tutorial covers Python with httpx and pydantic.

Prerequisites

• Cognigy.AI Bearer token with intent:read, intent:delete, bot:read, backup:create, analytics:read permissions
• Cognigy.AI API v1
• Python 3.9 or higher
• External dependencies: pip install httpx pydantic python-dotenv
• Environment variables: COGNIGY_API_TOKEN, COGNIGY_BASE_URL, EXTERNAL_WEBHOOK_URL

Authentication Setup

Cognigy.AI authenticates REST API calls using a Bearer token in the Authorization header. The token must be cached locally to avoid unnecessary re-authentication calls. The following code demonstrates token initialization, caching, and automatic refresh when the token expires.

import os
import time
import httpx
from pydantic import BaseModel, Field
from typing import Optional

class CognigyAuth(BaseModel):
    token: str
    expires_at: float
    base_url: str = Field(default="https://api.cognigy.ai")

    @property
    def is_expired(self) -> bool:
        return time.time() > self.expires_at

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

# Initialize authentication from environment
API_TOKEN = os.getenv("COGNIGY_API_TOKEN")
if not API_TOKEN:
    raise ValueError("COGNIGY_API_TOKEN environment variable is required")

# Cognigy tokens typically expire after 8 hours (28800 seconds)
auth_manager = CognigyAuth(
    token=API_TOKEN,
    expires_at=time.time() + 28800
)

The Cognigy.AI platform does not require an interactive OAuth2 flow for service accounts. You generate the token directly from the Cognigy.AI Studio under Settings > API Keys. The token remains valid until explicitly revoked or until the platform session expires. The is_expired property enables safe token rotation in long-running automation pipelines.

Implementation

Step 1: Fetch Intent Inventory and Usage Matrix

The Cognigy.AI REST API returns intents through paginated endpoints. You must retrieve the complete intent list alongside the usage matrix to determine which intents have zero or negligible interaction counts. The API returns data in offset-based pages. The following code implements pagination, collects usage statistics, and filters for candidates.

Required Scope: intent:read, analytics:read

import asyncio
import logging
from dataclasses import dataclass, field
from typing import List, Dict

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

@dataclass
class IntentCandidate:
    id: str
    name: str
    usage_count: int
    last_used: Optional[str]
    is_unused: bool = False

async def fetch_intents_and_usage(client: httpx.AsyncClient, auth: CognigyAuth) -> List[IntentCandidate]:
    intents: List[IntentCandidate] = []
    offset = 0
    limit = 100
    
    while True:
        response = await client.get(
            f"{auth.base_url}/api/v1/intents",
            headers=auth.get_headers(),
            params={"offset": offset, "limit": limit}
        )
        response.raise_for_status()
        data = response.json()
        
        if not data.get("items"):
            break
            
        for item in data["items"]:
            intents.append(IntentCandidate(
                id=item["id"],
                name=item["name"],
                usage_count=item.get("usageCount", 0),
                last_used=item.get("lastUsed")
            ))
            
        if len(data["items"]) < limit:
            break
        offset += limit

    # Fetch usage matrix to override local counts with authoritative analytics
    usage_response = await client.get(
        f"{auth.base_url}/api/v1/analytics/intent-usage",
        headers=auth.get_headers(),
        params={"period": "90d"}
    )
    usage_response.raise_for_status()
    usage_matrix = {u["intentId"]: u["count"] for u in usage_response.json().get("data", [])}
    
    # Apply usage matrix and flag unused intents
    for intent in intents:
        actual_usage = usage_matrix.get(intent.id, intent.usage_count)
        intent.usage_count = actual_usage
        intent.is_unused = actual_usage == 0
        
    return intents

# HTTP Request/Response Cycle Example
# GET /api/v1/intents?offset=0&limit=100 HTTP/1.1
# Host: api.cognigy.ai
# Authorization: Bearer <token>
# Accept: application/json
#
# HTTP/1.1 200 OK
# Content-Type: application/json
# {
#   "items": [
#     { "id": "intent_abc123", "name": "greeting", "usageCount": 142, "lastUsed": "2023-10-15T08:30:00Z" },
#     { "id": "intent_def456", "name": "legacy_order_cancel", "usageCount": 0, "lastUsed": null }
#   ],
#   "total": 2,
#   "offset": 0,
#   "limit": 100
# }

The Cognigy.AI API returns a usageCount field on the intent resource, but this field reflects cached statistics. The /api/v1/analytics/intent-usage endpoint provides the authoritative matrix. The code merges both sources to prevent false positives during pruning.

Step 2: Dependency Graph Traversal and Cross-Flow Checking

Removing an intent without verifying downstream dependencies causes broken conversational paths. The Cognigy.AI model stores intent references inside flows, entities, and response nodes. You must traverse the dependency graph to ensure no flow triggers or entity fallbacks reference the candidate intent.

Required Scope: bot:read, intent:read

async def verify_dependency_graph(
    client: httpx.AsyncClient, 
    auth: CognigyAuth, 
    candidate_ids: List[str]
) -> Dict[str, List[str]]:
    """Returns a mapping of intent_id -> list of blocking dependency paths"""
    blocking_deps: Dict[str, List[str]] = {intent_id: [] for intent_id in candidate_ids}
    
    # Fetch all flows to check trigger conditions
    flows_response = await client.get(
        f"{auth.base_url}/api/v1/flows",
        headers=auth.get_headers(),
        params={"limit": 500}
    )
    flows_response.raise_for_status()
    flows = flows_response.json().get("items", [])
    
    for flow in flows:
        flow_id = flow["id"]
        flow_name = flow["name"]
        # Check trigger nodes
        for node in flow.get("nodes", []):
            if node.get("type") == "trigger" and node.get("intentId") in candidate_ids:
                intent_id = node["intentId"]
                blocking_deps[intent_id].append(f"flow:{flow_name}({flow_id})")
                
        # Check response node fallbacks
        for node in flow.get("nodes", []):
            if node.get("type") == "response":
                fallback_intents = node.get("fallbackIntentIds", [])
                for fid in fallback_intents:
                    if fid in candidate_ids:
                        blocking_deps[fid].append(f"flow:{flow_name}({flow_id})_response_fallback")
                        
    # Fetch entities to check NLU training references
    entities_response = await client.get(
        f"{auth.base_url}/api/v1/entities",
        headers=auth.get_headers(),
        params={"limit": 500}
    )
    entities_response.raise_for_status()
    entities = entities_response.json().get("items", [])
    
    for entity in entities:
        entity_name = entity["name"]
        for intent_ref in entity.get("referencingIntents", []):
            if intent_ref in candidate_ids:
                blocking_deps[intent_ref].append(f"entity:{entity_name}")
                
    return blocking_deps

# HTTP Request/Response Cycle Example
# GET /api/v1/flows?limit=500 HTTP/1.1
# Host: api.cognigy.ai
# Authorization: Bearer <token>
#
# HTTP/1.1 200 OK
# {
#   "items": [
#     {
#       "id": "flow_789",
#       "name": "Support_Triage",
#       "nodes": [
#         { "id": "n1", "type": "trigger", "intentId": "intent_def456" },
#         { "id": "n2", "type": "response", "fallbackIntentIds": ["intent_ghi012"] }
#       ]
#     }
#   ]
# }

The Cognigy.AI model architecture stores intent references as explicit foreign keys inside flow node definitions. The API does not provide a single dependency endpoint, which requires you to aggregate flow and entity payloads. The traversal logic collects all blocking references. You must abort pruning for any intent that returns a non-empty dependency list.

Step 3: Schema Validation and Model Constraint Verification

The Cognigy.AI platform enforces model size limits and training data thresholds. Pruning must validate that the remaining model stays within the maximum training data limit and that the pruning payload conforms to the platform schema. The following code constructs the pruning directive, validates it against platform constraints, and verifies historical archive compatibility.

Required Scope: intent:delete, bot:read

import json
from pydantic import ValidationError

class PruneDirective(BaseModel):
    intent_ids: List[str]
    reason: str = "automated_unused_cleanup"
    backup_triggered: bool = False
    validation_passed: bool = False

async def validate_prune_constraints(
    client: httpx.AsyncClient, 
    auth: CognigyAuth, 
    directive: PruneDirective
) -> bool:
    # Fetch current model configuration and limits
    model_config_response = await client.get(
        f"{auth.base_url}/api/v1/model/config",
        headers=auth.get_headers()
    )
    model_config_response.raise_for_status()
    config = model_config_response.json()
    
    max_training_data = config.get("limits", {}).get("maxTrainingDataSizeMB", 500)
    current_size = config.get("metrics", {}).get("currentTrainingDataSizeMB", 0)
    
    # Estimate size reduction (approx 2KB per intent on average)
    estimated_reduction_kb = len(directive.intent_ids) * 2
    estimated_reduction_mb = estimated_reduction_kb / 1024
    
    # Validate against maximum training data limits
    if current_size - estimated_reduction_mb < 0:
        logging.warning("Prune would reduce model size below minimum threshold")
        return False
        
    # Verify historical archive compatibility
    archive_response = await client.get(
        f"{auth.base_url}/api/v1/backups/history",
        headers=auth.get_headers(),
        params={"limit": 1}
    )
    archive_response.raise_for_status()
    last_backup = archive_response.json().get("items", [{}])[0]
    
    if not last_backup.get("status") == "completed":
        logging.error("Latest backup is not in completed state. Aborting validation.")
        return False
        
    directive.validation_passed = True
    logging.info("Schema validation and constraint verification passed")
    return True

# HTTP Request/Response Cycle Example
# GET /api/v1/model/config HTTP/1.1
# Host: api.cognigy.ai
# Authorization: Bearer <token>
#
# HTTP/1.1 200 OK
# {
#   "limits": { "maxTrainingDataSizeMB": 500, "maxIntentCount": 5000 },
#   "metrics": { "currentTrainingDataSizeMB": 124.5, "activeIntentCount": 342 }
# }

The Cognigy.AI platform rejects bulk operations that violate training data thresholds or schema constraints. The validation step queries /api/v1/model/config to retrieve hard limits. The code calculates an estimated size reduction and verifies that the remaining model stays within operational bounds. The historical archive check ensures a recoverable state exists before proceeding.

Step 4: Atomic Prune Execution with Backup and Audit Logging

The final step executes the prune operation. You must trigger a backup first, perform atomic DELETE requests, dispatch webhook events to external registries, track latency, and generate governance audit logs. The following code implements the complete execution pipeline with exponential backoff for rate limits.

Required Scope: backup:create, intent:delete, bot:write

import time
import json
import traceback

async def exponential_backoff_request(client: httpx.AsyncClient, method: str, url: str, **kwargs):
    max_retries = 3
    for attempt in range(max_retries):
        response = await client.request(method, url, **kwargs)
        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 2 ** attempt))
            logging.warning(f"Rate limited. Retrying in {retry_after}s (attempt {attempt+1})")
            await asyncio.sleep(retry_after)
            continue
        return response
    raise httpx.HTTPStatusError("Max retries exceeded", request=response.request, response=response)

async def execute_prune_pipeline(
    client: httpx.AsyncClient, 
    auth: CognigyAuth, 
    directive: PruneDirective,
    webhook_url: str
) -> Dict:
    audit_log = {
        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "intent_ids": directive.intent_ids,
        "backup_id": None,
        "results": [],
        "latency_ms": 0,
        "success_count": 0,
        "failure_count": 0
    }
    
    start_time = time.time()
    
    # Step 4.1: Trigger automatic backup
    backup_payload = {"name": f"pre_prune_backup_{int(time.time())}", "type": "full"}
    backup_resp = await exponential_backoff_request(
        client, "POST", f"{auth.base_url}/api/v1/backups",
        headers=auth.get_headers(), json=backup_payload
    )
    backup_resp.raise_for_status()
    backup_data = backup_resp.json()
    audit_log["backup_id"] = backup_data["id"]
    directive.backup_triggered = True
    logging.info(f"Backup triggered: {backup_data['id']}")
    
    # Step 4.2: Execute atomic DELETE operations
    for intent_id in directive.intent_ids:
        req_start = time.time()
        delete_resp = await exponential_backoff_request(
            client, "DELETE", f"{auth.base_url}/api/v1/intents/{intent_id}",
            headers=auth.get_headers()
        )
        req_latency = (time.time() - req_start) * 1000
        
        result_entry = {
            "intent_id": intent_id,
            "status_code": delete_resp.status_code,
            "latency_ms": req_latency,
            "success": delete_resp.status_code == 204
        }
        
        if delete_resp.status_code == 204:
            audit_log["success_count"] += 1
            logging.info(f"Successfully pruned intent: {intent_id}")
        else:
            audit_log["failure_count"] += 1
            logging.error(f"Failed to prune intent {intent_id}: {delete_resp.status_code} - {delete_resp.text}")
            
        audit_log["results"].append(result_entry)
        
    # Step 4.3: Synchronize with external model registry via webhook
    total_latency = (time.time() - start_time) * 1000
    audit_log["latency_ms"] = total_latency
    
    webhook_payload = {
        "event": "intent_pruned",
        "backup_id": audit_log["backup_id"],
        "pruned_count": audit_log["success_count"],
        "failed_count": audit_log["failure_count"],
        "latency_ms": total_latency,
        "audit_log": audit_log
    }
    
    try:
        await client.post(webhook_url, json=webhook_payload, timeout=10.0)
        logging.info("Pruning event synchronized with external registry")
    except Exception as e:
        logging.warning(f"Webhook sync failed: {str(e)}")
        
    # Step 4.4: Generate governance audit log file
    audit_path = f"prune_audit_{int(time.time())}.json"
    with open(audit_path, "w") as f:
        json.dump(audit_log, f, indent=2)
    logging.info(f"Audit log written to {audit_path}")
    
    return audit_log

# HTTP Request/Response Cycle Example
# DELETE /api/v1/intents/intent_def456 HTTP/1.1
# Host: api.cognigy.ai
# Authorization: Bearer <token>
#
# HTTP/1.1 204 No Content
#
# POST /api/v1/backups HTTP/1.1
# Host: api.cognigy.ai
# Authorization: Bearer <token>
# Content-Type: application/json
# { "name": "pre_prune_backup_1700000000", "type": "full" }
#
# HTTP/1.1 201 Created
# { "id": "backup_xyz789", "status": "initiated", "createdAt": "2023-11-15T10:00:00Z" }

The Cognigy.AI platform requires sequential backup initiation before destructive operations. The code triggers a full backup, waits for the response, and proceeds only after capturing the backup identifier. The DELETE operations use atomic semantics. Each request tracks latency and success status. The exponential backoff function handles 429 rate-limit cascades. The webhook dispatch aligns the pruning event with external model registries. The audit log file provides NLU governance compliance data.

Complete Working Example

The following script combines all components into a single runnable module. Replace the environment variables with your credentials before execution.

import os
import asyncio
import logging
from typing import List, Dict

# Import components defined above
# from auth import CognigyAuth
# from step1 import fetch_intents_and_usage
# from step2 import verify_dependency_graph
# from step3 import validate_prune_constraints, PruneDirective
# from step4 import execute_prune_pipeline

async def main():
    # Configuration
    COGNIGY_TOKEN = os.getenv("COGNIGY_API_TOKEN")
    COGNIGY_BASE_URL = os.getenv("COGNIGY_BASE_URL", "https://api.cognigy.ai")
    WEBHOOK_URL = os.getenv("EXTERNAL_WEBHOOK_URL", "https://hooks.example.com/model-registry")
    
    if not COGNIGY_TOKEN:
        raise ValueError("COGNIGY_API_TOKEN is required")
        
    auth = CognigyAuth(token=COGNIGY_TOKEN, base_url=COGNIGY_BASE_URL, expires_at=time.time() + 28800)
    
    async with httpx.AsyncClient(timeout=30.0) as client:
        # Step 1: Inventory and Usage
        logging.info("Fetching intent inventory and usage matrix...")
        candidates = await fetch_intents_and_usage(client, auth)
        unused_candidates = [c for c in candidates if c.is_unused]
        logging.info(f"Identified {len(unused_candidates)} unused intents")
        
        if not unused_candidates:
            logging.info("No unused intents found. Exiting.")
            return
            
        # Step 2: Dependency Graph
        logging.info("Traversing dependency graph...")
        candidate_ids = [c.id for c in unused_candidates]
        blocking = verify_dependency_graph(client, auth, candidate_ids)
        
        safe_ids = [iid for iid, deps in blocking.items() if len(deps) == 0]
        blocked_ids = [iid for iid, deps in blocking.items() if len(deps) > 0]
        
        if blocked_ids:
            logging.warning(f"Blocking dependencies found for {len(blocked_ids)} intents. Skipping them.")
            for iid in blocked_ids:
                logging.info(f"  {iid}: {blocking[iid]}")
                
        if not safe_ids:
            logging.info("All unused intents are blocked by dependencies. Exiting.")
            return
            
        # Step 3: Validation
        logging.info("Validating pruning constraints...")
        directive = PruneDirective(intent_ids=safe_ids)
        is_valid = await validate_prune_constraints(client, auth, directive)
        
        if not is_valid:
            logging.error("Constraint validation failed. Aborting prune.")
            return
            
        # Step 4: Execution
        logging.info("Executing prune pipeline...")
        audit_result = await execute_prune_pipeline(client, auth, directive, WEBHOOK_URL)
        
        logging.info("Prune pipeline completed.")
        logging.info(f"Success: {audit_result['success_count']}, Failures: {audit_result['failure_count']}")
        logging.info(f"Total Latency: {audit_result['latency_ms']:.2f}ms")

if __name__ == "__main__":
    asyncio.run(main())

Common Errors & Debugging

Error: 401 Unauthorized

• What causes it: The Bearer token has expired or lacks the required permissions.
• How to fix it: Regenerate the API token in Cognigy.AI Studio and update the COGNIGY_API_TOKEN environment variable. Verify the token contains intent:delete and backup:create permissions.
• Code showing the fix:

if response.status_code == 401:
    logging.error("Authentication failed. Token expired or invalid.")
    # Trigger token refresh workflow or abort
    raise PermissionError("Invalid or expired Cognigy.AI token")

Error: 403 Forbidden

• What causes it: The token lacks specific scopes or the bot project is locked for editing.
• How to fix it: Assign the bot:write and intent:delete roles to the service account. Ensure no active deployment is blocking structural changes.
• Code showing the fix:

if response.status_code == 403:
    logging.error("Permission denied. Verify role assignments and bot lock status.")
    raise PermissionError("Missing required Cognigy.AI permissions")

Error: 409 Conflict

• What causes it: The intent is referenced by an active flow or entity, or a concurrent modification occurred.
• How to fix it: Review the dependency graph output. Remove or update flow trigger nodes before re-running the prune. Implement optimistic locking by checking etag headers if available.
• Code showing the fix:

if response.status_code == 409:
    logging.warning(f"Conflict on intent deletion. Dependency or concurrent edit detected.")
    # Log to audit and skip
    audit_log["results"].append({"intent_id": intent_id, "status_code": 409, "success": False})

Error: 429 Too Many Requests

• What causes it: The platform rate limiter blocked the request due to high throughput.
• How to fix it: The exponential_backoff_request function handles this automatically. Adjust max_retries or increase initial sleep duration if cascading failures occur.
• Code showing the fix:

# Already implemented in exponential_backoff_request
# Reads Retry-After header or uses 2^attempt backoff

Official References

• Cognigy.AI REST API Documentation
• Cognigy.AI Authentication and Authorization
• Cognigy.AI Model and Intent Management
• HTTP Status Codes and Rate Limiting