When to Use Conversations API vs Analytics API in Genesys Cloud

What You Will Build

• A Python script that retrieves real-time interaction metadata using the Conversations API and historical statistical aggregates using the Analytics API.
• This tutorial uses the Genesys Cloud REST API endpoints /api/v2/conversations and /api/v2/analytics/conversations/details/query.
• The implementation covers Python 3.9+ using the requests library and the official genesyscloud SDK.

Prerequisites

• OAuth Client Type: A Genesys Cloud OAuth Client with “Confidential” or “Public” client type.
• Required Scopes:
  • conversations:read (for /api/v2/conversations)
  • analytics:conversations:query (for /api/v2/analytics/conversations/details/query)
  • analytics:report:query (optional, for higher-level metric summaries)
• SDK Version: genesyscloud Python SDK v2.0+ (or direct REST calls).
• Runtime: Python 3.9 or later.
• Dependencies: pip install requests genesyscloud

Authentication Setup

Genesys Cloud uses OAuth 2.0. For server-to-server integrations, the Client Credentials flow is the standard. You must obtain an access token before making any API calls. The token expires after 3600 seconds (1 hour) and must be refreshed or re-requested.

The following class handles token acquisition and basic caching. It ensures that your application does not request a new token on every API call, which helps avoid rate-limiting.

import requests
import time
from typing import Optional

class GenesysAuth:
    def __init__(self, client_id: str, client_secret: str, org_id: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.org_id = org_id
        self.token_url = f"https://login.mypurecloud.com/oauth/token"
        self._token: Optional[str] = None
        self._token_expiry: float = 0

    def get_token(self) -> str:
        """
        Retrieves an OAuth token. Returns cached token if valid.
        """
        # Refresh if token is expired or does not exist
        if not self._token or time.time() >= self._token_expiry:
            self._refresh_token()
        return self._token

    def _refresh_token(self):
        """
        Performs the OAuth2 Client Credentials grant flow.
        """
        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }
        data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        try:
            response = requests.post(self.token_url, headers=headers, data=data)
            response.raise_for_status()
            token_data = response.json()
            
            self._token = token_data["access_token"]
            # Set expiry slightly before actual expiry to avoid edge-case failures
            self._token_expiry = time.time() + token_data["expires_in"] - 10
            
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise Exception("Invalid Client ID or Secret.")
            elif response.status_code == 429:
                raise Exception("Rate limited on token request. Wait and retry.")
            else:
                raise Exception(f"Failed to get token: {e}")
        except Exception as e:
            raise Exception(f"Unexpected error during authentication: {e}")

Implementation

Step 1: Understanding the Data Models

Before writing code, you must understand the fundamental difference in data structure.

1. Conversations API (/api/v2/conversations):

   • Purpose: Real-time or near-real-time state management.
   • Data Granularity: Individual interactions (calls, chats, emails).
   • Structure: Returns a Conversation object containing participants, wrapupCode, state, and type (e.g., “voice”, “chat”).
   • Use Case: Updating a wrap-up code, transferring a call, fetching the current status of an active chat, or logging a specific event.

2. Analytics API (/api/v2/analytics/conversations/details/query):

   • Purpose: Historical reporting and statistical analysis.
   • Data Granularity: Aggregated metrics or detailed historical records (with a delay).
   • Structure: Returns ConversationDetails or aggregated Metric objects. Includes fields like queueWaitTime, acwTime, totalHoldTime, and outcome.
   • Use Case: Calculating Average Handle Time (AHT), generating daily performance reports, or auditing completed interactions from last week.

Critical Distinction: The Conversations API is for control and current state. The Analytics API is for measurement and history. Never use Analytics to control a live call. Never use Conversations to generate a monthly report (it will time out or return incomplete data).

Step 2: Fetching Real-Time Conversation State

Use the Conversations API when you need to know what is happening right now or need to modify a live interaction.

