Querying Live Conversations vs. Historical Analytics in Genesys Cloud CX

What You Will Build

• This tutorial demonstrates how to retrieve real-time conversation data using the Conversations API versus querying historical interaction metrics using the Analytics API.
• This uses the Genesys Cloud CX REST API and the Python SDK (genesys-cloud-sdk).
• This covers Python 3.9+ with the requests library for raw HTTP examples and the official SDK for structured object handling.

Prerequisites

• OAuth Client Type: Service Account (recommended for server-to-server) or User-to-User.
• Required Scopes:
  • For Conversations API: conversation:read, conversation:monitor, conversation:view.
  • For Analytics API: analytics:conversations:view, analytics:query.
• SDK Version: genesys-cloud-sdk v13.0.0 or later.
• Runtime: Python 3.9+.
• Dependencies: pip install genesys-cloud-sdk requests

Authentication Setup

Genesys Cloud uses OAuth 2.0. For this tutorial, we assume a Service Account configuration. You must configure your Service Account with the specific scopes mentioned in the Prerequisites section. Missing scopes result in 403 Forbidden errors, which are distinct from authentication failures (401 Unauthorized).

Python SDK Authentication

The Python SDK handles the token management internally if you provide the client ID and secret.

import os
from purecloudplatformclientv2 import Configuration, ApiClient

def get_purecloud_api_client() -> ApiClient:
    """
    Initializes and returns a configured PureCloudPlatformClientV2 ApiClient.
    Uses environment variables for security.
    """
    config = Configuration(
        host=os.getenv("GENESYS_CLOUD_REGION_HOST", "https://api.mypurecloud.com"),
        client_id=os.getenv("GENESYS_CLOUD_CLIENT_ID"),
        client_secret=os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
    )
    
    # The API client automatically handles token acquisition and refresh
    api_client = ApiClient(configuration=config)
    return api_client

Raw HTTP Authentication (Requests Library)

If you are using raw HTTP requests, you must manage the token lifecycle manually.

import os
import requests

def get_access_token() -> str:
    """
    Retrieves an OAuth2 access token using the Client Credentials flow.
    """
    url = "https://api.mypurecloud.com/oauth/token"
    data = {
        "grant_type": "client_credentials",
        "client_id": os.getenv("GENESYS_CLOUD_CLIENT_ID"),
        "client_secret": os.getenv("GENESYS_CLOUD_CLIENT_SECRET"),
        "scope": "conversation:read analytics:conversations:view"
    }
    
    response = requests.post(url, data=data)
    response.raise_for_status() # Raises HTTPError for bad responses (4xx, 5xx)
    
    token_data = response.json()
    return token_data.get("access_token")

Implementation

The fundamental difference between these two endpoints lies in temporality and granularity.

1. /api/v2/conversations: This is the operational API. It returns the current state of active interactions. It is used for real-time monitoring, live dashboards, and immediate action (e.g., whispering to an agent). Data is transient. Once a conversation ends, it is eventually removed from this endpoint’s active scope.
2. /api/v2/analytics/conversations/details/query: This is the historical API. It returns aggregated or detailed records of past interactions. It is used for reporting, auditing, and trend analysis. Data is immutable and stored in the analytics data warehouse.

Step 1: Retrieving Active Conversations (Operational)

Use this endpoint when you need to know what is happening right now. This is critical for building a supervisor dashboard or an AI bot that needs context on an ongoing call.

Endpoint: GET /api/v2/conversations
Scope: conversation:read

Python SDK Example

from purecloudplatformclientv2 import ConversationApi, PaginationRequest

def list_active_conversations(api_client: ApiClient) -> list:
    """
    Retrieves the first page of active conversations.
    """
    conversation_api = ConversationApi(api_client)
    
    try:
        # PaginationRequest is optional but recommended to control page size
        pagination = PaginationRequest(page_size=10)
        
        response = conversation_api.get_conversations(pagination_request=pagination)
        
        if response.entities:
            for conv in response.entities:
                print(f"Conversation ID: {conv.id}")
                print(f"Type: {conv.type}")
                print(f"State: {conv.state}") # e.g., 'active', 'queued', 'wrapup'
                print(f"Start Time: {conv.start_time}")
                print("-" * 20)
        else:
            print("No active conversations found.")
            
        return response.entities
        
    except Exception as e:
        print(f"Error retrieving conversations: {e}")
        return []

Raw HTTP Example

import os
import requests

