Choosing Between Real-Time Conversations and Historical Analytics APIs in Genesys Cloud

What You Will Build

• You will build a Python script that retrieves active conversation details for immediate programmatic action using the Conversations API.
• You will build a separate Python script that queries historical conversation metrics and details for reporting using the Analytics API.
• You will implement logic to determine which API endpoint is appropriate based on the latency requirements and data granularity of your use case.

Prerequisites

• OAuth Client: A Genesys Cloud Private Client or Public Client with the following scopes:
  • conversation:read (for /api/v2/conversations)
  • analytics:conversation:read (for /api/v2/analytics/conversations/details/query)
• SDK: genesys-cloud-purecloud-platform-client version 140.0.0 or higher.
• Language: Python 3.9+.
• External Dependencies:
  • python-dotenv for managing credentials.
  • pandas (optional, for processing analytics data frames).

Install the required packages:

pip install genesys-cloud-purecloud-platform-client python-dotenv pandas

Authentication Setup

Genesys Cloud uses OAuth 2.0 Bearer tokens. All API calls require a valid Authorization: Bearer <token> header. The token expires after 1 hour. For production applications, you must implement token caching and refresh logic. This tutorial uses the SDK’s built-in authentication helper, which handles the initial token fetch.

Create a .env file in your project root:

GENESYS_REGION=us-east-1
GENESYS_CLIENT_ID=your_client_id_here
GENESYS_CLIENT_SECRET=your_client_secret_here

Initialize the client in your Python script:

import os
from dotenv import load_dotenv
from purecloud_platform_client import ApiClient, Configuration, ConversationApi, AnalyticsApi

load_dotenv()

def get_api_client():
    """
    Initializes and returns an authenticated PureCloud API client.
    """
    config = Configuration(
        host=os.getenv("GENESYS_REGION"),
        client_id=os.getenv("GENESYS_CLIENT_ID"),
        client_secret=os.getenv("GENESYS_CLIENT_SECRET")
    )
    
    with ApiClient(config) as client:
        # The SDK automatically fetches the OAuth token
        return client

# Example usage
api_client = get_api_client()

Implementation

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

The Conversations API is a real-time REST interface. It exposes the current state of active interactions (voice, chat, message, video). This API is designed for low-latency operations where you need to know what is happening right now.

Use cases:

• Detecting when an agent is currently on a call to trigger a screen pop.
• Retrieving the latest transcript of an ongoing chat session.
• Programmatically transferring or merging active calls.

Key Characteristic: It does not store history. Once a conversation ends, it is removed from this API’s scope. You cannot query conversations that ended 5 minutes ago using this endpoint.

Code Example: Fetching Active Conversations

We will use the ConversationApi to list all active conversations.

from purecloud_platform_client.rest import ApiException

def get_active_conversations():
    """
    Retrieves a list of currently active conversations.
    """
    with get_api_client() as client:
        api_instance = ConversationApi(client)
        
        try:
            # Fetch active conversations
            # The SDK handles pagination automatically if you iterate, 
            # but for a simple list, we fetch the first page.
            response = api_instance.get_conversations(
                limit=25,
                expand=["transcript", "wrapupcode"] # Optional: expand specific fields
            )
            
            if response.entities:
                print(f"Found {len(response.entities)} active conversations.")
                for conv in response.entities:
                    print(f"ID: {conv.id}, Type: {conv.type}, State: {conv.state}")
                    # Access real-time details
                    if conv.type == "chat" and conv.transcript:
                        last_msg = conv.transcript[-1] if conv.transcript else None
                        if last_msg:
                            print(f"  Last Message: {last_msg.text}")
            else:
                print("No active conversations found.")
                
        except ApiException as e:
            print(f"Exception when calling ConversationApi->get_conversations: {e}")
            # Handle 401 Unauthorized (expired token) or 403 Forbidden (missing scope)
            if e.status == 401:
                print("Token expired. Refresh required.")
            elif e.status == 403:
                print("Check 'conversation:read' scope.")

