Querying Live Conversations vs. Historical Analytics in Genesys Cloud

What You Will Build

• You will build two distinct Python scripts that demonstrate the fundamental difference between retrieving active, real-time conversation data and querying historical, aggregated analytics data.
• The first script uses the Conversations API (/api/v2/conversations) to fetch details of currently active interactions.
• The second script uses the Analytics API (/api/v2/analytics/conversations/details/query) to retrieve historical records with specific metrics and filtering capabilities.
• The tutorial covers Python using the official Genesys Cloud SDK (genesys-cloud-py-client) and the requests library for raw HTTP comparisons.

Prerequisites

• OAuth Client Type: Confidential Client (Client Credentials Flow).
• Required Scopes:
  • For Conversations API: conversation:read
  • For Analytics API: analytics:query
• SDK Version: genesys-cloud-py-client >= 1.0.0.
• Language/Runtime: Python 3.8+.
• External Dependencies:
  • genesys-cloud-py-client
  • requests
  • python-dotenv (for secure credential management)

Authentication Setup

Both APIs require a valid OAuth 2.0 bearer token. The Genesys Cloud SDK handles token acquisition and refresh automatically when initialized correctly. For the raw HTTP examples, you must manually manage the token lifecycle.

SDK Initialization

Install the SDK:

pip install genesys-cloud-py-client

Create a .env file:

GENESYS_REGION=us-east-1
GENESYS_CLIENT_ID=your_client_id
GENESYS_CLIENT_SECRET=your_client_secret

Initialize the client in your Python script:

import os
from purecloud_platform_client import Configuration, PureCloudPlatformClientV2
from dotenv import load_dotenv

load_dotenv()

def get_purecloud_client():
    configuration = Configuration()
    configuration.host = f"https://{os.getenv('GENESYS_REGION')}.mypurecloud.com/api/v2"
    configuration.client_id = os.getenv('GENESYS_CLIENT_ID')
    configuration.client_secret = os.getenv('GENESYS_CLIENT_SECRET')
    
    # The SDK handles OAuth token refresh automatically
    return PureCloudPlatformClientV2(configuration)

Raw HTTP Authentication (For Comparison)

If you are using requests directly, you must fetch the token first.

import requests
import os

def get_access_token():
    url = f"https://login.mypurecloud.com/oauth/token"
    data = {
        "grant_type": "client_credentials",
        "client_id": os.getenv('GENESYS_CLIENT_ID'),
        "client_secret": os.getenv('GENESYS_CLIENT_SECRET')
    }
    response = requests.post(url, data=data)
    response.raise_for_status()
    return response.json()['access_token']

Implementation

Step 1: Understanding /api/v2/conversations (Real-Time State)

The Conversations API is designed for operational use cases. It returns the current state of interactions. It is not designed for historical reporting, large-scale data extraction, or complex metric aggregation.

Use Case: Building a supervisor dashboard that shows which agents are currently on a call, or a chat bot that needs to know the current status of a user’s session.

Code Example: Fetching Active Conversations

We will use the get_conversations endpoint. Note that this endpoint does not support complex filtering by date range or metrics. It returns what is happening now.

from purecloud_platform_client import ConversationsApi
from purecloud_platform_client.rest import ApiException

def get_active_conversations(client: PureCloudPlatformClientV2):
    conversations_api = ConversationsApi(client)
    
    try:
        # Scope: conversation:read
        # This retrieves all conversations currently active in the organization
        # It does not accept a date range. It is strictly real-time.
        result = conversations_api.get_conversations(
            expand=['participants', 'wrappers']
        )
        
        print(f"Total Active Conversations: {len(result.entities)}")
        
        for conv in result.entities:
            print(f"- ID: {conv.id}")
            print(f"  Type: {conv.type}")
            print(f"  State: {conv.state}")
            print(f"  Started: {conv.start_time}")
            
            # Accessing participant details
            if conv.participants:
                for participant in conv.participants:
                    print(f"    Participant: {participant.name} (Type: {participant.type})")
            print("---")
            
    except ApiException as e:
        print(f"Exception when calling ConversationsApi->get_conversations: {e}\n")
        if e.status == 401:
            print("Error: Unauthorized. Check OAuth scopes (conversation:read).")
        elif e.status == 403:
            print("Error: Forbidden. Client lacks permission.")
        elif e.status == 429:
            print("Error: Rate Limited. Implement exponential backoff.")

Key Characteristics of This Endpoint:

• Latency: Low. Returns data within milliseconds.
• Data Freshness: Real-time.
• Pagination: Supports pageSize and pageNumber, but limited to currently active items.
• Limitation: You cannot query conversations that ended 5 minutes ago. They disappear from this endpoint immediately upon closure.

Step 2: Understanding /api/v2/analytics/conversations (Historical Data)