We will query for all active conversations of a specific type (e.g., “voice”) for a specific user.

Required Scope: conversations:read

def get_active_conversations(auth: GenesysAuth, user_id: str, conversation_type: str = "voice"):
    """
    Retrieves active conversations for a specific user.
    """
    base_url = f"https://{auth.org_id}.mypurecloud.com/api/v2/conversations"
    
    headers = {
        "Authorization": f"Bearer {auth.get_token()}",
        "Accept": "application/json"
    }
    
    params = {
        "type": conversation_type,
        "state": "active",  # Only fetch active conversations
        "userId": user_id
    }

    try:
        response = requests.get(base_url, headers=headers, params=params)
        response.raise_for_status()
        
        data = response.json()
        conversations = data.get("entities", [])
        
        print(f"Found {len(conversations)} active {conversation_type} conversations.")
        
        for conv in conversations:
            print(f"Conversation ID: {conv['id']}")
            print(f"State: {conv.get('state', 'Unknown')}")
            
            # Access participant details
            participants = conv.get("participants", [])
            for p in participants:
                if p.get("userId") == user_id:
                    print(f"  User Role: {p.get('role')}")
                    print(f"  Wrapup Code: {p.get('wrapupCode')}")
                    
        return conversations
        
    except requests.exceptions.HTTPError as e:
        if response.status_code == 403:
            print("Error: Missing 'conversations:read' scope.")
        elif response.status_code == 404:
            print("Error: User not found or no active conversations.")
        else:
            print(f"HTTP Error: {e}")
    except Exception as e:
        print(f"Error fetching conversations: {e}")
        return []

Key Parameters:

• type: Filter by “voice”, “chat”, “email”, “sms”, or “task”.
• state: Filter by “active”, “queued”, “waiting”, or “closed”. Note that “closed” conversations are only retained in the Conversations API for a short period (usually 24-48 hours) before being archived. For older data, you must use Analytics.
• userId: Restricts results to a specific user. If omitted, it returns all active conversations in the organization (use with caution due to pagination and rate limits).

Step 3: Querying Historical Analytics Data

Use the Analytics API when you need to analyze completed interactions. This endpoint accepts a JSON body defining the time range, filters, and metrics.

Required Scope: analytics:conversations:query

import json

def get_historical_analytics(auth: GenesysAuth, user_id: str, start_date: str, end_date: str):
    """
    Queries detailed conversation analytics for a specific user and time range.
    Dates must be in ISO 8601 format (e.g., "2023-10-01T00:00:00Z").
    """
    base_url = f"https://{auth.org_id}.mypurecloud.com/api/v2/analytics/conversations/details/query"
    
    headers = {
        "Authorization": f"Bearer {auth.get_token()}",
        "Accept": "application/json",
        "Content-Type": "application/json"
    }
    
    # Define the query body
    query_body = {
        "dateFrom": start_date,
        "dateTo": end_date,
        "interval": "P1D",  # Daily interval
        "filter": [
            {
                "type": "user",
                "id": user_id
            }
        ],
        "groupBy": [
            "user"
        ],
        "metrics": [
            "acwTime",
            "callDuration",
            "queueTime",
            "talkTime",
            "wrapUpCode"
        ],
        "limit": 100
    }

    try:
        response = requests.post(base_url, headers=headers, json=query_body)
        response.raise_for_status()
        
        data = response.json()
        
        # Analytics API returns a 'partitions' structure
        partitions = data.get("partitions", [])
        
        if not partitions:
            print("No analytics data found for the specified range.")
            return []
        
        for partition in partitions:
            print(f"Date Interval: {partition.get('dateFrom')} to {partition.get('dateTo')}")
            
            groups = partition.get("groups", [])
            for group in groups:
                print(f"User: {group.get('user', {}).get('name', 'Unknown')}")
                
                # Extract metrics
                metrics = group.get("metrics", {})
                print(f"  Average Call Duration: {metrics.get('callDuration', {}).get('average', 0):.2f} seconds")
                print(f"  Average ACW Time: {metrics.get('acwTime', {}).get('average', 0):.2f} seconds")
                print(f"  Total Calls: {metrics.get('callDuration', {}).get('count', 0)}")
                
        return data
        
    except requests.exceptions.HTTPError as e:
        if response.status_code == 400:
            print("Error: Invalid query body. Check date formats and metric names.")
        elif response.status_code == 403:
            print("Error: Missing 'analytics:conversations:query' scope.")
        else:
            print(f"HTTP Error: {e}")
    except Exception as e:
        print(f"Error fetching analytics: {e}")
        return []

