Choosing Between Real-Time Conversations and Historical Analytics in Genesys Cloud

What You Will Build

• This tutorial demonstrates how to retrieve active conversation details using the Real-Time API versus historical conversation metrics using the Analytics API.
• It utilizes the Genesys Cloud Platform REST API v2 endpoints /api/v2/conversations and /api/v2/analytics/conversations/details/query.
• The code examples are provided in Python using the genesys-cloud-purecloud-platform-client SDK.

Prerequisites

• OAuth Client Type: Confidential Client (Client Credentials Flow).
• Required Scopes:
  • For Real-Time: conversation:read, interaction:read.
  • For Analytics: analytics:conversations:read, analytics:report:read.
• SDK Version: genesys-cloud-purecloud-platform-client >= 170.0.0.
• Runtime: Python 3.9+.
• Dependencies: pip install genesys-cloud-purecloud-platform-client httpx.

Authentication Setup

Genesys Cloud requires OAuth 2.0 authentication for all API calls. The Real-Time API and Analytics API share the same authentication mechanism but require different scopes. Using the wrong scope results in a 403 Forbidden error.

The following code initializes the PureCloud client using the standard PureCloudPlatformClientV2 initialization pattern. This pattern handles token acquisition and refresh automatically.

import os
from purecloud_platform_client import PureCloudPlatformClientV2
from purecloud_platform_client.rest import ApiException

def get_purecloud_client():
    """
    Initializes and returns an authenticated PureCloud Platform Client.
    """
    # Load credentials from environment variables
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    environment = os.getenv("GENESYS_ENVIRONMENT", "mypurecloud.com")

    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")

    # Initialize the client
    # The SDK handles the OAuth token exchange internally
    client = PureCloudPlatformClientV2(
        client_id=client_id,
        client_secret=client_secret,
        environment=environment
    )

    return client

Implementation

Step 1: Retrieving Real-Time Conversations (/api/v2/conversations)

Use /api/v2/conversations when you need to know what is happening right now. This endpoint returns active interactions. It is stateful and volatile; data returned here disappears once the conversation ends.

Use Cases:

• Displaying a real-time dashboard of active agents.
• Triggering an immediate webhook when a conversation enters a specific state (e.g., “queued”).
• Fetching the current transcript of an ongoing call.

Required Scope: conversation:read

The following code fetches all currently active conversations. Note that this endpoint does not support pagination in the traditional sense because the dataset changes every millisecond. It returns a snapshot of the current state.

from purecloud_platform_client import ConversationApi, ConversationEventType
from purecloud_platform_client.rest import ApiException
import json

def get_active_conversations(client: PureCloudPlatformClientV2):
    """
    Fetches all currently active conversations.
    """
    api_instance = ConversationApi(client)

    try:
        # Parameters:
        # expand: Optional. Can include 'transcript', 'participants', 'metrics'.
        #         Be cautious: 'transcript' can be heavy on bandwidth.
        # types: Optional. Filter by conversation type (e.g., 'call', 'webchat').
        
        response = api_instance.get_conversations(
            expand="participants,metrics", 
            types="call,webchat"
        )

        # The response is a ConversationEntityListing
        print(f"Found {response.total} active conversations.")

        for conversation in response.entities:
            print(f"Conversation ID: {conversation.conversation_id}")
            print(f"Type: {conversation.type}")
            print(f"State: {conversation.state}")
            
            # Access participants
            if conversation.participants:
                for participant in conversation.participants:
                    print(f"  Participant: {participant.name} (Role: {participant.role})")
            
            print("-" * 40)

        return response.entities

    except ApiException as e:
        # Handle 401 (Unauthorized) or 403 (Forbidden)
        if e.status == 403:
            print("Error 403: Check if 'conversation:read' scope is granted.")
        elif e.status == 401:
            print("Error 401: Authentication failed. Check credentials.")
        else:
            print(f"API Error: {e.status} - {e.body}")
        raise

Step 2: Querying Historical Analytics (/api/v2/analytics/conversations/details/query)

Use /api/v2/analytics/conversations/details/query when you need to analyze past events. This endpoint is read-only and aggregates data over time. It is the source of truth for reporting.

