Choose the Right Genesys Cloud API: Real-Time Conversations vs. Analytics Data

What You Will Build

• You will build a Python application that retrieves active conversation details using the real-time API and historical conversation metrics using the analytics API.
• This tutorial uses the Genesys Cloud CX REST API v2 and the official genesys-cloud-python SDK.
• The implementation is written in Python 3.9+ using the requests library for raw HTTP examples and the SDK for structured data access.

Prerequisites

Before writing code, verify your environment meets these requirements:

• OAuth Client ID and Secret: You must have a Genesys Cloud application created in the Admin console.
• Required Scopes:
  • For /api/v2/conversations: conversation:view, conversation:participant:write (if modifying state).
  • For /api/v2/analytics/conversations/details/query: analytics:conversation:view.
• Python Environment: Python 3.9 or later.
• Dependencies:

  pip install requests genesys-cloud-python python-dotenv
  

• Environment Variables: Create a .env file with your credentials:

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

Authentication Setup

Genesys Cloud APIs require OAuth 2.0 client credentials flow. The token is short-lived (typically 1 hour), so production code must handle refresh or re-authentication. The following helper class manages this lifecycle.

import os
import time
import requests
from dotenv import load_dotenv

load_dotenv()

class GenesysAuth:
    def __init__(self, region: str = "us-east-1"):
        self.region = region
        self.client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
        self.client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
        self.auth_base_url = f"https://api.{region}.mypurecloud.com"
        self.access_token = None
        self.token_expiry = 0

    def get_token(self) -> str:
        # Check if we have a valid cached token
        if self.access_token and time.time() < self.token_expiry:
            return self.access_token

        # Request new token
        auth_url = f"{self.auth_base_url}/oauth/token"
        data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": "conversation:view analytics:conversation:view"
        }

        try:
            response = requests.post(auth_url, data=data)
            response.raise_for_status()
            token_data = response.json()
            
            self.access_token = token_data["access_token"]
            # Set expiry to slightly before actual expiry to avoid race conditions
            self.token_expiry = time.time() + (token_data["expires_in"] - 60)
            
            return self.access_token
        except requests.exceptions.HTTPError as e:
            print(f"Authentication failed: {e.response.status_code} - {e.response.text}")
            raise

# Initialize auth
auth = GenesysAuth()

This setup ensures that every subsequent API call uses a valid token without requiring manual token management in your business logic.

Implementation

The core distinction between the two endpoints lies in state versus history.

• /api/v2/conversations is operational. It returns the current state of active interactions (voice, chat, message, video). It is used for real-time actions: transferring, recording, adding participants, or updating presence.
• /api/v2/analytics/conversations/details/query is historical. It returns aggregated or detailed records of completed interactions. It is used for reporting, billing, quality assurance, and post-call analysis.

Step 1: Retrieving Real-Time Conversations

When you need to know what is happening right now, you query the conversations API. This endpoint supports filtering by user, queue, or conversation type.

Endpoint: GET /api/v2/conversations
Scope: conversation:view

import requests
from typing import List, Dict, Any

def get_active_conversations(auth: GenesysAuth, limit: int = 20) -> List[Dict[str, Any]]:
    """
    Retrieves a list of currently active conversations.
    """
    base_url = f"https://api.{auth.region}.mypurecloud.com"
    endpoint = "/api/v2/conversations"
    
    headers = {
        "Authorization": f"Bearer {auth.get_token()}",
        "Content-Type": "application/json"
    }
    
    params = {
        "limit": limit,
        "expand": "participants" # Include participant details in the response
    }
    
    try:
        response = requests.get(f"{base_url}{endpoint}", headers=headers, params=params)
        response.raise_for_status()
        return response.json()["entities"]
    except requests.exceptions.HTTPError as e:
        if response.status_code == 429:
            print("Rate limited. Wait and retry.")
            time.sleep(1) # Simple backoff
            return get_active_conversations(auth, limit)
        else:
            print(f"Error fetching conversations: {e.response.status_code}")
            raise

# Usage
active_convs = get_active_conversations(auth)
if active_convs:
    print(f"Found {len(active_convs)} active conversations.")
    for conv in active_convs:
        print(f"ID: {conv['id']}, Type: {conv['type']}, State: {conv['state']}")
else:
    print("No active conversations found.")

Key Response Fields:

• id: The unique identifier for the conversation.
• type: voice, chat, message, email, sms, or video.
• state: active, closed, ended, or unknown. For real-time API, active is the primary state of interest.
• participants: An array of objects containing userId, name, and state (e.g., ringing, talking, listening).

