Extracting CSAT Responses Linked to Interactions via the Genesys Cloud Quality API

What You Will Build

• You will build a Python script that queries the Genesys Cloud Quality API to retrieve evaluation results containing CSAT survey scores.
• The code uses the purecloudplatformclientv2 Python SDK to authenticate, query evaluations, and filter for specific interaction types.
• The tutorial covers Python, leveraging the official Genesys Cloud Python SDK for robust type safety and error handling.

Prerequisites

• OAuth Client: A Genesys Cloud OAuth Client with the quality:evaluation:read scope. For cross-referencing interaction details, conversation:view is also recommended.
• SDK Version: purecloudplatformclientv2 v115.0.0 or higher.
• Runtime: Python 3.9+
• Dependencies:

  pip install purecloudplatformclientv2 requests
  

Authentication Setup

Genesys Cloud APIs require OAuth 2.0 Bearer tokens. The recommended approach for server-to-server integration is the Client Credentials Grant. This flow provides a long-lived access token (typically 1 hour) that does not require user interaction.

The following code initializes the Genesys Cloud Platform Client with your environment and credentials.

import os
from purecloudplatformclientv2 import (
    PlatformClient,
    Configuration,
    ApiClient
)

def initialize_platform_client() -> PlatformClient:
    """
    Initializes and returns the Genesys Cloud Platform Client.
    Assumes environment variables are set:
    - GENESYS_ENVIRONMENT: e.g., 'us-east-1'
    - GENESYS_CLIENT_ID: Your OAuth Client ID
    - GENESYS_CLIENT_SECRET: Your OAuth Client Secret
    """
    env = os.getenv("GENESYS_ENVIRONMENT", "us-east-1")
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")

    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")

    # Construct the base URL for the API
    base_url = f"https://{env}.mypurecloud.com"
    
    # Create the configuration object
    config = Configuration(
        host=base_url,
        oauth_client_id=client_id,
        oauth_client_secret=client_secret
    )

    # Instantiate the ApiClient and PlatformClient
    api_client = ApiClient(configuration=config)
    platform_client = PlatformClient(api_client)

    # Verify connectivity by fetching the token immediately
    # This triggers the OAuth flow and caches the token
    try:
        # This is an implicit call within the SDK, but we can explicitly 
        # check if the client is ready by attempting a lightweight call
        # or simply relying on the lazy loading of the token in subsequent calls.
        # For explicit verification:
        api_client.get_token() 
        print("Authentication successful.")
    except Exception as e:
        print(f"Authentication failed: {e}")
        raise

    return platform_client

Implementation

Step 1: Querying Evaluations with CSAT Data

The Quality API does not have a single endpoint that returns “all CSAT scores.” Instead, CSAT data is embedded within Evaluation Results. When an interaction (voice, digital, email) is evaluated, and that evaluation template includes CSAT questions, the scores are stored in the scorecards array of the EvaluationResult object.

To extract this data, you must query the GET /api/v2/quality/evaluations endpoint.

Required Scope: quality:evaluation:read

The following function demonstrates how to query evaluations. We will filter by a specific date range to ensure the query returns manageable results.

from purecloudplatformclientv2 import QualityApi
from purecloudplatformclientv2.rest import ApiException
from datetime import datetime, timedelta
from typing import List, Dict, Any