def get_active_conversations_http():
    token = get_access_token()
    url = "https://api.mypurecloud.com/api/v2/conversations"
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/json"
    }
    
    params = {
        "pageSize": 10,
        "page": 1
    }
    
    response = requests.get(url, headers=headers, params=params)
    
    if response.status_code == 200:
        data = response.json()
        for conv in data.get("entities", []):
            print(f"Active Conv ID: {conv['id']}, State: {conv['state']}")
    else:
        print(f"Failed to fetch conversations: {response.status_code} {response.text}")

Key Observations:

• The response includes the state of the conversation (queued, active, wrapup, monitoring).
• The data is lightweight. It does not contain the full transcript or detailed metric breakdowns by default.
• If you need the transcript for an active chat, you must use /api/v2/conversations/{id}/messages or the specific chat API, not the general conversations list.

Step 2: Querying Historical Analytics (Historical)

Use this endpoint when you need to analyze what happened in the past. This is used for compliance reviews, quality management, or calculating Average Handle Time (AHT) over a month.

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

This endpoint requires a query body that defines the time range, date range, and specific metrics you want to retrieve. Unlike the GET request for active conversations, this is a POST request because the query parameters are complex and nested.

Python SDK Example

from purecloudplatformclientv2 import AnalyticsApi, ConversationDetailsQueryRequest, DateRange, QueryType, IntervalType

def query_historical_conversations(api_client: ApiClient):
    """
    Queries historical conversation data for the last 24 hours.
    """
    analytics_api = AnalyticsApi(api_client)
    
    # Define the time range: Last 24 hours
    date_range = DateRange(
        from_date="now-1d",  # ISO 8601 duration format supported
        to_date="now"
    )
    
    # Define the query structure
    query_request = ConversationDetailsQueryRequest(
        date_from=date_range.from_date,
        date_to=date_range.to_date,
        type_=QueryType("interval"), # 'interval' aggregates data, 'details' gets row-level data
        interval=IntervalType("hour"), # Aggregate by hour
        
        # Selectors determine which data points to return
        selectors=["segment", "interaction"]
        
        # Filters can be added here to narrow down by queue, skill, or user
        # filters=[...] 
    )
    
    try:
        # Execute the query
        response = analytics_api.post_analytics_conversations_details_query(body=query_request)
        
        if response.entities:
            for entity in response.entities:
                print(f"Interval: {entity.interval}")
                print(f"Total Interactions: {entity.segment.get('interactions', {}).get('total', 0)}")
                # Access other metrics like handle_time, wait_time, etc.
                handle_time = entity.segment.get('handle_time', {}).get('total', 0)
                print(f"Total Handle Time (ms): {handle_time}")
                print("-" * 20)
        else:
            print("No historical data found for the specified range.")
            
    except Exception as e:
        print(f"Error querying analytics: {e}")

Raw HTTP Example

import os
import requests

def get_historical_analytics_http():
    token = get_access_token()
    url = "https://api.mypurecloud.com/api/v2/analytics/conversations/details/query"
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/json",
        "Content-Type": "application/json"
    }
    
    # The body defines the query
    payload = {
        "dateFrom": "now-1d",
        "dateTo": "now",
        "type": "interval",
        "interval": "hour",
        "selectors": [
            "segment",
            "interaction"
        ]
    }
    
    response = requests.post(url, headers=headers, json=payload)
    
    if response.status_code == 200:
        data = response.json()
        for entity in data.get("entities", []):
            interval = entity.get("interval")
            interactions = entity.get("segment", {}).get("interactions", {}).get("total", 0)
            print(f"Hour: {interval}, Interactions: {interactions}")
    else:
        print(f"Failed to fetch analytics: {response.status_code} {response.text}")

Key Observations:

• The response is aggregated based on the interval and type.
• You must explicitly request the metrics you need via selectors.
• Data latency: Analytics data is not real-time. There is typically a latency of 5-15 minutes before a conversation appears in the analytics query results. Do not use this API for real-time monitoring.

Step 3: Processing Results and Handling Pagination

Both APIs support pagination, but they handle it differently.

Conversations API (GET):
Returns nextPageUri and previousPageUri in the response header or body. You must follow these URIs to get subsequent pages.

Analytics API (POST):
The response includes nextPageId. You must include this ID in the next POST request body to fetch the next page.

Handling Analytics Pagination in Python

