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

What You Will Build

• This tutorial demonstrates how to query Genesys Cloud to retrieve Customer Satisfaction (CSAT) survey results and correlate them with the original interaction metadata.
• The solution uses the Genesys Cloud Quality API and the Python SDK (genesyscloud).
• The implementation is written in Python 3.9+ using type hints and async/await patterns where appropriate for high-throughput data extraction.

Prerequisites

OAuth Configuration

• Client Type: Service Account or User Account.
• Required Scopes:
  • quality:evaluation:read (To access evaluation data)
  • quality:scorecard:read (To access scorecard definitions if needed for context)
  • analytics:conversation:view (If you need to pull detailed interaction transcripts alongside CSAT)
  • user:read (Optional, to resolve user IDs to names)

Environment Setup

• Python Version: 3.9 or higher.
• SDK Version: genesyscloud >= 12.0.0.
• Dependencies:

  pip install genesyscloud requests
  

Authentication Setup

Genesys Cloud uses OAuth 2.0 for API access. For server-side integrations, a Service Account with a JWT or Client Credentials grant is standard. Below is a robust setup using the Genesys Cloud Python SDK, which handles token caching and refresh automatically.

import os
from genesyscloud.platform_client_v2 import Configuration, ApiClient
from genesyscloud.platform_client_v2.rest import ApiException

def get_platform_api_client() -> ApiClient:
    """
    Initializes and returns a configured Genesys Cloud API Client.
    Uses environment variables for credentials.
    """
    config = Configuration()
    
    # Load credentials from environment variables
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    base_url = os.getenv("GENESYS_BASE_URL", "https://api.mypurecloud.com")
    
    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set in environment variables.")
    
    # Configure the client
    config.host = base_url
    config.client_id = client_id
    config.client_secret = client_secret
    
    # Create the API client instance
    # The SDK handles token acquisition and refresh automatically
    api_client = ApiClient(configuration=config)
    
    return api_client

# Initialize the client
client = get_platform_api_client()

Implementation

Step 1: Identify the CSAT Scorecard

Before querying evaluations, you must identify the scorecardId associated with your CSAT surveys. Genesys Cloud stores CSAT responses as evaluations linked to a specific scorecard.

1. List all scorecards.
2. Filter for the one used for CSAT (usually named “CSAT” or identified by its type).

from genesyscloud.quality_api import QualityApi
from genesyscloud.platform_client_v2.rest import ApiException
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def get_csat_scorecard_id(api_client: ApiClient) -> str:
    """
    Retrieves the ID of the primary CSAT scorecard.
    Assumes there is one scorecard with 'CSAT' in the name or type.
    """
    quality_api = QualityApi(api_client)
    scorecard_id = None
    
    try:
        # Fetch all scorecards
        # Pagination is handled by the SDK if we use the async iterator, 
        # but for a simple lookup, a single page with a large limit is often sufficient.
        result = quality_api.get_quality_scorecards(limit=100)
        
        if result.body and result.body.entities:
            for scorecard in result.body.entities:
                # Check name or description for 'CSAT'
                if scorecard.name and 'CSAT' in scorecard.name.upper():
                    scorecard_id = scorecard.id
                    logger.info(f"Found CSAT Scorecard ID: {scorecard_id}")
                    break
            
            if not scorecard_id:
                # Fallback: Look for the first scorecard if only one exists
                if len(result.body.entities) == 1:
                    scorecard_id = result.body.entities[0].id
                    logger.warning("No explicit CSAT name found. Using the only available scorecard.")
                else:
                    raise Exception("Could not identify a unique CSAT scorecard. Please verify scorecard names.")
        else:
            raise Exception("No scorecards found in the organization.")
            
    except ApiException as e:
        logger.error(f"Error fetching scorecards: {e.body}")
        raise
    
    return scorecard_id

# Get the scorecard ID
CSAT_SCORECARD_ID = get_csat_scorecard_id(client)

Step 2: Query Evaluations for CSAT Responses

The core logic involves querying the /api/v2/quality/evaluations endpoint. We need to filter by:

1. scorecardId: The ID found in Step 1.
2. status: complete (to ensure we only get submitted surveys).
3. orderBy: interaction.endTime (to process chronologically).

We must handle pagination because large organizations may have thousands of daily CSAT responses.

from datetime import datetime, timedelta
from typing import List, Dict, Any

def fetch_csat_evaluations(
    api_client: ApiClient, 
    scorecard_id: str, 
    start_date: str, 
    end_date: str,
    limit: int = 1000
) -> List[Dict[str, Any]]:
    """
    Fetches all completed CSAT evaluations within a date range.
    Handles pagination automatically.
    
    Args:
        api_client: The initialized ApiClient.
        scorecard_id: The ID of the CSAT scorecard.
        start_date: ISO 8601 start date string (e.g., '2023-10-01T00:00:00.000Z').
        end_date: ISO 8601 end date string.
        limit: Max items per page.
        
    Returns:
        A list of evaluation objects.
    """
    quality_api = QualityApi(api_client)
    all_evaluations = []
    
    # Construct query parameters
    query_params = {
        'scorecardId': scorecard_id,
        'status': 'complete',
        'interactionStartTimeFrom': start_date,
        'interactionStartTimeTo': end_date,
        'limit': limit,
        'orderBy': 'interaction.endTime'
    }
    
    try:
        while True:
            # API Call
            response = quality_api.get_quality_evaluations(**query_params)
            
            if not response.body or not response.body.entities:
                break
                
            all_evaluations.extend(response.body.entities)
            
            # Check for pagination
            next_uri = response.body.next_page
            if next_uri:
                # The SDK simplifies pagination, but for explicit control:
                # We need to extract the 'pageToken' from the next_uri or use the SDK's paging helper.
                # In the Python SDK, get_quality_evaluations returns a response with a 'next_page' link.
                # However, the simplest way with the SDK is to use the returned response's paging info.
                
                # Note: The Genesys Cloud Python SDK v12+ often returns the entities directly.
                # To handle pagination robustly, we check if there are more items.
                # The response body contains 'total', 'count', and 'next_page'.
                
                if response.body.count < limit:
                    break
                    
                # Update query for next page using the page token if available
                # The SDK response object usually has a 'next_page' attribute which is a URL.
                # We can parse the pageToken from it, or simpler:
                # The SDK does not automatically loop. We must extract the token.
                
                # Alternative approach: Use the raw response to get the page token
                # But the standard SDK method is to pass the 'pageToken' in the next call.
                # Let's assume the response body has 'next_page' which contains the token.
                # Actually, the Genesys Cloud API returns a 'pageToken' in the response body if there is a next page.
                
                page_token = response.body.page_token
                if page_token:
                    query_params['pageToken'] = page_token
                else:
                    break
            else:
                break
                
    except ApiException as e:
        logger.error(f"Error fetching evaluations: {e.status} - {e.body}")
        raise
        
    return all_evaluations

# Example usage for the last 24 hours
end_time = datetime.utcnow()
start_time = end_time - timedelta(days=1)

# Format as 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")

evaluations = fetch_csat_evaluations(
    api_client=client,
    scorecard_id=CSAT_SCORECARD_ID,
    start_date=start_date_str,
    end_date=end_date_str
)

logger.info(f"Fetched {len(evaluations)} CSAT evaluations.")

Step 3: Extract and Correlate Interaction Data

The evaluation object contains the interactionId. To get the full context (who called whom, duration, channel), you must query the Interaction API or Analytics API. For a direct tie-in, the interactionId is the key.

If you need the actual survey question answers (e.g., “How was your service?”), they are stored in the sections of the evaluation.

