How to Get Real-Time Queue Observation Data via the Statistics API

StarAdmin · May 28, 2026, 9:16am

How to Get Real-Time Queue Observation Data via the Statistics API

What You Will Build

You will build a script that queries the Genesys Cloud CX Statistics API to retrieve real-time metrics for a specific queue, including the number of waiting interactions and available agents.
This implementation uses the Genesys Cloud CX REST API endpoint /api/v2/analytics/queues/details/query.
The tutorial provides working examples in Python and JavaScript.

Prerequisites

OAuth Client Type: Machine-to-Machine (M2M) or Authorization Code Grant. M2M is recommended for backend services.
Required OAuth Scopes:
- analytics:queue:view (Required to query queue statistics)
- analytics:conversation:view (Optional, if you need conversation-level details within the query)
SDK Version:
- Python: genesyscloud SDK v2.10+
- JavaScript: @genesyscloud/api-analytics v2.0+
Language/Runtime Requirements:
- Python 3.9+
- Node.js 18+
External Dependencies:
- Python: pip install genesyscloud requests
- JavaScript: npm install @genesyscloud/api-analytics @genesyscloud/environment

Authentication Setup

Before querying data, you must obtain a valid OAuth access token. The Statistics API does not support anonymous access. The following examples demonstrate how to retrieve a token using a Machine-to-Machine (M2M) flow, which is the standard pattern for automated monitoring scripts.

Python Authentication

import os
import requests
from typing import Optional

class GenesysAuth:
    def __init__(self, client_id: str, client_secret: str, base_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = base_url.rstrip('/')
        self.token_url = f"{self.base_url}/oauth/token"

    def get_token(self) -> Optional[str]:
        """
        Retrieves an OAuth2 access token using Client Credentials flow.
        """
        payload = {
            'grant_type': 'client_credentials',
            'client_id': self.client_id,
            'client_secret': self.client_secret
        }

        try:
            response = requests.post(
                self.token_url,
                data=payload,
                timeout=10
            )
            response.raise_for_status()
            token_data = response.json()
            return token_data.get('access_token')
        except requests.exceptions.HTTPError as e:
            print(f"Authentication failed: {e.response.status_code} - {e.response.text}")
            return None
        except Exception as e:
            print(f"Unexpected error during auth: {e}")
            return None

# Usage
auth = GenesysAuth(
    client_id=os.getenv('GENESYS_CLIENT_ID'),
    client_secret=os.getenv('GENESYS_CLIENT_SECRET'),
    base_url=os.getenv('GENESYS_BASE_URL', 'https://api.mypurecloud.com')
)
access_token = auth.get_token()
if not access_token:
    exit(1)

JavaScript Authentication

const axios = require('axios');

class GenesysAuth {
    constructor(clientId, clientSecret, baseUrl) {
        this.clientId = clientId;
        this.clientSecret = clientSecret;
        this.baseUrl = baseUrl.replace(/\/$/, '');
        this.tokenUrl = `${this.baseUrl}/oauth/token`;
    }

    async getToken() {
        const payload = new URLSearchParams({
            grant_type: 'client_credentials',
            client_id: this.clientId,
            client_secret: this.clientSecret
        });

        try {
            const response = await axios.post(this.tokenUrl, payload, {
                headers: {
                    'Content-Type': 'application/x-www-form-urlencoded'
                }
            });
            return response.data.access_token;
        } catch (error) {
            if (error.response) {
                console.error(`Authentication failed: ${error.response.status} - ${error.response.data}`);
            } else {
                console.error(`Unexpected error during auth: ${error.message}`);
            }
            return null;
        }
    }
}

// Usage
const auth = new GenesysAuth(
    process.env.GENESYS_CLIENT_ID,
    process.env.GENESYS_CLIENT_SECRET,
    process.env.GENESYS_BASE_URL || 'https://api.mypurecloud.com'
);

(async () => {
    const accessToken = await auth.getToken();
    if (!accessToken) process.exit(1);
})();

Implementation

Step 1: Constructing the Query Payload

The Genesys Cloud Statistics API uses a query-based model rather than a simple GET parameter list. You must send a JSON body that defines the entities (queues), the time range, and the specific metrics (data points) you require.

To get real-time data, you set the since and until parameters to the current time. However, the API aggregates data in 15-second intervals. To get the “current” state, you query the most recent completed interval.

Key Parameters:

entities: An array of objects specifying the queue IDs you want to monitor.
timeGroup: Use fifteenSeconds for real-time granularity.
since / until: ISO 8601 timestamps. For real-time, set until to now and since to 15 seconds ago.
dataPoints: The specific metrics to return.
- waitingCount: Number of interactions waiting in the queue.
- availableAgents: Number of agents logged in and available.
- busyAgents: Number of agents currently on a call.

Python Implementation

import os
import time
from datetime import datetime, timedelta, timezone
import json

# Assuming 'access_token' and 'auth' from previous step are available

def get_realtime_queue_stats(queue_id: str, access_token: str, base_url: str):
    """
    Queries the Genesys Cloud Statistics API for real-time queue metrics.
    """
    endpoint = f"{base_url}/api/v2/analytics/queues/details/query"
    
    # Define the time window: Last 15 seconds to Now
    now = datetime.now(timezone.utc)
    since = now - timedelta(seconds=15)
    
    # Format timestamps to ISO 8601
    since_iso = since.isoformat()
    until_iso = now.isoformat()

    # Construct the query body
    body = {
        "entities": [
            {
                "id": queue_id
            }
        ],
        "timeGroup": "fifteenSeconds",
        "since": since_iso,
        "until": until_iso,
        "dataPoints": [
            "waitingCount",
            "availableAgents",
            "busyAgents",
            "abandonedCount"
        ]
    }

    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json"
    }

    try:
        response = requests.post(endpoint, json=body, headers=headers, timeout=15)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        print(f"API Error: {e.response.status_code} - {e.response.text}")
        return None
    except Exception as e:
        print(f"Request failed: {e}")
        return None