Why use this endpoint?
Use this when building a supervisor dashboard that shows who is talking now, or an IVR script that needs to check if an agent is currently on a call before routing.

Step 2: Querying Historical Analytics Data

When a conversation ends, its data moves to the analytics store. You cannot retrieve a closed conversation via /api/v2/conversations. You must use the analytics query API.

Endpoint: POST /api/v2/analytics/conversations/details/query
Scope: analytics:conversation:view

This endpoint uses a complex JSON body to define the query. It supports date ranges, filters, and sorting.

import json
from datetime import datetime, timedelta

def get_historical_conversations(auth: GenesysAuth, days_back: int = 7) -> Dict[str, Any]:
    """
    Queries completed conversations from the last N days.
    """
    base_url = f"https://api.{auth.region}.mypurecloud.com"
    endpoint = "/api/v2/anversations/details/query" # Note: Correct path is /analytics/conversations/details/query
    
    # Fix endpoint path
    endpoint = "/api/v2/analytics/conversations/details/query"

    headers = {
        "Authorization": f"Bearer {auth.get_token()}",
        "Content-Type": "application/json"
    }

    # Define the time range
    end_time = datetime.utcnow().isoformat() + "Z"
    start_time = (datetime.utcnow() - timedelta(days=days_back)).isoformat() + "Z"

    # Construct the query body
    query_body = {
        "dateFrom": start_time,
        "dateTo": end_time,
        "interval": "PT1H", # Aggregate by hour (optional, but recommended for large datasets)
        "view": "conversationDetail", # Specific view for detailed records
        "select": [
            "conversationId",
            "conversationType",
            "start",
            "end",
            "duration",
            "wrapupCode",
            "userId",
            "userName"
        ],
        "where": [
            {
                "path": "conversationType",
                "constraint": "voice" # Filter for voice calls only
            },
            {
                "path": "state",
                "constraint": "closed" # Only closed conversations
            }
        ],
        "size": 100 # Max records per page
    }

    try:
        response = requests.post(
            f"{base_url}{endpoint}", 
            headers=headers, 
            json=query_body
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        print(f"Analytics query failed: {e.response.status_code}")
        print(e.response.text)
        raise

# Usage
analytics_data = get_historical_conversations(auth, days_back=1)
if "entities" in analytics_data:
    print(f"Retrieved {len(analytics_data['entities'])} historical records.")
    for entity in analytics_data["entities"][:5]: # Show first 5
        print(f"Call ID: {entity.get('conversationId')}, Duration: {entity.get('duration')}")
else:
    print("No historical data found in the specified range.")

Key Differences in Request:

1. Method: POST instead of GET. The query parameters are too complex for a URL string.
2. Body: Uses a structured JSON object with select, where, and view clauses.
3. Data Scope: Returns only closed or ended conversations. Active conversations will not appear here until they terminate.

Why use this endpoint?
Use this for generating end-of-day reports, calculating Average Handle Time (AHT), or exporting call logs for compliance auditing.

Step 3: Handling Pagination and Large Datasets

The analytics API returns paginated results. If you request more data than the size limit (max 1000 for details query), you must handle the nextPageToken.

def get_all_historical_conversations(auth: GenesysAuth, days_back: int = 1) -> List[Dict]:
    """
    Fetches all pages of historical conversations.
    """
    base_url = f"https://api.{auth.region}.mypurecloud.com"
    endpoint = "/api/v2/analytics/conversations/details/query"
    headers = {
        "Authorization": f"Bearer {auth.get_token()}",
        "Content-Type": "application/json"
    }

    end_time = datetime.utcnow().isoformat() + "Z"
    start_time = (datetime.utcnow() - timedelta(days=days_back)).isoformat() + "Z"

    base_query = {
        "dateFrom": start_time,
        "dateTo": end_time,
        "view": "conversationDetail",
        "select": ["conversationId", "duration"],
        "where": [{"path": "state", "constraint": "closed"}],
        "size": 100
    }

    all_records = []
    page_token = None

    while True:
        query = base_query.copy()
        if page_token:
            query["pageToken"] = page_token

        response = requests.post(f"{base_url}{endpoint}", headers=headers, json=query)
        response.raise_for_status()
        data = response.json()

        if "entities" in data:
            all_records.extend(data["entities"])
        
        # Check for next page
        if data.get("nextPageToken"):
            page_token = data["nextPageToken"]
            # Add a small delay to respect rate limits
            time.sleep(0.5)
        else:
            break

    return all_records

# Usage
all_calls = get_all_historical_conversations(auth)
print(f"Total historical records fetched: {len(all_calls)}")

This pattern ensures you capture the entire dataset without manual intervention.

Complete Working Example

Combine the authentication and query logic into a single script that demonstrates both real-time and historical access.

import os
import time
import requests
from datetime import datetime, timedelta
from typing import List, Dict, Any
from dotenv import load_dotenv

load_dotenv()

class GenesysAPI:
    def __init__(self, region: str = "us-east-1"):
        self.region = region
        self.client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
        self.client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
        self.auth_base_url = f"https://api.{region}.mypurecloud.com"
        self.base_url = f"https://api.{region}.mypurecloud.com"
        self.access_token = None
        self.token_expiry = 0

    def get_token(self) -> str:
        if self.access_token and time.time() < self.token_expiry:
            return self.access_token

        auth_url = f"{self.auth_base_url}/oauth/token"
        data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": "conversation:view analytics:conversation:view"
        }

        try:
            response = requests.post(auth_url, data=data)
            response.raise_for_status()
            token_data = response.json()
            self.access_token = token_data["access_token"]
            self.token_expiry = time.time() + (token_data["expires_in"] - 60)
            return self.access_token
        except requests.exceptions.HTTPError as e:
            print(f"Auth Failed: {e.response.text}")
            raise

    def get_active_conversations(self) -> List[Dict[str, Any]]:
        headers = {"Authorization": f"Bearer {self.get_token()}"}
        params = {"limit": 5, "expand": "participants"}
        
        response = requests.get(
            f"{self.base_url}/api/v2/conversations", 
            headers=headers, 
            params=params
        )
        response.raise_for_status()
        return response.json().get("entities", [])

    def get_historical_conversations(self, days_back: int = 1) -> List[Dict]:
        headers = {"Authorization": f"Bearer {self.get_token()}", "Content-Type": "application/json"}
        
        end_time = datetime.utcnow().isoformat() + "Z"
        start_time = (datetime.utcnow() - timedelta(days=days_back)).isoformat() + "Z"

        query_body = {
            "dateFrom": start_time,
            "dateTo": end_time,
            "view": "conversationDetail",
            "select": ["conversationId", "conversationType", "duration"],
            "where": [{"path": "state", "constraint": "closed"}],
            "size": 10
        }

        response = requests.post(
            f"{self.base_url}/api/v2/analytics/conversations/details/query",
            headers=headers,
            json=query_body
        )
        response.raise_for_status()
        return response.json().get("entities", [])