def fetch_evaluations_with_csat(
    platform_client: PlatformClient,
    days_back: int = 7
) -> List[Dict[str, Any]]:
    """
    Fetches evaluations from the last N days and filters for those containing CSAT scores.
    
    Args:
        platform_client: The initialized PlatformClient instance.
        days_back: Number of days to look back for evaluations.
        
    Returns:
        A list of dictionaries containing interaction ID, evaluator, and CSAT scores.
    """
    quality_api = QualityApi(platform_client)
    
    # Define the time range
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(days=days_back)
    
    # Format dates for the API (ISO 8601)
    start_date_str = start_time.strftime("%Y-%m-%dT%H:%M:%S.000Z")
    end_date_str = end_time.strftime("%Y-%m-%dT%H:%M:%S.000Z")

    csat_results = []
    
    try:
        # Initial request
        # Note: The Quality API supports pagination. We must handle 'nextPageUri'.
        page_uri = None
        has_more_pages = True
        
        while has_more_pages:
            # Construct the request body
            # We filter by the date range to narrow down the search space
            request_body = {
                "from": start_date_str,
                "to": end_date_str,
                # Optional: Filter by specific interaction type if needed
                # "interactionTypes": ["voice", "digital"] 
            }
            
            if page_uri:
                # If we have a next page URI, we pass it directly.
                # The SDK's get_quality_evaluations method supports a 'page_uri' parameter.
                response = quality_api.get_quality_evaluations(page_uri=page_uri)
            else:
                # First page
                response = quality_api.get_quality_evaluations(body=request_body)

            # Process the current page of evaluations
            if response.entities:
                for evaluation in response.entities:
                    csat_data = extract_csat_from_evaluation(evaluation)
                    if csat_data:
                        csat_results.append(csat_data)
            
            # Check for pagination
            if response.next_page_uri:
                page_uri = response.next_page_uri
                has_more_pages = True
            else:
                has_more_pages = False

        return csat_results

    except ApiException as e:
        print(f"Exception when calling QualityApi->get_quality_evaluations: {e}")
        if e.status == 401:
            print("Unauthorized: Check your OAuth token and scopes.")
        elif e.status == 403:
            print("Forbidden: Your client lacks the 'quality:evaluation:read' scope.")
        elif e.status == 429:
            print("Rate Limited: Implement exponential backoff.")
        raise

Step 2: Parsing CSAT Scores from Evaluation Objects

The EvaluationResult object contains an array called scorecards. Each scorecard represents a section of the evaluation template. CSAT questions are typically found in a scorecard labeled “CSAT” or similar, depending on how the Quality Manager configured the template.

The structure of a scorecard item looks like this:

{
  "id": "question-id-123",
  "label": "How would you rate your experience?",
  "value": 5,
  "score": 100,
  "type": "rating"
}

We need to iterate through the scorecards, identify CSAT-related questions, and extract the numeric value.

from purecloudplatformclientv2.models import EvaluationResult

def extract_csat_from_evaluation(evaluation: EvaluationResult) -> Dict[str, Any] | None:
    """
    Inspects an EvaluationResult object to find CSAT scores.
    
    Args:
        evaluation: The EvaluationResult object from the API.
        
    Returns:
        A dictionary with CSAT data if found, otherwise None.
    """
    if not evaluation.scorecards:
        return None

    csat_scores = {}
    has_csat = False

    # Iterate through each scorecard in the evaluation
    for scorecard in evaluation.scorecards:
        # Check if the scorecard label contains 'CSAT' (case-insensitive)
        # Note: This depends on your Quality Template configuration.
        # You may want to use specific question IDs instead for robustness.
        if scorecard.label and "csat" in scorecard.label.lower():
            has_csat = True
            # Iterate through items in this scorecard
            if scorecard.items:
                for item in scorecard.items:
                    # Store the question label and its value
                    if item.value is not None:
                        csat_scores[item.label] = item.value
                    else:
                        # Handle non-numeric responses (e.g., text comments)
                        if item.text:
                            csat_scores[item.label] = item.text

    if has_csat and csat_scores:
        return {
            "interaction_id": evaluation.interaction_id,
            "evaluation_id": evaluation.id,
            "evaluator_id": evaluation.evaluator_id,
            "evaluator_name": evaluation.evaluator_name,
            "submit_date": evaluation.submit_date,
            "csat_scores": csat_scores
        }
    
    return None

Step 3: Enriching with Interaction Details (Optional but Recommended)

The Evaluation object provides the interaction_id. To understand what the CSAT score refers to (e.g., which agent, which queue, duration), you should query the Analytics Conversations API. This step connects the Quality data to the Operational data.

Required Scope: conversation:view

from purecloudplatformclientv2 import AnalyticsApi
from purecloudplatformclientv2.models import ConversationDetailQueryRequest