# Example Usage
QUEUE_ID = os.getenv('GENESYS_QUEUE_ID') # Replace with your actual Queue ID
stats = get_realtime_queue_stats(QUEUE_ID, access_token, auth.base_url)

if stats:
    print(json.dumps(stats, indent=2))

JavaScript Implementation

const axios = require('axios');

async function getRealtimeQueueStats(queueId, accessToken, baseUrl) {
    const endpoint = `${baseUrl}/api/v2/analytics/queues/details/query`;

    // Define the time window: Last 15 seconds to Now
    const now = new Date();
    const since = new Date(now.getTime() - 15000); // 15 seconds ago

    const body = {
        entities: [
            {
                id: queueId
            }
        ],
        timeGroup: "fifteenSeconds",
        since: since.toISOString(),
        until: now.toISOString(),
        dataPoints: [
            "waitingCount",
            "availableAgents",
            "busyAgents",
            "abandonedCount"
        ]
    };

    const headers = {
        Authorization: `Bearer ${accessToken}`,
        "Content-Type": "application/json"
    };

    try {
        const response = await axios.post(endpoint, body, { headers });
        return response.data;
    } catch (error) {
        if (error.response) {
            console.error(`API Error: ${error.response.status} - ${JSON.stringify(error.response.data)}`);
        } else {
            console.error(`Request failed: ${error.message}`);
        }
        return null;
    }
}

// Example Usage
const QUEUE_ID = process.env.GENESYS_QUEUE_ID; // Replace with your actual Queue ID

(async () => {
    const stats = await getRealtimeQueueStats(QUEUE_ID, accessToken, auth.baseUrl);
    if (stats) {
        console.log(JSON.stringify(stats, null, 2));
    }
})();

Step 2: Processing the Response

The response from /api/v2/analytics/queues/details/query is a nested structure. It contains a byEntity object where the keys are the queue IDs you queried. Inside each queue ID, you will find a byTime object. Because we queried a short time window (15 seconds), there may be one or more 15-second buckets depending on network latency and server processing time.

Expected Response Structure:

{
  "byEntity": {
    "a1b2c3d4-e5f6-7890-abcd-ef1234567890": {
      "entityId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
      "entityName": "Sales Support",
      "byTime": {
        "2023-10-27T10:00:00Z": {
          "waitingCount": 5,
          "availableAgents": 12,
          "busyAgents": 8,
          "abandonedCount": 0
        },
        "2023-10-27T10:00:15Z": {
          "waitingCount": 4,
          "availableAgents": 13,
          "busyAgents": 7,
          "abandonedCount": 1
        }
      }
    }
  }
}

To get the “current” state, you must extract the metric values from the most recent timestamp in the byTime object.

Python Data Extraction