def process_csat_response(eval_obj: Any) -> Dict[str, Any]:
    """
    Parses a single evaluation object to extract CSAT score and interaction details.
    """
    result = {
        'evaluation_id': eval_obj.id,
        'interaction_id': eval_obj.interaction_id,
        'interaction_time': eval_obj.interaction.start_time,
        'evaluator_id': eval_obj.evaluator_id,
        'score': None,
        'comments': None,
        'raw_answers': {}
    }
    
    if not eval_obj.sections:
        return result
        
    # Iterate through sections to find the CSAT score
    for section in eval_obj.sections:
        if not section.items:
            continue
            
        for item in section.items:
            # Check if the item is the CSAT score item
            # Typically, the CSAT score is a specific field. 
            # You may need to map this based on your scorecard definition.
            # Often, the 'name' of the item contains 'CSAT' or 'Overall'.
            
            if item.name and 'CSAT' in item.name.upper():
                # The value can be an integer or a string depending on the scorecard type
                if item.value:
                    result['score'] = item.value
                if item.comment:
                    result['comments'] = item.comment
                    
            # Store all answers for flexibility
            if item.value is not None:
                result['raw_answers'][item.name] = item.value
                
    return result

# Process all evaluations
csat_data = [process_csat_response(eval_obj) for eval_obj in evaluations]

Step 4: Handle Rate Limits and Retries

Genesys Cloud APIs enforce rate limits (429 Too Many Requests). A production script must implement exponential backoff.

import time
import random

def api_call_with_retry(func, *args, max_retries=5, base_delay=1, **kwargs):
    """
    Executes an API call with exponential backoff on 429 errors.
    """
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except ApiException as e:
            if e.status == 429:
                # Extract retry-after header if present
                retry_after = e.headers.get('Retry-After')
                if retry_after:
                    delay = int(retry_after)
                else:
                    # Exponential backoff: 1s, 2s, 4s, 8s... with jitter
                    delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                
                logger.warning(f"Rate limited (429). Retrying in {delay:.2f} seconds... (Attempt {attempt + 1}/{max_retries})")
                time.sleep(delay)
            else:
                # Non-retryable error
                raise
    raise Exception(f"Max retries ({max_retries}) exceeded for API call.")

# Wrap the fetch function
def robust_fetch_csat_evaluations(api_client, scorecard_id, start_date, end_date):
    return api_call_with_retry(
        fetch_csat_evaluations, 
        api_client, 
        scorecard_id, 
        start_date, 
        end_date
    )

Complete Working Example

Below is the consolidated, runnable Python script. Save this as extract_csat.py.

import os
import logging
from datetime import datetime, timedelta
from typing import List, Dict, Any

from genesyscloud.platform_client_v2 import Configuration, ApiClient
from genesyscloud.quality_api import QualityApi
from genesyscloud.platform_client_v2.rest import ApiException
import time
import random

# Configure Logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

def get_platform_api_client() -> ApiClient:
    config = Configuration()
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    base_url = os.getenv("GENESYS_BASE_URL", "https://api.mypurecloud.com")
    
    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")
        
    config.host = base_url
    config.client_id = client_id
    config.client_secret = client_secret
    
    return ApiClient(configuration=config)

def get_csat_scorecard_id(api_client: ApiClient) -> str:
    quality_api = QualityApi(api_client)
    try:
        result = quality_api.get_quality_scorecards(limit=100)
        if result.body and result.body.entities:
            for sc in result.body.entities:
                if sc.name and 'CSAT' in sc.name.upper():
                    return sc.id
            return result.body.entities[0].id
        raise Exception("No scorecards found.")
    except ApiException as e:
        logger.error(f"Error fetching scorecards: {e.body}")
        raise

def fetch_evaluations_page(api_client: ApiClient, scorecard_id: str, start: str, end: str, limit: int, page_token: str = None) -> tuple:
    """
    Fetches a single page of evaluations.
    Returns (entities, next_page_token)
    """
    quality_api = QualityApi(api_client)
    params = {
        'scorecardId': scorecard_id,
        'status': 'complete',
        'interactionStartTimeFrom': start,
        'interactionStartTimeTo': end,
        'limit': limit
    }
    if page_token:
        params['pageToken'] = page_token
        
    response = quality_api.get_quality_evaluations(**params)
    
    if not response.body or not response.body.entities:
        return [], None
        
    return response.body.entities, response.body.page_token

