Get Real-Time Queue Statistics via Genesys Cloud API

How-To-Articles
1

Get Real-Time Queue Statistics via Genesys Cloud API

What You Will Build

  • You will build a script that queries the Genesys Cloud Statistics API to retrieve real-time queue metrics, including the number of waiting conversations and available agents.
  • This tutorial uses the Genesys Cloud v2 API endpoint /api/v2/analytics/conversations/queues/details/query.
  • The implementation covers Python using the official genesyscloud SDK and JavaScript using the @genesyscloud/purecloud-platform-client-v2 library.

Prerequisites

  • OAuth Client Type: Confidential Client (Client Credentials Flow).
  • Required Scopes: analytics:conversation:queue:view or analytics:conversation:view.
  • SDK Version:
    • Python: genesyscloud >= 140.0.0
    • JavaScript: @genesyscloud/purecloud-platform-client-v2 >= 140.0.0
  • Runtime Requirements:
    • Python 3.9+
    • Node.js 18+
  • External Dependencies:
    • Python: pip install genesyscloud
    • JavaScript: npm install @genesyscloud/purecloud-platform-client-v2

Authentication Setup

Genesys Cloud APIs require an OAuth 2.0 bearer token. For server-side integrations, the Client Credentials flow is the standard approach. You must store your Organization ID, Client ID, and Client Secret securely.

Python Authentication

import os
from genesyscloud.platform.client import PlatformClient

def get_platform_client() -> PlatformClient:
    """
    Initializes and returns a configured Genesys Cloud Platform Client.
    """
    client = PlatformClient()
    
    # Set environment variables in your deployment
    org_id = os.getenv("GENESYS_ORG_ID")
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")

    if not all([org_id, client_id, client_secret]):
        raise EnvironmentError("Missing required environment variables: GENESYS_ORG_ID, GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET")

    # Configure the client with credentials
    client.set_default_configuration(
        org_id=org_id,
        client_id=client_id,
        client_secret=client_secret,
        use_oauth=True,
        disable_ssl_verification=False,
        debug=False
    )
    
    return client

JavaScript Authentication

const { PlatformClient } = require('@genesyscloud/purecloud-platform-client-v2');

/**
 * Initializes and returns a configured Genesys Cloud Platform Client.
 * @returns {Promise<PlatformClient>} The configured client instance.
 */
async function getPlatformClient() {
    const client = new PlatformClient();
    
    const orgId = process.env.GENESYS_ORG_ID;
    const clientId = process.env.GENESYS_CLIENT_ID;
    const clientSecret = process.env.GENESYS_CLIENT_SECRET;

    if (!orgId || !clientId || !clientSecret) {
        throw new Error('Missing required environment variables: GENESYS_ORG_ID, GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET');
    }

    // Configure the client with credentials
    client.setConfig({
        orgId: orgId,
        clientId: clientId,
        clientSecret: clientSecret,
        useOauth: true,
        disableSslVerification: false,
        debug: false
    });

    // Authenticate the client
    await client.authenticate();
    
    return client;
}

Implementation

Step 1: Identify Queue IDs

The Statistics API requires specific queue IDs to return data. You cannot query “all queues” in a single call without knowing their IDs. First, you must retrieve the list of queues.

Endpoint: GET /api/v2/queues
Scope: queue:view

Python

def get_queue_ids(client: PlatformClient) -> list[str]:
    """
    Retrieves all queue IDs from the Genesys Cloud organization.
    """
    queue_api = client.queue_api
    
    # Fetch the first page of queues
    response = queue_api.post_queuestoresearch(
        body={
            "query": {"filter": {"type": "all"}},
            "size": 250,
            "page": 1
        }
    )
    
    queue_ids = [q.id for q in response.entities]
    
    # Handle pagination if there are more than 250 queues
    while response.page < response.total:
        response = queue_api.post_queuestoresearch(
            body={
                "query": {"filter": {"type": "all"}},
                "size": 250,
                "page": response.page + 1
            }
        )
        queue_ids.extend([q.id for q in response.entities])
        
    return queue_ids

JavaScript

/**
 * Retrieves all queue IDs from the Genesys Cloud organization.
 * @param {PlatformClient} client - The authenticated Genesys Cloud client.
 * @returns {Promise<string[]>} Array of queue IDs.
 */
async function getQueueIds(client) {
    const queueApi = client.queueApi;
    let queueIds = [];
    let page = 1;
    let total = 0;
    const size = 250;

    do {
        const response = await queueApi.postQueuestoresearch({
            body: {
                query: { filter: { type: "all" } },
                size: size,
                page: page
            }
        });

        queueIds.push(...response.entities.map(q => q.id));
        total = response.total;
        page++;
    } while (page <= Math.ceil(total / size));

    return queueIds;
}

