Choosing Between /api/v2/conversations and /api/v2/analytics/conversations in Genesys Cloud

What You Will Build

• You will build a Python script that retrieves real-time conversation data using the Conversations API and historical analytics data using the Analytics API.
• You will understand the architectural distinction between live state management and historical reporting in Genesys Cloud.
• You will implement error handling and pagination for both endpoints using the official Genesys Cloud Python SDK.

Prerequisites

• OAuth Client Type: Private Private Key (for server-to-server authentication) or Public Client ID/Secret (for user context).
• Required Scopes:
  • For /api/v2/conversations: view:conversation, view:conversation:details
  • For /api/v2/analytics/conversations: view:analytics:conversation
• SDK Version: genesys-cloud-purecloud-platform-client >= 160.0.0
• Language/Runtime: Python 3.9+
• External Dependencies: pip install genesys-cloud-purecloud-platform-client

Authentication Setup

Authentication is the foundation of every API call. Genesys Cloud uses OAuth 2.0. For server-side scripts that do not require a specific user’s context (or require the full scope of the application), a Private Key flow is standard. If you need to act as a specific user (e.g., a supervisor viewing their team’s data), you must use the User Context flow.

This tutorial assumes a Private Key flow for simplicity, as it grants the broadest access for integration testing.

import os
import json
import logging
from datetime import datetime, timedelta
from genesyscloud.platform.client import PlatformClient
from genesyscloud.platform.client import AuthClient
from genesyscloud.conversations.api import ConversationApi
from genesyscloud.analytics.api import AnalyticsApi

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

def get_platform_client(client_id: str, private_key_str: str, org_id: str) -> PlatformClient:
    """
    Initializes and authenticates the Genesys Cloud Platform Client.
    """
    client = PlatformClient()
    auth = client.auth
    
    # Load private key from environment or file
    # private_key_str should be loaded securely, e.g., from a secret manager
    if not private_key_str:
        raise ValueError("Private key not provided")

    # Authenticate using private key
    try:
        auth.authenticate_private_key(
            client_id=client_id,
            private_key=private_key_str,
            org_id=org_id
        )
        logger.info("Authentication successful.")
        return client
    except Exception as e:
        logger.error(f"Authentication failed: {e}")
        raise

# Example Usage
# CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
# PRIVATE_KEY = os.getenv("GENESYS_PRIVATE_KEY") # PEM format string
# ORG_ID = os.getenv("GENESYS_ORG_ID")
# client = get_platform_client(CLIENT_ID, PRIVATE_KEY, ORG_ID)

Critical Note on Scopes: When creating your OAuth application in the Genesys Cloud Admin console, ensure the view:conversation and view:analytics:conversation scopes are checked. Without these, the SDK will return a 403 Forbidden error.

Implementation

Step 1: Understanding the Data Models

Before writing code, you must understand the fundamental difference between the two APIs.

1. /api/v2/conversations (Conversations API):

   • Purpose: Real-time state management.
   • Data: Returns the current state of active conversations. It includes media types (voice, chat, video), current participants, current state (ringing, connected, wrapping), and timestamps for the current session.
   • Retention: Data is transient. Once a conversation ends, it is removed from this endpoint after a short grace period.
   • Use Case: Building a live dashboard, triggering an action when a call connects, or updating CRM records in real-time.

2. /api/v2/analytics/conversations/details/query (Analytics API):

   • Purpose: Historical reporting and querying.
   • Data: Returns aggregated or detailed records of completed conversations. It includes metrics like wait time, handle time, wrap-up time, and final disposition.
   • Retention: Data is stored in the analytics warehouse. It is available for up to 90 days (or longer with specific retention settings).
   • Use Case: Generating end-of-day reports, calculating SLA breaches, or auditing call history.

Rule of Thumb: If the conversation is currently happening, use /api/v2/conversations. If the conversation happened yesterday, last week, or even 5 minutes ago (and is closed), use /api/v2/analytics/conversations.

Step 2: Retrieving Live Conversations

We will use the ConversationApi class to fetch currently active conversations.

Endpoint: GET /api/v2/conversations
Method: get_conversations

from genesyscloud.conversations.api import ConversationApi
from genesyscloud.conversations.model import Conversation

