Choosing Between Conversations API and Analytics API for Real-Time vs Historical Data

What You Will Build

• You will build a Python script that retrieves active conversation details using the Conversations API and historical conversation metrics using the Analytics API.
• This tutorial uses the Genesys Cloud PureCloud Platform Client V2 SDK and raw HTTP requests for clarity.
• The programming language covered is Python 3.9+.

Prerequisites

• OAuth Client Type: Public or Confidential client with appropriate scopes.
• Required Scopes:
  • conversation:read for accessing active conversations.
  • analytics:conversation:read for accessing historical analytics data.
• SDK Version: genesyscloud-py-client >= 1.0.0.
• Language/Runtime: Python 3.9 or later.
• External Dependencies:
  • genesyscloud-py-client
  • requests
  • python-dotenv (for secure credential management)

Authentication Setup

Genesys Cloud uses OAuth 2.0 for authentication. You must obtain an access token before making any API calls. For production applications, implement token caching and automatic refresh. For this tutorial, we will use the SDK’s built-in authentication helper which handles token retrieval and storage.

Create a .env file in your project root:

GENESYS_CLOUD_REGION=us-east-1
GENESYS_CLOUD_CLIENT_ID=your_client_id
GENESYS_CLOUD_CLIENT_SECRET=your_client_secret
GENESYS_CLOUD_ENVIRONMENT=us-east-1

Install the required packages:

pip install genesyscloud-py-client requests python-dotenv

Initialize the SDK in your Python script:

import os
from dotenv import load_dotenv
from purecloudplatformclientv2 import ApiClient, Configuration, Environment
from purecloudplatformclientv2.rest import ApiException

load_dotenv()

def get_purecloud_client():
    """
    Initializes the PureCloud API client using environment variables.
    """
    try:
        # Load environment variables
        client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
        client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
        environment = os.getenv("GENESYS_CLOUD_ENVIRONMENT", "us-east-1")
        
        # Configure the client
        configuration = Configuration()
        configuration.host = Environment.get_host(environment)
        
        # Initialize API client with OAuth2 client credentials flow
        api_client = ApiClient(configuration=configuration)
        api_client.configuration.access_token = api_client.authenticator.get_access_token(
            client_id=client_id,
            client_secret=client_secret
        )
        
        return api_client
        
    except Exception as e:
        print(f"Failed to initialize PureCloud client: {e}")
        raise

# Get the authenticated client
api_client = get_purecloud_client()

Implementation

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

The Conversations API provides real-time access to currently active conversations. It is designed for operational purposes, such as:

• Monitoring live interactions.
• Providing agent assist features.
• Updating conversation metadata during an interaction.
• Retrieving participant details for active sessions.

This API does not store historical data. Once a conversation ends, it is no longer accessible via this endpoint. The data is transient and exists only while the conversation is active.

Use Case: Retrieve Active Conversations for a Specific User

We will retrieve all active conversations for a specific user (e.g., an agent).

from purecloudplatformclientv2 import ConversationApi

def get_active_conversations_for_user(user_id: str, api_client: ApiClient) -> list:
    """
    Retrieves all active conversations for a specific user.
    
    Args:
        user_id: The ID of the user whose active conversations to retrieve.
        api_client: The initialized PureCloud API client.
        
    Returns:
        A list of active conversation objects.
    """
    conversation_api = ConversationApi(api_client)
    
    try:
        # Get active conversations for the user
        response = conversation_api.get_conversations_user_conversations(
            user_id=user_id,
            expand=["participants", "wraps"]  # Expand related entities
        )
        
        return response.entities if response.entities else []
        
    except ApiException as e:
        if e.status == 401:
            print("Authentication failed. Check your OAuth token.")
        elif e.status == 403:
            print("Forbidden. Ensure your client has 'conversation:read' scope.")
        elif e.status == 404:
            print(f"User {user_id} not found.")
        else:
            print(f"API Error {e.status}: {e.body}")
        raise

