Subscribing to Genesys Cloud Interaction Analytics Real-Time Metrics via WebSocket with Python SDK

What You Will Build

• A production-grade Python module that establishes a persistent WebSocket connection to Genesys Cloud real-time analytics streams, validates subscription payloads against stream engine constraints, and processes metric events with latency tracking, audit logging, and external dashboard synchronization.
• The implementation uses the official genesyscloud Python SDK for authentication and platform client initialization, combined with native websockets and pydantic for robust stream handling and schema validation.
• The tutorial covers Python 3.9+ with async/await patterns, OAuth 2.0 client credentials flow, and complete error recovery logic.

Prerequisites

• OAuth Client Type: Confidential Client (Client Credentials Grant)
• Required Scope: analytics:read (grants access to real-time analytics WebSocket endpoints)
• SDK Version: genesyscloud >= 2.15.0
• Runtime: Python 3.9 or newer
• External Dependencies: httpx==0.27.0, websockets==12.0, pydantic==2.6.0, pydantic-core==2.16.0
• Environment Variables: GENESYS_DOMAIN, GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET

Authentication Setup

Genesys Cloud requires a valid Bearer token for WebSocket handshake initialization. The client credentials flow exchanges client_id and client_secret for a short-lived token. Production implementations must cache tokens and handle 429 Too Many Requests responses during refresh.

import os
import time
import httpx
import logging
from typing import Optional

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

class OAuthTokenManager:
    def __init__(self, domain: str, client_id: str, client_secret: str):
        self.domain = domain.rstrip("/")
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = f"https://{self.domain}/oauth/token"
        self._token: Optional[str] = None
        self._expires_at: float = 0.0
        self._http_client = httpx.Client(timeout=10.0)

    def get_token(self) -> str:
        """Returns a valid Bearer token, refreshing if expired."""
        if self._token and time.time() < self._expires_at - 60:
            return self._token

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

        max_retries = 3
        for attempt in range(max_retries):
            try:
                response = self._http_client.post(self.token_url, data=payload)
                response.raise_for_status()
                data = response.json()
                self._token = data["access_token"]
                self._expires_at = time.time() + data["expires_in"]
                logger.info("OAuth token refreshed successfully.")
                return self._token
            except httpx.HTTPStatusError as e:
                if e.response.status_code == 429 and attempt < max_retries - 1:
                    retry_after = int(e.response.headers.get("Retry-After", 2 ** attempt))
                    logger.warning("OAuth 429 rate limit hit. Retrying in %s seconds.", retry_after)
                    time.sleep(retry_after)
                else:
                    raise RuntimeError(f"OAuth token acquisition failed: {e.response.status_code} {e.response.text}") from e
        raise RuntimeError("Max OAuth retries exceeded.")

Implementation

Step 1: Subscription Payload Construction and Schema Validation

The Genesys Cloud analytics stream engine enforces strict limits on payload size, metric type combinations, and time window boundaries. The subscription payload must reference valid metricTypes, define timeWindows with ISO 8601 durations, and specify granularity directives. Validation prevents 400 Bad Request rejections during the WebSocket handshake.

from pydantic import BaseModel, Field, field_validator
from typing import List, Dict, Any
import datetime

class TimeWindow(BaseModel):
    start: str
    end: str

    @field_validator("start", "end")
    @classmethod
    def validate_iso_duration(cls, v: str) -> str:
        # Genesys accepts relative durations like "now-PT1H" or absolute ISO 8601
        if not v.startswith("now-") and "T" not in v:
            raise ValueError("Time window must use 'now-PT...' or ISO 8601 format.")
        return v

