Extract CSAT Survey Responses Tied to Interactions via the Quality API

What You Will Build

This tutorial demonstrates how to retrieve Customer Satisfaction (CSAT) survey results and join them to specific conversation interactions using the Genesys Cloud Quality API. You will build a Python script that queries the Quality Analytics endpoint to fetch completed survey responses, extracts the associated interaction IDs, and retrieves the full conversation details to correlate agent performance with customer sentiment. The solution uses the Genesys Cloud Python SDK.

Prerequisites

• OAuth Client Type: You need a Genesys Cloud OAuth Client with the api type.
• Required Scopes:
  • quality:analytics:read - Required to query survey analytics data.
  • conversation:interaction:read - Required to fetch details about the specific interactions linked to the surveys.
  • user:read - Optional, but recommended if you wish to resolve agent IDs to names.
• SDK Version: genesys-cloud-py-sdk version 133.0.0 or later.
• Language/Runtime: Python 3.9+.
• External Dependencies:
  • genesys-cloud-py-sdk
  • python-dotenv (for secure credential management)

Install the dependencies using pip:

pip install genesys-cloud-py-sdk python-dotenv

Authentication Setup

Genesys Cloud uses OAuth 2.0 for authentication. The SDK handles the token acquisition and refresh cycles automatically when you initialize the PlatformClient. You must store your client credentials securely. For this tutorial, we will use environment variables loaded via .env.

Create a .env file in your project root:

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

The following code initializes the SDK client. Note that the SDK caches the access token and handles refresh tokens automatically.

import os
from dotenv import load_dotenv
from purecloud_platform_client import PlatformClient, Configuration

# Load environment variables
load_dotenv()

def get_platform_client() -> PlatformClient:
    """
    Initializes and returns the Genesys Cloud Platform Client.
    """
    # Configure the client with region and credentials
    configuration = Configuration(
        region=os.getenv("GENESYS_CLOUD_REGION"),
        client_id=os.getenv("GENESYS_CLOUD_CLIENT_ID"),
        client_secret=os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
    )
    
    return PlatformClient(configuration)

# Initialize the client
client = get_platform_client()

Implementation

Step 1: Querying Survey Analytics Data

The Quality API provides an analytics endpoint that aggregates survey responses. Unlike standard conversation queries, survey data is accessed via /api/v2/quality/analyses/surveys. This endpoint returns a list of survey response records. Each record contains the interactionId which serves as the foreign key to the conversation data.

You must construct a query body that specifies the time range, the survey ID (optional, but recommended for specificity), and the metrics you require.

OAuth Scope: quality:analytics:read

from purecloud_platform_client import QualityApi
from purecloud_platform_client.models import SurveyQueryRequest
from datetime import datetime, timedelta

def fetch_survey_responses(platform_client: PlatformClient, survey_id: str = None) -> list:
    """
    Fetches survey responses from the Quality Analytics API.
    
    Args:
        platform_client: The initialized PlatformClient instance.
        survey_id: Optional ID of the specific survey to query. If None, queries all active surveys.
        
    Returns:
        A list of survey response objects.
    """
    quality_api = QualityApi(platform_client)
    
    # Define the time window for the query
    # The API requires ISO 8601 format timestamps
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(days=7)
    
    # Construct the query request
    # The 'groupings' parameter allows you to aggregate data. 
    # For individual response details, we often query without grouping or group by interactionId.
    query_body = SurveyQueryRequest(
        start_time=start_time.isoformat() + "Z",
        end_time=end_time.isoformat() + "Z",
        groupings=["interactionId"] if survey_id else [], 
        # If we want specific survey, we filter by it. Otherwise, we might need to filter post-fetch.
        # Note: The API supports filtering by surveyId in some contexts, but often it is easier to 
        # fetch all and filter, or use the specific survey endpoint if available.
        # For this example, we assume we are querying a specific survey ID for precision.
        filters=[{"type": "surveyId", "values": [survey_id]}] if survey_id else []
    )
    
    try:
        # Execute the query
        # The post_quality_analyses_surveys method returns a SurveyAnalyticsResponse object
        response = quality_api.post_quality_analyses_surveys(body=query_body)
        
        # The response contains a 'results' list. 
        # Each item in results represents an aggregated row based on groupings.
        # If we grouped by interactionId, each result corresponds to one survey submission.
        
        if response.results:
            return response.results
        else:
            print("No survey responses found in the specified time window.")
            return []
            
    except Exception as e:
        print(f"Error fetching survey analytics: {e}")
        return []

# Example usage:
# survey_id = "your-survey-id-here"
# survey_results = fetch_survey_responses(client, survey_id)

Expected Response Structure:
The response.results list contains objects with fields such as:

• interactionId: The unique ID of the conversation.
• surveyId: The ID of the survey taken.
• score: The numerical score provided by the customer (e.g., 1-5).
• comments: Free-text feedback from the customer.
• submittedTime: When the survey was submitted.

Step 2: Resolving Interaction Details

Once you have the list of survey results, you need to fetch the conversation details for each interactionId. This allows you to see who the agent was, what channel was used, and the transcript.

OAuth Scope: conversation:interaction:read

Fetching interactions one by one can trigger rate limits. It is best practice to batch requests or use a retry mechanism. The SDK does not have a bulk fetch endpoint for interactions, so we will iterate through the survey results.

from purecloud_platform_client import ConversationsApi
import time

def fetch_interaction_details(platform_client: PlatformClient, interaction_ids: list[str]) -> dict:
    """
    Fetches detailed conversation data for a list of interaction IDs.
    
    Args:
        platform_client: The initialized PlatformClient instance.
        interaction_ids: A list of interaction ID strings.
        
    Returns:
        A dictionary mapping interactionId to its Conversation object.
    """
    conversations_api = ConversationsApi(platform_client)
    interaction_map = {}
    
    for interaction_id in interaction_ids:
        try:
            # Get the conversation details
            # The API returns a Conversation object
            conversation = conversations_api.get_conversations_interaction(interaction_id)
            interaction_map[interaction_id] = conversation
            
            # Small delay to respect rate limits (100ms)
            # Genesys Cloud allows 100 requests per second for this endpoint typically, 
            # but spreading them out prevents 429s in tight loops.
            time.sleep(0.1)
            
        except Exception as e:
            print(f"Error fetching interaction {interaction_id}: {e}")
            # Handle 404 (interaction not found) or 403 (permission denied)
            if "404" in str(e):
                print(f"Interaction {interaction_id} not found. It may have been deleted or expired.")
            elif "403" in str(e):
                print(f"Permission denied for interaction {interaction_id}. Check scopes.")
    
    return interaction_map

Step 3: Correlating and Processing Results

With both the survey data and the interaction details, you can now correlate the CSAT score with the specific agent and conversation metadata. We will create a structured output that combines these two data sources.

from typing import Dict, List, Any
from purecloud_platform_client.models import Conversation

def correlate_survey_and_interaction(
    survey_results: list, 
    interactions: Dict[str, Conversation]
) -> List[Dict[str, Any]]:
    """
    Merges survey analytics data with conversation interaction details.
    
    Args:
        survey_results: List of survey result objects from Quality API.
        interactions: Dictionary of interactionId to Conversation objects.
        
    Returns:
        A list of dictionaries containing merged data.
    """
    merged_data = []
    
    for survey in survey_results:
        interaction_id = survey.get("interactionId")
        
        if not interaction_id:
            continue
            
        conversation = interactions.get(interaction_id)
        
        if conversation:
            # Extract agent ID from the conversation
            # Conversations can have multiple participants. We look for the agent.
            agent_id = None
            if conversation.participants:
                for participant in conversation.participants:
                    if participant.type == "agent":
                        agent_id = participant.id
                        break
            
            # Construct the merged record
            record = {
                "interactionId": interaction_id,
                "surveyId": survey.get("surveyId"),
                "csatScore": survey.get("score"),
                "surveyComments": survey.get("comments"),
                "submittedTime": survey.get("submittedTime"),
                "channelType": conversation.type if conversation else None,
                "agentId": agent_id,
                "startTime": conversation.startTime if conversation else None,
                "durationSeconds": conversation.duration if conversation else None
            }
            merged_data.append(record)
        else:
            # Handle cases where interaction data is missing
            print(f"Warning: Interaction data missing for survey associated with {interaction_id}")
            
    return merged_data

Complete Working Example

The following script combines all steps into a single executable module. It fetches survey responses for a specific survey ID over the last 7 days, retrieves the corresponding interaction details, and prints a formatted summary.

import os
import time
from datetime import datetime, timedelta
from dotenv import load_dotenv
from purecloud_platform_client import PlatformClient, Configuration, QualityApi, ConversationsApi
from purecloud_platform_client.models import SurveyQueryRequest
from typing import Dict, List, Any

def get_platform_client() -> PlatformClient:
    load_dotenv()
    configuration = Configuration(
        region=os.getenv("GENESYS_CLOUD_REGION"),
        client_id=os.getenv("GENESYS_CLOUD_CLIENT_ID"),
        client_secret=os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
    )
    return PlatformClient(configuration)

def fetch_survey_responses(platform_client: PlatformClient, survey_id: str) -> list:
    quality_api = QualityApi(platform_client)
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(days=7)
    
    query_body = SurveyQueryRequest(
        start_time=start_time.isoformat() + "Z",
        end_time=end_time.isoformat() + "Z",
        groupings=["interactionId"],
        filters=[{"type": "surveyId", "values": [survey_id]}]
    )
    
    try:
        response = quality_api.post_quality_analyses_surveys(body=query_body)
        return response.results if response.results else []
    except Exception as e:
        print(f"Error fetching survey analytics: {e}")
        return []