# Example usage
USER_ID = "your_user_id_here"
active_conversations = get_active_conversations_for_user(USER_ID, api_client)

for conv in active_conversations:
    print(f"Active Conversation ID: {conv.id}")
    print(f"  Type: {conv.type}")
    print(f"  State: {conv.state}")
    print(f"  Created: {conv.created_time}")
    if conv.participants:
        for participant in conv.participants:
            print(f"    Participant: {participant.user.name} ({participant.role})")
    print("-" * 50)

Expected Response Structure (JSON):

{
  "entities": [
    {
      "id": "conv-12345678-1234-1234-1234-123456789012",
      "type": "voice",
      "state": "connected",
      "created_time": "2023-10-27T10:30:00.000Z",
      "participants": [
        {
          "user": {
            "id": "user-123",
            "name": "Agent Smith"
          },
          "role": "agent"
        },
        {
          "external_contact": {
            "phone_number": "+15551234567"
          },
          "role": "customer"
        }
      ]
    }
  ],
  "page_count": 1,
  "page_size": 25
}

Error Handling:

• 401 Unauthorized: Your OAuth token is expired or invalid. Refresh the token.
• 403 Forbidden: Your OAuth client lacks the conversation:read scope.
• 404 Not Found: The specified user_id does not exist.

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

The Analytics API provides historical data about conversations. It is designed for reporting, business intelligence, and post-interaction analysis. This API queries the data warehouse, which has a slight delay (typically 15-30 minutes) compared to real-time data.

This API supports:

• Aggregated metrics (e.g., average handle time, total conversations).
• Detailed conversation records with full metadata.
• Time-range filtering (e.g., last 7 days, last month).
• Complex filtering by queue, skill, user, or custom attributes.

Use Case: Retrieve Historical Conversation Details for the Last 7 Days

We will retrieve detailed conversation records for the last 7 days.

from purecloudplatformclientv2 import ConversationAnalyticsApi
from datetime import datetime, timedelta
from purecloudplatformclientv2.models import ConversationQueryRequest

def get_historical_conversations(api_client: ApiClient, days_back: int = 7) -> list:
    """
    Retrieves historical conversation details for the specified number of days.
    
    Args:
        api_client: The initialized PureCloud API client.
        days_back: Number of days to look back for conversation data.
        
    Returns:
        A list of historical conversation objects.
    """
    conversation_analytics_api = ConversationAnalyticsApi(api_client)
    
    # Define the time range
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(days=days_back)
    
    # Format times as ISO 8601 strings
    start_time_str = start_time.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
    end_time_str = end_time.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
    
    # Build the query request
    query_request = ConversationQueryRequest(
        date_from=start_time_str,
        date_to=end_time_str,
        size=25,  # Page size
        sort_by="created_time",
        sort_order="desc"
    )
    
    try:
        # Execute the query
        response = conversation_analytics_api.post_analytics_conversations_details_query(
            body=query_request
        )
        
        return response.entities if response.entities else []
        
    except ApiException as e:
        if e.status == 401:
            print("Authentication failed. Check your OAuth token.")
        elif e.status == 403:
            print("Forbidden. Ensure your client has 'analytics:conversation:read' scope.")
        elif e.status == 422:
            print("Unprocessable Entity. Check your query parameters.")
        else:
            print(f"API Error {e.status}: {e.body}")
        raise

# Example usage
historical_conversations = get_historical_conversations(api_client, days_back=7)

for conv in historical_conversations:
    print(f"Historical Conversation ID: {conv.id}")
    print(f"  Type: {conv.type}")
    print(f"  Created: {conv.created_time}")
    print(f"  Duration: {conv.duration_seconds} seconds")
    if conv.queue:
        print(f"  Queue: {conv.queue.name}")
    if conv.agents:
        for agent in conv.agents:
            print(f"  Agent: {agent.user.name}")
    print("-" * 50)

Expected Response Structure (JSON):

