Querying Conversations in Real-Time vs. Analytics: Choosing Between /conversations and /analytics

What You Will Build

• You will build two distinct Python scripts that retrieve conversation data from Genesys Cloud.
• One script uses the Real-Time Conversations API (/api/v2/conversations) to fetch active, in-progress interactions.
• The other script uses the Analytics API (/api/v2/analytics/conversations/details/query) to fetch historical, aggregated interaction data.
• The language covered is Python, utilizing the purecloudplatformclientv2 SDK and the requests library for direct HTTP calls.

Prerequisites

• OAuth Client Type: Service Account (Client Credentials Flow) or User Account (Authorization Code Flow).
• Required Scopes:
  • For Real-Time: conversation:read
  • For Analytics: analytics:read
• SDK Version: purecloudplatformclientv2 >= 168.0.0
• Runtime: Python 3.9+
• Dependencies:

  pip install purecloudplatformclientv2 requests python-dotenv
  

Authentication Setup

Before querying either endpoint, you must obtain a valid OAuth 2.0 access token. The Real-Time API and the Analytics API operate on different security boundaries and require different scopes. Using a token with the wrong scope will result in a 403 Forbidden response, even if the authentication itself is valid.

Below is a reusable function to authenticate using the Client Credentials flow. This is suitable for service accounts running backend integrations.

import os
import requests
from purecloudplatformclientv2 import (
    Configuration,
    ApiClient,
    ConversationApi,
    AnalyticsConversationsApi
)
from purecloudplatformclientv2.rest import ApiException

def get_auth_token(client_id: str, client_secret: str, scope: str) -> str:
    """
    Authenticates with Genesys Cloud using Client Credentials flow.
    
    Args:
        client_id: Your OAuth Client ID
        client_secret: Your OAuth Client Secret
        scope: The specific OAuth scope required for the operation
        
    Returns:
        Access token string
    """
    url = "https://login.mypurecloud.com/oauth/token"
    
    payload = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
        "scope": scope
    }
    
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    
    response = requests.post(url, data=payload, headers=headers)
    
    if response.status_code == 200:
        return response.json().get("access_token")
    else:
        raise Exception(f"Authentication failed: {response.status_code} - {response.text}")

def get_api_client(token: str) -> ApiClient:
    """
    Initializes the Genesys Cloud SDK ApiClient with the provided token.
    """
    configuration = Configuration()
    configuration.access_token = token
    api_client = ApiClient(configuration)
    return api_client

Implementation

Step 1: Understanding the Real-Time Conversations API (/api/v2/conversations)

The /api/v2/conversations endpoint is designed for operational visibility. It returns conversations that are currently active (live) or recently closed (within a short retention window, typically seconds to minutes, depending on system load). This API is used for screen pops, real-time agent assist, or immediate status updates.

Key Characteristics:

• Latency: Low (near real-time).
• Data Depth: Shallow. It contains the conversation ID, type, state, and participants. It does not contain detailed metrics like handle time, queue wait times, or wrap-up codes unless you drill down into specific sub-entities.
• Pagination: Supports cursor-based pagination for large volumes of active conversations.
• Scopes: conversation:read.

Code Example: Fetching Active Conversations

def list_active_conversations(api_client: ApiClient, max_records: int = 100) -> dict:
    """
    Retrieves a list of currently active conversations.
    
    Args:
        api_client: Initialized PureCloud ApiClient
        max_records: Maximum number of records to return
        
    Returns:
        Dictionary containing the conversation list
    """
    conversation_api = ConversationApi(api_client)
    
    try:
        # The SDK method corresponds to GET /api/v2/conversations
        # We use expand=None to keep the payload small for demonstration
        response = conversation_api.post_conversations_query(
            body={}, # Empty body returns default set
            max_records=max_records
        )
        
        return response.entities
        
    except ApiException as e:
        print(f"Exception when calling ConversationApi->post_conversations_query: {e}\n")
        if e.status == 401:
            print("Error: Invalid or expired token. Check your OAuth scope.")
        elif e.status == 403:
            print("Error: Forbidden. Ensure the token has 'conversation:read' scope.")
        return []

Expected Response Structure:
The response is a list of Conversation objects. Each object contains:

• id: The unique conversation identifier.
• type: e.g., “voice”, “chat”, “email”.
• state: e.g., “connected”, “queued”, “closed”.
• participants: A list of participant objects with their roles and states.

Error Handling:

• 401 Unauthorized: Token is expired or malformed.
• 403 Forbidden: The token lacks the conversation:read scope.
• 429 Too Many Requests: You have exceeded the rate limit. Implement exponential backoff.

Step 2: Understanding the Analytics Conversations API (/api/v2/analytics/conversations/details/query)

The /api/v2/analytics/conversations/details/query endpoint is designed for historical analysis. It queries the Genesys Cloud data warehouse. This API is used for reporting, compliance reviews, quality assurance, and post-call analytics.

Key Characteristics:

• Latency: High. Data can take 15-30 minutes to appear in Analytics after a conversation ends.
• Data Depth: Deep. You can request specific metrics (handle time, acw time, queue time), attributes, and segments.
• Pagination: Supports offset-based pagination and large result sets (up to 10,000 records per request, with cursors for more).
• Scopes: analytics:read.

Code Example: Querying Historical Conversations

def query_historical_conversations(api_client: ApiClient, start_time: str, end_time: str) -> dict:
    """
    Queries historical conversation data from the Analytics API.
    
    Args:
        api_client: Initialized PureCloud ApiClient
        start_time: ISO 8601 start timestamp (e.g., "2023-10-01T00:00:00Z")
        end_time: ISO 8601 end timestamp (e.g., "2023-10-01T23:59:59Z")
        
    Returns:
        Dictionary containing the query results
    """
    analytics_api = AnalyticsConversationsApi(api_client)
    
    # Define the query body
    # This is critical: you must specify what data you want
    query_body = {
        "groupBy": [], # No grouping; we want individual conversation details
        "interval": None, # No time intervals; we want raw events
        "dateRange": {
            "from": start_time,
            "to": end_time
        },
        "filters": {
            "types": ["voice"] # Only fetch voice conversations
        },
        "metrics": [
            "handleTime",
            "acwTime",
            "queueTime",
            "wrapUpCode"
        ],
        "segments": [], # No segments for this basic query
        "maxRecords": 100, # Limit results for demonstration
        "offset": 0
    }
    
    try:
        # The SDK method corresponds to POST /api/v2/analytics/conversations/details/query
        response = analytics_api.post_analytics_conversations_details_query(
            body=query_body
        )
        
        return response
        
    except ApiException as e:
        print(f"Exception when calling AnalyticsConversationsApi->post_analytics_conversations_details_query: {e}\n")
        if e.status == 401:
            print("Error: Invalid or expired token. Check your OAuth scope.")
        elif e.status == 403:
            print("Error: Forbidden. Ensure the token has 'analytics:read' scope.")
        elif e.status == 400:
            print("Error: Bad Request. Check your query body syntax.")
        return None

Expected Response Structure:
The response is a QueryResponse object. Key fields include:

• total: Total number of records matching the query.
• page: Current page number.
• pageSize: Number of records per page.
• entities: A list of ConversationDetails objects. Each object contains the requested metrics and segments.

Error Handling:

• 400 Bad Request: The query body is malformed or contains invalid date ranges.
• 401/403: Authentication/Authorization issues.
• 429 Rate Limit: Analytics queries are resource-intensive. Monitor your usage.

Step 3: Comparing Data Models and Use Cases

To understand when to use which API, you must look at the data model differences.

Real-Time (/conversations):

• Use Case: An agent is on a call. You need to know the customer’s email address to pop their CRM record. You query /api/v2/conversations/{id} to get the participant details, then use the email to fetch CRM data.
• Data Model: Focuses on state. state: "connected", direction: "inbound".
• Limitation: You cannot calculate average handle time (AHT) using this API because the call is not finished.

Analytics (/analytics/conversations):

• Use Case: A manager needs to review all calls from yesterday where the handle time exceeded 5 minutes. You query the Analytics API with a filter on handleTime > 300000 (milliseconds).
• Data Model: Focuses on metrics. handleTime: 350000, acwTime: 120000.
• Limitation: You cannot use this API to update the state of a live conversation. It is read-only and delayed.

Complete Working Example

Below is a complete Python script that demonstrates both APIs. It authenticates, fetches active conversations, and then fetches historical analytics data for the last hour.