class AnalyticsSubscriptionPayload(BaseModel):
    metricTypes: List[str] = Field(..., min_length=1, max_length=5)
    timeWindows: List[TimeWindow] = Field(..., max_length=3)
    granularity: str = Field(pattern=r"^PT(?:\d+M|\d+H|\d+S)$")
    metrics: List[str] = Field(..., min_length=1, max_length=20)
    entityFilters: Dict[str, Any] = Field(default_factory=dict)

    @field_validator("metricTypes")
    @classmethod
    def validate_metric_types(cls, v: List[str]) -> List[str]:
        allowed = {"queue", "agent", "interaction", "skill", "wrapup", "skillGroup"}
        invalid = set(v) - allowed
        if invalid:
            raise ValueError(f"Invalid metricTypes: {invalid}")
        return v

    @field_validator("metrics")
    @classmethod
    def validate_metrics(cls, v: List[str]) -> List[str]:
        # Subset of real Genesys analytics metrics
        valid_metrics = {
            "offered", "answered", "abandoned", "serviceLevelPercent",
            "serviceLevelDuration", "avgHandleTime", "avgWaitTime",
            "holdDuration", "transferCount", "wrapupCount"
        }
        invalid = set(v) - valid_metrics
        if invalid:
            raise ValueError(f"Invalid metrics requested: {invalid}")
        return v

    def to_dict(self) -> Dict[str, Any]:
        return self.model_dump(exclude_none=True)

Step 2: Atomic Handshake and Stream Engine Constraint Validation

The WebSocket handshake requires the subscription payload in the initial message. The stream engine validates concurrent connection limits per OAuth client and payload schema compliance. The implementation includes a handshake verification step that confirms the server returns a 200 acknowledgment before entering the streaming loop. Automatic reconnection triggers activate on connection drops or 429 stream-level throttling.

import asyncio
import json
import websockets
import uuid
from datetime import datetime, timezone
from typing import Callable, Optional, Dict, Any

class MetricStreamSubscriber:
    MAX_CONCURRENT_CONNECTIONS = 50
    PAYLOAD_SIZE_LIMIT_BYTES = 8192

    def __init__(self, domain: str, token_manager: OAuthTokenManager, 
                 subscription: AnalyticsSubscriptionPayload,
                 dashboard_callback: Callable[[Dict[str, Any]], None]):
        self.domain = domain.rstrip("/")
        self.token_manager = token_manager
        self.subscription = subscription
        self.dashboard_callback = dashboard_callback
        self.ws_url = f"wss://{self.domain}/api/v2/analytics/conversations/details/query"
        self._is_running = False
        self._connection_id = str(uuid.uuid4())
        self._latency_tracker: Dict[str, float] = {}
        self._audit_logger = logging.getLogger("analytics.audit")

    async def _validate_handshake(self, websocket: websockets.WebSocketClientProtocol, 
                                  payload: Dict[str, Any]) -> bool:
        """Sends subscription payload and verifies atomic handshake acknowledgment."""
        payload_bytes = json.dumps(payload).encode("utf-8")
        if len(payload_bytes) > self.PAYLOAD_SIZE_LIMIT_BYTES:
            raise ValueError(f"Payload exceeds stream engine limit: {len(payload_bytes)} bytes")

        await websocket.send(json.dumps(payload))
        
        try:
            response = await asyncio.wait_for(websocket.recv(), timeout=10.0)
            ack = json.loads(response)
            if ack.get("status") == 200 and ack.get("connectionId"):
                self._audit_logger.info("Handshake successful. Connection ID: %s", ack["connectionId"])
                return True
            raise RuntimeError(f"Handshake failed: {ack}")
        except asyncio.TimeoutError:
            raise RuntimeError("Handshake acknowledgment timed out.")
        except json.JSONDecodeError:
            raise RuntimeError("Invalid handshake response format.")

    async def _process_metric_event(self, raw_data: str) -> None:
        """Verifies format, tracks latency, and routes to dashboard callback."""
        try:
            event = json.loads(raw_data)
        except json.JSONDecodeError:
            self._audit_logger.error("Malformed metric event received.")
            return

        if "timestamp" not in event or "metrics" not in event:
            self._audit_logger.warning("Event missing required fields: %s", event.get("type", "unknown"))
            return

        # Latency tracking
        event_time = datetime.fromisoformat(event["timestamp"].replace("Z", "+00:00"))
        current_time = datetime.now(timezone.utc)
        latency_seconds = (current_time - event_time).total_seconds()
        
        metric_key = event.get("metricType", "unknown")
        self._latency_tracker[metric_key] = latency_seconds

        # Data freshness verification
        if latency_seconds > 30.0:
            self._audit_logger.warning("Stale data detected for %s: %.2f seconds latency.", metric_key, latency_seconds)

        # External dashboard synchronization
        try:
            self.dashboard_callback(event)
        except Exception as e:
            self._audit_logger.error("Dashboard callback failed: %s", e)

    async def start_streaming(self) -> None:
        """Enters the streaming loop with automatic reconnection logic."""
        self._is_running = True
        reconnect_delay = 1
        max_reconnect_delay = 60

        while self._is_running:
            try:
                token = self.token_manager.get_token()
                headers = {"Authorization": f"Bearer {token}"}
                
                async with websockets.connect(self.ws_url, extra_headers=headers) as ws:
                    await self._validate_handshake(ws, self.subscription.to_dict())
                    reconnect_delay = 1  # Reset on successful connection
                    self._audit_logger.info("Streaming active. Connection ID: %s", self._connection_id)

                    async for message in ws:
                        await self._process_metric_event(message)

            except websockets.ConnectionClosed as e:
                self._audit_logger.warning("WebSocket closed: %s. Reconnecting in %s seconds.", e, reconnect_delay)
                await asyncio.sleep(reconnect_delay)
                reconnect_delay = min(reconnect_delay * 2, max_reconnect_delay)
            except RuntimeError as e:
                if "429" in str(e):
                    self._audit_logger.warning("Stream engine rate limited. Backing off.")
                    await asyncio.sleep(reconnect_delay * 2)
                else:
                    self._audit_logger.error("Fatal error: %s", e)
                    break
            except Exception as e:
                self._audit_logger.error("Unexpected error: %s", e)
                await asyncio.sleep(reconnect_delay)

