Extract CSAT Survey Responses Tied to Interactions via the Genesys Cloud Quality API

Extract CSAT Survey Responses Tied to Interactions via the Genesys Cloud Quality API

What You Will Build

  • You will build a Python script that retrieves Customer Satisfaction (CSAT) survey responses and correlates them with the underlying voice or digital interactions.
  • This tutorial uses the Genesys Cloud Quality API (/api/v2/quality/evaluations) and the Conversations API for interaction details.
  • The implementation is written in Python 3.10+ using the official genesys-cloud-py-sdk.

Prerequisites

  • OAuth Client: A Genesys Cloud OAuth client with the following scopes:
    • quality:evaluation:read (to retrieve evaluation results)
    • analytics:conversation:details:read (optional, to fetch interaction transcripts/details)
    • user:read (to resolve user IDs if needed)
  • SDK Version: genesys-cloud-py-sdk version 13.0.0 or higher.
  • Runtime: Python 3.10 or higher.
  • Dependencies:
    pip install genesys-cloud-py-sdk python-dotenv
    
  • Environment Variables: You must have your GENESYS_CLOUD_REGION, GENESYS_CLOUD_CLIENT_ID, and GENESYS_CLOUD_CLIENT_SECRET configured.

Authentication Setup

Genesys Cloud uses OAuth 2.0 Client Credentials flow for server-to-server integrations. The SDK handles token refresh automatically, but you must initialize the PlatformClient correctly.

Create a .env file in your project root:

GENESYS_CLOUD_REGION=us-east-1
GENESYS_CLOUD_CLIENT_ID=your_client_id_here
GENESYS_CLOUD_CLIENT_SECRET=your_client_secret_here

Initialize the client in your Python script:

import os
from dotenv import load_dotenv
from genesyscloud.platform.client import PlatformClient
from genesyscloud.platform.client.auth import OAuthClientCredentials

# Load environment variables
load_dotenv()

def get_platform_client() -> PlatformClient:
    """
    Initializes and returns the Genesys Cloud PlatformClient.
    """
    region = os.getenv("GENESYS_CLOUD_REGION")
    client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")

    if not all([region, client_id, client_secret]):
        raise ValueError("Missing required environment variables: REGION, CLIENT_ID, CLIENT_SECRET")

    # Configure the OAuth provider
    oauth_provider = OAuthClientCredentials(
        client_id=client_id,
        client_secret=client_secret,
        base_url=f"https://{region}.mypurecloud.com"
    )

    # Initialize the platform client
    platform_client = PlatformClient(oauth_provider=oauth_provider)
    
    # Optional: Set a reasonable timeout for long-running queries
    platform_client.set_http_timeout(30)
    
    return platform_client

Implementation

Step 1: Querying CSAT Evaluations via the Quality API

The Quality API does not have a single endpoint that returns “CSAT responses.” Instead, CSAT scores are stored as evaluations. To extract these, you must query the /api/v2/quality/evaluations endpoint.

The critical parameter here is evaluatorId. In Genesys Cloud, CSAT surveys are evaluated by a system user. You must identify the evaluatorId associated with your CSAT survey. This is typically a system-generated ID. You can find this by:

  1. Going to Admin > Quality > Evaluations.
  2. Filtering by “Evaluator” and selecting the CSAT survey name.
  3. Copying the ID from the URL or the API response.

For this tutorial, assume you have already identified the evaluatorId.

from genesyscloud.quality.api.evaluation_api import EvaluationApi
from genesyscloud.quality.model import EvaluationQuery

def get_csat_evaluations(platform_client: PlatformClient, evaluator_id: str, date_from: str, date_to: str, limit: int = 100):
    """
    Retrieves CSAT evaluations for a specific date range.
    
    Args:
        platform_client: The initialized PlatformClient.
        evaluator_id: The ID of the CSAT survey evaluator.
        date_from: ISO 8601 start date (e.g., '2023-10-01T00:00:00Z').
        date_to: ISO 8601 end date (e.g., '2023-10-31T23:59:59Z').
        limit: Maximum number of results per page.
        
    Returns:
        A list of Evaluation objects.
    """
    evaluation_api = EvaluationApi(platform_client)
    
    # Construct the query
    # 'dateFrom' and 'dateTo' filter when the evaluation was created/completed
    query = EvaluationQuery(
        evaluator_id=evaluator_id,
        date_from=date_from,
        date_to=date_to,
        limit=limit
    )
    
    all_evaluations = []
    
    try:
        # The SDK handles pagination if we loop through the response
        # However, for precise control, we use the raw response structure
        response = evaluation_api.post_quality_evaluation_query(query=query)
        
        if response.body:
            all_evaluations.extend(response.body.entities)
            
            # Handle pagination if there are more results
            while response.body.next_page_token:
                query.next_page_token = response.body.next_page_token
                response = evaluation_api.post_quality_evaluation_query(query=query)
                if response.body:
                    all_evaluations.extend(response.body.entities)
                    
    except Exception as e:
        print(f"Error retrieving evaluations: {e}")
        # Check for specific HTTP errors
        if hasattr(e, 'status') and e.status == 401:
            print("Authentication failed. Check your OAuth credentials.")
        elif hasattr(e, 'status') and e.status == 429:
            print("Rate limited. Implement exponential backoff.")
            
    return all_evaluations