{
  "page_count": 1,
  "page_size": 25,
  "entities": [
    {
      "id": "conv-87654321-4321-4321-4321-210987654321",
      "type": "voice",
      "created_time": "2023-10-26T15:45:00.000Z",
      "duration_seconds": 120.5,
      "queue": {
        "id": "queue-123",
        "name": "Support Team"
      },
      "agents": [
        {
          "user": {
            "id": "user-456",
            "name": "Agent Doe"
          }
        }
      ],
      "wrap_up_code": "Resolved",
      "hold_time_seconds": 10.2,
      "talk_time_seconds": 110.3
    }
  ]
}

Error Handling:

• 401 Unauthorized: Your OAuth token is expired or invalid.
• 403 Forbidden: Your OAuth client lacks the analytics:conversation:read scope.
• 422 Unprocessable Entity: Your query parameters are invalid (e.g., invalid date format, missing required fields).
• 429 Too Many Requests: You have exceeded the rate limit. Implement exponential backoff.

Step 3: Processing Results and Pagination

Both APIs support pagination. The Conversations API returns a page_count and page_size, but for active conversations, the result set is usually small. The Analytics API can return large datasets, so pagination is critical.

Handling Pagination for Analytics API

from purecloudplatformclientv2.models import ConversationQueryRequest

def get_all_historical_conversations(api_client: ApiClient, days_back: int = 7) -> list:
    """
    Retrieves all historical conversation details with pagination handling.
    
    Args:
        api_client: The initialized PureCloud API client.
        days_back: Number of days to look back for conversation data.
        
    Returns:
        A list of all historical conversation objects.
    """
    conversation_analytics_api = ConversationAnalyticsApi(api_client)
    
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(days=days_back)
    
    start_time_str = start_time.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
    end_time_str = end_time.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
    
    all_conversations = []
    page_number = 1
    
    while True:
        query_request = ConversationQueryRequest(
            date_from=start_time_str,
            date_to=end_time_str,
            page=page_number,
            size=100,  # Max page size is 250, but 100 is safer for stability
            sort_by="created_time",
            sort_order="desc"
        )
        
        try:
            response = conversation_analytics_api.post_analytics_conversations_details_query(
                body=query_request
            )
            
            if not response.entities:
                break
                
            all_conversations.extend(response.entities)
            
            # Check if there are more pages
            if page_number >= response.page_count:
                break
                
            page_number += 1
            
        except ApiException as e:
            print(f"API Error on page {page_number}: {e.status} - {e.body}")
            break
            
    return all_conversations

# Example usage
all_historical_conversations = get_all_historical_conversations(api_client, days_back=30)
print(f"Total conversations retrieved: {len(all_historical_conversations)}")

Complete Working Example

Below is a complete, runnable Python script that demonstrates both APIs.

import os
from dotenv import load_dotenv
from purecloudplatformclientv2 import ApiClient, Configuration, Environment, ConversationApi, ConversationAnalyticsApi
from purecloudplatformclientv2.rest import ApiException
from purecloudplatformclientv2.models import ConversationQueryRequest
from datetime import datetime, timedelta
import json

load_dotenv()

def get_purecloud_client():
    """Initializes the PureCloud API client."""
    try:
        client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
        client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
        environment = os.getenv("GENESYS_CLOUD_ENVIRONMENT", "us-east-1")
        
        configuration = Configuration()
        configuration.host = Environment.get_host(environment)
        
        api_client = ApiClient(configuration=configuration)
        api_client.configuration.access_token = api_client.authenticator.get_access_token(
            client_id=client_id,
            client_secret=client_secret
        )
        
        return api_client
        
    except Exception as e:
        print(f"Failed to initialize PureCloud client: {e}")
        raise

def get_active_conversations(user_id: str, api_client: ApiClient) -> list:
    """Retrieves active conversations for a user."""
    conversation_api = ConversationApi(api_client)
    try:
        response = conversation_api.get_conversations_user_conversations(
            user_id=user_id,
            expand=["participants"]
        )
        return response.entities if response.entities else []
    except ApiException as e:
        print(f"Error fetching active conversations: {e.status} - {e.body}")
        return []