Step 3: Metric Availability Checking and Aggregation Verification Pipelines

Before committing to a long-lived connection, the subscriber must verify that requested metrics are available for the specified entity filters and time windows. This prevents silent data drops when queue IDs or skill groups lack active traffic. The verification pipeline queries the REST analytics endpoint to confirm metric availability before opening the WebSocket.

from genesyscloud.platform_client_v2 import PureCloudPlatformClientV2
from genesyscloud.rest import ApiException

class MetricAvailabilityVerifier:
    def __init__(self, domain: str, client_id: str, client_secret: str):
        self.client = PureCloudPlatformClientV2()
        self.client.set_access_token_mode("client_credentials")
        self.client.set_client_id(client_id)
        self.client.set_client_secret(client_secret)
        self.client.set_base_url(f"https://{domain}")

    def verify_metrics(self, subscription: AnalyticsSubscriptionPayload) -> bool:
        """Validates metric availability using the REST analytics query endpoint."""
        analytics_api = self.client.analytics_api
        
        # Construct a lightweight REST query mirroring the WebSocket subscription
        query_request = {
            "dateFrom": subscription.timeWindows[0].start if subscription.timeWindows else "now-PT1H",
            "dateTo": subscription.timeWindows[-1].end if subscription.timeWindows else "now",
            "metricTypes": subscription.metricTypes,
            "metrics": subscription.metrics[:3],  # Sample subset for verification
            "size": 1,
            "granularity": subscription.granularity
        }

        try:
            response = analytics_api.post_analytics_conversations_details_query(body=query_request)
            if response.entity and len(response.entity) > 0:
                self._log_verification_success(subscription)
                return True
            self._log_verification_warning(subscription)
            return False
        except ApiException as e:
            if e.status == 400:
                raise ValueError(f"Invalid subscription schema for REST verification: {e.body}")
            raise RuntimeError(f"Metric verification failed: {e.status} {e.body}") from e

    def _log_verification_success(self, sub: AnalyticsSubscriptionPayload):
        logger.info("Metric availability confirmed for types: %s, granularity: %s", sub.metricTypes, sub.granularity)

    def _log_verification_warning(self, sub: AnalyticsSubscriptionPayload):
        logger.warning("No data returned during verification. Metrics may be inactive or filters too restrictive.")

