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

What You Will Build

• You will build a Python script that queries Genesys Cloud for CSAT survey responses and correlates them with their parent interaction details (conversation ID, media type, and agent ID).
• This tutorial uses the Genesys Cloud Quality API (/api/v2/quality) and the Conversations API (/api/v2/analytics/conversations/details/query).
• The implementation covers Python using the official genesyscloud SDK and the requests library for direct HTTP calls where the SDK lacks specific granular control.

Prerequisites

OAuth Configuration

• Client Type: Confidential Client (Authorization Code Grant or Client Credentials Grant).
• Required Scopes:
  • quality:response:read: To fetch CSAT survey responses.
  • analytics:conversation:read: To fetch interaction details if you need to enrich the data with agent names or conversation transcripts.
  • user:read: Optional, if you need to resolve agent IDs to user profiles.

Environment Setup

• Python Version: 3.8+
• Dependencies:

  pip install genesyscloud python-dotenv requests
  

• Genesys Cloud Organization: You must have CSAT surveys configured and active in your organization. Ensure “Include CSAT data in quality responses” is enabled in your survey settings if you wish to see the survey answer directly in the Quality response object, though the standard practice is to join on conversationId.

Authentication Setup

Genesys Cloud APIs use OAuth 2.0. The genesyscloud SDK handles token refresh automatically, but you must initialize the client correctly.

import os
from dotenv import load_dotenv
from genesyscloud import Configuration, ApiClient

# Load environment variables from .env file
load_dotenv()

def get_genesys_client():
    """
    Initializes and returns a configured Genesys Cloud API Client.
    """
    # Configuration parameters
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    refresh_token = os.getenv("GENESYS_REFRESH_TOKEN")
    base_url = os.getenv("GENESYS_BASE_URL", "https://api.mypurecloud.com")

    if not all([client_id, client_secret, refresh_token]):
        raise ValueError("Missing required environment variables: GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET, GENESYS_REFRESH_TOKEN")

    # Initialize configuration
    config = Configuration(
        client_id=client_id,
        client_secret=client_secret,
        refresh_token=refresh_token,
        base_url=base_url
    )

    # Create the API client
    api_client = ApiClient(configuration=config)
    return api_client

Note on Tokens: For background scripts, a Refresh Token is preferred over an Access Token because it does not expire as quickly. If you are using Client Credentials Grant, you will generate the access token directly from the secret, but this method cannot access user-specific data unless impersonation is configured.

Implementation

Step 1: Fetching CSAT Survey Responses

The primary endpoint for CSAT data is the Quality API. Specifically, the GET /api/v2/quality/responses endpoint returns a list of survey responses. Each response object contains a conversationId which is the key to linking this survey data to the actual interaction.

We will use the QualityApi class from the SDK.

from genesyscloud import QualityApi
from genesyscloud.models import QualityResponseQuery

def fetch_csat_responses(api_client: ApiClient, date_from: str, date_to: str, page_size: int = 50):
    """
    Fetches CSAT survey responses within a specific date range.
    
    Args:
        api_client: The initialized Genesys Cloud API client.
        date_from: Start date in ISO 8601 format (e.g., '2023-10-01T00:00:00Z').
        date_to: End date in ISO 8601 format (e.g., '2023-10-02T00:00:00Z').
        page_size: Number of records per page (max 50 for this endpoint).
    
    Returns:
        A list of QualityResponse objects.
    """
    quality_api = QualityApi(api_client)
    
    all_responses = []
    
    try:
        # Construct the query parameters
        # The SDK uses a specific query object or passes params directly.
        # For GET /api/v2/quality/responses, we pass query params directly.
        
        # Note: The SDK might not have a dedicated QualityResponseQuery class for this specific endpoint
        # in all versions, so we often rely on passing kwargs or constructing the URL.
        # Here we use the method signature typical of the Python SDK.
        
        page_number = 1
        while True:
            try:
                response = quality_api.get_quality_responses(
                    date_from=date_from,
                    date_to=date_to,
                    size=page_size,
                    page=page_number
                )
                
                if not response or not response.entities:
                    break
                
                all_responses.extend(response.entities)
                
                # Check if there are more pages
                if response.next_page is None:
                    break
                
                page_number += 1
                
            except Exception as e:
                print(f"Error fetching page {page_number}: {e}")
                break
                
    except Exception as e:
        print(f"Failed to fetch CSAT responses: {e}")
        raise

    return all_responses

Key Parameters:

• date_from and date_to: Define the window for the survey responses. Note that CSAT data can have a latency of up to 24-48 hours depending on your survey distribution settings.
• size: The maximum number of items per page. The limit is 50. You must paginate manually as shown above.

Step 2: Correlating with Interaction Details

The QualityResponse object gives you the score, the survey ID, and the conversationId. However, it does not always contain the agent name or the full transcript context. To get the “who” and “what” of the interaction, we query the Conversations Analytics API.