def get_historical_conversations(api_client: ApiClient, days_back: int = 7) -> list:
    """Retrieves historical conversations for the last N days."""
    conversation_analytics_api = ConversationAnalyticsApi(api_client)
    
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(days=days_back)
    
    query_request = ConversationQueryRequest(
        date_from=start_time.strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
        date_to=end_time.strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
        size=25,
        sort_by="created_time",
        sort_order="desc"
    )
    
    try:
        response = conversation_analytics_api.post_analytics_conversations_details_query(
            body=query_request
        )
        return response.entities if response.entities else []
    except ApiException as e:
        print(f"Error fetching historical conversations: {e.status} - {e.body}")
        return []

def main():
    # Initialize client
    api_client = get_purecloud_client()
    
    # Define user ID for active conversations
    USER_ID = "your_user_id_here"
    
    # Fetch active conversations
    print("Fetching Active Conversations...")
    active_convs = get_active_conversations(USER_ID, api_client)
    
    if active_convs:
        print(f"Found {len(active_convs)} active conversation(s).")
        for conv in active_convs:
            print(json.dumps({
                "id": conv.id,
                "type": conv.type,
                "state": conv.state,
                "created_time": conv.created_time.isoformat()
            }, indent=2))
    else:
        print("No active conversations found.")
    
    print("\n" + "="*50 + "\n")
    
    # Fetch historical conversations
    print("Fetching Historical Conversations (Last 7 Days)...")
    historical_convs = get_historical_conversations(api_client, days_back=7)
    
    if historical_convs:
        print(f"Found {len(historical_convs)} historical conversation(s).")
        for conv in historical_convs:
            print(json.dumps({
                "id": conv.id,
                "type": conv.type,
                "created_time": conv.created_time.isoformat(),
                "duration_seconds": conv.duration_seconds,
                "queue_name": conv.queue.name if conv.queue else None
            }, indent=2))
    else:
        print("No historical conversations found.")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

Cause: The OAuth access token is expired, invalid, or missing.

Fix: Ensure your get_purecloud_client function correctly retrieves a new token. The SDK’s get_access_token method handles this, but if you are managing tokens manually, implement a refresh flow.

# If managing tokens manually
if api_client.configuration.access_token_expired:
    api_client.configuration.access_token = api_client.authenticator.get_access_token(
        client_id=client_id,
        client_secret=client_secret
    )

Error: 403 Forbidden

Cause: The OAuth client lacks the required scope.

Fix:

• For /api/v2/conversations, ensure the client has conversation:read.
• For /api/v2/analytics/conversations, ensure the client has analytics:conversation:read.

Check the client’s scopes in the Genesys Cloud Admin portal under Administration > Security > OAuth Clients.

Error: 422 Unprocessable Entity

Cause: Invalid query parameters in the Analytics API request.

Fix: Verify that date_from and date_to are valid ISO 8601 strings. Ensure the time range does not exceed the maximum allowed range (typically 90 days for detailed queries).

# Validate dates
if (end_time - start_time).days > 90:
    raise ValueError("Date range exceeds maximum allowed limit of 90 days.")

Error: 429 Too Many Requests

Cause: You have exceeded the API rate limit.

Fix: Implement exponential backoff. Do not retry immediately. Wait for a period that increases with each retry (e.g., 1s, 2s, 4s, 8s).

import time
import requests

def make_request_with_retry(url: str, headers: dict, max_retries: int = 3) -> requests.Response:
    """Makes an HTTP request with exponential backoff for 429 errors."""
    for attempt in range(max_retries):
        response = requests.get(url, headers=headers)
        if response.status_code == 429:
            wait_time = 2 ** attempt
            print(f"Rate limited. Waiting {wait_time} seconds...")
            time.sleep(wait_time)
        else:
            return response
    raise Exception("Max retries exceeded for 429 error.")

Official References

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