Use Cases:

• Calculating Average Handle Time (AHT) for the last week.
• Generating a list of all calls that exceeded a 5-minute wait time.
• Auditing conversation records for compliance.

Required Scope: analytics:conversations:read

The Analytics API uses a query-based model. You send a JSON body defining the time range, metric filters, and groupings. The response is paginated.

from purecloud_platform_client import AnalyticsApi, ConversationDetailsQuery
from purecloud_platform_client.rest import ApiException
from datetime import datetime, timedelta, timezone

def query_historical_conversations(client: PureCloudPlatformClientV2):
    """
    Queries historical conversation details for the last 24 hours.
    """
    api_instance = AnalyticsApi(client)

    # Define the time range
    end_time = datetime.now(timezone.utc)
    start_time = end_time - timedelta(days=1)

    # Format for ISO 8601
    start_str = start_time.isoformat()
    end_str = end_time.isoformat()

    # Construct the query body
    # This query fetches call details where the duration was greater than 60 seconds
    query_body = {
        "interval": "PT1H",  # Aggregate by hour
        "dateFrom": start_str,
        "dateTo": end_str,
        "types": ["call"],
        "metrics": ["conversationDuration", "holdDuration"],
        "groupBy": ["direction"],  # Group by inbound/outbound
        "filters": [
            {
                "type": "metric",
                "metric": "conversationDuration",
                "operation": "gt",
                "value": 60
            }
        ],
        "pageSize": 100,
        "pageNumber": 1
    }

    try:
        # The SDK method maps to POST /api/v2/analytics/conversations/details/query
        response = api_instance.post_analytics_conversations_details_query(
            body=query_body
        )

        print(f"Total records found: {response.total}")
        print(f"Page size: {response.page_size}")
        print(f"Page number: {response.page_number}")

        if response.entities:
            for detail in response.entities:
                print(f"Interval: {detail.interval}")
                print(f"Direction: {detail.direction}")
                print(f"Average Duration: {detail.metrics['conversationDuration']['average']} seconds")
                print(f"Count: {detail.metrics['conversationDuration']['count']}")
                print("-" * 40)
        else:
            print("No historical data found matching the criteria.")

        return response

    except ApiException as e:
        if e.status == 403:
            print("Error 403: Check if 'analytics:conversations:read' scope is granted.")
        elif e.status == 400:
            print("Error 400: Invalid query body. Check date formats and metric names.")
        else:
            print(f"API Error: {e.status} - {e.body}")
        raise

Step 3: Handling Pagination in Analytics

The Analytics API enforces pagination to prevent large payloads from timing out. The response object contains next_page, previous_page, and total_pages. You must loop through pages to retrieve all results if total exceeds page_size.

def query_all_historical_conversations(client: PureCloudPlatformClientV2):
    """
    Iterates through all pages of historical conversation data.
    """
    api_instance = AnalyticsApi(client)
    
    end_time = datetime.now(timezone.utc)
    start_time = end_time - timedelta(days=7) # Last 7 days

    query_body = {
        "interval": "PT1D",
        "dateFrom": start_time.isoformat(),
        "dateTo": end_time.isoformat(),
        "types": ["call"],
        "metrics": ["conversationDuration"],
        "pageSize": 200,
        "pageNumber": 1
    }

    all_results = []

    try:
        while True:
            response = api_instance.post_analytics_conversations_details_query(
                body=query_body
            )

            if response.entities:
                all_results.extend(response.entities)
            
            # Check if there are more pages
            if response.next_page is None:
                break
            
            # Update query for next page
            # Note: The SDK often provides a helper, but manually updating 
            # pageNumber is explicit and reliable.
            query_body["pageNumber"] += 1

        print(f"Successfully retrieved {len(all_results)} historical records.")
        return all_results

    except ApiException as e:
        print(f"Failed to retrieve analytics data: {e.body}")
        raise

Complete Working Example

This script combines authentication, real-time retrieval, and historical querying into a single executable module. It demonstrates the distinct patterns for each API type.

import os
import sys
import json
from datetime import datetime, timedelta, timezone
from purecloud_platform_client import PureCloudPlatformClientV2, ConversationApi, AnalyticsApi
from purecloud_platform_client.rest import ApiException

