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

What You Will Build

• You will build a Python script that demonstrates the distinct use cases for the Genesys Cloud Conversations API (/api/v2/conversations) versus the Analytics API (/api/v2/analytics/conversations).
• The script will fetch real-time conversation details for an active call and then query historical analytics data for a completed interaction.
• The tutorial covers Python using the genesys-cloud-python-sdk and the requests library for raw HTTP comparison.

Prerequisites

• OAuth Client Type: A Genesys Cloud OAuth Client with client_credentials grant type.
• Required Scopes:
  • conversation:read (for /api/v2/conversations)
  • analytics:conversation:view (for /api/v2/analytics/conversations)
• SDK Version: genesys-cloud-python-sdk v2.0.0+
• Runtime: Python 3.8+
• External Dependencies:
  • genesys-cloud-python-sdk
  • requests
  • python-dotenv (for credential management)

Authentication Setup

Both API surfaces require valid OAuth 2.0 access tokens. The token generation process is identical for both, but the scopes granted to your OAuth client determine which endpoints you can access.

You must ensure your OAuth client in the Genesys Cloud Admin Console has both conversation:read and analytics:conversation:view scopes enabled.

Here is the standard authentication setup using the Python SDK, which handles token caching and refresh automatically.

import os
from purecloud_platform_client import Configuration, ApiClient

def get_purecloud_api_client() -> ApiClient:
    """
    Initializes and returns a configured PureCloudPlatformClientV2 ApiClient.
    Uses environment variables for credentials.
    """
    configuration = Configuration()
    configuration.host = "https://api.mypurecloud.com"
    
    # Set client credentials
    configuration.client_id = os.getenv("GENESYS_CLIENT_ID")
    configuration.client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    
    # Initialize the API client
    api_client = ApiClient(configuration)
    
    # The SDK automatically handles OAuth token acquisition and refresh
    return api_client

For the raw HTTP examples later, you will need to fetch a token manually.

import requests

def get_access_token(client_id: str, client_secret: str) -> str:
    """
    Fetches an OAuth access token from Genesys Cloud.
    """
    url = "https://login.mypurecloud.com/oauth/token"
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    data = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret
    }
    
    response = requests.post(url, headers=headers, data=data)
    response.raise_for_status()
    
    return response.json()["access_token"]

Implementation

Step 1: Understanding the Data Model Difference

The core distinction lies in timeliness and granularity.

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

   • Purpose: Used for immediate actions, telephony control, and retrieving the current state of an interaction.
   • Data: Contains the live participant list, current media state (connected, ringing, hold), and immediate transcript snippets.
   • Retention: Data is available immediately upon conversation start. Historical data may be purged or archived depending on your license, but it is not optimized for long-term statistical analysis.
   • Use Case: “Which agents are currently on the phone?” or “Transfer this call now.”

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

   • Purpose: Used for reporting, compliance, quality assurance, and historical trend analysis.
   • Data: Contains finalized metrics (handle time, queue time, disposition), complete transcripts (if licensed), and sentiment scores.
   • Retention: Data is processed asynchronously. There is typically a delay (15-60 minutes) before a conversation appears in analytics. Data is retained for longer periods based on your analytics license.
   • Use Case: “What was the average handle time for last month?” or “Retrieve the transcript of a call from 3 days ago for QA.”

Step 2: Fetching Real-Time Conversation Details

To demonstrate the Conversations API, we will retrieve details for a specific conversation ID. This endpoint is synchronous and returns the current state.

Endpoint: GET /api/v2/conversations/{conversationId}
Scope: conversation:read

from purecloud_platform_client import ConversationApi
from purecloud_platform_client.rest import ApiException

def get_realtime_conversation(api_client: ApiClient, conversation_id: str) -> dict:
    """
    Retrieves real-time details for a specific conversation.
    """
    api_instance = ConversationApi(api_client)
    
    try:
        # The SDK returns a Conversation object
        response = api_instance.get_conversation(conversation_id)
        
        # Convert to dict for easier inspection
        conversation_data = {
            "id": response.id,
            "type": response.type,
            "created_time": response.created_time.isoformat() if response.created_time else None,
            "state": response.state, # e.g., 'connected', 'ringing', 'ended'
            "participants": []
        }
        
        if response.participants:
            for p in response.participants:
                conversation_data["participants"].append({
                    "id": p.id,
                    "name": p.name,
                    "state": p.state,
                    "role": p.role
                })
                
        return conversation_data

    except ApiException as e:
        print(f"Exception when calling ConversationApi->get_conversation: {e}\n")
        if e.status == 404:
            print("Conversation not found. It may have ended and been purged from real-time storage.")
        elif e.status == 401:
            print("Unauthorized. Check OAuth scopes (conversation:read).")
        return None

