Choosing the Right Endpoint: Real-Time Conversations vs. Analytics Data

What You Will Build

• You will build a Python script that retrieves active, real-time conversation details using the Conversations API and a separate script that queries historical conversation metrics using the Analytics API.
• This tutorial uses the Genesys Cloud CX REST API and the genesyscloud Python SDK.
• The examples cover Python, demonstrating the distinct data models, authentication scopes, and use cases for each endpoint family.

Prerequisites

• OAuth Client Type: A Genesys Cloud OAuth application with the client_credentials grant type.
• Required Scopes:
  • For Conversations API: conversation:read, presence:read (if checking agent status).
  • For Analytics API: analytics:conversations:read, analytics:reports:read.
• SDK Version: genesyscloud Python SDK v10.0.0 or later.
• Runtime: Python 3.9+.
• Dependencies:

  pip install genesyscloud
  

Authentication Setup

Both endpoints require a valid OAuth 2.0 Bearer token. The genesyscloud SDK handles token caching and refresh automatically when initialized correctly. You must set your environment variables or pass credentials directly.

import os
from purecloudplatformclientv2 import PlatformApiClient, Configuration

# Initialize the Platform API Client for authentication
# This client manages the OAuth token lifecycle
platform_api_client = PlatformApiClient(
    environment_url=os.environ.get("GENESYS_CLOUD_ENV", "https://api.mypurecloud.com"),
    client_id=os.environ.get("GENESYS_CLOUD_CLIENT_ID"),
    client_secret=os.environ.get("GENESYS_CLOUD_CLIENT_SECRET"),
    grant_type="client_credentials"
)

# Get the configuration object to pass to API clients
configuration = Configuration(
    access_token=platform_api_client.get_access_token(),
    host=os.environ.get("GENESYS_CLOUD_ENV", "https://api.mypurecloud.com")
)

The PlatformApiClient is the foundation. It does not make data requests itself. It provides the access_token property that other API clients use. If the token expires, subsequent API calls will trigger an automatic refresh if the SDK is configured with the client credentials.

Implementation

Step 1: Retrieving Real-Time Data with /api/v2/conversations

The Conversations API (/api/v2/conversations) represents the current state of the system. It is event-driven and stateful. When you query this endpoint, you receive data about conversations that are currently active (in progress) or recently completed (within a short retention window, typically 24-48 hours depending on configuration).

Use Case: You need to display a live dashboard of active calls, update a CRM with the current caller ID, or trigger a webhook when an agent answers.

Working Code: List Active Conversations

from purecloudplatformclientv2 import ConversationApi
from purecloudplatformclientv2.rest import ApiException
import logging

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

def get_active_conversations(configuration):
    """
    Retrieves currently active conversations.
    Scope Required: conversation:read
    """
    api_instance = ConversationApi(api_client=configuration)
    
    # Parameters for get_conversations
    # domain_id: Optional. If omitted, returns all domains.
    # expand: Optional. Use "routing", "wrapup", "media" to get more details.
    try:
        # Fetch active conversations
        response = api_instance.get_conversations(
            domain_id="default",
            expand=["routing", "wrapup"],
            limit=100,
            offset=0
        )
        
        if response.entities and len(response.entities) > 0:
            for conv in response.entities:
                logger.info(f"Active Conversation ID: {conv.id}")
                logger.info(f"  Type: {conv.type}")
                logger.info(f"  State: {conv.state}")
                logger.info(f"  Start Time: {conv.start_time}")
                
                # Access routing details if expanded
                if conv.routing:
                    logger.info(f"  Queue ID: {conv.routing.queue_id}")
                    logger.info(f"  Agent ID: {conv.routing.agent_id}")
        else:
            logger.info("No active conversations found.")
            
        return response.entities

    except ApiException as e:
        logger.error(f"Status: {e.status}")
        logger.error(f"Reason: {e.reason}")
        logger.error(f"Response: {e.body}")
        raise

# Execute
if __name__ == "__main__":
    # Assuming 'configuration' is initialized as shown in Authentication Setup
    # active_convs = get_active_conversations(configuration)
    pass

Expected Response Structure

The response from /api/v2/conversations is a ConversationEntityListing. Key fields include:

• id: The unique conversation identifier.
• type: e.g., “voice”, “chat”, “email”.
• state: e.g., “ringing”, “connected”, “wrapup”, “closed”.
• routing: Contains queue_id, agent_id, and wrapup_code.
• participants: A list of objects representing each leg of the conversation (customer, agent, IVR).

Error Handling

• 401 Unauthorized: The token has expired or is invalid. The SDK usually retries automatically if configured with client credentials. If not, you must re-authenticate.
• 403 Forbidden: The OAuth client lacks the conversation:read scope.
• 404 Not Found: The domain_id does not exist.

Step 2: Querying Historical Data with /api/v2/analytics/conversations

The Analytics API (/api/v2/analytics/conversations/details/query) is designed for batch processing and historical reporting. It does not return real-time state. It returns aggregated or detailed records of conversations that have been processed and stored in the analytics database. This data is typically available 24-48 hours after the conversation ends, but can be configured for near-real-time in some enterprise setups.