def main():
    # 1. Authentication
    try:
        client = get_purecloud_client()
        print("Authenticated successfully.")
    except Exception as e:
        print(f"Authentication failed: {e}")
        sys.exit(1)

    # 2. Real-Time Fetch
    print("\n--- Fetching Real-Time Conversations ---")
    try:
        active_convos = get_active_conversations(client)
        print(f"Retrieved {len(active_convos) if active_convos else 0} active conversations.")
    except ApiException as e:
        print(f"Real-Time API Error: {e}")

    # 3. Historical Fetch
    print("\n--- Fetching Historical Analytics ---")
    try:
        historical_data = query_all_historical_conversations(client)
        print(f"Retrieved {len(historical_data)} historical records.")
        
        # Example: Calculate total duration
        total_duration = 0
        for record in historical_data:
            if 'conversationDuration' in record.metrics:
                total_duration += record.metrics['conversationDuration']['sum'] or 0
        
        print(f"Total conversation duration in seconds: {total_duration}")

    except ApiException as e:
        print(f"Analytics API Error: {e}")

def get_purecloud_client():
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    environment = os.getenv("GENESYS_ENVIRONMENT", "mypurecloud.com")

    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")

    return PureCloudPlatformClientV2(
        client_id=client_id,
        client_secret=client_secret,
        environment=environment
    )

def get_active_conversations(client: PureCloudPlatformClientV2):
    api_instance = ConversationApi(client)
    try:
        response = api_instance.get_conversations(expand="participants")
        return response.entities
    except ApiException as e:
        raise

def query_all_historical_conversations(client: PureCloudPlatformClientV2):
    api_instance = AnalyticsApi(client)
    end_time = datetime.now(timezone.utc)
    start_time = end_time - timedelta(days=1)

    query_body = {
        "interval": "PT1H",
        "dateFrom": start_time.isoformat(),
        "dateTo": end_time.isoformat(),
        "types": ["call"],
        "metrics": ["conversationDuration"],
        "pageSize": 100,
        "pageNumber": 1
    }

    all_results = []
    try:
        while True:
            response = api_instance.post_analytics_conversations_details_query(body=query_body)
            if response.entities:
                all_results.extend(response.entities)
            if response.next_page is None:
                break
            query_body["pageNumber"] += 1
        return all_results
    except ApiException as e:
        raise

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden on /api/v2/analytics/...

• Cause: The OAuth token lacks the analytics:conversations:read scope. The Real-Time scope conversation:read does not grant access to analytics endpoints.
• Fix: Update your OAuth client configuration in the Genesys Cloud Admin UI. Navigate to Admin > Security > OAuth > Select Client > Scopes. Add analytics:conversations:read. Re-authenticate to get a new token.

Error: 429 Too Many Requests

• Cause: You are exceeding the rate limit. The Real-Time API has lower limits than the Analytics API because it is more resource-intensive to maintain state.
• Fix: Implement exponential backoff. The Genesys Cloud SDK does not handle retries automatically for 429s in all versions. Wrap your calls in a retry loop.

import time

def resilient_api_call(func, *args, max_retries=3):
    for attempt in range(max_retries):
        try:
            return func(*args)
        except ApiException as e:
            if e.status == 429:
                wait_time = 2 ** attempt
                print(f"Rate limited. Waiting {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise
    raise Exception("Max retries exceeded for 429 error.")

Error: Empty Response from Analytics Query

• Cause: The dateFrom and dateTo range is too narrow, or the filters are too restrictive.
• Fix: Verify the time zone. Genesys Cloud Analytics uses UTC. If you pass local time without converting to UTC, you may query a future or past empty range. Ensure interval is valid (e.g., PT1H, PT1D).

Error: TypeError in SDK Initialization

• Cause: Passing environment as a string without the correct format.
• Fix: Ensure the environment variable ends with .mypurecloud.com or is set to mypurecloud.com (the SDK handles the prefix). Avoid passing just us-east-1 unless using the specific region-aware constructor.

Official References

• Genesys Cloud API: Conversations
• Genesys Cloud API: Analytics Conversations Details
• Python SDK Documentation