Querying Live Conversations vs. Historical Analytics Data in Genesys Cloud

What You Will Build

• You will build two distinct data retrieval mechanisms: one to fetch real-time, active conversation details and another to aggregate historical interaction metrics.
• This tutorial utilizes the Genesys Cloud Platform API (/api/v2/conversations and /api/v2/analytics/conversations/details/query) and the official Python SDK.
• The implementation covers Python 3.8+ with the genesyscloud SDK and raw requests for direct HTTP comparison.

Prerequisites

• OAuth Client: A Genesys Cloud OAuth client with the following scopes:
  • conversation:read (for live conversations)
  • analytics:conversation:read (for historical analytics)
• SDK Version: genesyscloud Python SDK version 140.0.0 or higher.
• Runtime: Python 3.8+.
• Dependencies:

  pip install genesyscloud requests
  

Authentication Setup

Both endpoints require valid OAuth 2.0 Bearer tokens. The critical distinction lies in the scopes attached to the client. If you use a client with only conversation:read, the analytics query will return a 403 Forbidden. If you use a client with only analytics:conversation:read, the live conversation fetch will fail similarly.

For this tutorial, assume you have a service account or integration client configured. Below is a standard token refresh helper using requests to illustrate the raw HTTP requirement, which the SDK handles internally.

import requests
import time
import json