Step 4: Complete Integration with Audit Logging and Governance Compliance

The final component ties authentication, verification, streaming, and callback routing into a single executable module. Audit logs capture subscription lifecycle events, latency metrics, and data freshness rates for governance compliance. The module exposes a clean interface for automated analytics management.

import asyncio
import sys

def dashboard_sync_callback(event: Dict[str, Any]) -> None:
    """External dashboard synchronization handler."""
    metric_type = event.get("metricType", "unknown")
    metric_name = event.get("metrics", [{}])[0].get("name", "unknown") if event.get("metrics") else "unknown"
    value = event.get("metrics", [{}])[0].get("value", 0) if event.get("metrics") else 0
    timestamp = event.get("timestamp", "unknown")
    print(f"[DASHBOARD SYNC] {timestamp} | {metric_type}.{metric_name} = {value}")

async def run_analytics_subscriber(domain: str, client_id: str, client_secret: str):
    """Main execution pipeline for the metric subscriber."""
    token_manager = OAuthTokenManager(domain, client_id, client_secret)
    
    # Construct subscription payload with time window matrices and granularity directives
    subscription = AnalyticsSubscriptionPayload(
        metricTypes=["queue", "agent"],
        timeWindows=[TimeWindow(start="now-PT1H", end="now")],
        granularity="PT1M",
        metrics=["offered", "answered", "serviceLevelPercent", "avgWaitTime"],
        entityFilters={"queueIds": ["default"]}
    )

    # Verify metric availability before streaming
    verifier = MetricAvailabilityVerifier(domain, client_id, client_secret)
    if not verifier.verify_metrics(subscription):
        logger.error("Metric verification failed. Aborting subscription.")
        sys.exit(1)

    # Initialize subscriber with dashboard callback
    subscriber = MetricStreamSubscriber(
        domain=domain,
        token_manager=token_manager,
        subscription=subscription,
        dashboard_callback=dashboard_sync_callback
    )

    # Start streaming loop
    try:
        await subscriber.start_streaming()
    except KeyboardInterrupt:
        logger.info("Subscription terminated by user.")
        subscriber._is_running = False

if __name__ == "__main__":
    DOMAIN = os.getenv("GENESYS_DOMAIN")
    CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
    CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")

    if not all([DOMAIN, CLIENT_ID, CLIENT_SECRET]):
        sys.exit("Missing required environment variables: GENESYS_DOMAIN, GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET")

    asyncio.run(run_analytics_subscriber(DOMAIN, CLIENT_ID, CLIENT_SECRET))

Complete Working Example

The following script combines all components into a single runnable module. Save it as genesys_analytics_subscriber.py, set the environment variables, and execute with python genesys_analytics_subscriber.py.

import os
import time
import json
import asyncio
import logging
import uuid
import sys
import httpx
import websockets
from typing import Optional, List, Dict, Any, Callable
from datetime import datetime, timezone
from pydantic import BaseModel, Field, field_validator
from genesyscloud.platform_client_v2 import PureCloudPlatformClientV2
from genesyscloud.rest import ApiException

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)
audit_logger = logging.getLogger("analytics.audit")