Key Parameters:

• dateFrom / dateTo: The time window for the query. Analytics data is not real-time; there is typically a delay of 15-30 minutes for data to appear.
• interval: How to bucket the data (e.g., “P1D” for daily, “PT1H” for hourly).
• filter: Restricts data to specific users, queues, or skills.
• metrics: The specific data points to retrieve. Common metrics include callDuration, queueTime, talkTime, acwTime (After Call Work), and holdTime.

Step 4: Handling Pagination in Analytics

The Analytics API enforces strict pagination. The limit parameter controls the number of records per page. You must check the nextPageToken to fetch subsequent pages.

def get_all_analytics_pages(auth: GenesysAuth, user_id: str, start_date: str, end_date: str):
    """
    Iterates through all pages of analytics data.
    """
    base_url = f"https://{auth.org_id}.mypurecloud.com/api/v2/analytics/conversations/details/query"
    headers = {
        "Authorization": f"Bearer {auth.get_token()}",
        "Accept": "application/json",
        "Content-Type": "application/json"
    }
    
    query_body = {
        "dateFrom": start_date,
        "dateTo": end_date,
        "interval": "P1D",
        "filter": [{"type": "user", "id": user_id}],
        "groupBy": ["user"],
        "metrics": ["callDuration", "acwTime"],
        "limit": 100
    }
    
    all_data = []
    next_page_token = None
    page_count = 1
    
    while True:
        if next_page_token:
            query_body["pageToken"] = next_page_token
            
        try:
            response = requests.post(base_url, headers=headers, json=query_body)
            response.raise_for_status()
            
            data = response.json()
            all_data.extend(data.get("partitions", []))
            
            next_page_token = data.get("nextPageToken")
            print(f"Fetched page {page_count}. Total partitions so far: {len(all_data)}")
            
            if not next_page_token:
                break
                
            page_count += 1
            
        except requests.exceptions.HTTPError as e:
            print(f"Error on page {page_count}: {e}")
            break
            
    return all_data

Complete Working Example

This script combines authentication, real-time status checking, and historical analytics retrieval. Replace the placeholder credentials with your actual Genesys Cloud OAuth client details.

import requests
import time
from typing import Optional, List, Dict, Any