def fetch_interaction_details(platform_client: PlatformClient, interaction_ids: list[str]) -> Dict[str, Any]:
    conversations_api = ConversationsApi(platform_client)
    interaction_map = {}
    
    for interaction_id in interaction_ids:
        try:
            conversation = conversations_api.get_conversations_interaction(interaction_id)
            interaction_map[interaction_id] = conversation
            time.sleep(0.1) # Rate limit courtesy
        except Exception as e:
            print(f"Error fetching interaction {interaction_id}: {e}")
            
    return interaction_map

def correlate_survey_and_interaction(survey_results: list, interactions: Dict[str, Any]) -> List[Dict[str, Any]]:
    merged_data = []
    for survey in survey_results:
        interaction_id = survey.get("interactionId")
        if not interaction_id:
            continue
            
        conversation = interactions.get(interaction_id)
        if conversation:
            agent_id = None
            if conversation.participants:
                for participant in conversation.participants:
                    if participant.type == "agent":
                        agent_id = participant.id
                        break
            
            record = {
                "interactionId": interaction_id,
                "surveyId": survey.get("surveyId"),
                "csatScore": survey.get("score"),
                "surveyComments": survey.get("comments"),
                "submittedTime": survey.get("submittedTime"),
                "channelType": conversation.type if conversation else None,
                "agentId": agent_id,
                "startTime": conversation.startTime if conversation else None
            }
            merged_data.append(record)
    return merged_data

def main():
    # Replace with your actual Survey ID
    TARGET_SURVEY_ID = "your-survey-id-here"
    
    if TARGET_SURVEY_ID == "your-survey-id-here":
        print("Please update TARGET_SURVEY_ID in the script with a valid Genesys Cloud Survey ID.")
        return

    print("Initializing Genesys Cloud Client...")
    client = get_platform_client()
    
    print(f"Fetching survey responses for Survey ID: {TARGET_SURVEY_ID}...")
    survey_results = fetch_survey_responses(client, TARGET_SURVEY_ID)
    
    if not survey_results:
        print("No survey responses found.")
        return

    print(f"Found {len(survey_results)} survey responses. Fetching interaction details...")
    
    # Extract unique interaction IDs
    interaction_ids = list(set([s.get("interactionId") for s in survey_results if s.get("interactionId")]))
    
    interactions = fetch_interaction_details(client, interaction_ids)
    
    print("Correlating data...")
    merged_data = correlate_survey_and_interaction(survey_results, interactions)
    
    print("\n--- CSAT Survey Summary ---")
    for record in merged_data:
        print(f"Interaction: {record['interactionId']}")
        print(f"Agent ID: {record['agentId']}")
        print(f"Channel: {record['channelType']}")
        print(f"CSAT Score: {record['csatScore']}")
        print(f"Comments: {record['surveyComments']}")
        print("-" * 30)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token is invalid, expired, or the client credentials are incorrect.
• Fix: Verify that GENESYS_CLOUD_CLIENT_ID and GENESYS_CLOUD_CLIENT_SECRET are correct in your .env file. Ensure the OAuth Client is not disabled in the Genesys Cloud Admin portal. The SDK handles refresh tokens, so this error usually indicates initial credential failure.

Error: 403 Forbidden

• Cause: The OAuth Client lacks the required scopes.
• Fix: Navigate to the Genesys Cloud Admin portal → Settings → Integrations → OAuth Clients. Select your client and ensure the following scopes are checked:
  • quality:analytics:read
  • conversation:interaction:read
    Save the changes. Note that existing tokens may need to be refreshed to pick up new scopes.

Error: 429 Too Many Requests

• Cause: You are exceeding the API rate limit. The get_conversations_interaction endpoint is frequently called in this pattern.
• Fix: Implement exponential backoff or simple delays (as shown in the time.sleep(0.1) example). If processing large datasets, consider batching requests or using the post_conversations_interactions_search endpoint if available for bulk retrieval, though individual fetch is often required for full detail.

Error: Empty Results

• Cause: The time window does not contain any survey responses, or the surveyId filter is incorrect.
• Fix: Verify that the SurveyQueryRequest time window (start_time and end_time) overlaps with actual survey submission activity. Check the surveyId against the list of available surveys in your Genesys Cloud instance. You can remove the filters parameter to see all survey activity and debug which survey IDs are active.

Official References

• Quality Analytics API Documentation
• Conversations API Documentation
• Genesys Cloud Python SDK Reference
• OAuth Scopes Documentation