Deprovision Genesys Cloud SIP Trunks Safely via Telephony API with Python

What You Will Build

• This code safely disables and removes SIP trunk registrations from Genesys Cloud CX while validating active calls, session limits, and credential states.
• The implementation uses the Genesys Cloud Telephony API, Analytics API, and Platform Webhooks API via direct HTTP calls.
• The programming language covered is Python 3.9+ using httpx for synchronous and asynchronous request handling.

Prerequisites

• OAuth client type: Machine-to-Machine (M2M) with client_id and client_secret
• Required scopes: telephony:trunk:read, telephony:trunk:delete, telephony:provideredge:read, analytics:conversations:read, webhook:write
• API version: Genesys Cloud CX REST API (v2)
• Runtime: Python 3.9+
• External dependencies: httpx, pydantic, pyyaml, structlog

Authentication Setup

The Genesys Cloud platform requires OAuth 2.0 bearer tokens. The following code fetches a token, caches it in memory, and handles expiration by tracking the expires_in claim. The token request targets the environment-specific authorization server.

import httpx
import time
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class GenesysAuth:
    client_id: str
    client_secret: str
    environment: str = "mypurecloud.com"
    _token: Optional[str] = field(default=None, repr=False)
    _expires_at: float = field(default=0.0, repr=False)

    @property
    def token_url(self) -> str:
        return f"https://login.{self.environment}/oauth/token"

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

        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.post(
                self.token_url,
                data={
                    "grant_type": "client_credentials",
                    "client_id": self.client_id,
                    "client_secret": self.client_secret,
                    "scope": "telephony:trunk:read telephony:trunk:delete telephony:provideredge:read analytics:conversations:read webhook:write"
                }
            )
            response.raise_for_status()
            payload = response.json()
            self._token = payload["access_token"]
            self._expires_at = time.time() + payload["expires_in"]
            return self._token

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

Implementation

Step 1: Validate Trunk Reference and SIP Matrix Constraints

Before attempting deprovisioning, you must verify the trunk exists and extract its SIP matrix configuration. The /api/v2/telephony/users/trunks endpoint returns the full trunk registry. You must filter for the target trunk and validate its sipMatrix configuration against telephony constraints.

Scope: telephony:trunk:read

import httpx
import structlog
from typing import Any, Dict, List

logger = structlog.get_logger()

class TrunkValidator:
    def __init__(self, auth: GenesysAuth):
        self.auth = auth
        self.base_url = f"https://api.{auth.environment}"

    async def fetch_trunks(self) -> List[Dict[str, Any]]:
        """Fetches all trunk registrations with pagination support."""
        async with httpx.AsyncClient(timeout=15.0) as client:
            all_trunks = []
            page = 1
            while True:
                response = await client.get(
                    f"{self.base_url}/api/v2/telephony/users/trunks",
                    headers=self.auth.headers(),
                    params={"page": page, "pageSize": 25}
                )
                
                if response.status_code == 429:
                    retry_after = int(response.headers.get("Retry-After", 5))
                    logger.warning("Rate limited, retrying", retry_after=retry_after)
                    await asyncio.sleep(retry_after)
                    continue
                    
                response.raise_for_status()
                data = response.json()
                all_trunks.extend(data["entities"])
                
                if page >= data["pageCount"]:
                    break
                page += 1
            return all_trunks

    async def validate_trunk_reference(self, trunk_id: str) -> Dict[str, Any]:
        trunks = await self.fetch_trunks()
        target = next((t for t in trunks if t["id"] == trunk_id), None)
        
        if not target:
            raise ValueError(f"Trunk {trunk_id} not found in registry.")
            
        if not target.get("sipMatrix"):
            raise ValueError("Trunk lacks SIP matrix configuration. Cannot deprovision.")
            
        logger.info("Trunk validated", trunk_id=trunk_id, site_id=target["siteId"])
        return target

HTTP Request Cycle:

GET /api/v2/telephony/users/trunks?page=1&pageSize=25 HTTP/1.1
Host: api.mypurecloud.com
Authorization: Bearer <token>
Accept: application/json

HTTP Response:

{
  "entities": [
    {
      "id": "trunk-abc-123",
      "name": "Primary_SIP_Trunk",
      "siteId": "site-xyz-789",
      "sipMatrix": {
        "type": "manual",
        "proxyAddresses": ["sip.example.com"]
      },
      "maxConcurrentSessions": 500
    }
  ],
  "page": 1,
  "pageSize": 25,
  "pageCount": 1
}

Step 2: Check Active Calls and Route Dependencies

Deprovisioning a trunk with active media streams causes call drops. You must query the Analytics API to verify zero active conversations routed through the target edge or trunk. The query filters by edgeId and conversationState.

Scope: analytics:conversations:read