Step 2: Construct the Real-Time Query

The core of this tutorial is the /api/v2/analytics/conversations/queues/details/query endpoint. This endpoint accepts a complex JSON body defining what data to return.

To get real-time data, you must set intervalType to REALTIME. You also need to specify the metrics you want. For waiting count and agent availability, use queue-waiting-count and queue-available-agents.

Endpoint: POST /api/v2/analytics/conversations/queues/details/query
Scope: analytics:conversation:queue:view

Python

def build_realtime_query(queue_ids: list[str]) -> dict:
    """
    Constructs the request body for the real-time queue statistics query.
    """
    return {
        "intervalType": "REALTIME",
        "groupBy": ["queue"],
        "metrics": [
            "queue-waiting-count",
            "queue-available-agents",
            "queue-active-conversations",
            "queue-abandoned-count"
        ],
        "entityIds": queue_ids,
        "timePeriod": {
            "relativeTo": "now"
        }
    }

def fetch_realtime_stats(client: PlatformClient, query_body: dict) -> dict:
    """
    Executes the real-time statistics query.
    """
    analytics_api = client.analytics_api
    
    try:
        response = analytics_api.post_analytics_conversations_queues_details_query(
            body=query_body
        )
        return response
    except Exception as e:
        print(f"Error fetching analytics: {e}")
        raise

JavaScript

/**
 * Executes the real-time statistics query.
 * @param {PlatformClient} client - The authenticated Genesys Cloud client.
 * @param {string[]} queueIds - Array of queue IDs to query.
 * @returns {Promise<Object>} The analytics response object.
 */
async function fetchRealtimeStats(client, queueIds) {
    const analyticsApi = client.analyticsApi;
    
    const queryBody = {
        intervalType: "REALTIME",
        groupBy: ["queue"],
        metrics: [
            "queue-waiting-count",
            "queue-available-agents",
            "queue-active-conversations",
            "queue-abandoned-count"
        ],
        entityIds: queueIds,
        timePeriod: {
            relativeTo: "now"
        }
    };

    try {
        const response = await analyticsApi.postAnalyticsConversationsQueuesDetailsQuery({
            body: queryBody
        });
        return response;
    } catch (error) {
        console.error('Error fetching analytics:', error);
        throw error;
    }
}

Step 3: Process Results

The response from the Statistics API is a nested structure. The entities array contains one object per queue. Inside each entity, the metrics array contains the actual values. Each metric object has a metricName and a value.

You must map these metrics back to the queue ID for readable output.

Python

def parse_queue_stats(response: dict) -> list[dict]:
    """
    Parses the analytics response into a list of readable queue statistics.
    """
    queue_stats = []
    
    if not hasattr(response, 'entities') or not response.entities:
        return queue_stats

    for entity in response.entities:
        queue_id = entity.id
        queue_name = entity.name if hasattr(entity, 'name') else "Unknown"
        
        metrics = {m.metricName: m.value for m in entity.metrics}
        
        queue_data = {
            "queue_id": queue_id,
            "queue_name": queue_name,
            "waiting_count": metrics.get("queue-waiting-count", 0),
            "available_agents": metrics.get("queue-available-agents", 0),
            "active_conversations": metrics.get("queue-active-conversations", 0),
            "abandoned_count": metrics.get("queue-abandoned-count", 0)
        }
        queue_stats.append(queue_data)
        
    return queue_stats

JavaScript

/**
 * Parses the analytics response into an array of readable queue statistics.
 * @param {Object} response - The analytics response object.
 * @returns {Object[]} Array of queue statistics objects.
 */
function parseQueueStats(response) {
    const queueStats = [];

    if (!response.entities || response.entities.length === 0) {
        return queueStats;
    }

    for (const entity of response.entities) {
        const queueId = entity.id;
        const queueName = entity.name || "Unknown";

        // Map metrics to a dictionary for easier access
        const metricsMap = {};
        for (const m of entity.metrics) {
            metricsMap[m.metricName] = m.value;
        }

        queueStats.push({
            queueId: queueId,
            queueName: queueName,
            waitingCount: metricsMap["queue-waiting-count"] || 0,
            availableAgents: metricsMap["queue-available-agents"] || 0,
            activeConversations: metricsMap["queue-active-conversations"] || 0,
            abandonedCount: metricsMap["queue-abandoned-count"] || 0
        });
    }

    return queueStats;
}