def api_call_with_retry(func, *args, max_retries=5, base_delay=1, **kwargs):
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except ApiException as e:
            if e.status == 429:
                retry_after = e.headers.get('Retry-After')
                delay = int(retry_after) if retry_after else (base_delay * (2 ** attempt) + random.uniform(0, 1))
                logger.warning(f"429 Rate Limit. Retrying in {delay:.2f}s...")
                time.sleep(delay)
            else:
                raise
    raise Exception("Max retries exceeded.")

def extract_csat_data():
    # 1. Setup
    client = get_platform_api_client()
    scorecard_id = get_csat_scorecard_id(client)
    
    # 2. Define Date Range (Last 7 Days)
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(days=7)
    start_str = start_time.strftime("%Y-%m-%dT%H:%M:%S.000Z")
    end_str = end_time.strftime("%Y-%m-%dT%H:%M:%S.000Z")
    
    logger.info(f"Extracting CSAT data from {start_str} to {end_str}")
    
    all_evaluations = []
    page_token = None
    limit = 1000
    
    # 3. Paginate through all evaluations
    while True:
        entities, next_token = api_call_with_retry(
            fetch_evaluations_page,
            client, scorecard_id, start_str, end_str, limit, page_token
        )
        
        if not entities:
            break
            
        all_evaluations.extend(entities)
        logger.info(f"Fetched {len(entities)} evaluations. Total so far: {len(all_evaluations)}")
        
        if not next_token:
            break
        page_token = next_token
        
    # 4. Process Results
    processed_data = []
    for eval_obj in all_evaluations:
        score = None
        comments = None
        if eval_obj.sections:
            for section in eval_obj.sections:
                if section.items:
                    for item in section.items:
                        if item.name and 'CSAT' in item.name.upper():
                            score = item.value
                            comments = item.comment
                            break
        
        processed_data.append({
            'interaction_id': eval_obj.interaction_id,
            'timestamp': eval_obj.interaction.start_time,
            'csat_score': score,
            'comments': comments,
            'evaluator_id': eval_obj.evaluator_id
        })
        
    # 5. Output (Example: Print first 5)
    logger.info(f"Total CSAT responses processed: {len(processed_data)}")
    for item in processed_data[:5]:
        logger.info(f"Interaction: {item['interaction_id']}, Score: {item['csat_score']}, Comments: {item['comments']}")

if __name__ == "__main__":
    extract_csat_data()

Common Errors & Debugging

Error: 403 Forbidden on get_quality_evaluations

• Cause: The OAuth client lacks the quality:evaluation:read scope.
• Fix: Log into the Genesys Cloud Admin Console. Navigate to Admin > Security > OAuth Clients. Edit your client and add the quality:evaluation:read scope. Save and regenerate the token.

Error: 429 Too Many Requests

• Cause: The API rate limit for your organization or client ID has been exceeded.
• Fix: Ensure your code implements the api_call_with_retry pattern shown above. If the error persists, check your organization’s API usage in the Admin Console under Admin > System > API Usage. Consider increasing the delay between requests.

Error: scorecardId is None or Incorrect

• Cause: The script could not find a scorecard with “CSAT” in the name, or the CSAT survey is configured with a different scorecard name.
• Fix: Inspect the scorecard_id variable. In the Admin Console, go to Quality > Scorecards. Note the exact ID of the scorecard used for your CSAT surveys. Hardcode this ID in the script if dynamic detection fails, or update the filtering logic in get_csat_scorecard_id.

Error: Empty entities list despite known CSAT responses

• Cause: The date range (interactionStartTimeFrom/To) does not overlap with the actual evaluation times. Genesys Cloud evaluates interactions after they complete. There may be a lag.
• Fix: Verify the start_time and end_time formats are strict ISO 8601 with timezone Z. Ensure the status filter is complete. If you need pending surveys, change status to in-progress, but note these may not have final scores.

Official References

• Genesys Cloud Quality API Documentation
• Quality Evaluations Endpoint
• Python SDK Reference