How to extract CSAT survey responses tied to specific interactions via the Quality API

What You Will Build

• You will build a script that queries the Genesys Cloud Quality API to retrieve survey responses and join them with their underlying interaction metadata (transcripts, participant IDs, and timestamps).
• This tutorial uses the Genesys Cloud PureCloud Platform Client V2 SDK.
• The implementation is provided in Python 3.9+, using the official genesys-cloud-sdk package.

Prerequisites

• OAuth Client Type: client_credentials or jwt. For production batch jobs, client_credentials is preferred as it does not require a user context.
• Required Scopes:
  • quality:survey:read (to read survey responses)
  • analytics:conversations:read (optional, if you need to enrich with conversation metrics)
  • conversation:read (optional, if you need full transcript data via the Conversation API)
• SDK Version: genesys-cloud-sdk>=2.0.0
• Runtime: Python 3.9 or higher.
• Dependencies:

  pip install genesys-cloud-sdk python-dotenv
  

Authentication Setup

Genesys Cloud uses OAuth 2.0 for authentication. The SDK handles the token exchange and refresh logic automatically when initialized correctly. You must store your client ID and secret securely. Never hardcode credentials in source code.

Create a .env file in your project root:

GENESYS_REGION=us-east-1
GENESYS_CLIENT_ID=your_client_id_here
GENESYS_CLIENT_SECRET=your_client_secret_here

Initialize the SDK in your Python script. The Configuration object manages the API base URL based on the region.

import os
from dotenv import load_dotenv
from purecloudplatformclientv2 import (
    Configuration,
    ApiClient,
    QualityApi,
    ConversationApi
)

# Load environment variables
load_dotenv()

def get_quality_api_instance() -> QualityApi:
    """
    Initializes and returns a configured QualityApi instance.
    """
    # Create configuration object
    config = Configuration(
        region=os.getenv('GENESYS_REGION', 'us-east-1'),
        client_id=os.getenv('GENESYS_CLIENT_ID'),
        client_secret=os.getenv('GENESYS_CLIENT_SECRET')
    )
    
    # Create API client
    api_client = ApiClient(configuration=config)
    
    # Initialize the Quality API resource
    quality_api = QualityApi(api_client)
    
    return quality_api

def get_conversation_api_instance() -> ConversationApi:
    """
    Initializes and returns a configured ConversationApi instance.
    """
    config = Configuration(
        region=os.getenv('GENESYS_REGION', 'us-east-1'),
        client_id=os.getenv('GENESYS_CLIENT_ID'),
        client_secret=os.getenv('GENESYS_CLIENT_SECRET')
    )
    
    api_client = Api_client(configuration=config)
    return ConversationApi(api_client)

Implementation

Step 1: Query Survey Responses

The core entry point is get_quality_surveysurveyresponse. This endpoint returns a list of survey responses. Unlike the Analytics API, which aggregates data, the Quality API returns individual response records.

You must filter by date range. The API does not support “last N days” relative filters; you must provide absolute ISO 8601 timestamps.

from purecloudplatformclientv2 import (
    GetQualitySurveySurveyResponseRequest,
    SurveyResponseQuery
)
from datetime import datetime, timedelta
import pytz

def fetch_survey_responses(quality_api: QualityApi, days_back: int = 7):
    """
    Fetches survey responses from the last N days.
    
    Args:
        quality_api: The initialized QualityApi instance.
        days_back: Number of days to look back.
        
    Returns:
        List of SurveyResponse objects.
    """
    # Define time range
    now = datetime.now(pytz.utc)
    start_time = now - timedelta(days=days_back)
    
    # Format as ISO 8601 string required by the API
    start_time_str = start_time.strftime('%Y-%m-%dT%H:%M:%SZ')
    end_time_str = now.strftime('%Y-%m-%dT%H:%M:%SZ')
    
    # Build the query object
    query = SurveyResponseQuery(
        start_date=start_time_str,
        end_date=end_time_str
    )
    
    # Prepare the request
    # max_records controls pagination. Default is 25, max is 500.
    request = GetQualitySurveySurveyResponseRequest(
        query=query,
        max_records=500
    )
    
    all_responses = []
    continuation_token = None
    
    while True:
        try:
            # Execute the API call
            response = quality_api.get_quality_survey_survey_response(
                query=query,
                max_records=500,
                continuation_token=continuation_token
            )
            
            # Append results
            if response.entities:
                all_responses.extend(response.entities)
            
            # Check for pagination
            if response.continuation_token:
                continuation_token = response.continuation_token
            else:
                break
                
        except Exception as e:
            print(f"Error fetching survey responses: {e}")
            break
            
    return all_responses

Key Parameters:

• query: The SurveyResponseQuery object filters by start_date and end_date. You can also filter by survey_id if you only want responses from a specific survey template.
• max_records: Set this to 500 to minimize API calls. The API enforces a hard limit of 500 per request.
• continuation_token: Essential for handling large datasets. If more records exist, the API returns a token. You must pass this token in the next request to get the next page.

Step 2: Extract Interaction IDs and Handle Nulls

