Querying Live Conversations vs. Analytics History in Genesys Cloud CX

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.
• This tutorial uses the Genesys Cloud CX REST API v2 and the genesys-cloud-sdk-python package.
• The implementation covers Python 3.9+ with httpx for raw HTTP requests to illustrate the underlying mechanics.

Prerequisites

• OAuth Client: A Genesys Cloud CX OAuth Client with confidential access type.
• Required Scopes:
  • For Conversations API: conversation:read, conversation:monitoring:read
  • For Analytics API: analytics:conversation:read, analytics:interaction:read
• SDK Version: genesys-cloud-sdk-python >= 130.0.0
• Language/Runtime: Python 3.9+
• External Dependencies:
  • httpx (for raw HTTP examples)
  • python-dotenv (for credential management)
  • genesys-cloud-sdk-python

Authentication Setup

Genesys Cloud CX uses OAuth 2.0. The Conversations API and Analytics API share the same authentication mechanism, but they serve fundamentally different purposes. The Conversations API operates on the live event bus, while the Analytics API queries the data warehouse.

You must obtain a bearer token before making any requests.

import httpx
import os
from dotenv import load_dotenv

load_dotenv()

CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")
ORGANIZATION_ID = os.getenv("GENESYS_ORG_ID")

def get_access_token() -> str:
    """
    Obtains an OAuth 2.0 bearer token from Genesys Cloud.
    """
    url = f"https://api.mypurecloud.com/oauth/token"
    
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    
    data = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET
    }
    
    # Use httpx for synchronous requests in this example
    with httpx.Client() as client:
        response = client.post(url, headers=headers, data=data)
        
        if response.status_code != 200:
            raise Exception(f"Failed to obtain token: {response.text}")
            
        token_data = response.json()
        return token_data["access_token"]

ACCESS_TOKEN = get_access_token()
AUTH_HEADERS = {
    "Authorization": f"Bearer {ACCESS_TOKEN}",
    "Content-Type": "application/json",
    "Accept": "application/json"
}

Implementation

Step 1: Understanding the Conversations API (Real-Time)

The /api/v2/conversations endpoints are designed for operational use cases. They return data about conversations that are currently active or have recently concluded (typically within the last 24-48 hours, depending on the specific endpoint).

Use this API when you need to:

• Monitor active calls in a dashboard.
• Trigger webhooks based on real-time conversation events.
• Retrieve metadata for a conversation that is happening right now.

Endpoint: GET /api/v2/conversations

This endpoint returns a list of conversations. It does not return detailed metrics (like handle time) unless you drill down into specific conversation types (e.g., /api/v2/conversations/calls/{id}).

def list_active_conversations(token: str) -> dict:
    """
    Retrieves a list of active conversations.
    Scope required: conversation:read
    """
    url = "https://api.mypurecloud.com/api/v2/conversations"
    
    params = {
        "pageSize": 100,
        "pageNumber": 1
    }
    
    with httpx.Client() as client:
        response = client.get(url, headers=AUTH_HEADERS, params=params)
        
        if response.status_code == 401:
            raise Exception("Unauthorized. Token is invalid or expired.")
        if response.status_code == 403:
            raise Exception("Forbidden. Check OAuth scopes.")
        if response.status_code == 429:
            # Retry logic should be implemented in production
            raise Exception("Rate limited. Wait before retrying.")
            
        return response.json()

# Execute
active_convs = list_active_conversations(ACCESS_TOKEN)
print(f"Found {len(active_convs.get('entities', []))} active conversations.")

Expected Response Structure:

{
  "entities": [
    {
      "id": "conv-12345-67890",
      "type": "call",
      "createdTimestamp": "2023-10-27T10:00:00.000Z",
      "updatedTimestamp": "2023-10-27T10:05:00.000Z",
      "participants": [
        {
          "id": "part-111",
          "routing": {
            "queueName": "Support Queue",
            "state": "connected"
          }
        }
      ]
    }
  ],
  "pageSize": 100,
  "pageNumber": 1,
  "pageCount": 1
}