class GenesysAuth:
    def __init__(self, client_id: str, client_secret: str, org_id: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.org_id = org_id
        self.token_url = f"https://api.mypurecloud.com/oauth/token"
        self.access_token = None
        self.token_expiry = 0

    def get_token(self) -> str:
        if self.access_token and time.time() < self.token_expiry:
            return self.access_token
        
        data = {
            "grant_type": "client_credentials",
            "scope": "conversation:read analytics:conversation:read"
        }
        auth = (self.client_id, self.client_secret)
        
        response = requests.post(self.token_url, data=data, auth=auth)
        response.raise_for_status()
        
        token_data = response.json()
        self.access_token = token_data["access_token"]
        # Subtract 30 seconds to ensure we refresh before actual expiry
        self.token_expiry = time.time() + token_data["expires_in"] - 30
        
        return self.access_token

In the SDK, this is handled by the PlatformClient initialization. Ensure you pass the correct scopes or use a client that possesses both.

from genesyscloud.platform.client import PlatformClient
from genesyscloud.auth.oauth import OAuthClientCredentials

# Initialize Auth
oauth = OAuthClientCredentials(
    client_id="YOUR_CLIENT_ID",
    client_secret="YOUR_CLIENT_SECRET",
    org_id="YOUR_ORG_ID"
)

# Initialize Platform Client
platform_client = PlatformClient(auth=oauth)

Implementation

Step 1: Understanding /api/v2/conversations (Live State)

The /api/v2/conversations endpoint is designed for stateful, real-time operations. It returns conversations that are currently active in the system. This includes calls in progress, chats open, callbacks waiting, and tasks currently assigned to agents.

Key Characteristics:

• Latency: Near real-time (seconds).
• Data Scope: Only active sessions. Once a conversation ends, it disappears from this endpoint.
• Use Case: Dashboarding current queue depth, triggering real-time workflows, or fetching the current status of a specific interaction for an agent assist tool.
• Pagination: Supports cursor-based pagination via cursor parameter.

Fetching Active Conversations via SDK

We will retrieve all currently active conversations of type voice.

from genesyscloud.conversations.api import ConversationsApi
from genesyscloud.rest import ApiException

def get_active_voice_conversations(platform_client: PlatformClient):
    conversations_api = ConversationsApi(platform_client)
    
    all_conversations = []
    next_page_token = None
    
    try:
        while True:
            # Note: 'types' is a list of strings. 
            # 'pageSize' defaults to 25. Max is 100.
            response = conversations_api.get_conversations(
                types=["voice"],
                page_size=100,
                next_page_token=next_page_token
            )
            
            if response.conversations:
                all_conversations.extend(response.conversations)
                next_page_token = response.next_page_token
            
            # If no next_page_token, we have exhausted the list
            if not next_page_token:
                break
                
    except ApiException as e:
        print(f"Exception when calling ConversationsApi->get_conversations: {e}")
        if e.status == 429:
            print("Rate limited. Implement exponential backoff.")
        raise

    return all_conversations

Raw HTTP Equivalent

To understand exactly what is being sent, here is the equivalent requests call.

import requests

def get_active_voice_conversations_http(token: str):
    url = "https://api.mypurecloud.com/api/v2/conversations"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    params = {
        "types": "voice",
        "pageSize": "100"
    }
    
    response = requests.get(url, headers=headers, params=params)
    response.raise_for_status()
    
    data = response.json()
    return data.get("conversations", [])

Response Structure (Truncated):

{
  "conversations": [
    {
      "id": "conv-123-456",
      "type": "voice",
      "state": "connected",
      "start_time": "2023-10-27T10:00:00.000Z",
      "participants": [
        {
          "id": "part-789",
          "name": "John Doe",
          "external_contact": {
            "phone_number": "+15551234567"
          }
        }
      ]
    }
  ],
  "next_page_token": "abc123token..."
}

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

The /api/v2/analytics/conversations/details/query endpoint is designed for historical analysis. It does not return “live” data. It returns aggregated metrics and details for conversations that have already been closed and processed by the analytics pipeline.

Key Characteristics:

• Latency: Delayed. Data is typically available 15-30 minutes after conversation closure.
• Data Scope: Historical records. You query by time range (from, to).
• Use Case: Reporting, calculating Service Level (SL), Average Handle Time (AHT), or auditing past interactions.
• Query Mechanism: Uses a complex JSON body to define filters, groupings, and metrics.

Building an Analytics Query via SDK

We will query for all voice conversations from the last 24 hours, grouping by user (agent) to see how many calls each agent handled.

from genesyscloud.analytics.api import AnalyticsApi
from genesyscloud.models import ConversationDetailsQuery
from datetime import datetime, timedelta

def get_historical_voice_metrics(platform_client: PlatformClient):
    analytics_api = AnalyticsApi(platform_client)
    
    # Define time window: Last 24 hours
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(hours=24)
    
    # Construct the query body
    # Note: The SDK uses specific model objects for the query body
    query_body = ConversationDetailsQuery(
        date_from=start_time.isoformat() + "Z",
        date_to=end_time.isoformat() + "Z",
        size=100, # Max results per page for details
        type="voice",
        # We want to group by agent to see individual performance
        group_by=["user"],
        # We want specific metrics
        metrics=["handle_time", "hold_time", "talk_time"],
        # Optional: Filter only completed conversations
        query_filter="state:closed" 
    )
    
    try:
        response = analytics_api.post_analytics_conversations_details_query(
            body=query_body
        )
        
        return response
        
    except ApiException as e:
        print(f"Exception when calling AnalyticsApi->post_analytics_conversations_details_query: {e}")
        raise

Raw HTTP Equivalent

The Analytics API uses a POST request with a JSON body, unlike the GET request for live conversations.

import requests

def get_historical_voice_metrics_http(token: str):
    url = "https://api.mypurecloud.com/api/v2/analytics/conversations/details/query"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    # Calculate dates dynamically
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(hours=24)
    
    payload = {
        "dateFrom": start_time.isoformat() + "Z",
        "dateTo": end_time.isoformat() + "Z",
        "size": 100,
        "type": "voice",
        "groupBy": ["user"],
        "metrics": ["handleTime", "holdTime", "talkTime"],
        "queryFilter": "state:closed"
    }
    
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    
    return response.json()

Response Structure (Truncated):

{
  "pageSize": 100,
  "total": 450,
  "page": 1,
  "entities": [
    {
      "id": "user-id-1",
      "name": "Agent Smith",
      "handleTime": {
        "avg": 120.5,
        "min": 30.0,
        "max": 400.0,
        "sum": 12050.0,
        "count": 100
      },
      "holdTime": {
        "avg": 5.2,
        "count": 100
      }
    }
  ],
  "nextPageUri": "/api/v2/analytics/conversations/details/query?cursor=..."
}

Step 3: Processing Results and Handling Pagination

Both APIs support pagination, but they handle it differently. Live conversations use a next_page_token string in the response header/body. Analytics uses a nextPageUri or a cursor mechanism depending on the specific endpoint variant. For details/query, the SDK often handles the cursor internally if you use the iterator patterns, but manual handling is safer for production.

Handling Live Conversation Pagination (SDK)

The get_conversations method returns a response object. You must check response.next_page_token.

def fetch_all_live_conversations(platform_client: PlatformClient):
    conversations_api = ConversationsApi(platform_client)
    all_convs = []
    cursor = None
    
    while True:
        try:
            resp = conversations_api.get_conversations(types=["voice"], page_size=100, next_page_token=cursor)
            if resp.conversations:
                all_convs.extend(resp.conversations)
            cursor = resp.next_page_token
            if not cursor:
                break
        except ApiException as e:
            if e.status == 429:
                time.sleep(1) # Simple retry logic
                continue
            raise
    return all_convs

Handling Analytics Pagination (SDK)

The Analytics API response contains a nextPageUri. However, the Python SDK for Analytics often requires you to parse the cursor from the URI or use the get_analytics_conversations_details_query (GET) variant if you already have a cursor. The POST endpoint is for initiating the query. Subsequent pages are fetched via GET using the cursor.

Note: The SDK abstracts this somewhat, but for robustness, check response.next_page_token if available in the specific SDK version, or parse the nextPageUri.

def fetch_all_analytics_results(platform_client: PlatformClient, query_body: ConversationDetailsQuery):
    analytics_api = AnalyticsApi(platform_client)
    all_entities = []
    
    # Initial POST request
    response = analytics_api.post_analytics_conversations_details_query(body=query_body)
    all_entities.extend(response.entities if response.entities else [])
    
    next_page_token = response.next_page_token
    
    # Subsequent GET requests
    while next_page_token:
        try:
            # Use the GET endpoint for pagination
            resp = analytics_api.get_analytics_conversations_details_query(
                query_filter=query_body.query_filter, # Must match original filters
                date_from=query_body.date_from,
                date_to=query_body.date_to,
                type=query_body.type,
                group_by=query_body.group_by,
                metrics=query_body.metrics,
                size=query_body.size,
                next_page_token=next_page_token
            )
            all_entities.extend(resp.entities if resp.entities else [])
            next_page_token = resp.next_page_token
        except ApiException as e:
            print(f"Pagination error: {e}")
            break
            
    return all_entities

Complete Working Example

This script combines both approaches. It first checks for any live voice conversations. If there are none, it falls back to retrieving the last 10 completed voice conversations from the analytics engine to demonstrate the data difference.

import time
import sys
from datetime import datetime, timedelta
from genesyscloud.platform.client import PlatformClient
from genesyscloud.auth.oauth import OAuthClientCredentials
from genesyscloud.conversations.api import ConversationsApi
from genesyscloud.analytics.api import AnalyticsApi
from genesyscloud.models import ConversationDetailsQuery
from genesyscloud.rest import ApiException

def main():
    # Configuration
    CLIENT_ID = "YOUR_CLIENT_ID"
    CLIENT_SECRET = "YOUR_CLIENT_SECRET"
    ORG_ID = "YOUR_ORG_ID"
    
    # 1. Authentication
    try:
        oauth = OAuthClientCredentials(client_id=CLIENT_ID, client_secret=CLIENT_SECRET, org_id=ORG_ID)
        platform_client = PlatformClient(auth=oauth)
        print("Authentication successful.")
    except Exception as e:
        print(f"Authentication failed: {e}")
        sys.exit(1)

    # 2. Fetch Live Conversations
    conversations_api = ConversationsApi(platform_client)
    live_convs = []
    
    print("\n--- Fetching Live Voice Conversations ---")
    try:
        # Fetch first page
        resp = conversations_api.get_conversations(types=["voice"], page_size=10)
        if resp.conversations:
            live_convs = resp.conversations
            print(f"Found {len(live_convs)} live conversations.")
            for conv in live_convs[:2]: # Print first 2
                print(f"  ID: {conv.id}, State: {conv.state}, Started: {conv.start_time}")
        else:
            print("No live voice conversations found.")
            
    except ApiException as e:
        print(f"Error fetching live conversations: {e}")

    # 3. Fetch Historical Analytics
    analytics_api = AnalyticsApi(platform_client)
    
    print("\n--- Fetching Historical Voice Analytics (Last 24 Hours) ---")
    
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(hours=24)
    
    query_body = ConversationDetailsQuery(
        date_from=start_time.isoformat() + "Z",
        date_to=end_time.isoformat() + "Z",
        size=10,
        type="voice",
        group_by=[], # No grouping to get individual conversation details
        metrics=["handle_time"],
        query_filter="state:closed"
    )
    
    try:
        resp = analytics_api.post_analytics_conversations_details_query(body=query_body)
        
        if resp.entities:
            print(f"Found {len(resp.entities)} historical records.")
            for entity in resp.entities[:2]: # Print first 2
                # In analytics details, the entity structure differs from live convos
                # It usually contains the metrics requested
                handle_time = entity.get("handleTime", {}).get("avg", "N/A")
                print(f"  Entity ID: {entity.get('id')}, Avg Handle Time: {handle_time}")
        else:
            print("No historical records found in the last 24 hours.")
            
    except ApiException as e:
        print(f"Error fetching analytics: {e}")
        if e.status == 403:
            print("Check OAuth scopes: analytics:conversation:read is required.")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden on /api/v2/analytics/conversations/details/query

• Cause: The OAuth client used for authentication lacks the analytics:conversation:read scope.
• Fix: Go to the Genesys Cloud Admin UI > Administration > Security > OAuth Clients. Edit your client and add analytics:conversation:read to the scopes. Re-generate the token.

Error: 403 Forbidden on /api/v2/conversations

• Cause: The OAuth client lacks conversation:read.
• Fix: Add conversation:read to the OAuth client scopes.

Error: 429 Too Many Requests

• Cause: Genesys Cloud enforces strict rate limits. The Analytics API is particularly heavy and has lower throughput limits than the Conversations API.
• Fix: Implement exponential backoff. Do not poll the live conversation endpoint faster than every 1-2 seconds unless necessary. For analytics, cache results.

import time

def retry_with_backoff(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: Empty Results in Analytics

• Cause: The time window (dateFrom/dateTo) may be in the future, or no conversations match the queryFilter.
• Fix: Ensure dateFrom is strictly before dateTo. Verify that conversations actually exist in that timeframe. Remember that analytics data has a propagation delay (15-30 mins). Do not query for “now”.

Error: ConversationDetailsQuery Model Errors

• Cause: Passing invalid metric names or group-by fields.
• Fix: Refer to the Genesys Cloud Analytics API Documentation for valid metric keys. Common mistakes include using handle_time (snake_case) instead of handleTime (camelCase) in raw JSON, though the SDK models often handle this mapping. In the Python SDK, use the model properties as defined in the generated code.

Official References

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