class OAuthTokenManager:
    def __init__(self, domain: str, client_id: str, client_secret: str):
        self.domain = domain.rstrip("/")
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = f"https://{self.domain}/oauth/token"
        self._token: Optional[str] = None
        self._expires_at: float = 0.0
        self._http_client = httpx.Client(timeout=10.0)

    def get_token(self) -> str:
        if self._token and time.time() < self._expires_at - 60:
            return self._token
        payload = {"grant_type": "client_credentials", "client_id": self.client_id, "client_secret": self.client_secret}
        max_retries = 3
        for attempt in range(max_retries):
            try:
                response = self._http_client.post(self.token_url, data=payload)
                response.raise_for_status()
                data = response.json()
                self._token = data["access_token"]
                self._expires_at = time.time() + data["expires_in"]
                logger.info("OAuth token refreshed successfully.")
                return self._token
            except httpx.HTTPStatusError as e:
                if e.response.status_code == 429 and attempt < max_retries - 1:
                    retry_after = int(e.response.headers.get("Retry-After", 2 ** attempt))
                    logger.warning("OAuth 429 rate limit hit. Retrying in %s seconds.", retry_after)
                    time.sleep(retry_after)
                else:
                    raise RuntimeError(f"OAuth token acquisition failed: {e.response.status_code} {e.response.text}") from e
        raise RuntimeError("Max OAuth retries exceeded.")

class TimeWindow(BaseModel):
    start: str
    end: str
    @field_validator("start", "end")
    @classmethod
    def validate_iso_duration(cls, v: str) -> str:
        if not v.startswith("now-") and "T" not in v:
            raise ValueError("Time window must use 'now-PT...' or ISO 8601 format.")
        return v

class AnalyticsSubscriptionPayload(BaseModel):
    metricTypes: List[str] = Field(..., min_length=1, max_length=5)
    timeWindows: List[TimeWindow] = Field(..., max_length=3)
    granularity: str = Field(pattern=r"^PT(?:\d+M|\d+H|\d+S)$")
    metrics: List[str] = Field(..., min_length=1, max_length=20)
    entityFilters: Dict[str, Any] = Field(default_factory=dict)

    @field_validator("metricTypes")
    @classmethod
    def validate_metric_types(cls, v: List[str]) -> List[str]:
        allowed = {"queue", "agent", "interaction", "skill", "wrapup", "skillGroup"}
        invalid = set(v) - allowed
        if invalid:
            raise ValueError(f"Invalid metricTypes: {invalid}")
        return v

    @field_validator("metrics")
    @classmethod
    def validate_metrics(cls, v: List[str]) -> List[str]:
        valid_metrics = {"offered", "answered", "abandoned", "serviceLevelPercent", "serviceLevelDuration", "avgHandleTime", "avgWaitTime", "holdDuration", "transferCount", "wrapupCount"}
        invalid = set(v) - valid_metrics
        if invalid:
            raise ValueError(f"Invalid metrics requested: {invalid}")
        return v

    def to_dict(self) -> Dict[str, Any]:
        return self.model_dump(exclude_none=True)

class MetricAvailabilityVerifier:
    def __init__(self, domain: str, client_id: str, client_secret: str):
        self.client = PureCloudPlatformClientV2()
        self.client.set_access_token_mode("client_credentials")
        self.client.set_client_id(client_id)
        self.client.set_client_secret(client_secret)
        self.client.set_base_url(f"https://{domain}")

    def verify_metrics(self, subscription: AnalyticsSubscriptionPayload) -> bool:
        analytics_api = self.client.analytics_api
        query_request = {
            "dateFrom": subscription.timeWindows[0].start if subscription.timeWindows else "now-PT1H",
            "dateTo": subscription.timeWindows[-1].end if subscription.timeWindows else "now",
            "metricTypes": subscription.metricTypes,
            "metrics": subscription.metrics[:3],
            "size": 1,
            "granularity": subscription.granularity
        }
        try:
            response = analytics_api.post_analytics_conversations_details_query(body=query_request)
            if response.entity and len(response.entity) > 0:
                logger.info("Metric availability confirmed for types: %s, granularity: %s", subscription.metricTypes, subscription.granularity)
                return True
            logger.warning("No data returned during verification. Metrics may be inactive or filters too restrictive.")
            return False
        except ApiException as e:
            if e.status == 400:
                raise ValueError(f"Invalid subscription schema for REST verification: {e.body}")
            raise RuntimeError(f"Metric verification failed: {e.status} {e.body}") from e