def enrich_with_interaction_details(
    platform_client: PlatformClient,
    csat_results: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
    """
    Fetches interaction details for each CSAT result to add context (agent, duration, etc.).
    
    Args:
        platform_client: The initialized PlatformClient instance.
        csat_results: List of CSAT data dictionaries from Step 2.
        
    Returns:
        The enriched list of CSAT results.
    """
    analytics_api = AnalyticsApi(platform_client)
    
    # Batch processing to avoid excessive API calls
    # The Analytics API allows querying multiple interactions in one request
    batch_size = 100
    enriched_results = []

    for i in range(0, len(csat_results), batch_size):
        batch = csat_results[i : i + batch_size]
        interaction_ids = [item["interaction_id"] for item in batch]
        
        try:
            # Create the query body
            query_body = ConversationDetailQueryRequest(
                interaction_ids=interaction_ids
            )
            
            # Query the conversations
            response = analytics_api.post_analytics_conversations_details_query(body=query_body)
            
            # Map the interaction details back to the CSAT results
            # Create a lookup map for efficiency
            interaction_lookup = {
                conv.interaction_id: conv for conv in response.entities
            }
            
            for result in batch:
                inter_id = result["interaction_id"]
                if inter_id in interaction_lookup:
                    conv = interaction_lookup[inter_id]
                    result["agent_name"] = conv.to_attributes.get("agent_name", "Unknown")
                    result["queue_name"] = conv.to_attributes.get("queue_name", "Unknown")
                    result["duration_seconds"] = conv.to_attributes.get("duration_seconds", 0)
                
                enriched_results.append(result)

        except ApiException as e:
            print(f"Error fetching interaction details: {e}")
            # Fallback: Add the original result without enrichment
            enriched_results.extend(batch)

    return enriched_results

Complete Working Example

The following script combines all steps into a single executable module. It authenticates, fetches evaluations, extracts CSAT, enriches the data with interaction details, and outputs the result as JSON.

import json
import os
import sys
from datetime import datetime
from purecloudplatformclientv2 import (
    PlatformClient,
    Configuration,
    ApiClient
)
from purecloudplatformclientv2.rest import ApiException
from purecloudplatformclientv2.models import EvaluationResult

# --- Import helper functions from previous steps ---
# In a real project, these would be in separate modules or classes.
# For this tutorial, they are included inline for copy-paste functionality.

def initialize_platform_client() -> PlatformClient:
    env = os.getenv("GENESYS_ENVIRONMENT", "us-east-1")
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")

    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")

    base_url = f"https://{env}.mypurecloud.com"
    config = Configuration(
        host=base_url,
        oauth_client_id=client_id,
        oauth_client_secret=client_secret
    )
    api_client = ApiClient(configuration=config)
    platform_client = PlatformClient(api_client)
    
    try:
        api_client.get_token() 
    except Exception as e:
        print(f"Authentication failed: {e}")
        raise
    return platform_client

def extract_csat_from_evaluation(evaluation: EvaluationResult):
    if not evaluation.scorecards:
        return None

    csat_scores = {}
    has_csat = False

    for scorecard in evaluation.scorecards:
        if scorecard.label and "csat" in scorecard.label.lower():
            has_csat = True
            if scorecard.items:
                for item in scorecard.items:
                    if item.value is not None:
                        csat_scores[item.label] = item.value
                    elif item.text:
                        csat_scores[item.label] = item.text

    if has_csat and csat_scores:
        return {
            "interaction_id": evaluation.interaction_id,
            "evaluation_id": evaluation.id,
            "evaluator_id": evaluation.evaluator_id,
            "evaluator_name": evaluation.evaluator_name,
            "submit_date": evaluation.submit_date.isoformat() if evaluation.submit_date else None,
            "csat_scores": csat_scores
        }
    return None

def fetch_evaluations_with_csat(platform_client, days_back: int = 7):
    from purecloudplatformclientv2 import QualityApi
    quality_api = QualityApi(platform_client)
    
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(days=days_back)
    
    start_date_str = start_time.strftime("%Y-%m-%dT%H:%M:%S.000Z")
    end_date_str = end_time.strftime("%Y-%m-%dT%H:%M:%S.000Z")

    csat_results = []
    page_uri = None
    has_more_pages = True
    
    while has_more_pages:
        request_body = {
            "from": start_date_str,
            "to": end_date_str
        }
        
        try:
            if page_uri:
                response = quality_api.get_quality_evaluations(page_uri=page_uri)
            else:
                response = quality_api.get_quality_evaluations(body=request_body)

            if response.entities:
                for evaluation in response.entities:
                    csat_data = extract_csat_from_evaluation(evaluation)
                    if csat_data:
                        csat_results.append(csat_data)
            
            if response.next_page_uri:
                page_uri = response.next_page_uri
            else:
                has_more_pages = False
        except ApiException as e:
            print(f"API Error: {e}")
            raise

    return csat_results

def enrich_with_interaction_details(platform_client, csat_results):
    from purecloudplatformclientv2 import AnalyticsApi
    from purecloudplatformclientv2.models import ConversationDetailQueryRequest
    
    analytics_api = AnalyticsApi(platform_client)
    enriched_results = []
    batch_size = 100

    for i in range(0, len(csat_results), batch_size):
        batch = csat_results[i : i + batch_size]
        interaction_ids = [item["interaction_id"] for item in batch]
        
        try:
            query_body = ConversationDetailQueryRequest(interaction_ids=interaction_ids)
            response = analytics_api.post_analytics_conversations_details_query(body=query_body)
            
            interaction_lookup = {
                conv.interaction_id: conv for conv in response.entities
            }
            
            for result in batch:
                inter_id = result["interaction_id"]
                if inter_id in interaction_lookup:
                    conv = interaction_lookup[inter_id]
                    # Accessing attributes safely
                    attrs = conv.to_attributes if hasattr(conv, 'to_attributes') else {}
                    result["agent_name"] = attrs.get("agent_name", "Unknown")
                    result["queue_name"] = attrs.get("queue_name", "Unknown")
                    result["duration_seconds"] = attrs.get("duration_seconds", 0)
                
                enriched_results.append(result)

        except ApiException as e:
            print(f"Error enriching: {e}")
            enriched_results.extend(batch)

    return enriched_results

# --- Main Execution ---

if __name__ == "__main__":
    from datetime import timedelta
    
    print("Initializing Genesys Cloud Client...")
    pc = initialize_platform_client()
    
    print("Fetching evaluations with CSAT data...")
    csat_data = fetch_evaluations_with_csat(pc, days_back=7)
    
    print(f"Found {len(csat_data)} evaluations with CSAT data.")
    
    if csat_data:
        print("Enriching with interaction details...")
        enriched_data = enrich_with_interaction_details(pc, csat_data)
        
        # Output result
        print(json.dumps(enriched_data, indent=2, default=str))
    else:
        print("No CSAT evaluations found in the specified time range.")

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token is expired, invalid, or the client credentials are incorrect.
• Fix: Verify GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET. Ensure the token has not expired (tokens last 1 hour). The SDK handles refresh automatically for client credentials, but if you are using a user-based flow, ensure the refresh token is valid.

Error: 403 Forbidden

• Cause: The OAuth Client lacks the required scope.
• Fix: Go to Genesys Cloud Admin > Settings > OAuth Clients. Select your client and ensure quality:evaluation:read and conversation:view are added to the scopes. Save and regenerate the token.

Error: 429 Too Many Requests

• Cause: You have exceeded the rate limit for the Quality or Analytics API.

• Fix: Implement exponential backoff. The Genesys Cloud API returns a Retry-After header.

  import time
  
  except ApiException as e:
      if e.status == 429:
          retry_after = e.headers.get("Retry-After", 10)
          print(f"Rate limited. Waiting {retry_after} seconds...")
          time.sleep(int(retry_after))
          # Retry the request
  

Error: Empty csat_scores

• Cause: The evaluation template does not use the label “CSAT” or the questions are not configured as rating/text types.
• Fix: Inspect the raw evaluation.scorecards in the API response. Identify the exact scorecard.label used in your Quality Template and update the extract_csat_from_evaluation function to match it. Alternatively, filter by specific question_id if known.

Official References

• Quality API Evaluation Endpoints
• Analytics Conversations Details Query
• Python SDK Documentation