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

What You Will Build

• This tutorial builds a Python script that retrieves detailed CSAT survey results and joins them with their original interaction metadata using the Genesys Cloud Quality API.
• This uses the Genesys Cloud v2 Quality API endpoints (/api/v2/quality/surveys/results/query and /api/v2/quality/interactions/details/query) and the genesys-cloud-py-sdk.
• The implementation covers Python 3.10+ using the official SDK and raw httpx for advanced pagination handling.

Prerequisites

• OAuth Client: A Genesys Cloud OAuth client with the following scopes:
  • quality:interaction:read
  • quality:survey:result:read
  • analytics:conversation:read (optional, if you need deep conversation context)
• SDK Version: genesys-cloud-py-sdk version 140.0.0 or higher.
• Runtime: Python 3.10 or higher.
• Dependencies:
  • genesys-cloud-py-sdk
  • httpx (for robust async HTTP handling in the raw API section)
  • pydantic (for data validation)

Install the dependencies:

pip install genesys-cloud-py-sdk httpx pydantic

Authentication Setup

Genesys Cloud uses OAuth 2.0. For server-to-server integrations, you must use the Client Credentials grant flow. This flow issues a token that is valid for 15 minutes. You must cache this token and refresh it when it expires.

The following class handles authentication using the SDK’s built-in OAuthClient.

import os
import logging
from typing import Optional
from genesyscloud.auth.api_client import ApiClient
from genesyscloud.auth.oauth_client import OAuthClient
from genesyscloud.api_client import ApiClient as CoreApiClient

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