Key Observation: The state field in the response reflects the current status. If the call is active, you will see connected. If you use this endpoint for a call that ended 10 minutes ago, you might still get the record, but the data is not optimized for aggregation.

Step 3: Querying Historical Analytics Data

To demonstrate the Analytics API, we will use the POST /api/v2/analytics/conversations/details/query endpoint. This is the primary method for retrieving historical conversation data. It uses a query-based approach rather than a direct ID lookup, which is critical for performance.

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

from purecloud_platform_client import AnalyticsApi
from purecloud_platform_client.models import ConversationDetailQueryRequest
from purecloud_platform_client.rest import ApiException
import datetime

def query_historical_conversations(api_client: ApiClient, conversation_id: str) -> list:
    """
    Queries historical analytics data for a specific conversation ID.
    Note: There is a delay between conversation end and analytics availability.
    """
    api_instance = AnalyticsApi(api_client)
    
    # Define the time range (last 24 hours)
    now = datetime.datetime.now(datetime.timezone.utc)
    start_time = now - datetime.timedelta(days=1)
    
    # Build the query body
    body = ConversationDetailQueryRequest(
        filter=conversation_id, # Filter by specific ID
        interval=f"{start_time.isoformat()}/{now.isoformat()}",
        size=25, # Max records per page
        select=[
            "id",
            "type",
            "state",
            "created_time",
            "answered_time",
            "ended_time",
            "duration_seconds",
            "participants"
        ]
    )
    
    try:
        # The SDK returns a ConversationDetailResponse object
        response = api_instance.post_analytics_conversations_details_query(body)
        
        if not response.entities:
            return []
            
        results = []
        for entity in response.entities:
            result = {
                "id": entity.id,
                "type": entity.type,
                "state": entity.state, # Usually 'ended' in analytics
                "created_time": entity.created_time.isoformat() if entity.created_time else None,
                "answered_time": entity.answered_time.isoformat() if entity.answered_time else None,
                "ended_time": entity.ended_time.isoformat() if entity.ended_time else None,
                "duration_seconds": entity.duration_seconds,
                "participants": []
            }
            
            if entity.participants:
                for p in entity.participants:
                    result["participants"].append({
                        "id": p.id,
                        "name": p.name,
                        "role": p.role,
                        "duration_seconds": p.duration_seconds
                    })
            
            results.append(result)
            
        return results

    except ApiException as e:
        print(f"Exception when calling AnalyticsApi->post_analytics_conversations_details_query: {e}\n")
        if e.status == 400:
            print("Bad Request. Check query syntax and date format.")
        elif e.status == 401:
            print("Unauthorized. Check OAuth scopes (analytics:conversation:view).")
        return []

Key Observation: The Analytics API requires an interval. You cannot simply ask for “this ID” without a time range. This is by design to prevent full-table scans. The response includes calculated fields like duration_seconds and answered_time, which are not always available in the real-time API until the call concludes.

Step 4: Comparing the Outputs Side-by-Side

The following function runs both queries and highlights the differences in data availability and structure.

def compare_api_responses(api_client: ApiClient, conversation_id: str):
    """
    Demonstrates the difference between real-time and analytics data.
    """
    print(f"--- Fetching Real-Time Data for {conversation_id} ---")
    realtime_data = get_realtime_conversation(api_client, conversation_id)
    
    if realtime_data:
        print(f"Real-Time State: {realtime_data.get('state')}")
        print(f"Real-Time Created: {realtime_data.get('created_time')}")
        print(f"Participants Count: {len(realtime_data.get('participants', []))}")
    else:
        print("Real-Time data not found.")
        
    print("\n--- Fetching Historical Analytics Data for {conversation_id} ---")
    analytics_data = query_historical_conversations(api_client, conversation_id)
    
    if analytics_data:
        for record in analytics_data:
            print(f"Analytics State: {record.get('state')}")
            print(f"Analytics Duration: {record.get('duration_seconds')} seconds")
            print(f"Analytics Answered Time: {record.get('answered_time')}")
    else:
        print("Analytics data not found. (Note: Analytics data may take up to 60 minutes to populate after call end.)")