Complete Working Example

Python

import os
import json
from genesyscloud.platform.client import PlatformClient

def get_platform_client() -> PlatformClient:
    client = PlatformClient()
    org_id = os.getenv("GENESYS_ORG_ID")
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")

    if not all([org_id, client_id, client_secret]):
        raise EnvironmentError("Missing required environment variables.")

    client.set_default_configuration(
        org_id=org_id,
        client_id=client_id,
        client_secret=client_secret,
        use_oauth=True,
        disable_ssl_verification=False,
        debug=False
    )
    return client

def get_queue_ids(client: PlatformClient) -> list[str]:
    queue_api = client.queue_api
    response = queue_api.post_queuestoresearch(
        body={"query": {"filter": {"type": "all"}}, "size": 250, "page": 1}
    )
    queue_ids = [q.id for q in response.entities]
    while response.page < response.total:
        response = queue_api.post_queuestoresearch(
            body={"query": {"filter": {"type": "all"}}, "size": 250, "page": response.page + 1}
        )
        queue_ids.extend([q.id for q in response.entities])
    return queue_ids

def build_realtime_query(queue_ids: list[str]) -> dict:
    return {
        "intervalType": "REALTIME",
        "groupBy": ["queue"],
        "metrics": ["queue-waiting-count", "queue-available-agents", "queue-active-conversations", "queue-abandoned-count"],
        "entityIds": queue_ids,
        "timePeriod": {"relativeTo": "now"}
    }

def fetch_realtime_stats(client: PlatformClient, query_body: dict) -> dict:
    analytics_api = client.analytics_api
    return analytics_api.post_analytics_conversations_queues_details_query(body=query_body)

def parse_queue_stats(response: dict) -> list[dict]:
    queue_stats = []
    if not hasattr(response, 'entities') or not response.entities:
        return queue_stats

    for entity in response.entities:
        metrics = {m.metricName: m.value for m in entity.metrics}
        queue_stats.append({
            "queue_id": entity.id,
            "queue_name": getattr(entity, 'name', "Unknown"),
            "waiting_count": metrics.get("queue-waiting-count", 0),
            "available_agents": metrics.get("queue-available-agents", 0),
            "active_conversations": metrics.get("queue-active-conversations", 0),
            "abandoned_count": metrics.get("queue-abandoned-count", 0)
        })
    return queue_stats

def main():
    try:
        client = get_platform_client()
        print("Authenticating and fetching queue IDs...")
        queue_ids = get_queue_ids(client)
        
        if not queue_ids:
            print("No queues found.")
            return

        print(f"Found {len(queue_ids)} queues. Querying real-time statistics...")
        query_body = build_realtime_query(queue_ids)
        response = fetch_realtime_stats(client, query_body)
        
        stats = parse_queue_stats(response)
        
        print("\nReal-Time Queue Statistics:")
        print("-" * 60)
        print(f"{'Queue Name':<20} | {'Waiting':<8} | {'Available':<10} | {'Active':<8} | {'Abandoned':<10}")
        print("-" * 60)
        
        for stat in stats:
            print(f"{stat['queue_name']:<20} | {stat['waiting_count']:<8} | {stat['available_agents']:<10} | {stat['active_conversations']:<8} | {stat['abandoned_count']:<10}")
            
    except Exception as e:
        print(f"Execution failed: {e}")

if __name__ == "__main__":
    main()

JavaScript

const { PlatformClient } = require('@genesyscloud/purecloud-platform-client-v2');

async function getPlatformClient() {
    const client = new PlatformClient();
    const orgId = process.env.GENESYS_ORG_ID;
    const clientId = process.env.GENESYS_CLIENT_ID;
    const clientSecret = process.env.GENESYS_CLIENT_SECRET;

    if (!orgId || !clientId || !clientSecret) {
        throw new Error('Missing required environment variables.');
    }

    client.setConfig({
        orgId,
        clientId,
        clientSecret,
        useOauth: true,
        disableSslVerification: false,
        debug: false
    });

    await client.authenticate();
    return client;
}

async function getQueueIds(client) {
    const queueApi = client.queueApi;
    let queueIds = [];
    let page = 1;
    let total = 0;
    const size = 250;

    do {
        const response = await queueApi.postQueuestoresearch({
            body: {
                query: { filter: { type: "all" } },
                size,
                page
            }
        });
        queueIds.push(...response.entities.map(q => q.id));
        total = response.total;
        page++;
    } while (page <= Math.ceil(total / size));

    return queueIds;
}