Step 2: Extracting CSAT Scores and Interaction IDs

Each Evaluation object contains the CSAT score and the conversationId. The conversationId is the key to linking the survey response back to the actual voice or digital interaction.

The structure of a CSAT evaluation varies slightly depending on the survey design, but typically the score is stored in the score field or within a custom attribute. For standard Genesys CSAT, the score field usually contains the numeric value (1-5).

from genesyscloud.quality.model import Evaluation

def extract_csat_data(evaluations: list[Evaluation]) -> list[dict]:
    """
    Parses Evaluation objects to extract relevant CSAT data.
    
    Args:
        evaluations: A list of Evaluation objects from the Quality API.
        
    Returns:
        A list of dictionaries containing csat_score, conversation_id, and survey_time.
    """
    csat_records = []
    
    for ev in evaluations:
        if not ev.conversation:
            continue
            
        conversation_id = ev.conversation.id
        
        # Extract the score
        # Note: The exact field for CSAT score can vary. 
        # Standard CSAT usually populates the 'score' field.
        csat_score = ev.score
        
        # Extract the date the survey was completed
        survey_completed_date = ev.date if ev.date else None
        
        # Extract respondent info if available (often masked for privacy)
        respondent_id = ev.respondent.id if ev.respondent else None
        
        record = {
            "conversation_id": conversation_id,
            "csat_score": csat_score,
            "survey_completed_date": survey_completed_date,
            "respondent_id": respondent_id,
            "evaluator_id": ev.evaluator.id
        }
        
        csat_records.append(record)
        
    return csat_records

Step 3: Correlating with Interaction Details (Optional)

To understand why a customer gave a certain score, you often need the interaction transcript or metadata. The conversation_id from the evaluation allows you to query the Analytics Conversations API or the Conversations API.

For voice interactions, you might want the call duration and agent ID. For digital, you might want the transcript. This example fetches basic conversation details using the Analytics API, which is more performant for historical data than the real-time Conversations API.

from genesyscloud.analytics.api.conversation_details_api import ConversationDetailsApi
from genesyscloud.analytics.model import ConversationDetailsQuery

def get_conversation_details(platform_client: PlatformClient, conversation_ids: list[str]) -> dict:
    """
    Fetches details for a batch of conversation IDs.
    
    Args:
        platform_client: The initialized PlatformClient.
        conversation_ids: A list of conversation IDs.
        
    Returns:
        A dictionary mapping conversation_id to its details.
    """
    conversation_api = ConversationDetailsApi(platform_client)
    conversation_map = {}
    
    # The Analytics API allows querying up to 100 conversations per request
    batch_size = 100
    
    for i in range(0, len(conversation_ids), batch_size):
        batch_ids = conversation_ids[i:i + batch_size]
        
        # Construct the query
        query = ConversationDetailsQuery(
            conversations=batch_ids,
            include_data=True  # Include transcript and other data
        )
        
        try:
            response = conversation_api.post_analytics_conversations_details_query(query=query)
            
            if response.body and response.body.conversations:
                for conv in response.body.conversations:
                    conversation_map[conv.id] = conv
                    
        except Exception as e:
            print(f"Error fetching conversation details for batch: {e}")
            continue
            
    return conversation_map

Complete Working Example

This complete script combines authentication, CSAT extraction, and interaction correlation. It requires the evaluator_id for your specific CSAT survey.

import os
import json
from dotenv import load_dotenv
from genesyscloud.platform.client import PlatformClient
from genesyscloud.platform.client.auth import OAuthClientCredentials
from genesyscloud.quality.api.evaluation_api import EvaluationApi
from genesyscloud.quality.model import EvaluationQuery
from genesyscloud.analytics.api.conversation_details_api import ConversationDetailsApi
from genesyscloud.analytics.model import ConversationDetailsQuery

def load_env():
    load_dotenv()
    return {
        "region": os.getenv("GENESYS_CLOUD_REGION"),
        "client_id": os.getenv("GENESYS_CLOUD_CLIENT_ID"),
        "client_secret": os.getenv("GENESYS_CLOUD_CLIENT_SECRET"),
    }

def init_client(config: dict) -> PlatformClient:
    oauth_provider = OAuthClientCredentials(
        client_id=config["client_id"],
        client_secret=config["client_secret"],
        base_url=f"https://{config['region']}.mypurecloud.com"
    )
    client = PlatformClient(oauth_provider=oauth_provider)
    return client