Use Case: You need to generate a weekly report on average handle time (AHT), calculate first call resolution rates, or export conversation transcripts for compliance auditing.

Working Code: Query Historical Conversation Details

from purecloudplatformclientv2 import AnalyticsApi
from purecloudplatformclientv2.rest import ApiException
from purecloudplatformclientv2.models import ConversationDetailsQuery
import logging
from datetime import datetime, timedelta

def get_historical_conversations(configuration):
    """
    Queries historical conversation details.
    Scope Required: analytics:conversations:read
    """
    api_instance = AnalyticsApi(api_client=configuration)
    
    # Define the time range
    # Analytics API requires a start and end time
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(days=7)
    
    # Construct the query body
    # The body must be a ConversationDetailsQuery object
    query_body = ConversationDetailsQuery(
        date_from=start_time.isoformat(),
        date_to=end_time.isoformat(),
        # Filter by conversation type
        type=["voice"],
        # Select the fields you want to retrieve
        # This is crucial for performance. Only request what you need.
        view="conversation",
        group_by=[],
        interval=None,
        # Pagination
        page_size=100,
        page_token=None
    )
    
    try:
        # Execute the query
        # Note: The endpoint is post_analytics_conversations_details_query
        response = api_instance.post_analytics_conversations_details_query(
            body=query_body
        )
        
        if response.entities and len(response.entities) > 0:
            for entity in response.entities:
                # The structure of 'entity' depends on the 'view' and 'group_by'
                # For view="conversation", entity is a ConversationDetailsRecord
                logger.info(f"Historical Conversation ID: {entity.id}")
                logger.info(f"  Start Time: {entity.start_time}")
                logger.info(f"  Duration: {entity.duration_seconds} seconds")
                logger.info(f"  Queue: {entity.queue_id}")
                logger.info(f"  Agent: {entity.agent_id}")
                
                # Access specific metrics if available
                if entity.total_talk_time_seconds:
                    logger.info(f"  Talk Time: {entity.total_talk_time_seconds}s")
                if entity.total_hold_time_seconds:
                    logger.info(f"  Hold Time: {entity.total_hold_time_seconds}s")
        else:
            logger.info("No historical conversations found for the specified range.")
            
        return response.entities

    except ApiException as e:
        logger.error(f"Status: {e.status}")
        logger.error(f"Reason: {e.reason}")
        logger.error(f"Response: {e.body}")
        raise

# Execute
if __name__ == "__main__":
    # Assuming 'configuration' is initialized as shown in Authentication Setup
    # historical_convs = get_historical_conversations(configuration)
    pass

Expected Response Structure

The response from /api/v2/analytics/conversations/details/query is a ConversationDetailsRecord. Key fields include:

• id: The unique conversation identifier.
• start_time: ISO 8601 timestamp.
• duration_seconds: Total duration of the conversation.
• total_talk_time_seconds: Sum of all talk segments.
• total_hold_time_seconds: Sum of all hold segments.
• wrap_up_code: The code selected by the agent.
• queue_id: The queue the conversation was routed to.
• agent_id: The ID of the agent who handled the conversation.

Error Handling

• 400 Bad Request: The query body is malformed. Common causes include invalid date formats, missing date_from/date_to, or requesting too many fields in a grouped query.
• 429 Too Many Requests: Analytics queries are heavy on the database. If you exceed the rate limit, the API returns a 429. You must implement exponential backoff.
• 500 Internal Server Error: The analytics database is under heavy load. Retry with a delay.

Step 3: Processing Results and Pagination

Both APIs support pagination, but they handle it differently.

Conversations API Pagination

The Conversations API uses limit and offset parameters.

def get_all_active_conversations_paginated(configuration):
    api_instance = ConversationApi(api_client=configuration)
    all_conversations = []
    limit = 100
    offset = 0
    
    while True:
        try:
            response = api_instance.get_conversations(
                domain_id="default",
                limit=limit,
                offset=offset
            )
            
            if not response.entities or len(response.entities) == 0:
                break
                
            all_conversations.extend(response.entities)
            
            # Check if there are more pages
            if response.next_page is None:
                break
                
            # The next_page URL contains the new offset
            # You can parse it or simply increment offset
            offset += limit
            
        except ApiException as e:
            logger.error(f"Pagination error: {e.reason}")
            break
            
    return all_conversations

Analytics API Pagination

The Analytics API uses page_token in the request body and returns next_page in the response.

def get_all_historical_conversations_paginated(configuration):
    api_instance = AnalyticsApi(api_client=configuration)
    all_conversations = []
    page_token = None
    
    while True:
        query_body = ConversationDetailsQuery(
            date_from=(datetime.utcnow() - timedelta(days=7)).isoformat(),
            date_to=datetime.utcnow().isoformat(),
            type=["voice"],
            view="conversation",
            page_size=100,
            page_token=page_token
        )
        
        try:
            response = api_instance.post_analytics_conversations_details_query(
                body=query_body
            )
            
            if not response.entities or len(response.entities) == 0:
                break
                
            all_conversations.extend(response.entities)
            
            # Check for next page
            if response.next_page is None:
                break
                
            # Extract the page token from the next_page URL
            # The next_page is a full URL. You must parse the page_token query parameter.
            from urllib.parse import urlparse, parse_qs
            parsed_url = urlparse(response.next_page)
            page_token = parse_qs(parsed_url.query).get("page_token", [None])[0]
            
            if page_token is None:
                break
                
        except ApiException as e:
            logger.error(f"Pagination error: {e.reason}")
            break
            
    return all_conversations