async function fetchRealtimeStats(client, queueIds) {
    const analyticsApi = client.analyticsApi;
    const queryBody = {
        intervalType: "REALTIME",
        groupBy: ["queue"],
        metrics: [
            "queue-waiting-count",
            "queue-available-agents",
            "queue-active-conversations",
            "queue-abandoned-count"
        ],
        entityIds: queueIds,
        timePeriod: {
            relativeTo: "now"
        }
    };

    const response = await analyticsApi.postAnalyticsConversationsQueuesDetailsQuery({
        body: queryBody
    });
    return response;
}

function parseQueueStats(response) {
    const queueStats = [];
    if (!response.entities || response.entities.length === 0) return queueStats;

    for (const entity of response.entities) {
        const metricsMap = {};
        for (const m of entity.metrics) {
            metricsMap[m.metricName] = m.value;
        }
        queueStats.push({
            queueId: entity.id,
            queueName: entity.name || "Unknown",
            waitingCount: metricsMap["queue-waiting-count"] || 0,
            availableAgents: metricsMap["queue-available-agents"] || 0,
            activeConversations: metricsMap["queue-active-conversations"] || 0,
            abandonedCount: metricsMap["queue-abandoned-count"] || 0
        });
    }
    return queueStats;
}

async function main() {
    try {
        const client = await getPlatformClient();
        console.log("Authenticating and fetching queue IDs...");
        const queueIds = await getQueueIds(client);

        if (queueIds.length === 0) {
            console.log("No queues found.");
            return;
        }

        console.log(`Found ${queueIds.length} queues. Querying real-time statistics...`);
        const response = await fetchRealtimeStats(client, queueIds);
        const stats = parseQueueStats(response);

        console.log("\nReal-Time Queue Statistics:");
        console.log("-".repeat(60));
        console.log(`${"Queue Name".padEnd(20)} | ${"Waiting".padEnd(8)} | ${"Available".padEnd(10)} | ${"Active".padEnd(8)} | ${"Abandoned".padEnd(10)}`);
        console.log("-".repeat(60));

        for (const stat of stats) {
            console.log(
                `${stat.queueName.padEnd(20)} | ${String(stat.waitingCount).padEnd(8)} | ${String(stat.availableAgents).padEnd(10)} | ${String(stat.activeConversations).padEnd(8)} | ${String(stat.abandonedCount).padEnd(10)}`
            );
        }
    } catch (error) {
        console.error("Execution failed:", error);
    }
}

main();

Common Errors & Debugging

Error: 401 Unauthorized

  • Cause: The OAuth token is missing, expired, or invalid. The Client ID or Secret is incorrect.
  • Fix: Verify that GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET are correct. Ensure the PlatformClient has been authenticated (in JS, await client.authenticate() must be called before API calls). In Python, the SDK handles token refresh automatically, but initial configuration must be correct.

Error: 403 Forbidden

  • Cause: The OAuth client does not have the required scope.
  • Fix: Ensure the OAuth client in the Genesys Cloud Admin console has the analytics:conversation:queue:view scope assigned. If using a custom OAuth client, check the “Scopes” tab.

Error: 422 Unprocessable Entity

  • Cause: The request body is malformed or contains invalid metric names.
  • Fix: Verify that intervalType is set to REALTIME (case-sensitive). Check that all metric strings (e.g., queue-waiting-count) are spelled exactly as documented. Ensure entityIds is a list of valid UUIDs.

Error: 429 Too Many Requests

  • Cause: You have exceeded the API rate limit for the Analytics endpoint.
  • Fix: Implement exponential backoff retry logic. The Genesys Cloud API returns a Retry-After header in the response.

Python Retry Example

import time
import requests

def fetch_with_retry(client, query_body, max_retries=3):
    analytics_api = client.analytics_api
    for attempt in range(max_retries):
        try:
            response = analytics_api.post_analytics_conversations_queues_details_query(body=query_body)
            return response
        except Exception as e:
            if "429" in str(e) or hasattr(e, 'status') and e.status == 429:
                retry_after = 2 ** attempt
                print(f"Rate limited. Retrying in {retry_after} seconds...")
                time.sleep(retry_after)
            else:
                raise
    raise Exception("Max retries exceeded")

Error: Empty Entities List

  • Cause: The queue IDs provided do not exist, or the queues have no data for the requested metrics.
  • Fix: Verify that the queue IDs returned by get_queue_ids are valid. Check that the queues are active and have been used recently. Real-time data may be empty if no conversations have occurred in the last few seconds.

Official References