def get_active_conversations(client: PlatformClient, media_type: str = None) -> list[Conversation]:
    """
    Retrieves currently active conversations.
    
    Args:
        client: The authenticated PlatformClient.
        media_type: Optional filter for media type (e.g., 'voice', 'chat').
        
    Returns:
        A list of Conversation objects.
    """
    api_instance = ConversationApi(client)
    
    try:
        # Parameters
        # media_type: Optional. Filter by media type.
        # expand: Optional. Include additional details like 'participants', 'wrapup'.
        
        params = {}
        if media_type:
            params['mediaType'] = media_type
        else:
            params['mediaType'] = 'voice' # Default to voice for this example
            
        params['expand'] = 'participants,wrapup' # Get participant details
        
        logger.info(f"Fetching active conversations for media type: {params['mediaType']}")
        
        response = api_instance.get_conversations(**params)
        
        if response and response.conversations:
            logger.info(f"Found {len(response.conversations)} active conversation(s).")
            return response.conversations
        else:
            logger.info("No active conversations found.")
            return []
            
    except Exception as e:
        logger.error(f"Error fetching active conversations: {e}")
        # Handle specific errors
        if hasattr(e, 'status_code') and e.status_code == 401:
            logger.error("Authentication expired. Please refresh the token.")
        elif hasattr(e, 'status_code') and e.status_code == 403:
            logger.error("Insufficient permissions. Check OAuth scopes.")
        raise

# Example Usage
# active_convs = get_active_conversations(client, media_type='chat')
# for conv in active_convs:
#     print(f"Conversation ID: {conv.id}, State: {conv.state}")

Key Parameters:

• mediaType: Filters by voice, chat, video, webchat, etc.
• expand: Use participants to see who is talking. Use wrapup to see if the agent has finished wrapping up.

Error Handling:

• 401 Unauthorized: The OAuth token has expired. The SDK does not auto-refresh for all flows. You must re-authenticate.
• 429 Too Many Requests: Genesys Cloud enforces rate limits. The SDK includes basic retry logic, but for high-volume polling, implement exponential backoff.

Step 3: Querying Historical Analytics Data

We will use the AnalyticsApi class to query past conversations. This uses a POST request with a query body, which is more flexible than GET parameters.

Endpoint: POST /api/v2/analytics/conversations/details/query
Method: post_analytics_conversations_details_query

from genesyscloud.analytics.api import AnalyticsApi
from genesyscloud.analytics.model import ConversationDetailQueryBody
from datetime import datetime

def get_historical_conversations(client: PlatformClient, start_time: datetime, end_time: datetime, limit: int = 100) -> list:
    """
    Retrieves historical conversation details within a time range.
    
    Args:
        client: The authenticated PlatformClient.
        start_time: Start of the query window.
        end_time: End of the query window.
        limit: Maximum number of records to return (max 10000).
        
    Returns:
        A list of conversation detail objects.
    """
    api_instance = AnalyticsApi(client)
    
    # Define the query body
    # The Analytics API uses a specific JSON structure for queries
    query_body = ConversationDetailQueryBody(
        interval=f"{start_time.isoformat()}/{end_time.isoformat()}",
        view="conversation",
        metrics=["waitTime", "handleTime", "totalTime"],
        group_by=["mediaType", "routingQueueId"],
        size=limit
    )
    
    # To get detailed records (not just aggregates), we need to use the 'details' endpoint
    # Note: The 'details' endpoint returns individual conversation records
    
    try:
        logger.info(f"Querying analytics from {start_time} to {end_time}")
        
        # The SDK method for detailed query
        response = api_instance.post_analytics_conversations_details_query(body=query_body)
        
        if response and response.entities:
            logger.info(f"Retrieved {len(response.entities)} historical conversation record(s).")
            return response.entities
        else:
            logger.info("No historical conversations found for the specified period.")
            return []
            
    except Exception as e:
        logger.error(f"Error querying analytics: {e}")
        if hasattr(e, 'status_code') and e.status_code == 400:
            logger.error("Bad Request. Check the query body structure and date format.")
        raise

# Example Usage
# end_time = datetime.utcnow()
# start_time = end_time - timedelta(hours=24)
# historical_convs = get_historical_conversations(client, start_time, end_time, limit=10)

Critical Distinction in Query Structure:
The Analytics API does not use simple GET parameters. It requires a ConversationDetailQueryBody.

• interval: ISO 8601 formatted time range.
• metrics: The specific data points you want (e.g., waitTime, handleTime).
• size: The maximum number of records to return.

Pagination:
If you need more than limit records, the response includes a nextPageToken. You must pass this token in the subsequent request’s query_body to fetch the next page.

Step 4: Processing Results and Edge Cases

When processing results, you must handle the different data structures.

Live Conversation Object (Conversation):

• id: Unique identifier.
• state: Current state (ringing, connected, wrapping, closed).
• participants: List of participants with their current roles and states.

Analytics Detail Object (ConversationDetail):

• conversationId: Unique identifier (matches the Live ID).
• startTime: When the conversation started.
• endTime: When the conversation ended.
• metrics: Dictionary containing the requested metrics (e.g., waitTime: 12000 ms).