Not all survey responses are tied to a specific interaction ID in a straightforward way. Some surveys are sent via email or SMS without a direct conversation link in the survey metadata, or the link is implicit.

The SurveyResponse object contains an interactions array. Each item in this array represents a linked conversation.

def extract_interaction_ids(responses: list) -> list:
    """
    Extracts interaction IDs from survey responses.
    
    Args:
        responses: List of SurveyResponse objects.
        
    Returns:
        List of dictionaries with survey_id and interaction_id.
    """
    extracted_data = []
    
    for response in responses:
        survey_id = response.id
        response_date = response.responded_at
        
        # Check if interactions are linked
        if response.interactions:
            for interaction in response.interactions:
                # interaction.id is the conversation ID
                interaction_id = interaction.id
                
                extracted_data.append({
                    'survey_id': survey_id,
                    'interaction_id': interaction_id,
                    'responded_at': response_date,
                    'survey_score': response.score,
                    'comments': response.comments
                })
        else:
            # Handle unlinked surveys
            extracted_data.append({
                'survey_id': survey_id,
                'interaction_id': None,
                'responded_at': response_date,
                'survey_score': response.score,
                'comments': response.comments
            })
            
    return extracted_data

Edge Case:

• Null Interactions: If response.interactions is empty or null, the survey was likely sent independently (e.g., a standalone email survey). In this case, interaction_id will be None. You cannot fetch transcript data for these without additional mapping logic (e.g., matching email addresses).

Step 3: Enrich with Conversation Metadata

To get the “specific interaction” details (like participant IDs, channel type, or transcript), you must call the ConversationApi. The Quality API does not return full conversation transcripts.

You will use the interaction IDs extracted in Step 2 to fetch conversation details.

def fetch_conversation_details(conversation_api: ConversationApi, interaction_id: str) -> dict:
    """
    Fetches details for a single conversation.
    
    Args:
        conversation_api: The initialized ConversationApi instance.
        interaction_id: The ID of the conversation.
        
    Returns:
        Dictionary with conversation metadata.
    """
    try:
        # Fetch conversation details
        conv_response = conversation_api.get_conversation(
            conversation_id=interaction_id
        )
        
        # Extract relevant fields
        return {
            'conversation_id': conv_response.id,
            'type': conv_response.type,
            'start_time': conv_response.start_time,
            'end_time': conv_response.end_time,
            'participants': [p.id for p in conv_response.participants if p.id]
        }
        
    except Exception as e:
        # Handle 404 (conversation deleted) or 403 (no access)
        print(f"Could not fetch conversation {interaction_id}: {e}")
        return None

Step 4: Batch Processing and Rate Limiting

Calling the Conversation API for every single survey response can trigger 429 Rate Limit errors. Genesys Cloud enforces rate limits per API endpoint. You must implement throttling or batch requests.

The ConversationApi supports get_conversations to fetch multiple conversations at once. This is significantly more efficient.

import time

def fetch_bulk_conversations(conversation_api: ConversationApi, interaction_ids: list, batch_size: int = 100) -> dict:
    """
    Fetches details for multiple conversations in batches.
    
    Args:
        conversation_api: The initialized ConversationApi instance.
        interaction_ids: List of conversation IDs.
        batch_size: Number of IDs to fetch per request.
        
    Returns:
        Dictionary mapping interaction_id to conversation metadata.
    """
    conversation_map = {}
    
    # Split IDs into batches
    for i in range(0, len(interaction_ids), batch_size):
        batch_ids = interaction_ids[i:i+batch_size]
        
        try:
            # Fetch batch
            response = conversation_api.get_conversations(
                conversation_ids=batch_ids
            )
            
            # Map results
            for conv in response.entities:
                conversation_map[conv.id] = {
                    'conversation_id': conv.id,
                    'type': conv.type,
                    'start_time': conv.start_time,
                    'end_time': conv.end_time,
                    'participants': [p.id for p in conv.participants if p.id]
                }
                
            # Respect rate limits: wait briefly between batches
            time.sleep(1)
            
        except Exception as e:
            print(f"Error fetching batch: {e}")
            # Implement exponential backoff here for production
            time.sleep(5)
            
    return conversation_map

Complete Working Example

This script combines all steps. It fetches survey responses, extracts interaction IDs, enriches them with conversation data, and prints the result.

import os
import json
from dotenv import load_dotenv
from purecloudplatformclientv2 import (
    Configuration,
    ApiClient,
    QualityApi,
    ConversationApi,
    SurveyResponseQuery,
    GetQualitySurveySurveyResponseRequest
)
from datetime import datetime, timedelta
import pytz
import time

# Load environment variables
load_dotenv()

def init_apis():
    """Initialize and return Quality and Conversation API instances."""
    config = Configuration(
        region=os.getenv('GENESYS_REGION', 'us-east-1'),
        client_id=os.getenv('GENESYS_CLIENT_ID'),
        client_secret=os.getenv('GENESYS_CLIENT_SECRET')
    )
    
    api_client = ApiClient(configuration=config)
    quality_api = QualityApi(api_client)
    conversation_api = ConversationApi(api_client)
    
    return quality_api, conversation_api

