Choosing the Right Conversation API: Real-Time vs. Analytics in Genesys Cloud

What You Will Build

• Two distinct data retrieval patterns: one for real-time conversation state and one for historical, aggregated metrics.
• This tutorial uses the Genesys Cloud Platform API v2 and the Python SDK genesyscloud.
• The programming language covered is Python 3.9+.

Prerequisites

• OAuth Client Type: Service Account or Client Credentials flow.
• Required Scopes:
  • For /api/v2/conversations: conversation:view (and conversation:read if you need media content).
  • For /api/v2/analytics/conversations: analytics:conversation:view.
• SDK Version: genesyscloud >= 14.0.0.
• Runtime: Python 3.9 or higher.
• Dependencies: genesyscloud, pandas (optional, for data manipulation demonstration).

Authentication Setup

Before accessing either endpoint, you must establish a valid OAuth 2.0 access token. The Genesys Cloud Python SDK handles token refresh automatically, but you must initialize the PlatformClient with valid credentials.

import os
from genesyscloud import Configuration, ApiClient

def get_platform_client() -> ApiClient:
    """
    Initializes and returns a configured ApiClient for Genesys Cloud.
    """
    # Retrieve credentials from environment variables
    client_id = os.environ.get("GENESYS_CLIENT_ID")
    client_secret = os.environ.get("GENESYS_CLIENT_SECRET")
    org_id = os.environ.get("GENESYS_ORG_ID")

    if not all([client_id, client_secret, org_id]):
        raise ValueError("Missing required environment variables: GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET, GENESYS_ORG_ID")

    # Configure the client
    configuration = Configuration(
        client_id=client_id,
        client_secret=client_secret,
        org_id=org_id
    )
    
    api_client = ApiClient(configuration)
    return api_client

Implementation

Step 1: Understanding the Data Model Difference

The core confusion between these two endpoints stems from the nature of the data they return.

1. /api/v2/conversations (Real-Time/Transaction):

   • Purpose: Retrieves the live state of a conversation or the history of a specific interaction ID.
   • Data Structure: Returns a Conversation object. This object contains the participants (users, customers, bots), the wrapup codes, the media type (voice, chat, email), and the specific transcripts or call recordings associated with that single instance.
   • Use Case: “What is happening right now in this chat?” or “Show me the transcript of this specific phone call.”

2. /api/v2/analytics/conversations (Historical/Aggregated):

   • Purpose: Retrieves aggregated metrics and filtered lists of past conversations for reporting.
   • Data Structure: Returns an AnalyticsConversationsDetailsQueryResponse. This contains entities (a list of conversation summaries) and metrics (aggregated data like average handle time, wait times, etc.).
   • Use Case: “Show me all calls from last week where the handle time exceeded 5 minutes” or “Calculate the average wait time for the Sales queue.”

Step 2: Fetching Real-Time Conversation Details

We will use the ConversationsApi to fetch details for a specific conversation ID. This is useful when you have a notification (via Webhooks or Streaming) that a conversation has ended, and you need to pull the specific details.

Required Scope: conversation:view

from genesyscloud import ConversationsApi
from genesyscloud.rest import ApiException
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def get_conversation_details(api_client: ApiClient, conversation_id: str) -> dict:
    """
    Fetches detailed information for a specific conversation ID.
    
    Args:
        api_client: The initialized ApiClient.
        conversation_id: The UUID of the conversation.
        
    Returns:
        A dictionary containing the conversation details.
    """
    api_instance = ConversationsApi(api_client)
    
    try:
        # Expand parameters allow you to include related data like transcripts
        # 'transcripts' is heavy; only include if necessary
        expand = ["participants", "wrapup"] 
        
        response = api_instance.get_conversation(
            conversation_id=conversation_id,
            expand=expand
        )
        
        logger.info(f"Successfully retrieved conversation {conversation_id}")
        return response.to_dict()
        
    except ApiException as e:
        logger.error(f"Exception when calling ConversationsApi->get_conversation: {e}")
        if e.status == 404:
            logger.error(f"Conversation {conversation_id} not found.")
        elif e.status == 403:
            logger.error("Forbidden. Check if 'conversation:view' scope is granted.")
        raise

# Example Usage
# client = get_platform_client()
# details = get_conversation_details(client, "123e4567-e89b-12d3-a456-426614174000")
# print(details.get('participants'))

Key Insight: The get_conversation endpoint does not support date range filtering. You must know the exact conversation_id. If you do not have the ID, this endpoint is useless for discovery.

Step 3: Querying Historical Analytics Data