class GenesysClient:
    def __init__(self, client_id: str, client_secret: str, org_id: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.org_id = org_id
        self.base_url = f"https://{org_id}.mypurecloud.com"
        self.token_url = f"https://login.mypurecloud.com/oauth/token"
        self._token: Optional[str] = None
        self._token_expiry: float = 0

    def _get_headers(self, include_auth: bool = True) -> Dict[str, str]:
        headers = {
            "Content-Type": "application/json",
            "Accept": "application/json"
        }
        if include_auth:
            headers["Authorization"] = f"Bearer {self._get_token()}"
        return headers

    def _get_token(self) -> str:
        if not self._token or time.time() >= self._token_expiry:
            self._refresh_token()
        return self._token

    def _refresh_token(self):
        data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }
        response = requests.post(self.token_url, data=data)
        response.raise_for_status()
        token_data = response.json()
        self._token = token_data["access_token"]
        self._token_expiry = time.time() + token_data["expires_in"] - 10

    def get_active_voice_conversations(self, user_id: str) -> List[Dict[str, Any]]:
        """
        Retrieves active voice conversations for a user.
        Uses /api/v2/conversations
        """
        url = f"{self.base_url}/api/v2/conversations"
        params = {
            "type": "voice",
            "state": "active",
            "userId": user_id
        }
        
        response = requests.get(url, headers=self._get_headers(), params=params)
        response.raise_for_status()
        return response.json().get("entities", [])

    def get_user_analytics(self, user_id: str, date_from: str, date_to: str) -> Dict[str, Any]:
        """
        Retrieves historical analytics for a user.
        Uses /api/v2/analytics/conversations/details/query
        """
        url = f"{self.base_url}/api/v2/analytics/conversations/details/query"
        
        body = {
            "dateFrom": date_from,
            "dateTo": date_to,
            "interval": "P1D",
            "filter": [{"type": "user", "id": user_id}],
            "groupBy": ["user"],
            "metrics": ["callDuration", "queueTime", "talkTime", "acwTime"],
            "limit": 50
        }
        
        response = requests.post(url, headers=self._get_headers(), json=body)
        response.raise_for_status()
        return response.json()

def main():
    # Configuration
    CLIENT_ID = "YOUR_CLIENT_ID"
    CLIENT_SECRET = "YOUR_CLIENT_SECRET"
    ORG_ID = "YOUR_ORG_ID"
    USER_ID = "TARGET_USER_ID"
    
    # Date range for analytics (ISO 8601)
    DATE_FROM = "2023-10-01T00:00:00Z"
    DATE_TO = "2023-10-31T23:59:59Z"

    try:
        client = GenesysClient(CLIENT_ID, CLIENT_SECRET, ORG_ID)
        
        print("--- Checking Active Conversations ---")
        active_convs = client.get_active_voice_conversations(USER_ID)
        if active_convs:
            for conv in active_convs:
                print(f"Active Call ID: {conv['id']}")
        else:
            print("No active voice conversations found.")
            
        print("\n--- Retrieving Historical Analytics ---")
        analytics_data = client.get_user_analytics(USER_ID, DATE_FROM, DATE_TO)
        
        if analytics_data.get("partitions"):
            for partition in analytics_data["partitions"]:
                for group in partition.get("groups", []):
                    metrics = group.get("metrics", {})
                    call_dur = metrics.get("callDuration", {})
                    print(f"Total Calls: {call_dur.get('count', 0)}")
                    print(f"Avg Duration: {call_dur.get('average', 0):.2f}s")
        else:
            print("No analytics data available for the selected period.")
            
    except Exception as e:
        print(f"Fatal Error: {e}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

• Cause: The OAuth token lacks the required scope.
• Fix: Verify that the OAuth Client in Genesys Cloud Admin has conversations:read for real-time queries and analytics:conversations:query for historical data. Re-generate the token after updating scopes.

Error: 400 Bad Request (Analytics)

• Cause: Invalid date format or unsupported metric name.
• Fix: Ensure dateFrom and dateTo are in strict ISO 8601 format with ‘Z’ suffix. Check the Genesys Cloud Analytics API documentation for valid metric names (e.g., use callDuration not duration).

Error: 429 Too Many Requests

• Cause: Exceeding the rate limit for the API endpoint.
• Fix: Implement exponential backoff. For Analytics queries, avoid polling too frequently. Cache results if possible. The Conversations API has higher rate limits but still enforces them per client.

Error: Empty Result Set

• Cause: No data matches the filter, or the data has not yet propagated to Analytics.
• Fix: For Analytics, wait at least 15-30 minutes after the interaction ends before querying. For Conversations, ensure the conversation is actually active and the user ID is correct.

Official References

• Genesys Cloud Conversations API Documentation
• Genesys Cloud Analytics Conversations Query Documentation
• OAuth 2.0 Client Credentials Flow