async def check_active_calls(self, edge_id: str) -> int:
    """Queries active conversations bound to the trunk edge."""
    query_payload = {
        "interval": "now-5m/now",
        "groupBy": ["edgeId"],
        "filter": [
            {"dimension": "edgeId", "operator": "equals", "value": edge_id},
            {"dimension": "conversationState", "operator": "in", "value": ["active", "alerting", "ringing"]}
        ],
        "aggregations": [{"name": "count", "type": "count"}],
        "pageSize": 100
    }

    async with httpx.AsyncClient(timeout=15.0) as client:
        response = await client.post(
            f"{self.base_url}/api/v2/analytics/conversations/details/query",
            headers=self.auth.headers(),
            json=query_payload
        )
        
        if response.status_code == 429:
            await asyncio.sleep(int(response.headers.get("Retry-After", 5)))
            return await self.check_active_calls(edge_id)
            
        response.raise_for_status()
        data = response.json()
        
        # Sum counts across paginated results
        total_active = 0
        for entity in data.get("entities", []):
            total_active += entity.get("count", 0)
            
        if total_active > 0:
            logger.warning("Active calls detected", edge_id=edge_id, count=total_active)
        return total_active

Step 3: Validate Session Limits and Health Check Status

The trunk must pass a health evaluation before revocation. You retrieve the edge configuration to verify maxConcurrentSessions alignment and confirm the health check status is not in a degraded state. This prevents deprovisioning failures caused by orphaned SIP registrations.

Scope: telephony:provideredge:read

async def evaluate_edge_health(self, edge_id: str) -> Dict[str, Any]:
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(
            f"{self.base_url}/api/v2/telephony/providers/edges/{edge_id}",
            headers=self.auth.headers()
        )
        response.raise_for_status()
        edge_config = response.json()
        
        health_status = edge_config.get("healthCheck", {}).get("status", "unknown")
        if health_status in ["degraded", "down"]:
            raise RuntimeError(f"Edge {edge_id} health check failed. Status: {health_status}")
            
        return {
            "maxConcurrentSessions": edge_config.get("maxConcurrentSessions", 0),
            "healthStatus": health_status,
            "credentialRotated": edge_config.get("credentialRotation", {}).get("enabled", False)
        }

Step 4: Construct Revoke Payload and Execute Atomic DELETE

The deprovisioning operation requires a two-phase commit. First, you update the trunk status to disabled with a revoke directive. Second, you issue the DELETE request. The SDK does not expose a direct revoke method, so you construct the payload manually. The code implements exponential backoff for 429 responses and verifies the response format before proceeding.

Scope: telephony:trunk:delete, telephony:trunk:read

import asyncio

async def deprovision_trunk(self, trunk_id: str, edge_id: str) -> Dict[str, Any]:
    """Executes atomic disable and delete with retry logic."""
    revoke_payload = {
        "id": trunk_id,
        "status": "disabled",
        "revokeDirective": "force",
        "sipMatrix": {
            "type": "manual",
            "proxyAddresses": [],
            "disableKeepAlive": True
        }
    }

    async with httpx.AsyncClient(timeout=15.0) as client:
        # Phase 1: Disable trunk
        put_resp = await client.put(
            f"{self.base_url}/api/v2/telephony/users/trunks/{trunk_id}",
            headers=self.auth.headers(),
            json=revoke_payload
        )
        
        if put_resp.status_code == 429:
            await asyncio.sleep(2)
            return await self.deprovision_trunk(trunk_id, edge_id)
        put_resp.raise_for_status()
        
        # Phase 2: Atomic DELETE
        del_resp = await client.delete(
            f"{self.base_url}/api/v2/telephony/users/trunks/{trunk_id}",
            headers=self.auth.headers()
        )
        
        if del_resp.status_code == 409:
            raise RuntimeError("Trunk deletion blocked by active route dependency.")
        if del_resp.status_code == 429:
            await asyncio.sleep(3)
            return await self.deprovision_trunk(trunk_id, edge_id)
            
        del_resp.raise_for_status()
        
        return {
            "trunkId": trunk_id,
            "status": "deprovisioned",
            "deletedAt": asyncio.get_event_loop().time(),
            "responseCode": del_resp.status_code
        }

HTTP Request Cycle:

PUT /api/v2/telephony/users/trunks/trunk-abc-123 HTTP/1.1
Host: api.mypurecloud.com
Authorization: Bearer <token>
Content-Type: application/json

{
  "id": "trunk-abc-123",
  "status": "disabled",
  "revokeDirective": "force",
  "sipMatrix": {
    "type": "manual",
    "proxyAddresses": [],
    "disableKeepAlive": true
  }
}

Step 5: Synchronize SBC Webhooks and Generate Audit Logs

After successful deletion, you must notify external Session Border Controllers (SBC) and record the event for network governance. You create a platform webhook that fires on trunk state changes and generate a structured audit log entry with latency tracking.

Scope: webhook:write