To find conversations based on criteria (date, queue, user), you must use the Analytics API. This endpoint uses a query body to filter results.

Required Scope: analytics:conversation:view

from genesyscloud import AnalyticsApi
from genesyscloud.models.analytics_conversations_details_query import AnalyticsConversationsDetailsQuery
from genesyscloud.models.analytics_filter import AnalyticsFilter
from genesyscloud.models.analytics_entity_filter import AnalyticsEntityFilter
from datetime import datetime, timedelta
import logging

def query_historical_conversations(api_client: ApiClient, days_back: int = 7) -> list:
    """
    Queries the Analytics API for conversations from the last N days.
    
    Args:
        api_client: The initialized ApiClient.
        days_back: Number of days to look back.
        
    Returns:
        A list of conversation entities.
    """
    api_instance = AnalyticsApi(api_client)
    
    # Calculate date range
    end_date = datetime.utcnow()
    start_date = end_date - timedelta(days=days_back)
    
    # Format dates as ISO 8601 strings
    start_date_str = start_date.isoformat() + 'Z'
    end_date_str = end_date.isoformat() + 'Z'
    
    # Define the filter criteria
    # We want to filter by date range
    date_filter = AnalyticsEntityFilter(
        type="date",
        field="createdDate",
        operator="gte",
        value=[start_date_str]
    )
    
    # Optional: Filter by specific queue ID
    # queue_filter = AnalyticsEntityFilter(
    #     type="id",
    #     field="queueId",
    #     value=["QUEUE_UUID_HERE"]
    # )
    
    filters = [date_filter]
    
    # Construct the query object
    query = AnalyticsConversationsDetailsQuery(
        filters=filters,
        sort_by=["-createdDate"], # Sort by newest first
        size=25, # Page size (max 1000 for details, 5000 for summary)
        page=1
    )
    
    try:
        response = api_instance.post_analytics_conversations_details_query(
            body=query
        )
        
        logger.info(f"Retrieved {response.total} conversations from analytics.")
        
        if response.entities:
            return response.entities
        else:
            return []
            
    except ApiException as e:
        logger.error(f"Exception when calling AnalyticsApi->post_analytics_conversations_details_query: {e}")
        if e.status == 400:
            logger.error("Bad Request. Check filter syntax and date formats.")
        elif e.status == 403:
            logger.error("Forbidden. Check if 'analytics:conversation:view' scope is granted.")
        raise

# Example Usage
# client = get_platform_client()
# conversations = query_historical_conversations(client, days_back=1)
# for conv in conversations:
#     print(f"Conv ID: {conv.id}, Type: {conv.type}")

Key Insight: The Analytics API returns summaries of conversations by default in the entities list. It does not return full transcripts or participant details in the initial query response. If you need the transcript for a specific conversation found in the analytics query, you must take the conversation_id from the analytics response and call the /api/v2/conversations/{id} endpoint from Step 2.

Step 4: Handling Pagination in Analytics

Analytics queries can return thousands of records. The size parameter limits results per page. You must implement pagination to retrieve all data.

def get_all_analytics_conversations(api_client: ApiClient, days_back: int = 7) -> list:
    """
    Paginates through all analytics conversations for the given date range.
    """
    all_conversations = []
    page = 1
    max_pages = 10 # Safety break to prevent infinite loops
    
    while page <= max_pages:
        query = AnalyticsConversationsDetailsQuery(
            filters=[
                AnalyticsEntityFilter(
                    type="date",
                    field="createdDate",
                    operator="gte",
                    value=[(datetime.utcnow() - timedelta(days=days_back)).isoformat() + 'Z']
                )
            ],
            size=1000, # Max recommended for details
            page=page
        )
        
        try:
            response = api_instance.post_analytics_conversations_details_query(body=query)
            
            if not response.entities:
                break # No more data
                
            all_conversations.extend(response.entities)
            logger.info(f"Fetched page {page}, total entities so far: {len(all_conversations)}")
            
            # Check if there are more pages
            if response.page >= response.total_pages:
                break
                
            page += 1
            
        except ApiException as e:
            logger.error(f"Pagination error on page {page}: {e}")
            break
            
    return all_conversations

Complete Working Example

This script demonstrates the “Hybrid Approach”: using Analytics to find relevant conversations and then using the Conversations API to fetch detailed data for those specific IDs.

import os
import sys
from genesyscloud import Configuration, ApiClient, ConversationsApi, AnalyticsApi
from genesyscloud.models.analytics_conversations_details_query import AnalyticsConversationsDetailsQuery
from genesyscloud.models.analytics_entity_filter import AnalyticsEntityFilter
from datetime import datetime, timedelta