class GenesysAuth:
    """
    Handles OAuth2 Client Credentials flow for Genesys Cloud.
    """
    def __init__(self, client_id: str, client_secret: str, environment: str = "mypurecloud.com"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.environment = environment
        self.token: Optional[str] = None
        self.api_client: Optional[CoreApiClient] = None
        
    def authenticate(self) -> CoreApiClient:
        """
        Authenticates and returns a configured ApiClient.
        """
        try:
            # Create the OAuth client
            oauth_client = OAuthClient(self.client_id, self.client_secret, self.environment)
            
            # Generate the access token
            # The SDK handles the initial token fetch automatically upon first API call
            # but we can explicitly trigger it for clarity
            self.token = oauth_client.get_access_token()
            
            # Create the core API client
            self.api_client = CoreApiClient(
                client_id=self.client_id,
                client_secret=self.client_secret,
                environment=self.environment,
                oauth_client=oauth_client
            )
            
            logger.info("Authentication successful.")
            return self.api_client
            
        except Exception as e:
            logger.error(f"Authentication failed: {e}")
            raise

# Example Usage
# auth = GenesysAuth(
#     client_id=os.getenv("GENESYS_CLIENT_ID"),
#     client_secret=os.getenv("GENESYS_CLIENT_SECRET")
# )
# api_client = auth.authenticate()

Implementation

Step 1: Querying CSAT Survey Results

The Quality API does not return a flat list of all surveys. It uses a query-based endpoint that accepts a JSON body defining the date range, filter criteria, and sorting. This is critical for performance because survey data can be massive.

We will use the SurveyResultApi class from the SDK.

Required Scope: quality:survey:result:read

from genesyscloud.quality.survey_result_api import SurveyResultApi
from genesyscloud.models import SurveyResultQuery, SurveyResultFilter, SurveyResultSort

def get_csat_results(api_client: CoreApiClient, start_date: str, end_date: str) -> list:
    """
    Retrieves CSAT survey results within a date range.
    
    Args:
        api_client: The authenticated ApiClient.
        start_date: ISO 8601 start date (e.g., "2023-10-01T00:00:00.000Z")
        end_date: ISO 8601 end date (e.g., "2023-10-02T00:00:00.000Z")
        
    Returns:
        A list of SurveyResult objects.
    """
    survey_api = SurveyResultApi(api_client)
    
    # Define the query body
    query_body = SurveyResultQuery(
        start_date=start_date,
        end_date=end_date,
        # Optional: Filter only by specific survey IDs if known
        # survey_ids=["your-survey-id-1", "your-survey-id-2"],
        # Optional: Filter by specific score range (1-5)
        # filters=[
        #     SurveyResultFilter(
        #         field="score",
        #         operator="greater_than_or_equal",
        #         value=4
        #     )
        # ],
        # Sort by most recent first
        sort=[SurveyResultSort(field="completed_time", order="desc")],
        page_size=100
    )
    
    results = []
    next_page_token = None
    
    while True:
        try:
            # Execute the query
            response = survey_api.post_quality_survey_results_query(
                body=query_body,
                page_token=next_page_token
            )
            
            # Add results to our list
            if response.entities:
                results.extend(response.entities)
            
            # Check for pagination
            if response.next_page_token:
                next_page_token = response.next_page_token
                logger.info(f"Retrieved {len(results)} results so far. Fetching next page...")
            else:
                break
                
        except Exception as e:
            logger.error(f"Error fetching survey results: {e}")
            raise
            
    return results

# Example Usage:
# results = get_csat_results(api_client, "2023-10-01T00:00:00.000Z", "2023-10-31T23:59:59.999Z")

Step 2: Extracting Interaction IDs from Survey Responses

Each SurveyResult object contains an interaction_id field. This is the unique identifier that links the survey to the original conversation (voice, chat, email, etc.). However, the survey result itself does not contain the conversation details (like agent name, queue, or transcript).

We must extract these IDs and prepare them for a batch lookup in the Interaction Details API.

from typing import List

def extract_interaction_ids(survey_results: List[object]) -> List[str]:
    """
    Extracts unique interaction IDs from survey results.
    
    Args:
        survey_results: List of SurveyResult objects.
        
    Returns:
        List of unique interaction IDs.
    """
    interaction_ids = set()
    
    for result in survey_results:
        if result.interaction_id:
            interaction_ids.add(result.interaction_id)
        else:
            logger.warning(f"Survey result {result.id} has no interaction_id. Skipping.")
            
    return list(interaction_ids)

# Example Usage:
# ids = extract_interaction_ids(results)

Step 3: Fetching Interaction Details via Quality API

Now we need to fetch the details of these interactions. The Genesys Cloud Quality API provides an endpoint to retrieve interaction details by ID. This is more efficient than querying the Analytics Conversation API for simple metadata lookups.

Required Scope: quality:interaction:read

We will use the InteractionApi class. Note that the API supports fetching up to 100 interactions at a time.

from genesyscloud.quality.interaction_api import InteractionApi
from genesyscloud.models import InteractionDetailQuery

def get_interaction_details(api_client: CoreApiClient, interaction_ids: List[str]) -> dict:
    """
    Retrieves interaction details for a list of interaction IDs.
    
    Args:
        api_client: The authenticated ApiClient.
        interaction_ids: List of interaction IDs.
        
    Returns:
        A dictionary mapping interaction_id to InteractionDetail object.
    """
    interaction_api = InteractionApi(api_client)
    details_map = {}
    
    # Batch size limit for the API is 100
    batch_size = 100
    
    for i in range(0, len(interaction_ids), batch_size):
        batch_ids = interaction_ids[i:i + batch_size]
        
        try:
            # Create the query body
            query_body = InteractionDetailQuery(
                ids=batch_ids
            )
            
            # Fetch details
            response = interaction_api.post_quality_interactions_details_query(body=query_body)
            
            if response.entities:
                for interaction in response.entities:
                    details_map[interaction.id] = interaction
                    
        except Exception as e:
            logger.error(f"Error fetching interaction details for batch: {e}")
            continue
            
    return details_map

# Example Usage:
# interaction_details = get_interaction_details(api_client, ids)

Step 4: Joining Survey Results with Interaction Data

Finally, we join the data. This step is crucial for creating a meaningful report. We want to know not just the score, but who handled the interaction, which queue it was in, and when it occurred.

from datetime import datetime
import json

def join_survey_and_interaction_data(survey_results: list, interaction_details: dict) -> list:
    """
    Joins survey results with their corresponding interaction details.
    
    Args:
        survey_results: List of SurveyResult objects.
        interaction_details: Dict mapping interaction_id to InteractionDetail.
        
    Returns:
        A list of dictionaries containing joined data.
    """
    joined_data = []
    
    for survey in survey_results:
        interaction_id = survey.interaction_id
        
        # Check if we have interaction details for this survey
        if interaction_id and interaction_id in interaction_details:
            interaction = interaction_details[interaction_id]
            
            # Extract relevant fields
            # Note: Field names may vary slightly based on interaction type (voice, chat, etc.)
            agent_name = "Unknown"
            queue_name = "Unknown"
            start_time = None
            
            if interaction.type == "voice":
                if interaction.voice_details:
                    if interaction.voice_details.agent_id:
                        # In a real app, you might want to look up the agent name via Ids API
                        # For now, we use the ID
                        agent_name = interaction.voice_details.agent_id
                    if interaction.voice_details.queue_id:
                        queue_name = interaction.voice_details.queue_id
                    start_time = interaction.voice_details.start_time
            elif interaction.type == "chat":
                if interaction.chat_details:
                    if interaction.chat_details.agent_id:
                        agent_name = interaction.chat_details.agent_id
                    if interaction.chat_details.queue_id:
                        queue_name = interaction.chat_details.queue_id
                    start_time = interaction.chat_details.start_time
            
            # Construct the joined record
            record = {
                "survey_id": survey.id,
                "score": survey.score,
                "completed_time": survey.completed_time,
                "interaction_id": interaction_id,
                "interaction_type": interaction.type,
                "agent_id": agent_name,
                "queue_id": queue_name,
                "interaction_start_time": start_time
            }
            
            joined_data.append(record)
        else:
            # Handle cases where interaction details are missing
            logger.warning(f"No interaction details found for survey {survey.id}")
            
    return joined_data

# Example Usage:
# final_data = join_survey_and_interaction_data(results, interaction_details)
# print(json.dumps(final_data, indent=2, default=str))

Complete Working Example

Below is the complete, runnable Python script. It combines authentication, survey querying, interaction fetching, and data joining into a single cohesive module.

import os
import logging
import json
from typing import List, Optional, Dict
from datetime import datetime, timedelta

# Genesys Cloud SDK Imports
from genesyscloud.auth.api_client import ApiClient
from genesyscloud.auth.oauth_client import OAuthClient
from genesyscloud.api_client import ApiClient as CoreApiClient
from genesyscloud.quality.survey_result_api import SurveyResultApi
from genesyscloud.quality.interaction_api import InteractionApi
from genesyscloud.models import SurveyResultQuery, SurveyResultSort, InteractionDetailQuery

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

class CsatExtractor:
    """
    Extracts CSAT survey results and joins them with interaction details.
    """
    
    def __init__(self, client_id: str, client_secret: str, environment: str = "mypurecloud.com"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.environment = environment
        self.api_client: Optional[CoreApiClient] = None
        
    def authenticate(self) -> None:
        """
        Authenticates with Genesys Cloud.
        """
        try:
            oauth_client = OAuthClient(self.client_id, self.client_secret, self.environment)
            self.api_client = CoreApiClient(
                client_id=self.client_id,
                client_secret=self.client_secret,
                environment=self.environment,
                oauth_client=oauth_client
            )
            logger.info("Authentication successful.")
        except Exception as e:
            logger.error(f"Authentication failed: {e}")
            raise
    
    def get_csat_results(self, days_back: int = 30) -> List:
        """
        Retrieves CSAT survey results for the last N days.
        """
        if not self.api_client:
            raise Exception("Not authenticated. Call authenticate() first.")
            
        end_date = datetime.utcnow()
        start_date = end_date - timedelta(days=days_back)
        
        start_str = start_date.strftime("%Y-%m-%dT%H:%M:%S.000Z")
        end_str = end_date.strftime("%Y-%m-%dT%H:%M:%S.000Z")
        
        survey_api = SurveyResultApi(self.api_client)
        
        query_body = SurveyResultQuery(
            start_date=start_str,
            end_date=end_str,
            sort=[SurveyResultSort(field="completed_time", order="desc")],
            page_size=100
        )
        
        results = []
        next_page_token = None
        
        while True:
            try:
                response = survey_api.post_quality_survey_results_query(
                    body=query_body,
                    page_token=next_page_token
                )
                
                if response.entities:
                    results.extend(response.entities)
                
                if response.next_page_token:
                    next_page_token = response.next_page_token
                else:
                    break
                    
            except Exception as e:
                logger.error(f"Error fetching survey results: {e}")
                raise
                
        logger.info(f"Retrieved {len(results)} survey results.")
        return results
    
    def get_interaction_details(self, interaction_ids: List[str]) -> Dict:
        """
        Retrieves interaction details for a list of IDs.
        """
        if not self.api_client:
            raise Exception("Not authenticated. Call authenticate() first.")
            
        interaction_api = InteractionApi(self.api_client)
        details_map = {}
        batch_size = 100
        
        for i in range(0, len(interaction_ids), batch_size):
            batch_ids = interaction_ids[i:i + batch_size]
            
            try:
                query_body = InteractionDetailQuery(ids=batch_ids)
                response = interaction_api.post_quality_interactions_details_query(body=query_body)
                
                if response.entities:
                    for interaction in response.entities:
                        details_map[interaction.id] = interaction
                        
            except Exception as e:
                logger.error(f"Error fetching interaction details: {e}")
                continue
                
        return details_map
    
    def extract_and_join(self, days_back: int = 30) -> List[Dict]:
        """
        Main execution method: Fetches surveys, gets interactions, and joins data.
        """
        logger.info("Starting CSAT extraction process...")
        
        # Step 1: Get Survey Results
        survey_results = self.get_csat_results(days_back)
        
        if not survey_results:
            logger.warning("No survey results found.")
            return []
        
        # Step 2: Extract Interaction IDs
        interaction_ids = []
        for survey in survey_results:
            if survey.interaction_id:
                interaction_ids.append(survey.interaction_id)
        
        if not interaction_ids:
            logger.warning("No interaction IDs found in survey results.")
            return []
        
        logger.info(f"Found {len(interaction_ids)} unique interaction IDs.")
        
        # Step 3: Get Interaction Details
        interaction_details = self.get_interaction_details(interaction_ids)
        
        # Step 4: Join Data
        joined_data = []
        for survey in survey_results:
            interaction_id = survey.interaction_id
            
            if interaction_id and interaction_id in interaction_details:
                interaction = interaction_details[interaction_id]
                
                record = {
                    "survey_id": survey.id,
                    "score": survey.score,
                    "completed_time": survey.completed_time,
                    "interaction_id": interaction_id,
                    "interaction_type": interaction.type,
                    "agent_id": self._get_agent_id(interaction),
                    "queue_id": self._get_queue_id(interaction),
                    "start_time": self._get_start_time(interaction)
                }
                joined_data.append(record)
        
        logger.info(f"Successfully joined {len(joined_data)} records.")
        return joined_data
    
    def _get_agent_id(self, interaction) -> str:
        """Helper to extract agent ID from interaction."""
        if interaction.type == "voice" and interaction.voice_details:
            return interaction.voice_details.agent_id or "Unknown"
        elif interaction.type == "chat" and interaction.chat_details:
            return interaction.chat_details.agent_id or "Unknown"
        return "Unknown"
    
    def _get_queue_id(self, interaction) -> str:
        """Helper to extract queue ID from interaction."""
        if interaction.type == "voice" and interaction.voice_details:
            return interaction.voice_details.queue_id or "Unknown"
        elif interaction.type == "chat" and interaction.chat_details:
            return interaction.chat_details.queue_id or "Unknown"
        return "Unknown"
    
    def _get_start_time(self, interaction) -> Optional[str]:
        """Helper to extract start time from interaction."""
        if interaction.type == "voice" and interaction.voice_details:
            return interaction.voice_details.start_time
        elif interaction.type == "chat" and interaction.chat_details:
            return interaction.chat_details.start_time
        return None

if __name__ == "__main__":
    # Load credentials from environment variables
    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 environment variables are required.")
    
    try:
        # Initialize and authenticate
        extractor = CsatExtractor(CLIENT_ID, CLIENT_SECRET)
        extractor.authenticate()
        
        # Extract and join data for the last 30 days
        final_data = extractor.extract_and_join(days_back=30)
        
        # Output results
        if final_data:
            print(json.dumps(final_data, indent=2, default=str))
        else:
            print("No data found.")
            
    except Exception as e:
        logger.critical(f"Fatal error: {e}")
        exit(1)

Common Errors & Debugging

Error: 401 Unauthorized

Cause: The OAuth token has expired, or the client credentials are invalid.
Fix: Ensure the GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET are correct. The SDK handles token refresh automatically, but if you are using raw HTTP requests, you must implement token caching.

Error: 403 Forbidden

Cause: The OAuth client lacks the required scopes.
Fix: Verify that the OAuth client has quality:survey:result:read and quality:interaction:read scopes enabled in the Genesys Cloud Admin Console under Admin > Security > OAuth clients.

Error: 429 Too Many Requests

Cause: You are hitting the API rate limit. The Quality API has strict rate limits.
Fix: Implement exponential backoff. In the complete example, the while loop for pagination should include a small delay (time.sleep(1)) between pages if you are fetching large datasets. For production systems, use a retry library like tenacity.

import time

# Inside the pagination loop
if response.next_page_token:
    time.sleep(0.5)  # Small delay to avoid rate limiting
    next_page_token = response.next_page_token

Error: Missing Interaction Details

Cause: The survey result references an interaction that is no longer accessible or has been archived beyond the retention period.
Fix: Check the interaction_id in the survey result. If it exists in the survey but not in the interaction details, the interaction may have been purged. Log these IDs for manual investigation.

Official References

• Genesys Cloud Quality Survey Results API Documentation
• Genesys Cloud Quality Interactions API Documentation
• Genesys Cloud Python SDK Reference