import os
import time
from datetime import datetime, timedelta
from purecloudplatformclientv2 import (
    Configuration,
    ApiClient,
    ConversationApi,
    AnalyticsConversationsApi
)
from purecloudplatformclientv2.rest import ApiException
import requests

# Load environment variables
CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")

def get_auth_token(client_id: str, client_secret: str, scope: str) -> str:
    url = "https://login.mypurecloud.com/oauth/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
        "scope": scope
    }
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    response = requests.post(url, data=payload, headers=headers)
    if response.status_code == 200:
        return response.json().get("access_token")
    else:
        raise Exception(f"Auth failed: {response.status_code} - {response.text}")

def get_api_client(token: str) -> ApiClient:
    configuration = Configuration()
    configuration.access_token = token
    return ApiClient(configuration)

def fetch_realtime_conversations(api_client: ApiClient):
    print("--- Fetching Real-Time Conversations ---")
    conversation_api = ConversationApi(api_client)
    try:
        # Get first 10 active conversations
        response = conversation_api.post_conversations_query(body={}, max_records=10)
        
        if response.entities:
            for conv in response.entities:
                print(f"ID: {conv.id}, Type: {conv.type}, State: {conv.state}")
        else:
            print("No active conversations found.")
    except ApiException as e:
        print(f"Real-Time API Error: {e}")

def fetch_analytics_conversations(api_client: ApiClient):
    print("\n--- Fetching Historical Analytics ---")
    analytics_api = AnalyticsConversationsApi(api_client)
    
    # Define time range: Last 1 hour
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(hours=1)
    
    start_str = start_time.strftime("%Y-%m-%dT%H:%M:%SZ")
    end_str = end_time.strftime("%Y-%m-%dT%H:%M:%SZ")
    
    query_body = {
        "groupBy": [],
        "interval": None,
        "dateRange": {"from": start_str, "to": end_str},
        "filters": {"types": ["voice"]},
        "metrics": ["handleTime", "acwTime"],
        "segments": [],
        "maxRecords": 5,
        "offset": 0
    }
    
    try:
        response = analytics_api.post_analytics_conversations_details_query(body=query_body)
        
        if response.entities:
            print(f"Total records found: {response.total}")
            for conv in response.entities:
                handle_time = conv.metrics.get("handleTime", 0)
                print(f"ID: {conv.id}, Handle Time (ms): {handle_time}")
        else:
            print("No historical conversations found in the last hour.")
    except ApiException as e:
        print(f"Analytics API Error: {e}")

def main():
    if not CLIENT_ID or not CLIENT_SECRET:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET environment variables are required.")
    
    # 1. Authenticate for Real-Time
    rt_token = get_auth_token(CLIENT_ID, CLIENT_SECRET, "conversation:read")
    rt_client = get_api_client(rt_token)
    fetch_realtime_conversations(rt_client)
    
    # 2. Authenticate for Analytics
    analytics_token = get_auth_token(CLIENT_ID, CLIENT_SECRET, "analytics:read")
    analytics_client = get_api_client(analytics_token)
    fetch_analytics_conversations(analytics_client)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden on Analytics Query

• Cause: The OAuth token does not have the analytics:read scope. This is the most common error when switching from Real-Time to Analytics.
• Fix: Ensure you request the correct scope during token generation. If using a single token for both, request conversation:read analytics:read.

Error: 400 Bad Request on Analytics Query

• Cause: The dateRange is invalid or the metrics array contains unsupported field names.
• Fix: Validate your ISO 8601 timestamps. Ensure from is before to. Check the Analytics API documentation for valid metric names.

Error: Empty Response from Real-Time API

• Cause: There are no active conversations, or the conversation state is closed and has been purged from the real-time buffer.
• Fix: Real-time data is ephemeral. If you need data from a conversation that ended 5 minutes ago, use the Analytics API.

Error: 429 Too Many Requests

• Cause: You have exceeded the rate limit for either the Conversations or Analytics API.
• Fix: Implement exponential backoff. For Analytics, reduce the frequency of queries or cache results locally.

Official References

• Genesys Cloud Real-Time Conversations API Documentation
• Genesys Cloud Analytics Conversations API Documentation
• Genesys Cloud Python SDK Reference