def fetch_survey_responses(quality_api, days_back=7):
    """Fetch all survey responses from the last N days."""
    now = datetime.now(pytz.utc)
    start_time = now - timedelta(days=days_back)
    
    query = SurveyResponseQuery(
        start_date=start_time.strftime('%Y-%m-%dT%H:%M:%SZ'),
        end_date=now.strftime('%Y-%m-%dT%H:%M:%SZ')
    )
    
    all_responses = []
    continuation_token = None
    
    while True:
        try:
            response = quality_api.get_quality_survey_survey_response(
                query=query,
                max_records=500,
                continuation_token=continuation_token
            )
            
            if response.entities:
                all_responses.extend(response.entities)
            
            if response.continuation_token:
                continuation_token = response.continuation_token
            else:
                break
                
        except Exception as e:
            print(f"Error fetching survey responses: {e}")
            break
            
    return all_responses

def process_and_enrich(quality_api, conversation_api):
    """Main processing logic."""
    print("Fetching survey responses...")
    responses = fetch_survey_responses(quality_api, days_back=1)
    
    if not responses:
        print("No survey responses found.")
        return

    print(f"Found {len(responses)} survey responses. Extracting interaction IDs...")
    
    # Step 1: Extract Interaction IDs
    interaction_ids = []
    survey_mapping = {}
    
    for resp in responses:
        if resp.interactions:
            for interaction in resp.interactions:
                interaction_id = interaction.id
                if interaction_id:
                    interaction_ids.append(interaction_id)
                    survey_mapping[interaction_id] = {
                        'survey_id': resp.id,
                        'score': resp.score,
                        'comments': resp.comments,
                        'responded_at': resp.responded_at
                    }
    
    # Deduplicate interaction IDs
    unique_interaction_ids = list(set(interaction_ids))
    
    if not unique_interaction_ids:
        print("No linked interactions found.")
        return

    print(f"Enriching {len(unique_interaction_ids)} interactions...")
    
    # Step 2: Fetch Conversation Details in Batches
    conversation_map = {}
    batch_size = 100
    
    for i in range(0, len(unique_interaction_ids), batch_size):
        batch_ids = unique_interaction_ids[i:i+batch_size]
        try:
            response = conversation_api.get_conversations(conversation_ids=batch_ids)
            for conv in response.entities:
                conversation_map[conv.id] = {
                    'type': conv.type,
                    'start_time': conv.start_time,
                    'end_time': conv.end_time,
                    'participants': [p.id for p in conv.participants if p.id]
                }
            time.sleep(1) # Rate limiting
        except Exception as e:
            print(f"Error fetching conversations: {e}")
            time.sleep(5)

    # Step 3: Merge Data
    final_data = []
    for interaction_id, survey_data in survey_mapping.items():
        conv_data = conversation_map.get(interaction_id, {})
        
        final_record = {
            'interaction_id': interaction_id,
            'survey_id': survey_data['survey_id'],
            'survey_score': survey_data['score'],
            'survey_comments': survey_data['comments'],
            'survey_responded_at': survey_data['responded_at'],
            'conversation_type': conv_data.get('type'),
            'conversation_start_time': conv_data.get('start_time'),
            'conversation_end_time': conv_data.get('end_time'),
            'participants': conv_data.get('participants', [])
        }
        final_data.append(final_record)

    # Output Result
    print("\nFinal Enriched Data:")
    print(json.dumps(final_data, indent=2, default=str))

if __name__ == "__main__":
    quality_api, conversation_api = init_apis()
    process_and_enrich(quality_api, conversation_api)

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: Invalid client ID, secret, or expired token.
• Fix: Verify your .env file values. Ensure the OAuth client is active in the Genesys Cloud Admin console. Check that the client_credentials grant type is enabled for the client.

Error: 403 Forbidden

• Cause: Missing scopes or insufficient permissions.
• Fix: Ensure the OAuth client has the quality:survey:read scope. If using a JWT user, ensure the user role has “Read survey responses” permission in the Admin console under Admin > Users > Roles.

Error: 429 Too Many Requests

• Cause: Exceeding the rate limit for get_conversations or get_quality_survey_survey_response.
• Fix: Implement exponential backoff. In the example above, a time.sleep(1) is used. For high-volume jobs, implement a retry loop with jitter:

import random

def api_call_with_retry(func, *args, retries=3, base_delay=1):
    for attempt in range(retries):
        try:
            return func(*args)
        except Exception as e:
            if "429" in str(e) or "RateLimit" in str(e):
                delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                print(f"Rate limited. Retrying in {delay:.2f} seconds...")
                time.sleep(delay)
            else:
                raise
    raise Exception("Max retries exceeded")

Error: Empty interactions Array

• Cause: The survey was not linked to a conversation.
• Fix: This is expected behavior for standalone surveys. If you expect a link, verify the survey template configuration in Genesys Cloud. Ensure the survey is configured to attach to the conversation context (e.g., via genesys.cloud.survey integration).

Official References

• Genesys Cloud Quality API Documentation
• Survey Response Object Definition
• Conversation API Documentation
• Python SDK Repository