The Analytics API is designed for reporting and historical analysis. It queries a data warehouse that aggregates conversation events. It supports complex filters (date ranges, metric thresholds, specific agents) and returns structured metric data.

Use Case: Generating an end-of-day report on average handle time (AHT), calculating first-call resolution rates, or exporting data to a BI tool like Tableau.

Code Example: Querying Historical Conversation Details

We will use the post_analytics_conversations_details_query endpoint. This is a POST request because the filter criteria are complex and passed in the request body.

from purecloud_platform_client import AnalyticsApi
from purecloud_platform_client.rest import ApiException
from purecloud_platform_client.models import (
    ConversationDetailQueryRequest,
    ConversationDetailQueryFilter,
    ConversationDetailQueryMetric,
    ConversationDetailQuerySort
)
from datetime import datetime, timedelta

def query_historical_conversations(client: PureCloudPlatformClientV2):
    analytics_api = AnalyticsApi(client)
    
    # Define the time range (last 24 hours)
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(hours=24)
    
    # Define the metrics we want to retrieve
    # Note: Not all metrics are available in 'details'. Check docs for availability.
    metrics = [
        ConversationDetailQueryMetric(name="waitTime", rollup="sum"),
        ConversationDetailQueryMetric(name="talkTime", rollup="sum"),
        ConversationDetailQueryMetric(name="wrapUpTime", rollup="sum")
    ]
    
    # Define filters
    filters = ConversationDetailQueryFilter(
        from_date=start_time.isoformat(),
        to_date=end_time.isoformat(),
        types=["voice"], # Filter by conversation type
        # Example: Filter by specific queue ID
        # queue_ids=["your-queue-id"]
    )
    
    # Define sorting
    sort = ConversationDetailQuerySort(field="startTime", order="desc")
    
    # Construct the request body
    request_body = ConversationDetailQueryRequest(
        metrics=metrics,
        filters=filters,
        sort=sort,
        size=50 # Max rows per page
    )
    
    try:
        # Scope: analytics:query
        result = analytics_api.post_analytics_conversations_details_query(
            body=request_body
        )
        
        print(f"Total Records Returned: {result.total_records}")
        print(f"Page Size: {result.page_size}")
        
        for entity in result.entities:
            print(f"- Conversation ID: {entity.id}")
            print(f"  Start Time: {entity.start_time}")
            print(f"  Type: {entity.type}")
            
            # Extract metrics
            for metric in entity.metrics:
                print(f"  Metric: {metric.name} = {metric.value}")
            print("---")
            
        # Handle pagination if more data exists
        if result.next_page_id:
            print(f"More data available. Next Page ID: {result.next_page_id}")
            
    except ApiException as e:
        print(f"Exception when calling AnalyticsApi->post_analytics_conversations_details_query: {e}\n")
        if e.status == 400:
            print("Error: Bad Request. Check metric names and date format.")
        elif e.status == 401:
            print("Error: Unauthorized. Check OAuth scopes (analytics:query).")
        elif e.status == 429:
            print("Error: Rate Limited. Analytics API has strict rate limits.")

Key Characteristics of This Endpoint:

• Latency: Higher. Queries can take seconds to minutes depending on data volume.
• Data Freshness: Near real-time (usually 5-15 minute delay).
• Pagination: Uses nextPageId for cursor-based pagination.
• Capability: Supports complex filtering, aggregation, and historical lookback.

Step 3: Comparing Raw HTTP Requests

To understand the architectural difference, let us look at the raw HTTP requests. This highlights why the Analytics API uses POST and why the Conversations API uses GET.

Conversations API (GET)

import requests

def get_active_conversations_raw(token: str):
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    url = f"https://us-east-1.mypurecloud.com/api/v2/conversations"
    params = {
        "expand": "participants"
    }
    
    try:
        response = requests.get(url, headers=headers, params=params)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 429:
            # Implement retry logic here
            pass
        raise

Analytics API (POST)

def query_analytics_raw(token: str, start_time: str, end_time: str):
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    url = f"https://us-east-1.mypurecloud.com/api/v2/analytics/conversations/details/query"
    
    payload = {
        "metrics": [
            {"name": "waitTime", "rollup": "sum"},
            {"name": "talkTime", "rollup": "sum"}
        ],
        "filters": {
            "fromDate": start_time,
            "toDate": end_time,
            "types": ["voice"]
        },
        "size": 50
    }
    
    try:
        response = requests.post(url, headers=headers, json=payload)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 429:
            # Analytics API has lower rate limits. Be careful.
            pass
        raise

Complete Working Example

Below is a complete, runnable Python script that demonstrates both approaches. It requires the genesys-cloud-py-client and python-dotenv packages.

import os
import sys
from datetime import datetime, timedelta
from purecloud_platform_client import Configuration, PureCloudPlatformClientV2, ConversationsApi, AnalyticsApi
from purecloud_platform_client.rest import ApiException
from purecloud_platform_client.models import (
    ConversationDetailQueryRequest,
    ConversationDetailQueryFilter,
    ConversationDetailQueryMetric,
    ConversationDetailQuerySort
)
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