Complete Working Example

Below is a complete, runnable Python script that demonstrates both endpoints. It requires the GENESYS_CLOUD_CLIENT_ID and GENESYS_CLOUD_CLIENT_SECRET environment variables to be set.

import os
import logging
from datetime import datetime, timedelta
from urllib.parse import urlparse, parse_qs

from purecloudplatformclientv2 import PlatformApiClient, Configuration, ConversationApi, AnalyticsApi
from purecloudplatformclientv2.models import ConversationDetailsQuery
from purecloudplatformclientv2.rest import ApiException

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def initialize_api_clients():
    """
    Initializes the Platform API client and returns the Configuration object.
    """
    env_url = os.environ.get("GENESYS_CLOUD_ENV", "https://api.mypurecloud.com")
    client_id = os.environ.get("GENESYS_CLOUD_CLIENT_ID")
    client_secret = os.environ.get("GENESYS_CLOUD_CLIENT_SECRET")
    
    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLOUD_CLIENT_ID and GENESYS_CLOUD_CLIENT_SECRET must be set.")
        
    platform_api_client = PlatformApiClient(
        environment_url=env_url,
        client_id=client_id,
        client_secret=client_secret,
        grant_type="client_credentials"
    )
    
    return Configuration(
        access_token=platform_api_client.get_access_token(),
        host=env_url
    )

def fetch_active_conversations(configuration):
    """
    Fetches the first 10 active conversations.
    """
    api_instance = ConversationApi(api_client=configuration)
    try:
        response = api_instance.get_conversations(
            domain_id="default",
            limit=10,
            offset=0,
            expand=["routing"]
        )
        return response.entities if response.entities else []
    except ApiException as e:
        logger.error(f"Failed to fetch active conversations: {e.reason}")
        return []

def fetch_historical_conversations(configuration):
    """
    Fetches the first 10 voice conversations from the last 24 hours.
    """
    api_instance = AnalyticsApi(api_client=configuration)
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(hours=24)
    
    query_body = ConversationDetailsQuery(
        date_from=start_time.isoformat(),
        date_to=end_time.isoformat(),
        type=["voice"],
        view="conversation",
        page_size=10,
        page_token=None
    )
    
    try:
        response = api_instance.post_analytics_conversations_details_query(
            body=query_body
        )
        return response.entities if response.entities else []
    except ApiException as e:
        logger.error(f"Failed to fetch historical conversations: {e.reason}")
        return []

def main():
    try:
        configuration = initialize_api_clients()
        
        logger.info("=== Fetching Active Conversations ===")
        active_convs = fetch_active_conversations(configuration)
        for conv in active_convs:
            logger.info(f"ID: {conv.id}, State: {conv.state}, Agent: {conv.routing.agent_id if conv.routing else 'None'}")
            
        logger.info("=== Fetching Historical Conversations (Last 24h) ===")
        historical_convs = fetch_historical_conversations(configuration)
        for conv in historical_convs:
            logger.info(f"ID: {conv.id}, Duration: {conv.duration_seconds}s, Agent: {conv.agent_id}")
            
    except Exception as e:
        logger.error(f"An unexpected error occurred: {str(e)}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden on Analytics API

Cause: The OAuth client does not have the analytics:conversations:read scope. The Conversations API and Analytics API require different scopes. A client with conversation:read cannot access analytics data.

Fix:

1. Go to the Genesys Cloud Admin Console.
2. Navigate to Setup > Applications > OAuth Applications.
3. Select your application.
4. In the Scopes tab, add analytics:conversations:read.
5. Save the changes.
6. Re-generate the access token.

Error: 429 Too Many Requests on Analytics API

Cause: You are sending too many requests per second. The Analytics API has stricter rate limits than the Conversations API because it queries large datasets.

Fix: Implement exponential backoff.

import time

def api_call_with_retry(func, *args, max_retries=5, **kwargs):
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except ApiException as e:
            if e.status == 429:
                wait_time = 2 ** attempt  # Exponential backoff: 2, 4, 8, 16, 32 seconds
                logger.warning(f"Rate limited (429). Waiting {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise
    raise Exception("Max retries exceeded for 429 error")

# Usage
# response = api_call_with_retry(api_instance.post_analytics_conversations_details_query, body=query_body)

Error: 400 Bad Request on Analytics Query

Cause: The date_from or date_to fields are missing, or the view parameter is invalid for the selected group_by fields.

Fix: Ensure the ConversationDetailsQuery object is fully populated.

• date_from and date_to are mandatory.
• If group_by is empty, view must be “conversation” or “summary”.
• If group_by is set, view must match the grouping level (e.g., “agent”, “queue”).

Official References

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