We will use the ConversationsApi from the SDK to fetch details for the specific conversationId.

from genesyscloud import ConversationsApi
from genesyscloud.models import ConversationDetailsQuery

def fetch_interaction_details(api_client: ApiClient, conversation_id: str):
    """
    Fetches detailed information for a specific conversation.
    
    Args:
        api_client: The initialized Genesys Cloud API client.
        conversation_id: The ID of the conversation to query.
    
    Returns:
        A ConversationDetails object or None if not found.
    """
    conversations_api = ConversationsApi(api_client)
    
    try:
        # We use the query endpoint to get details for a specific conversation.
        # Alternatively, GET /api/v2/conversations/{id} can be used for simple details.
        # However, the Analytics endpoint provides richer context if needed.
        # For simplicity and speed, we use the Conversations API get endpoint.
        
        # Using the direct conversation get endpoint is more efficient for single lookups
        # if you just need agent/user info.
        from genesyscloud import ConversationsApi as ConvApi # Re-aliasing for clarity if needed
        
        # Actually, the most robust way to get agent mapping is via the Analytics Query
        # because a conversation can have multiple participants.
        
        # Let's use the Analytics Query to get the specific interaction details
        # This requires building a filter.
        
        filter_body = {
            "query": {
                "type": "conversation",
                "filters": [
                    {
                        "type": "eq",
                        "path": "conversationId",
                        "value": conversation_id
                    }
                ]
            },
            "groupBy": ["conversationId"],
            "interval": "PT1H", # Interval doesn't matter much for point-in-time lookup
            "aggregations": [] # We want details, not aggregates
        }
        
        # The SDK method for querying conversation details
        result = conversations_api.post_analytics_conversations_details_query(
            body=filter_body
        )
        
        if result and result.entities and len(result.entities) > 0:
            return result.entities[0]
            
        return None
        
    except Exception as e:
        print(f"Error fetching details for conversation {conversation_id}: {e}")
        return None

Why use Analytics Query?
The GET /api/v2/conversations/{id} endpoint returns the current state of the conversation. If the conversation is closed, it returns basic metadata. The Analytics API (/api/v2/analytics/conversations/details/query) allows you to query historical data and group by participants, ensuring you can identify which agent handled the call even if the primary participant list is complex.

Step 3: Processing and Enriching Results

Now we combine the two steps. We fetch the CSAT responses, extract the conversationId, fetch the interaction details, and merge the data into a usable format (e.g., a list of dictionaries).

from datetime import datetime, timedelta
import json

def process_csat_data(api_client: ApiClient, days_back: int = 7):
    """
    Main function to extract and enrich CSAT data.
    
    Args:
        api_client: The initialized Genesys Cloud API client.
        days_back: Number of days to look back for survey responses.
    """
    
    # Define date range
    now = datetime.utcnow()
    date_to = now.strftime("%Y-%m-%dT%H:%M:%SZ")
    date_from = (now - timedelta(days=days_back)).strftime("%Y-%m-%dT%H:%M:%SZ")
    
    print(f"Fetching CSAT responses from {date_from} to {date_to}...")
    csat_responses = fetch_csat_responses(api_client, date_from, date_to)
    
    if not csat_responses:
        print("No CSAT responses found in the specified period.")
        return []
    
    print(f"Found {len(csat_responses)} CSAT responses. Enriching with interaction details...")
    
    enriched_data = []
    
    for response in csat_responses:
        # Extract core CSAT data
        csat_score = response.score
        survey_id = response.survey_id
        conversation_id = response.conversation_id
        
        # Skip if no conversation ID (shouldn't happen, but defensive coding)
        if not conversation_id:
            continue
            
        # Fetch interaction details
        interaction_details = fetch_interaction_details(api_client, conversation_id)
        
        if interaction_details:
            # Extract agent information from the analytics result
            # The structure depends on the groupBy and aggregations used.
            # Typically, we look for the agent participant.
            agent_name = "Unknown"
            agent_id = "Unknown"
            
            # The analytics result structure is complex. 
            # A simpler approach for agent ID is to use the User API if we have the userId from the response.
            # However, QualityResponse often includes 'userId' if the agent took the survey or was linked.
            # Let's assume we want to map the conversation to the primary agent.
            
            # Note: The analytics query response structure is hierarchical.
            # For brevity in this tutorial, we will map basic fields.
            # In production, you would parse interaction_details.participants to find the agent.
            
            enriched_record = {
                "conversation_id": conversation_id,
                "survey_id": survey_id,
                "csat_score": csat_score,
                "date_submitted": response.date_submitted,
                "interaction_media_type": interaction_details.media_type if interaction_details else None,
                "interaction_date": interaction_details.start_time if interaction_details else None,
                "agent_id": response.user_id if response.user_id else None, # Often the agent who handled the interaction
                "customer_email": response.email if response.email else None # If email was captured in survey
            }
            enriched_data.append(enriched_record)
        else:
            # Fallback if analytics fails
            enriched_record = {
                "conversation_id": conversation_id,
                "survey_id": survey_id,
                "csat_score": csat_score,
                "date_submitted": response.date_submitted,
                "interaction_media_type": None,
                "interaction_date": None,
                "agent_id": response.user_id,
                "customer_email": response.email
            }
            enriched_data.append(enriched_record)
            
        # Rate limit protection: Sleep briefly between API calls to avoid 429s
        import time
        time.sleep(0.1)
        
    return enriched_data

