How to Build a Real-Time Queue Dashboard with the Statistics API
What You Will Build
You will build a Python script that polls the Genesys Cloud Statistics API to retrieve real-time queue metrics, specifically the number of waiting interactions and available agents. This tutorial covers the complete lifecycle from OAuth authentication to parsing the JSON response and handling rate limits. The code is written in Python using the official genesys-cloud-purecloud-platform-client SDK.
Prerequisites
Before writing code, ensure you have the following credentials and environment setup.
- Genesys Cloud Organization: You need access to an organization with at least one configured Queue and at least one Agent assigned to that Queue.
- API User: Create a Service Account or API User in the Genesys Cloud Admin Portal.
- Type: Service Account (recommended for server-to-server integrations) or API User.
- Scopes: The user must have the
analytics:realtime:viewscope. Without this, the API will return a403 Forbiddenerror.
- Python Environment: Python 3.8 or higher.
- Dependencies:
genesys-cloud-purecloud-platform-client: The official Genesys Cloud Python SDK.python-dotenv: For managing environment variables securely.
Install the dependencies using pip:
pip install genesys-cloud-purecloud-platform-client python-dotenv
Create a .env file in your project root with the following variables:
GENESYS_CLIENT_ID=your_client_id
GENESYS_CLIENT_SECRET=your_client_secret
GENESYS_REGION=us-east-1
QUEUE_ID=your_queue_id
Replace your_queue_id with the actual ID of the queue you wish to monitor. You can find this ID in the Admin Portal under Routing > Queues by clicking on a queue and looking at the URL path (e.g., .../routing/queues/123e4567-e89b-12d3-a456-426614174000).
Authentication Setup
Genesys Cloud uses OAuth 2.0 for authentication. The SDK handles the token exchange and refresh logic automatically, but you must initialize the client correctly. The PureCloudPlatformClientV2 class provides a static method login_client_credential for service accounts.
import os
from dotenv import load_dotenv
from platform_sdk.exceptions import ApiException
from platform_sdk import PureCloudPlatformClientV2
# Load environment variables
load_dotenv()
def get_platform_client():
"""
Initializes and returns a configured Genesys Cloud Platform Client.
Uses Client Credentials flow for service accounts.
"""
client_id = os.getenv("GENESYS_CLIENT_ID")
client_secret = os.getenv("GENESYS_CLIENT_SECRET")
region = os.getenv("GENESYS_REGION", "us-east-1")
if not client_id or not client_secret:
raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set in .env")
try:
# The SDK handles token caching and automatic refresh
client = PureCloudPlatformClientV2()
client.login_client_credential(client_id, client_secret)
# Set the region if it is not the default
if region != "us-east-1":
client.set_base_url(f"https://{region}.mygen.com")
return client
except ApiException as e:
print(f"Authentication failed: {e.body}")
raise
This function returns a fully authenticated PureCloudPlatformClientV2 instance. The SDK stores the access token in memory and refreshes it automatically before expiration. You should instantiate this client once and reuse it for all API calls.
Implementation
Step 1: Retrieve Queue Real-Time Statistics
The core of this tutorial is the StatisticsApi.get_queue_queuestatistics method. This endpoint returns a snapshot of the queue’s current state.
The API path is GET /api/v2/analytics/queues/{queueId}/statistics.
Key parameters:
queue_id: The UUID of the queue.group_by: Determines how data is aggregated. For real-time queue stats, we typically usenoneto get the aggregate for the entire queue.interval: For real-time data, userealtime.metrics: A comma-separated list of specific metrics to retrieve. To reduce payload size and improve performance, specify only the metrics you need. Common metrics include:waiting_count: Number of interactions waiting in the queue.available_count: Number of agents currently available (status: Available).busy_count: Number of agents currently on a call or task.wrapup_count: Number of agents in wrap-up.
from platform_sdk.api.statistics_api import StatisticsApi
from platform_sdk.model.queue_realtime_statistics import QueueRealtimeStatistics
def get_queue_stats(client: PureCloudPlatformClientV2, queue_id: str):
"""
Fetches real-time statistics for a specific queue.
Args:
client: Authenticated PureCloudPlatformClientV2 instance.
queue_id: UUID of the target queue.
Returns:
QueueRealtimeStatistics object or None if an error occurs.
"""
statistics_api = StatisticsApi(client)
# Define the metrics we want to retrieve
# Using specific metrics reduces latency and payload size
metrics = "waiting_count,available_count,busy_count,wrapup_count"
try:
# Call the API
# group_by='none' aggregates all agents in the queue
# interval='realtime' ensures we get the current snapshot
response = statistics_api.get_queue_queuestatistics(
queue_id=queue_id,
group_by="none",
interval="realtime",
metrics=metrics
)
return response.body
except ApiException as e:
print(f"Error fetching queue stats: {e.status_code} - {e.body}")
return None
Step 2: Parse the Response
The response body is a QueueRealtimeStatistics object. The structure contains an entities array. Even with group_by=none, the SDK returns an array because the underlying API design supports grouping by agent, skill, or other dimensions.
The entities array will contain a single object representing the aggregate queue data. Each entity has a metrics object containing the requested values.
def parse_queue_metrics(stats_response: QueueRealtimeStatistics):
"""
Extracts waiting count and available agents from the API response.
Args:
stats_response: The QueueRealtimeStatistics object from the API call.
Returns:
A dictionary with 'waiting_count' and 'available_count'.
"""
if not stats_response or not stats_response.entities:
print("No statistics entities found in response.")
return {
"waiting_count": 0,
"available_count": 0,
"busy_count": 0,
"wrapup_count": 0
}
# The first entity contains the aggregated queue data
queue_entity = stats_response.entities[0]
metrics = queue_entity.metrics
# Access specific metric values
# The SDK maps JSON keys to snake_case attributes
waiting = metrics.waiting_count if metrics.waiting_count is not None else 0
available = metrics.available_count if metrics.available_count is not None else 0
busy = metrics.busy_count if metrics.busy_count is not None else 0
wrapup = metrics.wrapup_count if metrics.wrapup_count is not None else 0
return {
"waiting_count": waiting,
"available_count": available,
"busy_count": busy,
"wrapup_count": wrapup
}
Step 3: Implement Polling with Rate Limit Handling
Real-time data changes frequently. To build a dashboard, you need to poll the API. However, Genesys Cloud enforces rate limits. If you exceed the limit, you receive a 429 Too Many Requests response. The SDK does not automatically retry 429 errors, so you must implement exponential backoff or a simple delay.
For queue statistics, polling every 5-10 seconds is usually sufficient and safe. Polling every second will likely trigger rate limits.
import time
import sys
def poll_queue_stats(client: PureCloudPlatformClientV2, queue_id: str, interval_seconds: int = 5):
"""
Continuously polls queue statistics with rate limit handling.
Args:
client: Authenticated PureCloudPlatformClientV2 instance.
queue_id: UUID of the target queue.
interval_seconds: Time to wait between polls.
"""
print(f"Starting polling for Queue ID: {queue_id}")
print(f"Interval: {interval_seconds} seconds")
print("-" * 40)
while True:
try:
stats_response = get_queue_stats(client, queue_id)
if stats_response:
metrics = parse_queue_metrics(stats_response)
# Display the current state
print(f"\n[{time.strftime('%H:%M:%S')}] Queue Status:")
print(f" Waiting Interactions: {metrics['waiting_count']}")
print(f" Available Agents: {metrics['available_count']}")
print(f" Busy Agents: {metrics['busy_count']}")
print(f" Wrap-up Agents: {metrics['wrapup_count']}")
# Calculate SLA risk if needed
if metrics['waiting_count'] > 0 and metrics['available_count'] == 0:
print(" *** ALERT: No available agents for waiting queue! ***")
else:
print("Failed to retrieve stats. Retrying in next interval.")
except KeyboardInterrupt:
print("\nPolling stopped by user.")
sys.exit(0)
except Exception as e:
print(f"Unexpected error: {e}")
# Wait before next poll to respect rate limits
time.sleep(interval_seconds)
Complete Working Example
Combine the previous sections into a single runnable script. Save this as queue_monitor.py.
import os
import time
import sys
from dotenv import load_dotenv
from platform_sdk.exceptions import ApiException
from platform_sdk import PureCloudPlatformClientV2
from platform_sdk.api.statistics_api import StatisticsApi
from platform_sdk.model.queue_realtime_statistics import QueueRealtimeStatistics
# Load environment variables from .env file
load_dotenv()
def get_platform_client():
"""
Initializes and returns a configured Genesys Cloud Platform Client.
"""
client_id = os.getenv("GENESYS_CLIENT_ID")
client_secret = os.getenv("GENESYS_CLIENT_SECRET")
region = os.getenv("GENESYS_REGION", "us-east-1")
if not client_id or not client_secret:
raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set in .env")
try:
client = PureCloudPlatformClientV2()
client.login_client_credential(client_id, client_secret)
if region != "us-east-1":
client.set_base_url(f"https://{region}.mygen.com")
return client
except ApiException as e:
print(f"Authentication failed: {e.body}")
raise
def get_queue_stats(client: PureCloudPlatformClientV2, queue_id: str):
"""
Fetches real-time statistics for a specific queue.
"""
statistics_api = StatisticsApi(client)
metrics = "waiting_count,available_count,busy_count,wrapup_count"
try:
response = statistics_api.get_queue_queuestatistics(
queue_id=queue_id,
group_by="none",
interval="realtime",
metrics=metrics
)
return response.body
except ApiException as e:
print(f"Error fetching queue stats: {e.status_code} - {e.body}")
return None
def parse_queue_metrics(stats_response: QueueRealtimeStatistics):
"""
Extracts waiting count and available agents from the API response.
"""
if not stats_response or not stats_response.entities:
return {
"waiting_count": 0,
"available_count": 0,
"busy_count": 0,
"wrapup_count": 0
}
queue_entity = stats_response.entities[0]
metrics = queue_entity.metrics
waiting = metrics.waiting_count if metrics.waiting_count is not None else 0
available = metrics.available_count if metrics.available_count is not None else 0
busy = metrics.busy_count if metrics.busy_count is not None else 0
wrapup = metrics.wrapup_count if metrics.wrapup_count is not None else 0
return {
"waiting_count": waiting,
"available_count": available,
"busy_count": busy,
"wrapup_count": wrapup
}
def poll_queue_stats(client: PureCloudPlatformClientV2, queue_id: str, interval_seconds: int = 5):
"""
Continuously polls queue statistics.
"""
print(f"Starting polling for Queue ID: {queue_id}")
print(f"Interval: {interval_seconds} seconds")
print("-" * 40)
while True:
try:
stats_response = get_queue_stats(client, queue_id)
if stats_response:
metrics = parse_queue_metrics(stats_response)
print(f"\n[{time.strftime('%H:%M:%S')}] Queue Status:")
print(f" Waiting Interactions: {metrics['waiting_count']}")
print(f" Available Agents: {metrics['available_count']}")
print(f" Busy Agents: {metrics['busy_count']}")
print(f" Wrap-up Agents: {metrics['wrapup_count']}")
if metrics['waiting_count'] > 0 and metrics['available_count'] == 0:
print(" *** ALERT: No available agents for waiting queue! ***")
else:
print("Failed to retrieve stats. Retrying in next interval.")
except KeyboardInterrupt:
print("\nPolling stopped by user.")
sys.exit(0)
except Exception as e:
print(f"Unexpected error: {e}")
time.sleep(interval_seconds)
if __name__ == "__main__":
queue_id = os.getenv("QUEUE_ID")
if not queue_id:
print("Error: QUEUE_ID must be set in .env")
sys.exit(1)
try:
client = get_platform_client()
poll_queue_stats(client, queue_id, interval_seconds=5)
except Exception as e:
print(f"Fatal error: {e}")
sys.exit(1)
Run the script with:
python queue_monitor.py
You will see output similar to:
Starting polling for Queue ID: 123e4567-e89b-12d3-a456-426614174000
Interval: 5 seconds
----------------------------------------
[14:23:05] Queue Status:
Waiting Interactions: 0
Available Agents: 3
Busy Agents: 1
Wrap-up Agents: 0
[14:23:10] Queue Status:
Waiting Interactions: 2
Available Agents: 2
Busy Agents: 1
Wrap-up Agents: 0
Common Errors & Debugging
Error: 403 Forbidden
Cause: The API user does not have the analytics:realtime:view scope.
Fix: Go to the Genesys Cloud Admin Portal, navigate to Users > [Your API User] > Permissions, and ensure the analytics:realtime:view permission is checked. Save and wait for the permissions to propagate (usually immediate, but may take up to 5 minutes).
Error: 404 Not Found
Cause: The queue_id is invalid or the queue does not exist.
Fix: Verify the QUEUE_ID in your .env file. Ensure it is a valid UUID. Check that the queue is not archived.
Error: 429 Too Many Requests
Cause: You are polling the API too frequently. Genesys Cloud has strict rate limits for real-time endpoints.
Fix: Increase the interval_seconds in the poll_queue_stats function. A minimum of 5 seconds is recommended. If you need to monitor multiple queues, stagger the requests or use the get_queues_queuestatistics endpoint to fetch stats for multiple queues in a single call.
Error: None Values in Metrics
Cause: The queue has no agents assigned, or no interactions are occurring.
Fix: This is normal behavior. The code handles None values by defaulting to 0. Ensure agents are logged in and assigned to the queue to see non-zero values for available_count or busy_count.