# Usage Example
if __name__ == "__main__":
    client = get_purecloud_api_client()
    # Replace with a valid conversation ID from your environment
    TARGET_CONVERSATION_ID = "your-conversation-id-here"
    compare_api_responses(client, TARGET_CONVERSATION_ID)

Complete Working Example

This is a complete, runnable script. Save it as compare_conversations.py. Ensure you have a .env file with GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET.

import os
import datetime
from purecloud_platform_client import Configuration, ApiClient, ConversationApi, AnalyticsApi
from purecloud_platform_client.models import ConversationDetailQueryRequest
from purecloud_platform_client.rest import ApiException

def get_purecloud_api_client() -> ApiClient:
    configuration = Configuration()
    configuration.host = "https://api.mypurecloud.com"
    configuration.client_id = os.getenv("GENESYS_CLIENT_ID")
    configuration.client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    return ApiClient(configuration)

def get_realtime_conversation(api_client: ApiClient, conversation_id: str) -> dict:
    api_instance = ConversationApi(api_client)
    try:
        response = api_instance.get_conversation(conversation_id)
        return {
            "id": response.id,
            "state": response.state,
            "created_time": response.created_time.isoformat() if response.created_time else None,
            "participants_count": len(response.participants) if response.participants else 0
        }
    except ApiException as e:
        print(f"Real-Time Error: {e}")
        return None

def query_analytics_conversation(api_client: ApiClient, conversation_id: str) -> list:
    api_instance = AnalyticsApi(api_client)
    now = datetime.datetime.now(datetime.timezone.utc)
    start_time = now - datetime.timedelta(days=7) # Look back 7 days
    
    body = ConversationDetailQueryRequest(
        filter=conversation_id,
        interval=f"{start_time.isoformat()}/{now.isoformat()}",
        size=1,
        select=["id", "state", "duration_seconds", "created_time"]
    )
    
    try:
        response = api_instance.post_analytics_conversations_details_query(body)
        if not response.entities:
            return []
        
        results = []
        for entity in response.entities:
            results.append({
                "id": entity.id,
                "state": entity.state,
                "duration_seconds": entity.duration_seconds,
                "created_time": entity.created_time.isoformat() if entity.created_time else None
            })
        return results
    except ApiException as e:
        print(f"Analytics Error: {e}")
        return []

def main():
    # Load environment variables
    if not os.getenv("GENESYS_CLIENT_ID") or not os.getenv("GENESYS_CLIENT_SECRET"):
        print("Error: GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET environment variables are required.")
        return

    api_client = get_purecloud_api_client()
    target_id = "INSERT_CONVERSATION_ID_HERE"
    
    print(f"Comparing APIs for Conversation ID: {target_id}\n")
    
    # 1. Real-Time Check
    print("1. Real-Time API (/api/v2/conversations):")
    rt_data = get_realtime_conversation(api_client, target_id)
    if rt_data:
        print(f"   State: {rt_data['state']}")
        print(f"   Created: {rt_data['created_time']}")
        print(f"   Participants: {rt_data['participants_count']}")
    else:
        print("   Not found or already purged from real-time store.")
        
    print("\n2. Analytics API (/api/v2/analytics/conversations):")
    an_data = query_analytics_conversation(api_client, target_id)
    if an_data:
        for record in an_data:
            print(f"   State: {record['state']}")
            print(f"   Duration: {record['duration_seconds']}s")
            print(f"   Created: {record['created_time']}")
    else:
        print("   Not found. (May not be processed yet or outside retention period).")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token does not have the required scope.
• Fix: Ensure your OAuth client has conversation:read for the real-time API and analytics:conversation:view for the analytics API. Regenerate the token after adding scopes.

Error: 404 Not Found (Real-Time API)

• Cause: The conversation ID is invalid, or the conversation ended long enough ago that it was purged from the real-time database.
• Fix: Real-time data is ephemeral. For historical data, always use the Analytics API. Check the conversation ID for typos.

Error: Empty Response (Analytics API)

• Cause: The conversation has not yet been processed by the analytics engine.
• Fix: Analytics data is not available immediately. Wait 15-60 minutes after the conversation ends. Ensure the interval in your query covers the time the conversation occurred.

Error: 429 Too Many Requests

• Cause: You have exceeded the rate limit for the API endpoint.
• Fix: Implement exponential backoff. The Analytics API is more resource-intensive; limit your query frequency. Use pagination (size parameter) to reduce the load per request.

Official References

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