def setup_api_client():
    config = Configuration(
        client_id=os.environ["GENESYS_CLIENT_ID"],
        client_secret=os.environ["GENESYS_CLIENT_SECRET"],
        org_id=os.environ["GENESYS_ORG_ID"]
    )
    return ApiClient(config)

def find_recent_failed_calls(api_client: ApiClient, days_back: int = 1) -> list:
    """
    Finds calls from the last N days that were disconnected or had issues.
    Note: Filtering by specific wrapup codes or disconnect reasons requires 
    understanding your specific Genesys Cloud configuration.
    Here we simply fetch all voice conversations for demonstration.
    """
    analytics_api = AnalyticsApi(api_client)
    start_date = (datetime.utcnow() - timedelta(days=days_back)).isoformat() + 'Z'
    
    query = AnalyticsConversationsDetailsQuery(
        filters=[
            AnalyticsEntityFilter(type="date", field="createdDate", operator="gte", value=[start_date]),
            AnalyticsEntityFilter(type="string", field="type", operator="eq", value=["voice"])
        ],
        size=10, # Limit for demo
        page=1
    )
    
    try:
        response = analytics_api.post_analytics_conversations_details_query(body=query)
        return response.entities if response.entities else []
    except Exception as e:
        print(f"Analytics Error: {e}")
        return []

def fetch_full_details(api_client: ApiClient, conv_ids: list[str]) -> list[dict]:
    """
    Fetches full details for a list of conversation IDs using the Conversations API.
    """
    conv_api = ConversationsApi(api_client)
    details_list = []
    
    for conv_id in conv_ids:
        try:
            # Expand participants to see who was involved
            resp = conv_api.get_conversation(conversation_id=conv_id, expand=["participants"])
            details_list.append({
                "id": resp.id,
                "type": resp.type,
                "start_time": resp.start_time,
                "participants": [p.user_id for p in resp.participants if p.user_id]
            })
        except Exception as e:
            print(f"Failed to fetch details for {conv_id}: {e}")
            
    return details_list

def main():
    if not all(k in os.environ for k in ["GENESYS_CLIENT_ID", "GENESYS_CLIENT_SECRET", "GENESYS_ORG_ID"]):
        print("Missing environment variables.")
        sys.exit(1)
        
    client = setup_api_client()
    
    # Step 1: Use Analytics to find IDs
    print("Querying Analytics API for recent voice conversations...")
    recent_convs = find_recent_failed_calls(client, days_back=1)
    
    if not recent_convs:
        print("No conversations found.")
        return
        
    conv_ids = [c.id for c in recent_convs]
    print(f"Found {len(conv_ids)} conversations. Fetching details...")
    
    # Step 2: Use Conversations API to get details
    full_details = fetch_full_details(client, conv_ids)
    
    # Step 3: Output results
    for detail in full_details:
        print(f"Conversation: {detail['id']}")
        print(f"  Type: {detail['type']}")
        print(f"  Start: {detail['start_time']}")
        print(f"  Participants (User IDs): {detail['participants']}")
        print("-" * 40)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden on Analytics Endpoint

• Cause: The OAuth token lacks the analytics:conversation:view scope. The conversation:view scope is insufficient for the Analytics API.
• Fix: Regenerate your OAuth token ensuring the client has the analytics:conversation:view scope assigned in the Genesys Cloud Admin UI under Administration > Security > OAuth Clients.

Error: 400 Bad Request with “Invalid Filter”

• Cause: The AnalyticsEntityFilter syntax is incorrect. Common issues include using the wrong operator for the type (e.g., using eq on a date field instead of gte/lte).
• Fix: Verify the field names in the Analytics Conversation Details Query documentation. Ensure date values are ISO 8601 with ‘Z’ suffix.

Error: 429 Too Many Requests

• Cause: Exceeding the rate limit for either API. The Conversations API and Analytics API have separate rate limits.
• Fix: Implement exponential backoff. The Genesys Cloud Python SDK does not automatically retry 429s for all methods. You must wrap your API calls in a retry loop.

import time

def retry_on_rate_limit(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
                logger.warning(f"Rate limited. Retrying in {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise
    raise Exception("Max retries exceeded")

Error: Empty Entities in Analytics Response

• Cause: The date range is too narrow, or the filters are too restrictive.
• Fix: Broaden the date range or remove specific entity filters (like queueId) to verify if any data exists for the selected time period.

Official References

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