if __name__ == "__main__":
    api = GenesysAPI()
    
    print("--- Real-Time Conversations ---")
    try:
        active = api.get_active_conversations()
        if active:
            for conv in active:
                print(f"Active: {conv['id']} ({conv['type']})")
        else:
            print("No active conversations.")
    except Exception as e:
        print(f"Error fetching active: {e}")

    print("\n--- Historical Conversations (Last 1 Day) ---")
    try:
        history = api.get_historical_conversations(days_back=1)
        if history:
            for rec in history:
                print(f"History: {rec.get('conversationId')} ({rec.get('conversationType')})")
        else:
            print("No historical conversations found.")
    except Exception as e:
        print(f"Error fetching history: {e}")

Common Errors & Debugging

Error: 403 Forbidden

Cause: The OAuth token lacks the required scope.
Fix: Ensure your client application in Genesys Cloud Admin has the conversation:view scope for real-time APIs and analytics:conversation:view for analytics APIs. Regenerate the token after updating scopes.

Error: 429 Too Many Requests

Cause: You exceeded the API rate limit. The conversations API has a lower rate limit than analytics.
Fix: Implement exponential backoff. Do not retry immediately. Wait 1 second, then 2, then 4, etc.

def retry_with_backoff(func, *args, max_retries=3, **kwargs):
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429:
                wait_time = 2 ** attempt
                print(f"Rate limited. Waiting {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise

Error: Empty Entities in Analytics Query

Cause: The date range is in the future, or the state filter excludes all records.
Fix: Verify dateFrom and dateTo are in ISO 8601 format with Z suffix. Ensure state is set to closed for historical data. If you query for active state in analytics, you will get no results because active conversations are not in the analytics store yet.

Error: Conversation ID Not Found in Analytics

Cause: The conversation is still active or recently closed. Analytics data is not real-time. There is typically a latency of 1-5 minutes before a closed conversation appears in the analytics API.
Fix: If you need immediate data after a call ends, use the Real-Time API until the state changes to closed, then switch to Analytics for long-term storage. Do not poll analytics for recent calls.

Official References

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