class MetricStreamSubscriber:
    MAX_CONCURRENT_CONNECTIONS = 50
    PAYLOAD_SIZE_LIMIT_BYTES = 8192

    def __init__(self, domain: str, token_manager: OAuthTokenManager, 
                 subscription: AnalyticsSubscriptionPayload,
                 dashboard_callback: Callable[[Dict[str, Any]], None]):
        self.domain = domain.rstrip("/")
        self.token_manager = token_manager
        self.subscription = subscription
        self.dashboard_callback = dashboard_callback
        self.ws_url = f"wss://{self.domain}/api/v2/analytics/conversations/details/query"
        self._is_running = False
        self._connection_id = str(uuid.uuid4())
        self._latency_tracker: Dict[str, float] = {}

    async def _validate_handshake(self, websocket: websockets.WebSocketClientProtocol, 
                                  payload: Dict[str, Any]) -> bool:
        payload_bytes = json.dumps(payload).encode("utf-8")
        if len(payload_bytes) > self.PAYLOAD_SIZE_LIMIT_BYTES:
            raise ValueError(f"Payload exceeds stream engine limit: {len(payload_bytes)} bytes")
        await websocket.send(json.dumps(payload))
        try:
            response = await asyncio.wait_for(websocket.recv(), timeout=10.0)
            ack = json.loads(response)
            if ack.get("status") == 200 and ack.get("connectionId"):
                audit_logger.info("Handshake successful. Connection ID: %s", ack["connectionId"])
                return True
            raise RuntimeError(f"Handshake failed: {ack}")
        except asyncio.TimeoutError:
            raise RuntimeError("Handshake acknowledgment timed out.")
        except json.JSONDecodeError:
            raise RuntimeError("Invalid handshake response format.")

    async def _process_metric_event(self, raw_data: str) -> None:
        try:
            event = json.loads(raw_data)
        except json.JSONDecodeError:
            audit_logger.error("Malformed metric event received.")
            return
        if "timestamp" not in event or "metrics" not in event:
            audit_logger.warning("Event missing required fields: %s", event.get("type", "unknown"))
            return
        event_time = datetime.fromisoformat(event["timestamp"].replace("Z", "+00:00"))
        current_time = datetime.now(timezone.utc)
        latency_seconds = (current_time - event_time).total_seconds()
        metric_key = event.get("metricType", "unknown")
        self._latency_tracker[metric_key] = latency_seconds
        if latency_seconds > 30.0:
            audit_logger.warning("Stale data detected for %s: %.2f seconds latency.", metric_key, latency_seconds)
        try:
            self.dashboard_callback(event)
        except Exception as e:
            audit_logger.error("Dashboard callback failed: %s", e)

    async def start_streaming(self) -> None:
        self._is_running = True
        reconnect_delay = 1
        max_reconnect_delay = 60
        while self._is_running:
            try:
                token = self.token_manager.get_token()
                headers = {"Authorization": f"Bearer {token}"}
                async with websockets.connect(self.ws_url, extra_headers=headers) as ws:
                    await self._validate_handshake(ws, self.subscription.to_dict())
                    reconnect_delay = 1
                    audit_logger.info("Streaming active. Connection ID: %s", self._connection_id)
                    async for message in ws:
                        await self._process_metric_event(message)
            except websockets.ConnectionClosed as e:
                audit_logger.warning("WebSocket closed: %s. Reconnecting in %s seconds.", e, reconnect_delay)
                await asyncio.sleep(reconnect_delay)
                reconnect_delay = min(reconnect_delay * 2, max_reconnect_delay)
            except RuntimeError as e:
                if "429" in str(e):
                    audit_logger.warning("Stream engine rate limited. Backing off.")
                    await asyncio.sleep(reconnect_delay * 2)
                else:
                    audit_logger.error("Fatal error: %s", e)
                    break
            except Exception as e:
                audit_logger.error("Unexpected error: %s", e)
                await asyncio.sleep(reconnect_delay)