Step 2: Understanding the Analytics API (Historical)

The /api/v2/analytics/conversations endpoints are designed for reporting use cases. They query the Genesys Cloud data warehouse. Data availability typically has a delay of 1-4 hours (real-time analytics may have lower latency, but detailed analytics are batched).

Use this API when you need to:

• Generate end-of-day reports.
• Calculate metrics like Average Handle Time (AHT), Abandon Rate, or Service Level.
• Retrieve conversation data older than 48 hours.

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

Unlike the Conversations API, the Analytics API uses a query body rather than query parameters. This allows for complex filtering and aggregation.

def query_historical_conversations(token: str) -> dict:
    """
    Retrieves detailed historical conversation data.
    Scope required: analytics:conversation:read
    """
    url = "https://api.mypurecloud.com/api/v2/analytics/conversations/details/query"
    
    # Define the time range (ISO 8601 format)
    # Analytics data is not available for the current minute; use a past range.
    query_body = {
        "interval": "2023-10-26T00:00:00.000Z/2023-10-26T23:59:59.999Z",
        "groupBy": ["conversationType"],
        "metrics": [
            "handled",
            "abandoned",
            "total",
            "serviceLevel"
        ],
        "filter": {
            "type": "and",
            "clauses": [
                {
                    "type": "eq",
                    "field": "routing.queue.name",
                    "value": "Support Queue"
                }
            ]
        }
    }
    
    with httpx.Client() as client:
        response = client.post(url, headers=AUTH_HEADERS, json=query_body)
        
        if response.status_code == 400:
            print(f"Bad Request: {response.text}")
            raise Exception("Invalid query body. Check interval format or metric names.")
        if response.status_code == 401:
            raise Exception("Unauthorized.")
            
        return response.json()

# Execute
analytics_data = query_historical_conversations(ACCESS_TOKEN)
print(f"Analytics result keys: {analytics_data.keys()}")

Expected Response Structure:

{
  "interval": "2023-10-26T00:00:00.000Z/2023-10-26T23:59:59.999Z",
  "groupBy": ["conversationType"],
  "metrics": [
    "handled",
    "abandoned",
    "total",
    "serviceLevel"
  ],
  "results": [
    {
      "conversationType": "call",
      "metrics": {
        "handled": {
          "value": 150,
          "unit": "count"
        },
        "abandoned": {
          "value": 5,
          "unit": "count"
        },
        "total": {
          "value": 155,
          "unit": "count"
        },
        "serviceLevel": {
          "value": 0.85,
          "unit": "ratio"
        }
      }
    }
  ]
}

Step 3: Processing Results and Handling Pagination

Both APIs support pagination, but they handle it differently.

Conversations API Pagination:
Uses pageSize and pageNumber or nextPageId. It is cursor-based or page-based depending on the endpoint.

Analytics API Pagination:
Uses nextPageToken in the response. You must include this token in the nextPageToken field of your next request body.

def get_all_analytics_pages(token: str, initial_query: dict) -> list:
    """
    Iterates through all pages of analytics results.
    """
    all_results = []
    current_query = initial_query.copy()
    
    with httpx.Client() as client:
        while True:
            response = client.post(
                "https://api.mypurecloud.com/api/v2/analytics/conversations/details/query",
                headers=AUTH_HEADERS,
                json=current_query
            )
            
            if response.status_code != 200:
                raise Exception(f"Analytics query failed: {response.status_code} {response.text}")
                
            data = response.json()
            all_results.extend(data.get("results", []))
            
            # Check for next page
            next_page_token = data.get("nextPageToken")
            if not next_page_token:
                break
                
            current_query["nextPageToken"] = next_page_token
            
    return all_results

Complete Working Example

This script demonstrates both APIs side-by-side. It first checks for active conversations, then queries historical data for the previous day.

import httpx
import os
import json
from datetime import datetime, timedelta
from dotenv import load_dotenv