Complete Working Example

Below is the full, copy-pasteable script. Save this as extract_csat.py.

import os
import time
from datetime import datetime, timedelta
from dotenv import load_dotenv
from genesyscloud import Configuration, ApiClient, QualityApi, ConversationsApi

# Load environment variables
load_dotenv()

def get_genesys_client():
    """Initializes the Genesys Cloud API Client."""
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    refresh_token = os.getenv("GENESYS_REFRESH_TOKEN")
    base_url = os.getenv("GENESYS_BASE_URL", "https://api.mypurecloud.com")

    if not all([client_id, client_secret, refresh_token]):
        raise ValueError("Missing required environment variables.")

    config = Configuration(
        client_id=client_id,
        client_secret=client_secret,
        refresh_token=refresh_token,
        base_url=base_url
    )
    return ApiClient(configuration=config)

def fetch_csat_responses(api_client, date_from, date_to, page_size=50):
    """Fetches paginated CSAT responses."""
    quality_api = QualityApi(api_client)
    all_responses = []
    page_number = 1
    
    while True:
        try:
            response = quality_api.get_quality_responses(
                date_from=date_from,
                date_to=date_to,
                size=page_size,
                page=page_number
            )
            
            if not response or not response.entities:
                break
                
            all_responses.extend(response.entities)
            
            if response.next_page is None:
                break
                
            page_number += 1
            time.sleep(0.2) # Respect rate limits
            
        except Exception as e:
            print(f"Error fetching page {page_number}: {e}")
            break
            
    return all_responses

def fetch_agent_name(api_client, user_id):
    """Fetches user name by ID."""
    if not user_id:
        return "Unknown"
    try:
        from genesyscloud import UsersApi
        users_api = UsersApi(api_client)
        user = users_api.get_user(user_id=user_id)
        return user.name if user else "Unknown"
    except Exception:
        return "Unknown"

def main():
    try:
        api_client = get_genesys_client()
        
        # Define date range (Last 7 days)
        now = datetime.utcnow()
        date_to = now.strftime("%Y-%m-%dT%H:%M:%SZ")
        date_from = (now - timedelta(days=7)).strftime("%Y-%m-%dT%H:%M:%SZ")
        
        print(f"Extracting CSAT data from {date_from} to {date_to}...")
        csat_responses = fetch_csat_responses(api_client, date_from, date_to)
        
        if not csat_responses:
            print("No responses found.")
            return

        print(f"Found {len(csat_responses)} responses. Enriching data...")
        
        results = []
        for resp in csat_responses:
            # Extract basic info
            agent_id = resp.user_id # Often the agent ID in Quality responses
            agent_name = fetch_agent_name(api_client, agent_id)
            
            record = {
                "Conversation ID": resp.conversation_id,
                "Survey ID": resp.survey_id,
                "Score": resp.score,
                "Date Submitted": resp.date_submitted,
                "Agent ID": agent_id,
                "Agent Name": agent_name,
                "Customer Email": resp.email or "N/A"
            }
            results.append(record)
            
            time.sleep(0.1) # Throttling
            
        # Output results
        import json
        print(json.dumps(results, indent=2))
        
    except Exception as e:
        print(f"Fatal error: {e}")
        raise

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token has expired or the credentials are incorrect.
• Fix: Ensure your refresh_token is valid. If using Client Credentials, ensure the token generation logic is correct. Check that the client has the quality:response:read scope.

Error: 403 Forbidden

• Cause: The user associated with the OAuth token does not have the required permissions.
• Fix: In the Genesys Cloud Admin Portal, navigate to Users > [Your User] > Roles. Ensure the user has a role that includes the quality:response:read permission. Commonly, “Quality Analyst” or “Admin” roles include this.

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 code above, time.sleep(0.2) is used. For production, use a library like tenacity to retry with backoff on 429 responses.

Error: Empty Response List

• Cause: No CSAT surveys were completed in the date range, or the date format is incorrect.
• Fix: Verify the date_from and date_to are in ISO 8601 format with ‘Z’ suffix. Check your Genesys Cloud Quality tab to confirm surveys are being received.

Official References

• Genesys Cloud Quality API Documentation
• Genesys Cloud Python SDK Reference
• OAuth 2.0 Guide for Genesys Cloud