def extract_latest_stats(api_response: dict, queue_id: str) -> dict:
    """
    Extracts the most recent metric values from the API response.
    """
    if not api_response or 'byEntity' not in api_response:
        return {}

    queue_data = api_response['byEntity'].get(queue_id)
    if not queue_data or 'byTime' not in queue_data:
        print(f"No data found for queue {queue_id}")
        return {}

    by_time = queue_data['byTime']
    if not by_time:
        print("No time buckets returned.")
        return {}

    # Get all timestamps and find the latest one
    timestamps = list(by_time.keys())
    if not timestamps:
        return {}

    latest_timestamp = max(timestamps)
    latest_data = by_time[latest_timestamp]

    return {
        "queueId": queue_id,
        "timestamp": latest_timestamp,
        "waitingCount": latest_data.get("waitingCount", 0),
        "availableAgents": latest_data.get("availableAgents", 0),
        "busyAgents": latest_data.get("busyAgents", 0),
        "abandonedCount": latest_data.get("abandonedCount", 0)
    }

# Usage
latest_stats = extract_latest_stats(stats, QUEUE_ID)
print(f"Latest Stats: {latest_stats}")

JavaScript Data Extraction

function extractLatestStats(apiResponse, queueId) {
    if (!apiResponse || !apiResponse.byEntity) {
        return {};
    }

    const queueData = apiResponse.byEntity[queueId];
    if (!queueData || !queueData.byTime) {
        console.log(`No data found for queue ${queueId}`);
        return {};
    }

    const byTime = queueData.byTime;
    const timestamps = Object.keys(byTime);
    
    if (timestamps.length === 0) {
        return {};
    }

    // Find the latest timestamp
    const latestTimestamp = timestamps.reduce((a, b) => a > b ? a : b);
    const latestData = byTime[latestTimestamp];

    return {
        queueId,
        timestamp: latestTimestamp,
        waitingCount: latestData.waitingCount || 0,
        availableAgents: latestData.availableAgents || 0,
        busyAgents: latestData.busyAgents || 0,
        abandonedCount: latestData.abandonedCount || 0
    };
}

// Usage
const latestStats = extractLatestStats(stats, QUEUE_ID);
console.log(`Latest Stats: ${JSON.stringify(latestStats, null, 2)}`);

Step 3: Handling Pagination and Rate Limits

The /api/v2/analytics/queues/details/query endpoint generally returns all requested time buckets in a single response for short time windows (like 15 seconds). However, if you query a longer duration (e.g., 1 hour with 15-second intervals), the response might be large. The API supports pagination via the pageToken field in the response if the result set is truncated.

For real-time monitoring, pagination is rarely needed because the time window is small. However, you must handle 429 Too Many Requests errors. Genesys Cloud enforces strict rate limits on analytics endpoints.

Retry Logic for 429 Errors:

import time

def get_stats_with_retry(queue_id: str, access_token: str, base_url: str, max_retries: int = 3):
    for attempt in range(max_retries):
        stats = get_realtime_queue_stats(queue_id, access_token, base_url)
        if stats:
            return stats
        
        # Check if the error was a 429
        # Note: In the previous function, we returned None on HTTPError. 
        # For robust retry logic, we should inspect the status code specifically.
        
        # Simplified retry delay
        wait_time = 2 ** attempt
        print(f"Rate limited or error. Retrying in {wait_time} seconds...")
        time.sleep(wait_time)
        
    return None

Complete Working Example

Below is a complete, runnable Python script that authenticates, queries the queue statistics, and prints the real-time data.

#!/usr/bin/env python3
"""
Genesys Cloud Real-Time Queue Monitor
Monitors waiting count and agent availability for a specific queue.
"""

import os
import sys
import json
import time
import requests
from datetime import datetime, timedelta, timezone