load_dotenv()

CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")

def get_token() -> str:
    url = "https://api.mypurecloud.com/oauth/token"
    data = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET
    }
    with httpx.Client() as client:
        res = client.post(url, data=data, headers={"Content-Type": "application/x-www-form-urlencoded"})
        res.raise_for_status()
        return res.json()["access_token"]

def fetch_active_conversations(token: str) -> list:
    """Fetches active conversations using Conversations API."""
    url = "https://api.mypurecloud.com/api/v2/conversations"
    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/json"
    }
    params = {"pageSize": 20}
    
    with httpx.Client() as client:
        res = client.get(url, headers=headers, params=params)
        if res.status_code == 401:
            raise Exception("Token Invalid")
        res.raise_for_status()
        return res.json().get("entities", [])

def fetch_historical_metrics(token: str, queue_name: str = "Support Queue") -> dict:
    """Fetches historical metrics using Analytics API."""
    url = "https://api.mypurecloud.com/api/v2/analytics/conversations/details/query"
    
    # Define yesterday's date range
    yesterday = datetime.utcnow() - timedelta(days=1)
    start = yesterday.replace(hour=0, minute=0, second=0, microsecond=0).isoformat() + "Z"
    end = yesterday.replace(hour=23, minute=59, second=59, microsecond=999999).isoformat() + "Z"
    
    query = {
        "interval": f"{start}/{end}",
        "groupBy": ["conversationType"],
        "metrics": ["handled", "abandoned"],
        "filter": {
            "type": "and",
            "clauses": [
                {
                    "type": "eq",
                    "field": "routing.queue.name",
                    "value": queue_name
                }
            ]
        }
    }
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }
    
    with httpx.Client() as client:
        res = client.post(url, headers=headers, json=query)
        if res.status_code == 400:
            print(f"Query Error: {res.text}")
            return {}
        res.raise_for_status()
        return res.json()

def main():
    token = get_token()
    
    # 1. Real-Time Check
    print("--- Real-Time Conversations ---")
    try:
        active = fetch_active_conversations(token)
        print(f"Active Conversations: {len(active)}")
        if active:
            print(f"Sample ID: {active[0]['id']}")
    except Exception as e:
        print(f"Error fetching active conversations: {e}")
        
    # 2. Historical Check
    print("\n--- Historical Analytics (Yesterday) ---")
    try:
        history = fetch_historical_metrics(token)
        if history.get("results"):
            for r in history["results"]:
                print(f"Type: {r['conversationType']}")
                print(f"  Handled: {r['metrics']['handled']['value']}")
                print(f"  Abandoned: {r['metrics']['abandoned']['value']}")
        else:
            print("No historical data found for the specified filter.")
    except Exception as e:
        print(f"Error fetching analytics: {e}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 400 Bad Request on Analytics API

Cause: The interval format is incorrect, or the requested metrics are not valid for the selected groupBy fields.
Fix: Ensure the interval is in ISO 8601 format with a / separator. Verify that the metrics exist in the Genesys Cloud Analytics dictionary. Use the Analytics API explorer in the developer portal to validate metric names.

Error: 401 Unauthorized on Conversations API

Cause: The OAuth token is missing the conversation:read scope.
Fix: Regenerate the token with the correct scopes. Check the OAuth client configuration in Genesys Cloud Admin > Security > OAuth Clients.

Error: 429 Too Many Requests

Cause: You have exceeded the rate limit for the API endpoint. Conversations API has a lower rate limit than Analytics API.
Fix: Implement exponential backoff. Do not poll the Conversations API more than once every 5-10 seconds per tenant. For Analytics, use caching for repeated queries.

Error: Empty Results in Analytics

Cause: The time interval is too recent (data delay) or the filter is too restrictive.
Fix: Query data from at least 24 hours ago. Broaden the filter to check if any data exists at all.

Official References

• Genesys Cloud Conversations API Documentation
• Genesys Cloud Analytics API Documentation
• Python SDK Reference