get_active_conversations()

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

The Analytics API is a data warehouse interface. It aggregates historical data from Genesys Cloud. It is not real-time. There is a delay (typically 15-30 minutes) between an event occurring in the system and it becoming queryable in Analytics.

Use cases:

• Generating weekly performance reports for agents.
• Calculating average handle time (AHT) over the last quarter.
• Analyzing conversation outcomes (resolved vs. unresolved) for trend analysis.

Key Characteristic: It provides aggregated metrics (counts, averages, sums) and detailed historical records (if requested via the details query). It supports time-series data and complex filtering by date range, skill, queue, or custom attributes.

Code Example: Querying Historical Conversation Details

We will use the AnalyticsApi to query historical conversation details. This requires constructing a specific request body (QueryConversationDetailsRequest).

from purecloud_platform_client.rest import ApiException
from purecloud_platform_client.models import QueryConversationDetailsRequest

def get_historical_conversations():
    """
    Queries historical conversation details from the last 24 hours.
    """
    with get_api_client() as client:
        api_instance = AnalyticsApi(client)
        
        try:
            # Construct the query body
            body = QueryConversationDetailsRequest(
                date_from="2023-10-01T00:00:00.000Z", # Replace with dynamic date
                date_to="2023-10-02T00:00:00.000Z",   # Replace with dynamic date
                filter=[
                    {
                        "type": "attribute",
                        "attribute": "conversationType",
                        "operator": "eq",
                        "value": "voice"
                    }
                ],
                expand=["wrapupcode", "acdskill", "queuename", "agent"]
            )
            
            # Execute the query
            # Note: This endpoint returns a job ID first if the dataset is large,
            # but for small datasets, it may return data synchronously.
            # The SDK handles the polling for asynchronous jobs automatically.
            response = api_instance.post_analytics_conversations_details_query(body)
            
            if response.entities:
                print(f"Retrieved {len(response.entities)} historical conversations.")
                for conv in response.entities:
                    print(f"ID: {conv.id}, Start Time: {conv.start_time}, Duration: {conv.duration_seconds}s")
                    if conv.agent:
                        print(f"  Agent: {conv.agent.name}")
            else:
                print("No historical conversations found matching the criteria.")
                
        except ApiException as e:
            print(f"Exception when calling AnalyticsApi->post_analytics_conversations_details_query: {e}")
            if e.status == 400:
                print("Bad Request. Check date format and filter syntax.")
            elif e.status == 403:
                print("Check 'analytics:conversation:read' scope.")

get_historical_conversations()

Step 3: Comparing Data Granularity and Latency

To make the correct choice, you must evaluate two factors: Latency and Granularity.

Latency Comparison

Granularity Comparison

Complete Working Example

The following script demonstrates a hybrid approach. It first checks for any active voice conversations. If none are found, it falls back to retrieving the most recent completed voice conversation from the analytics store to display in a “Last Call” widget.

import os
from datetime import datetime, timedelta
from dotenv import load_dotenv
from purecloud_platform_client import ApiClient, Configuration, ConversationApi, AnalyticsApi
from purecloud_platform_client.rest import ApiException
from purecloud_platform_client.models import QueryConversationDetailsRequest

load_dotenv()

def get_api_client():
    config = Configuration(
        host=os.getenv("GENESYS_REGION"),
        client_id=os.getenv("GENESYS_CLIENT_ID"),
        client_secret=os.getenv("GENESYS_CLIENT_SECRET")
    )
    return ApiClient(config)

def get_active_voice_conversations():
    """Fetches all active voice conversations."""
    client = get_api_client()
    api = ConversationApi(client)
    
    try:
        response = api.get_conversations(limit=100, expand=["participants"])
        voice_calls = [c for c in response.entities if c.type == "voice"]
        return voice_calls
    except ApiException as e:
        print(f"Error fetching active conversations: {e}")
        return []
    finally:
        client.close()