class GenesysQueueMonitor:
    def __init__(self, client_id: str, client_secret: str, base_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = base_url.rstrip('/')
        self.token_url = f"{self.base_url}/oauth/token"
        self.stats_url = f"{self.base_url}/api/v2/analytics/queues/details/query"

    def authenticate(self) -> str:
        """Retrieves an OAuth2 access token."""
        payload = {
            'grant_type': 'client_credentials',
            'client_id': self.client_id,
            'client_secret': self.client_secret
        }
        
        try:
            response = requests.post(self.token_url, data=payload, timeout=10)
            response.raise_for_status()
            return response.json().get('access_token')
        except Exception as e:
            print(f"Authentication failed: {e}")
            sys.exit(1)

    def get_queue_stats(self, queue_id: str, access_token: str) -> dict:
        """Queries real-time statistics for a specific queue."""
        now = datetime.now(timezone.utc)
        since = now - timedelta(seconds=15)

        body = {
            "entities": [{"id": queue_id}],
            "timeGroup": "fifteenSeconds",
            "since": since.isoformat(),
            "until": now.isoformat(),
            "dataPoints": [
                "waitingCount",
                "availableAgents",
                "busyAgents",
                "abandonedCount"
            ]
        }

        headers = {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json"
        }

        try:
            response = requests.post(self.stats_url, json=body, headers=headers, timeout=15)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429:
                print("Rate limited. Please wait before retrying.")
            else:
                print(f"API Error: {e.response.status_code} - {e.response.text}")
            return None
        except Exception as e:
            print(f"Request failed: {e}")
            return None

    def process_stats(self, api_response: dict, queue_id: str) -> dict:
        """Extracts the latest metrics from the API response."""
        if not api_response or 'byEntity' not in api_response:
            return {}

        queue_data = api_response.get('byEntity', {}).get(queue_id)
        if not queue_data or 'byTime' not in queue_data:
            return {}

        by_time = queue_data['byTime']
        if not by_time:
            return {}

        # Get the most recent timestamp
        latest_ts = max(by_time.keys())
        latest_data = by_time[latest_ts]

        return {
            "queueId": queue_id,
            "timestamp": latest_ts,
            "waitingCount": latest_data.get("waitingCount", 0),
            "availableAgents": latest_data.get("availableAgents", 0),
            "busyAgents": latest_data.get("busyAgents", 0),
            "abandonedCount": latest_data.get("abandonedCount", 0)
        }

def main():
    # Load credentials from environment variables
    client_id = os.getenv('GENESYS_CLIENT_ID')
    client_secret = os.getenv('GENESYS_CLIENT_SECRET')
    base_url = os.getenv('GENESYS_BASE_URL', 'https://api.mypurecloud.com')
    queue_id = os.getenv('GENESYS_QUEUE_ID')

    if not all([client_id, client_secret, queue_id]):
        print("Error: Missing environment variables. Set GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET, and GENESYS_QUEUE_ID.")
        sys.exit(1)

    monitor = GenesysQueueMonitor(client_id, client_secret, base_url)
    access_token = monitor.authenticate()

    if not access_token:
        print("Failed to obtain access token.")
        sys.exit(1)

    print(f"Monitoring Queue: {queue_id}")
    print("-" * 40)

    # Single query execution
    raw_stats = monitor.get_queue_stats(queue_id, access_token)
    
    if raw_stats:
        latest_metrics = monitor.process_stats(raw_stats, queue_id)
        if latest_metrics:
            print(json.dumps(latest_metrics, indent=2))
        else:
            print("No recent data available for this queue.")
    else:
        print("Failed to retrieve statistics.")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

Cause: The OAuth token is invalid, expired, or missing the required scope.
Fix: Ensure your M2M client has the analytics:queue:view scope assigned in the Genesys Cloud Admin Console under Security > Applications. Verify the token is passed in the Authorization: Bearer <token> header.

Error: 403 Forbidden

Cause: The user associated with the OAuth token does not have permission to view statistics for the specified queue.
Fix: In Genesys Cloud, navigate to Users > [Your User] > Permissions. Ensure the user has the Analytics > View Statistics permission. Also, verify the user is part of the team or routing group that owns the queue.

Error: 429 Too Many Requests

Cause: You have exceeded the rate limit for the Statistics API. Genesys Cloud enforces limits per client ID and per user.
Fix: Implement exponential backoff in your retry logic. Do not poll faster than every 15 seconds for real-time data, as the data granularity is 15 seconds. Querying more frequently provides no new data and wastes rate limit budget.

Error: Empty `byTime` Object

Cause: The queue has no activity in the specified time window, or the time window is in the future.
Fix: Ensure the since and until timestamps are in the past. If the queue is new or inactive, there may be no data points yet. Wait for at least one 15-second interval to pass after the query time.

How to Get Real-Time Queue Observation Data via the Statistics API

How to Get Real-Time Queue Observation Data via the Statistics API

What You Will Build

Prerequisites

Authentication Setup

Python Authentication

JavaScript Authentication

Implementation

Step 1: Constructing the Query Payload

Python Implementation

JavaScript Implementation

Step 2: Processing the Response

Python Data Extraction

JavaScript Data Extraction

Step 3: Handling Pagination and Rate Limits

Complete Working Example

Common Errors & Debugging

Error: 401 Unauthorized

Error: 403 Forbidden

Error: 429 Too Many Requests

Error: Empty byTime Object

Official References

Error: Empty `byTime` Object