def initialize_client():
    """Initializes the Genesys Cloud SDK client."""
    configuration = Configuration()
    region = os.getenv('GENESYS_REGION', 'us-east-1')
    configuration.host = f"https://{region}.mypurecloud.com/api/v2"
    configuration.client_id = os.getenv('GENESYS_CLIENT_ID')
    configuration.client_secret = os.getenv('GENESYS_CLIENT_SECRET')
    return PureCloudPlatformClientV2(configuration)

def fetch_active_conversations(client: PureCloudPlatformClientV2):
    """
    Fetches currently active conversations using the Conversations API.
    Use case: Real-time monitoring, supervisor dashboards.
    Scope: conversation:read
    """
    print("\n--- Fetching Active Conversations (/api/v2/conversations) ---")
    conversations_api = ConversationsApi(client)
    
    try:
        result = conversations_api.get_conversations(expand=['participants'])
        
        if not result.entities:
            print("No active conversations found.")
            return

        for conv in result.entities:
            print(f"[ACTIVE] ID: {conv.id} | Type: {conv.type} | State: {conv.state}")
            
    except ApiException as e:
        print(f"Error fetching active conversations: {e}")

def fetch_historical_conversations(client: PureCloudPlatformClientV2):
    """
    Fetches historical conversation details using the Analytics API.
    Use case: Reporting, AHT calculation, historical analysis.
    Scope: analytics:query
    """
    print("\n--- Fetching Historical Conversations (/api/v2/analytics/conversations) ---")
    analytics_api = AnalyticsApi(client)
    
    # Define time range: Last 1 hour
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(hours=1)
    
    # Define metrics
    metrics = [
        ConversationDetailQueryMetric(name="waitTime", rollup="sum"),
        ConversationDetailQueryMetric(name="talkTime", rollup="sum"),
        ConversationDetailQueryMetric(name="wrapUpTime", rollup="sum")
    ]
    
    # Define filters
    filters = ConversationDetailQueryFilter(
        from_date=start_time.isoformat(),
        to_date=end_time.isoformat(),
        types=["voice", "chat"]
    )
    
    # Define sort
    sort = ConversationDetailQuerySort(field="startTime", order="desc")
    
    # Construct request
    request_body = ConversationDetailQueryRequest(
        metrics=metrics,
        filters=filters,
        sort=sort,
        size=10
    )
    
    try:
        result = analytics_api.post_analytics_conversations_details_query(body=request_body)
        
        if not result.entities:
            print("No historical conversations found in the last hour.")
            return

        for entity in result.entities:
            print(f"[HISTORICAL] ID: {entity.id} | Start: {entity.start_time}")
            for metric in entity.metrics:
                print(f"  -> {metric.name}: {metric.value}")
                
    except ApiException as e:
        print(f"Error fetching historical conversations: {e}")

def main():
    if not os.getenv('GENESYS_CLIENT_ID') or not os.getenv('GENESYS_CLIENT_SECRET'):
        print("Error: GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set in .env file.")
        sys.exit(1)

    client = initialize_client()
    
    # 1. Get Real-Time Data
    fetch_active_conversations(client)
    
    # 2. Get Historical Data
    fetch_historical_conversations(client)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

Cause: The OAuth token is missing, expired, or the client does not have the required scope.

Fix:

1. Verify GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET are correct.
2. Check the OAuth Client settings in Genesys Cloud Admin.
3. Ensure the client has the specific scope:
   • For /api/v2/conversations: Add conversation:read.
   • For /api/v2/analytics/conversations: Add analytics:query.

Error: 400 Bad Request (Analytics API)

Cause: Invalid metric name, incorrect date format, or unsupported filter combination.

Fix:

1. Validate metric names against the Analytics Metrics Documentation.
2. Ensure fromDate and toDate are in ISO 8601 format.
3. Check that the metric is available for the requested conversation type (e.g., some voice metrics do not apply to chat).

Error: 429 Too Many Requests

Cause: Exceeding the rate limit for the API endpoint. The Analytics API has stricter rate limits than the Conversations API.

Fix: Implement exponential backoff.

import time

def make_api_call_with_retry(func, *args, max_retries=3, **kwargs):
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        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")

Error: No Data Returned (Analytics API)

Cause: The time range is too recent. Analytics data is not real-time.

Fix:

1. Analytics data typically has a latency of 5-15 minutes.
2. If querying for the last 5 minutes, you may get empty results.
3. Adjust the fromDate to be at least 15 minutes in the past.

Official References

• Genesys Cloud Conversations API Documentation
• Genesys Cloud Analytics API Documentation
• Genesys Cloud Python SDK Reference
• OAuth 2.0 Client Credentials Flow