def dashboard_sync_callback(event: Dict[str, Any]) -> None:
    metric_type = event.get("metricType", "unknown")
    metric_name = event.get("metrics", [{}])[0].get("name", "unknown") if event.get("metrics") else "unknown"
    value = event.get("metrics", [{}])[0].get("value", 0) if event.get("metrics") else 0
    timestamp = event.get("timestamp", "unknown")
    print(f"[DASHBOARD SYNC] {timestamp} | {metric_type}.{metric_name} = {value}")

async def run_analytics_subscriber(domain: str, client_id: str, client_secret: str):
    token_manager = OAuthTokenManager(domain, client_id, client_secret)
    subscription = AnalyticsSubscriptionPayload(
        metricTypes=["queue", "agent"],
        timeWindows=[TimeWindow(start="now-PT1H", end="now")],
        granularity="PT1M",
        metrics=["offered", "answered", "serviceLevelPercent", "avgWaitTime"],
        entityFilters={"queueIds": ["default"]}
    )
    verifier = MetricAvailabilityVerifier(domain, client_id, client_secret)
    if not verifier.verify_metrics(subscription):
        logger.error("Metric verification failed. Aborting subscription.")
        sys.exit(1)
    subscriber = MetricStreamSubscriber(
        domain=domain,
        token_manager=token_manager,
        subscription=subscription,
        dashboard_callback=dashboard_sync_callback
    )
    try:
        await subscriber.start_streaming()
    except KeyboardInterrupt:
        logger.info("Subscription terminated by user.")
        subscriber._is_running = False

if __name__ == "__main__":
    DOMAIN = os.getenv("GENESYS_DOMAIN")
    CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
    CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")
    if not all([DOMAIN, CLIENT_ID, CLIENT_SECRET]):
        sys.exit("Missing required environment variables: GENESYS_DOMAIN, GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET")
    asyncio.run(run_analytics_subscriber(DOMAIN, CLIENT_ID, CLIENT_SECRET))

Common Errors & Debugging

Error: 401 Unauthorized during WebSocket Handshake

• Cause: The Bearer token expired before the WebSocket connection stabilized, or the OAuth client lacks the analytics:read scope.
• Fix: Verify the token manager refreshes tokens with a 60-second safety margin. Confirm the OAuth client in the Genesys Cloud admin console has analytics:read assigned.
• Code Fix: The OAuthTokenManager implements automatic refresh before expiration. If 401 persists, add explicit scope validation to the token response assertion.

Error: 400 Bad Request Payload Schema Validation

• Cause: The subscription payload contains unsupported metricTypes, invalid granularity formats, or exceeds the PAYLOAD_SIZE_LIMIT_BYTES constraint.
• Fix: Use the AnalyticsSubscriptionPayload Pydantic model to catch schema violations before transmission. Ensure granularity matches ISO 8601 duration format (e.g., PT1M, PT5M).
• Code Fix: The field_validator decorators reject invalid inputs at initialization time, preventing handshake failures.

Error: 429 Too Many Requests on Stream Engine

• Cause: The tenant has reached the maximum concurrent WebSocket connection limit for the OAuth client, or the metric query frequency exceeds stream engine quotas.
• Fix: Implement exponential backoff with jitter. Reduce timeWindows overlap or consolidate metricTypes into fewer connections.
• Code Fix: The start_streaming loop catches RuntimeError containing 429, applies backoff, and resets the reconnect delay on successful reconnection.

Error: Stale Data or High Latency Warnings

• Cause: Network partition, server-side aggregation delays, or incorrect granularity alignment with timeWindows.
• Fix: Adjust granularity to match dashboard update cycles. Verify entityFilters target active queues or skills. The latency tracker logs events exceeding 30 seconds for operational review.

Official References

• Genesys Cloud Analytics API Documentation
• Genesys Cloud Python SDK Reference
• OAuth 2.0 Client Credentials Flow
• WebSocket Event Streams Specification