def get_last_completed_voice_conversation():
    """Fetches the most recent completed voice conversation from Analytics."""
    client = get_api_client()
    api = AnalyticsApi(client)
    
    # Define time window: Last 24 hours
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(hours=24)
    
    try:
        body = QueryConversationDetailsRequest(
            date_from=start_time.isoformat() + "Z",
            date_to=end_time.isoformat() + "Z",
            filter=[
                {
                    "type": "attribute",
                    "attribute": "conversationType",
                    "operator": "eq",
                    "value": "voice"
                },
                {
                    "type": "attribute",
                    "attribute": "conversationState",
                    "operator": "eq",
                    "value": "completed"
                }
            ],
            expand=["agent", "wrapupcode"],
            sort_by="start_time",
            sort_order="desc",
            max_records=1 # Only need the most recent one
        )
        
        response = api.post_analytics_conversations_details_query(body)
        
        if response.entities:
            return response.entities[0]
        return None
        
    except ApiException as e:
        print(f"Error fetching analytics data: {e}")
        return None
    finally:
        client.close()

def main():
    print("--- Checking Real-Time Conversations ---")
    active_calls = get_active_voice_conversations()
    
    if active_calls:
        print(f"Found {len(active_calls)} active voice call(s).")
        for call in active_calls:
            print(f"  Call ID: {call.id}")
            # Extract agent name from participants
            agents = [p for p in call.participants if p.role == "agent"]
            if agents:
                print(f"  Agent: {agents[0].user.name}")
    else:
        print("No active voice calls. Fetching last completed call from Analytics...")
        print("--- Checking Historical Analytics ---")
        last_call = get_last_completed_voice_conversation()
        
        if last_call:
            print(f"Last Completed Call ID: {last_call.id}")
            print(f"  Start Time: {last_call.start_time}")
            print(f"  Duration: {last_call.duration_seconds} seconds")
            if last_call.agent:
                print(f"  Agent: {last_call.agent.name}")
        else:
            print("No completed voice calls found in the last 24 hours.")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden on Analytics API

Cause: You are using a client with insufficient scopes. The Conversations API requires conversation:read, but the Analytics API requires analytics:conversation:read. These are separate scopes.

Fix:

1. Go to the Genesys Cloud Admin Console.
2. Navigate to Developers > OAuth Clients.
3. Edit your client.
4. Add analytics:conversation:read to the Scopes list.
5. Save and regenerate your client secret if necessary.

Error: 429 Too Many Requests

Cause: You have exceeded the rate limit for either the Conversations API or the Analytics API. Analytics queries are expensive and have lower rate limits than standard REST calls.

Fix: Implement exponential backoff. Do not poll the Analytics API in a tight loop.

import time

def retry_with_backoff(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
                print(f"Rate limited. Waiting {wait_time} seconds before retry...")
                time.sleep(wait_time)
            else:
                raise
    raise Exception("Max retries exceeded")

Error: Empty Response from Analytics API

Cause: The data has not yet been processed. Analytics data is not real-time. If you query for a conversation that ended 2 minutes ago, it will not appear.

Fix:

1. Verify the date_from and date_to parameters are in ISO 8601 format with a ‘Z’ suffix.
2. Ensure the time window includes the event.
3. If testing with recent data, wait at least 30 minutes after the conversation ends before querying Analytics. For immediate data, use the Conversations API.

Error: “Invalid Filter” in Analytics Query

Cause: The filter attribute name is incorrect. Analytics uses specific internal attribute names, not always the UI labels.

Fix:

• Use conversationType for filtering by voice/chat/message.
• Use conversationState for filtering by active/completed.
• Use agent.id or agent.name for agent filtering.
• Refer to the Analytics Conversation Details Documentation for the complete list of valid attributes.

Official References

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