Edge Case: Overlapping Data
There is a small window where a conversation is “closed” in the Live API but not yet available in the Analytics API. Analytics data is processed asynchronously. It may take 1-5 minutes for a closed conversation to appear in the Analytics query. Do not rely on the Analytics API for real-time status checks.

Complete Working Example

This script combines both APIs. It first checks for active voice conversations. If none are found, it queries the last hour of historical voice conversations.

import os
import json
import logging
from datetime import datetime, timedelta
from genesyscloud.platform.client import PlatformClient
from genesyscloud.conversations.api import ConversationApi
from genesyscloud.analytics.api import AnalyticsApi
from genesyscloud.analytics.model import ConversationDetailQueryBody

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

def main():
    # 1. Authentication
    client_id = os.getenv("GENESYS_CLIENT_ID")
    private_key = os.getenv("GENESYS_PRIVATE_KEY")
    org_id = os.getenv("GENESYS_ORG_ID")
    
    if not all([client_id, private_key, org_id]):
        logger.error("Environment variables GENESYS_CLIENT_ID, GENESYS_PRIVATE_KEY, and GENESYS_ORG_ID are required.")
        return

    client = PlatformClient()
    try:
        client.auth.authenticate_private_key(
            client_id=client_id,
            private_key=private_key,
            org_id=org_id
        )
    except Exception as e:
        logger.error(f"Authentication failed: {e}")
        return

    # 2. Fetch Active Conversations
    logger.info("--- Fetching Active Voice Conversations ---")
    conv_api = ConversationApi(client)
    try:
        active_response = conv_api.get_conversations(media_type='voice', expand='participants')
        if active_response and active_response.conversations:
            for conv in active_response.conversations:
                logger.info(f"Active Conv ID: {conv.id}, State: {conv.state}")
                # Check participants
                if conv.participants:
                    for p in conv.participants:
                        logger.info(f"  Participant: {p.name} ({p.role})")
        else:
            logger.info("No active voice conversations found.")
    except Exception as e:
        logger.error(f"Error fetching active conversations: {e}")

    # 3. Fetch Historical Conversations (Last 1 Hour)
    logger.info("--- Fetching Historical Voice Conversations (Last 1 Hour) ---")
    analytics_api = AnalyticsApi(client)
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(hours=1)
    
    query_body = ConversationDetailQueryBody(
        interval=f"{start_time.isoformat()}/{end_time.isoformat()}",
        view="conversation",
        metrics=["waitTime", "handleTime", "totalTime", "queueTime"],
        group_by=[], # No grouping for detailed records
        size=10
    )
    
    try:
        historical_response = analytics_api.post_analytics_conversations_details_query(body=query_body)
        if historical_response and historical_response.entities:
            for entity in historical_response.entities:
                logger.info(f"Historical Conv ID: {entity.conversationId}")
                logger.info(f"  Start: {entity.startTime}, End: {entity.endTime}")
                if entity.metrics:
                    logger.info(f"  Wait Time: {entity.metrics.get('waitTime', 'N/A')} ms")
                    logger.info(f"  Handle Time: {entity.metrics.get('handleTime', 'N/A')} ms")
        else:
            logger.info("No historical voice conversations found in the last hour.")
    except Exception as e:
        logger.error(f"Error fetching historical conversations: {e}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

Cause: The OAuth application lacks the required scopes.
Fix:

1. Go to the Genesys Cloud Admin Console.
2. Navigate to Integrations > OAuth 2.0 Applications.
3. Edit your application.
4. Ensure view:conversation and view:analytics:conversation are checked in the Scopes section.
5. Save and re-authenticate.

Error: 400 Bad Request (Analytics API)

Cause: The interval format is invalid or the view parameter is incorrect.
Fix:

• Ensure the interval is in ISO 8601 format: YYYY-MM-DDTHH:MM:SS.fffZ/YYYY-MM-DDTHH:MM:SS.fffZ.
• Ensure the view is set to conversation for detailed queries.
• Check that metrics are valid for the conversation view.

Error: 429 Too Many Requests

Cause: You have exceeded the rate limit for the API endpoint.
Fix:

• Implement exponential backoff in your code.
• Reduce the frequency of polling the /api/v2/conversations endpoint.
• Use Webhooks (/api/v2/platform/webhooks) for real-time updates instead of polling.

Error: Empty Results from Analytics API

Cause: The conversation has not yet been processed into the analytics warehouse.
Fix:

• Wait 1-5 minutes after the conversation ends.
• Verify the time range in the query body.
• Check if the conversation was filtered out by other criteria (e.g., media type).

Official References

• Genesys Cloud Conversations API Documentation
• Genesys Cloud Analytics Conversations API Documentation
• Genesys Cloud Python SDK Repository
• OAuth 2.0 Authentication Guide