def fetch_csat_evaluations(client: PlatformClient, evaluator_id: str, start_date: str, end_date: str):
    """
    Fetches all CSAT evaluations within the date range.
    """
    evaluation_api = EvaluationApi(client)
    all_evaluations = []
    
    query = EvaluationQuery(
        evaluator_id=evaluator_id,
        date_from=start_date,
        date_to=end_date,
        limit=100
    )
    
    try:
        # Initial query
        response = evaluation_api.post_quality_evaluation_query(query=query)
        if response.body:
            all_evaluations.extend(response.body.entities)
            
            # Paginate through all results
            while response.body.next_page_token:
                query.next_page_token = response.body.next_page_token
                response = evaluation_api.post_quality_evaluation_query(query=query)
                if response.body:
                    all_evaluations.extend(response.body.entities)
                    
    except Exception as e:
        print(f"Failed to fetch evaluations: {e}")
        return []
        
    return all_evaluations

def enrich_with_interaction_data(client: PlatformClient, evaluations):
    """
    Enriches evaluation data with conversation details.
    """
    conversation_ids = []
    eval_map = {}
    
    # Map evaluations to conversation IDs for batch processing
    for ev in evaluations:
        if ev.conversation and ev.conversation.id:
            conversation_ids.append(ev.conversation.id)
            eval_map[ev.conversation.id] = ev
            
    if not conversation_ids:
        return []

    # Fetch conversation details in batches
    conversation_api = ConversationDetailsApi(client)
    conversation_details = {}
    
    batch_size = 100
    for i in range(0, len(conversation_ids), batch_size):
        batch_ids = conversation_ids[i:i + batch_size]
        query = ConversationDetailsQuery(
            conversations=batch_ids,
            include_data=True
        )
        
        try:
            response = conversation_api.post_analytics_conversations_details_query(query=query)
            if response.body and response.body.conversations:
                for conv in response.body.conversations:
                    conversation_details[conv.id] = conv
        except Exception as e:
            print(f"Error fetching conversation details: {e}")
            continue

    # Combine data
    enriched_results = []
    for conv_id, ev in eval_map.items():
        conv_detail = conversation_details.get(conv_id)
        
        result = {
            "conversation_id": conv_id,
            "csat_score": ev.score,
            "survey_date": ev.date,
            "evaluator_id": ev.evaluator.id if ev.evaluator else None,
            "conversation_type": conv_detail.conversation_type if conv_detail else None,
            "duration_seconds": conv_detail.duration_seconds if conv_detail else None,
            "agent_id": conv_detail.participants[0].user.id if conv_detail and conv_detail.participants else None,
        }
        enriched_results.append(result)
        
    return enriched_results

def main():
    # Configuration
    config = load_env()
    if not all(config.values()):
        raise ValueError("Missing environment variables.")
        
    # Replace with your actual CSAT Evaluator ID
    # Find this in Admin > Quality > Evaluations > Select CSAT Survey
    EVALUATOR_ID = "your_csat_evaluator_id_here" 
    
    # Date range (ISO 8601)
    START_DATE = "2023-10-01T00:00:00Z"
    END_DATE = "2023-10-31T23:59:59Z"
    
    if EVALUATOR_ID == "your_csat_evaluator_id_here":
        print("Please set the EVALUATOR_ID in the code.")
        return

    # Initialize Client
    client = init_client(config)
    
    # Step 1: Fetch Evaluations
    print("Fetching CSAT evaluations...")
    evaluations = fetch_csat_evaluations(client, EVALUATOR_ID, START_DATE, END_DATE)
    print(f"Found {len(evaluations)} evaluations.")
    
    if not evaluations:
        print("No evaluations found. Check the date range and evaluator ID.")
        return

    # Step 2: Enrich with Interaction Data
    print("Enriching with interaction details...")
    enriched_data = enrich_with_interaction_data(client, evaluations)
    
    # Step 3: Output Results
    print("Processing complete. Outputting JSON...")
    print(json.dumps(enriched_data, indent=2, default=str))

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

  • Cause: Invalid Client ID, Client Secret, or expired token.
  • Fix: Verify your .env values. Ensure the OAuth client has the quality:evaluation:read scope. If using a personal access token, ensure it is not expired.

Error: 403 Forbidden

  • Cause: The OAuth client lacks the required scopes.
  • Fix: Go to Admin > Integrations > API Access > OAuth clients. Edit your client and add:
    • quality:evaluation:read
    • analytics:conversation:details:read (if fetching conversation details)
    • user:read (if resolving user names)

Error: 429 Too Many Requests

  • Cause: Exceeding Genesys Cloud rate limits. The Quality API has strict limits on evaluation queries.
  • Fix: Implement exponential backoff. The SDK does not automatically retry 429s. Wrap your API calls in a retry loop:
import time

def api_call_with_retry(func, *args, max_retries=3, **kwargs):
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            if hasattr(e, 'status') and e.status == 429:
                wait_time = 2 ** attempt  # Exponential backoff
                print(f"Rate limited. Waiting {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise
    raise Exception("Max retries exceeded")

Error: Evaluation Score is Null

  • Cause: The CSAT survey is configured to store scores in custom attributes rather than the standard score field.
  • Fix: Inspect the raw Evaluation object. Look for ev.custom_attributes or specific fields defined in your survey design. You may need to parse the answers array if the survey uses free-text or complex logic.

Official References