async def trigger_sbc_webhook(self, trunk_id: str, sbc_url: str) -> str:
    webhook_payload = {
        "name": f"SBC_Sync_{trunk_id}",
        "enabled": True,
        "type": "rest",
        "url": sbc_url,
        "eventFilters": [
            {
                "eventDefinitionId": "com.mypurecloud.platform.events.telephony.trunk.deleted",
                "eventFilter": {
                    "dimensions": [{"dimension": "trunkId", "operator": "equals", "value": trunk_id}]
                }
            }
        ],
        "restConfig": {
            "httpMethod": "POST",
            "headers": {"X-Genesys-Event": "trunk-deprovisioned"},
            "payloadFormat": "application/json"
        }
    }

    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.post(
            f"{self.base_url}/api/v2/platform/webhooks",
            headers=self.auth.headers(),
            json=webhook_payload
        )
        response.raise_for_status()
        return response.json()["id"]

def generate_audit_log(self, result: Dict[str, Any], latency_ms: float) -> Dict[str, Any]:
    return {
        "auditType": "TRUNK_DEPROVISION",
        "trunkId": result["trunkId"],
        "status": result["status"],
        "latencyMs": latency_ms,
        "timestamp": isoformat(datetime.utcnow()),
        "networkIsolation": True,
        "callDropRisk": "mitigated"
    }

Complete Working Example

The following script combines all components into a production-ready trunk deprovisioner. It handles credential rotation checks, health validation, active call verification, atomic deletion, webhook synchronization, and audit logging.

import asyncio
import time
from datetime import datetime, timezone
from typing import Dict, Any

class GenesysTrunkDeprovisioner:
    def __init__(self, client_id: str, client_secret: str, environment: str = "mypurecloud.com"):
        self.auth = GenesysAuth(client_id, client_secret, environment)
        self.validator = TrunkValidator(self.auth)
        self.base_url = f"https://api.{environment}"

    async def run_deprovision(self, trunk_id: str, edge_id: str, sbc_url: str) -> Dict[str, Any]:
        start_time = time.perf_counter()
        
        # 1. Validate trunk reference and SIP matrix
        trunk_data = await self.validator.validate_trunk_reference(trunk_id)
        
        # 2. Check active calls
        active_count = await self.validator.check_active_calls(edge_id)
        if active_count > 0:
            raise RuntimeError(f"Cannot deprovision. {active_count} active calls detected on edge {edge_id}.")
            
        # 3. Evaluate edge health and session limits
        edge_health = await self.validator.evaluate_edge_health(edge_id)
        if edge_health["maxConcurrentSessions"] < 1:
            raise ValueError("Edge max concurrent sessions is zero. Skipping deprovision.")
            
        # 4. Execute atomic DELETE
        deprovision_result = await self.validator.deprovision_trunk(trunk_id, edge_id)
        
        # 5. Trigger SBC webhook
        webhook_id = await self.trigger_sbc_webhook(trunk_id, sbc_url)
        
        latency_ms = (time.perf_counter() - start_time) * 1000
        audit_entry = self.generate_audit_log(deprovision_result, latency_ms)
        
        return {
            "success": True,
            "deprovisionResult": deprovision_result,
            "webhookId": webhook_id,
            "auditLog": audit_entry,
            "latencyMs": latency_ms
        }

if __name__ == "__main__":
    async def main():
        deprovisioner = GenesysTrunkDeprovisioner(
            client_id="YOUR_CLIENT_ID",
            client_secret="YOUR_CLIENT_SECRET"
        )
        try:
            result = await deprovisioner.run_deprovision(
                trunk_id="trunk-abc-123",
                edge_id="edge-xyz-789",
                sbc_url="https://sbc.internal/api/v1/trunks/sync"
            )
            print(result)
        except Exception as e:
            print(f"Deprovision failed: {e}")
            
    asyncio.run(main())

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token expired or the client credentials are invalid.
• Fix: Verify the client_id and client_secret match a registered M2M application. Ensure the token cache refreshes before expiration. The GenesysAuth class handles automatic refresh, but network timeouts may interrupt the flow. Add a try-except block around get_token() to force a re-fetch.

Error: 403 Forbidden

• Cause: The OAuth token lacks the required scope or the application permission is not granted in the Genesys Cloud admin console.
• Fix: Confirm the M2M application has telephony:trunk:delete and analytics:conversations:read scopes enabled. Re-authorize the application and regenerate the token.

Error: 409 Conflict

• Cause: The trunk is referenced by an active routing strategy or IVR flow.
• Fix: Query /api/v2/telephony/routing/strategies to identify dependent resources. Remove the trunk from all routing strategies before calling the DELETE endpoint. The code raises a RuntimeError on 409 to prevent silent failures.

Error: 429 Too Many Requests

• Cause: The API rate limit is exceeded, typically during bulk deprovisioning or concurrent analytics queries.
• Fix: The implementation reads the Retry-After header and pauses execution. For high-volume operations, implement a token bucket algorithm or stagger requests by 500ms. The httpx retry transport can be swapped in for automatic handling.

Error: 503 Service Unavailable

• Cause: The target edge or SBC is undergoing maintenance.
• Fix: Verify edge health via /api/v2/telephony/providers/edges/{edgeId}. Wait for the health check status to return to up before retrying. Implement exponential backoff with a maximum of three retries.

Official References

• Telephony Users Trunks API
• Telephony Providers Edges API
• Analytics Conversations Details Query API
• Platform Webhooks API
• Genesys Cloud Python SDK Reference