def query_all_historical_pages(api_client: ApiClient, page_size: int = 50):
    """
    Iterates through all pages of historical conversation data.
    """
    analytics_api = AnalyticsApi(api_client)
    
    # Initial request
    query_request = ConversationDetailsQueryRequest(
        date_from="now-1d",
        date_to="now",
        type_=QueryType("details"), # Use 'details' for row-level data
        page_size=page_size,
        selectors=["segment", "interaction"]
    )
    
    all_results = []
    
    try:
        response = analytics_api.post_analytics_conversations_details_query(body=query_request)
        all_results.extend(response.entities)
        
        # Loop through pages
        while response.next_page_id:
            # Update the request with the next page ID
            query_request.page_id = response.next_page_id
            
            # Fetch next page
            response = analytics_api.post_analytics_conversations_details_query(body=query_request)
            all_results.extend(response.entities)
            
        print(f"Total records fetched: {len(all_results)}")
        
    except Exception as e:
        print(f"Pagination error: {e}")

Complete Working Example

This script combines both approaches to demonstrate a hybrid use case: fetching an active conversation, then immediately querying its historical context if it has prior interactions.

import os
import sys
from purecloudplatformclientv2 import Configuration, ApiClient, ConversationApi, AnalyticsApi, ConversationDetailsQueryRequest, QueryType

def main():
    # 1. Setup Authentication
    if not os.getenv("GENESYS_CLOUD_CLIENT_ID") or not os.getenv("GENESYS_CLOUD_CLIENT_SECRET"):
        print("Error: Missing GENESYS_CLOUD_CLIENT_ID or GENESYS_CLOUD_CLIENT_SECRET environment variables.")
        sys.exit(1)

    config = Configuration(
        host=os.getenv("GENESYS_CLOUD_REGION_HOST", "https://api.mypurecloud.com"),
        client_id=os.getenv("GENESYS_CLOUD_CLIENT_ID"),
        client_secret=os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
    )
    api_client = ApiClient(configuration=config)

    # 2. Fetch Active Conversations
    conversation_api = ConversationApi(api_client)
    try:
        active_response = conversation_api.get_conversations(page_size=5)
        if not active_response.entities:
            print("No active conversations found. Exiting.")
            return

        print("=== Active Conversations ===")
        for conv in active_response.entities:
            print(f"ID: {conv.id} | Type: {conv.type} | State: {conv.state}")
            
            # 3. For each active conversation, check for recent history
            # Note: In production, you would likely filter analytics by the specific conversation ID if available
            # However, Analytics API queries are usually aggregated. 
            # To get history for a specific ID, you might use /api/v2/analytics/conversations/details/query 
            # with a filter on the conversation ID if the data model supports it, 
            # or more commonly, use the Conversation API to get the transcript for the current active session.
            
    except Exception as e:
        print(f"Error fetching active conversations: {e}")

    # 4. Fetch Historical Analytics
    print("\n=== Historical Analytics (Last 24 Hours) ===")
    analytics_api = AnalyticsApi(api_client)
    try:
        query_request = ConversationDetailsQueryRequest(
            date_from="now-1d",
            date_to="now",
            type_=QueryType("interval"),
            interval="hour",
            selectors=["segment"]
        )
        
        analytics_response = analytics_api.post_analytics_conversations_details_query(body=query_request)
        
        if analytics_response.entities:
            for entity in analytics_response.entities:
                interactions = entity.segment.get("interactions", {}).get("total", 0)
                print(f"Interval: {entity.interval} | Total Interactions: {interactions}")
        else:
            print("No historical data available.")
            
    except Exception as e:
        print(f"Error fetching analytics: {e}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

• Cause: The OAuth token lacks the required scope.
• Fix: Verify that the Service Account has conversation:read for the Conversations API and analytics:conversations:view for the Analytics API.
• Debug: Print the scope field from the OAuth token response.

# Check scopes in token response
token_data = response.json()
print(f"Scopes: {token_data.get('scope')}")

Error: 429 Too Many Requests

• Cause: You have exceeded the API rate limit.
• Fix: Implement exponential backoff. The response header Retry-After indicates the number of seconds to wait.
• Debug: Check the x-rate-limit-remaining header.

import time

def make_request_with_retry(url, headers, max_retries=3):
    for attempt in range(max_retries):
        response = requests.get(url, headers=headers)
        if response.status_code == 429:
            retry_after = int(response.headers.get('Retry-After', 5))
            print(f"Rate limited. Waiting {retry_after} seconds...")
            time.sleep(retry_after)
        else:
            return response
    raise Exception("Max retries exceeded")

Error: 400 Bad Request (Analytics API)

• Cause: The query body is malformed or the date range is invalid.
• Fix: Ensure dateFrom is before dateTo. Ensure the type is either interval or details. Ensure selectors are valid for the chosen type.
• Debug: Validate the JSON payload against the API documentation.

Official References

• Genesys Cloud Conversations API Documentation
• Genesys Cloud Analytics API Documentation
• Python SDK Reference