Extracting CSAT Survey Responses Tied to Interactions via the Quality API

What You Will Build

• You will build a Python script that queries Genesys Cloud for specific conversation IDs and retrieves the associated Customer Satisfaction (CSAT) survey data.
• This tutorial uses the Genesys Cloud CX Quality API (/api/v2/quality/conversations) and the Analytics API to correlate interaction metadata with survey scores.
• The implementation is written in Python using the requests library and the official genesyscloud SDK for type safety and convenience.

Prerequisites

OAuth Configuration

• Client Type: Confidential Client (Client Credentials Grant).
• Required Scopes:
  • quality:conversation:view (To access conversation quality details and survey data).
  • analytics:conversation:view (To query conversation details if needed for metadata enrichment).
  • user:read (Optional, if you need to resolve user IDs to names).

Environment Setup

• Python Version: 3.9 or higher.
• Dependencies:
  • genesyscloud: The official Python SDK.
  • requests: For raw HTTP interactions where the SDK lacks specific granularity.
  • python-dotenv: For secure credential management.

Install dependencies via pip:

pip install genesyscloud requests python-dotenv

Authentication Setup

Genesys Cloud uses OAuth 2.0 for authentication. The most common pattern for server-to-server integrations is the Client Credentials Grant. You must cache the access token and handle expiration.

import os
import time
import requests
from dotenv import load_dotenv

load_dotenv()

class GenesysAuth:
    def __init__(self):
        self.client_id = os.getenv("GENESYS_CLIENT_ID")
        self.client_secret = os.getenv("GENESYS_CLIENT_SECRET")
        self.env_url = os.getenv("GENESYS_ENV_URL", "https://api.mypurecloud.com")
        self.access_token = None
        self.token_expiry = 0

    def get_token(self) -> str:
        """
        Retrieves an OAuth2 access token using Client Credentials Grant.
        Implements basic caching to avoid unnecessary token requests.
        """
        # Return cached token if not expired (add 5 minute buffer)
        if self.access_token and time.time() < (self.token_expiry - 300):
            return self.access_token

        token_url = f"{self.env_url}/oauth/token"
        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }
        data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        try:
            response = requests.post(token_url, headers=headers, data=data)
            response.raise_for_status()
            token_data = response.json()
            
            self.access_token = token_data["access_token"]
            self.token_expiry = time.time() + token_data["expires_in"]
            return self.access_token
            
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise Exception("Authentication failed: Invalid Client ID or Secret.")
            raise Exception(f"Failed to obtain token: {e}")
        except requests.exceptions.RequestException as e:
            raise Exception(f"Network error during authentication: {e}")

# Initialize Auth
auth = GenesysAuth()

Implementation

Step 1: Query Conversation Details to Identify Survey Eligibility

Before fetching survey data, you often need to identify which conversations actually have surveys attached. The Genesys Cloud Quality API does not have a direct “list all surveys” endpoint that filters by arbitrary conversation IDs efficiently without first knowing the conversation exists in the quality context.

However, the most direct path to CSAT data is via the Quality Conversations API. This endpoint returns conversation details including the survey object if a survey was completed for that interaction.

Endpoint: GET /api/v2/quality/conversations/{conversationId}
Scope: quality:conversation:view

import json
from typing import Optional, Dict, Any

class QualityService:
    def __init__(self, auth: GenesysAuth):
        self.auth = auth
        self.base_url = auth.env_url

    def get_conversation_survey_data(self, conversation_id: str) -> Optional[Dict[str, Any]]:
        """
        Fetches quality conversation details for a specific ID.
        Extracts the survey data if present.
        """
        endpoint = f"{self.base_url}/api/v2/quality/conversations/{conversation_id}"
        token = self.auth.get_token()
        
        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }

        try:
            response = requests.get(endpoint, headers=headers)
            
            # Handle 404: Conversation not found or not eligible for quality review
            if response.status_code == 404:
                print(f"Conversation {conversation_id} not found in Quality API.")
                return None
            
            # Handle 403: Insufficient permissions
            if response.status_code == 403:
                raise Exception(f"Forbidden: Check OAuth scopes for quality:conversation:view")
            
            response.raise_for_status()
            data = response.json()
            
            # The survey data is nested within the conversation details
            # Structure: data['survey'] -> {'id': ..., 'score': ..., 'comments': ...}
            survey_data = data.get("survey")
            
            if not survey_data:
                print(f"No survey data found for conversation {conversation_id}")
                return None
                
            return {
                "conversation_id": conversation_id,
                "survey_id": survey_data.get("id"),
                "score": survey_data.get("score"),
                "comments": survey_data.get("comments"),
                "timestamp": survey_data.get("timestamp"),
                "raw_survey": survey_data
            }

        except requests.exceptions.HTTPError as e:
            print(f"HTTP Error fetching conversation {conversation_id}: {e}")
            return None
        except Exception as e:
            print(f"Unexpected error: {e}")
            return None

# Initialize Service
quality_service = QualityService(auth)

Step 2: Batch Processing and Pagination Considerations

The Quality API endpoint for a single conversation (/api/v2/quality/conversations/{id}) does not support pagination because it targets a specific resource. However, if you have a list of conversation IDs (e.g., from an Analytics query), you must iterate through them.

A critical performance consideration is Rate Limiting (429 Too Many Requests). Genesys Cloud enforces strict rate limits. You must implement exponential backoff.

import time
import random

class RateLimitedQualityService(QualityService):
    def __init__(self, auth: GenesysAuth):
        super().__init__(auth)
        self.min_delay = 0.5
        self.max_delay = 10.0

    def fetch_survey_batch(self, conversation_ids: list[str]) -> list[Dict[str, Any]]:
        """
        Fetches survey data for a batch of conversation IDs with rate limit handling.
        """
        results = []
        
        for conv_id in conversation_ids:
            try:
                # Small random jitter to avoid thundering herd
                time.sleep(self.min_delay + random.uniform(0, 0.5))
                
                survey_info = self.get_conversation_survey_data(conv_id)
                if survey_info:
                    results.append(survey_info)
                    
            except Exception as e:
                # If the auth token expires during the batch, refresh it
                if "401" in str(e) or "403" in str(e):
                    print("Token expired during batch. Refreshing...")
                    self.auth.access_token = None # Force refresh on next call
                    # Retry once
                    survey_info = self.get_conversation_survey_data(conv_id)
                    if survey_info:
                        results.append(survey_info)
                else:
                    print(f"Skipping {conv_id} due to error: {e}")
        
        return results

# Initialize Rate-Limited Service
rl_quality_service = RateLimitedQualityService(auth)

Step 3: Enriching Data with Analytics API

Often, the CSAT score alone is not enough. You want to know the agent name, channel (voice/chat), and duration. The Quality API returns basic metadata, but the Analytics API provides richer interaction context.

Endpoint: POST /api/v2/analytics/conversations/details/query
Scope: analytics:conversation:view

class AnalyticsService:
    def __init__(self, auth: GenesysAuth):
        self.auth = auth
        self.base_url = auth.env_url

    def get_conversation_details(self, conversation_ids: list[str]) -> Dict[str, Dict]:
        """
        Queries Analytics API for detailed conversation metadata.
        Returns a dictionary mapping conversation_id to its details.
        """
        endpoint = f"{self.base_url}/api/v2/analytics/conversations/details/query"
        token = self.auth.get_token()

        # Construct the query payload
        # We filter by the specific conversation IDs
        # Note: The API supports a list of IDs in the 'conversationIds' filter
        payload = {
            "groupBy": [],
            "interval": "PT1H",
            "dateFrom": "2023-01-01T00:00:00.000Z", # Adjust as needed
            "dateTo": "2025-12-31T23:59:59.999Z",   # Adjust as needed
            "metrics": [],
            "filter": {
                "type": "and",
                "clauses": [
                    {
                        "dimension": "conversationId",
                        "type": "in",
                        "values": conversation_ids[:100] # API limit on list size
                    }
                ]
            },
            "size": 250
        }

        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }

        try:
            response = requests.post(endpoint, headers=headers, json=payload)
            response.raise_for_status()
            data = response.json()
            
            # Map results by conversation ID for easy lookup
            details_map = {}
            for entity in data.get("entities", []):
                conv_id = entity["id"]
                details_map[conv_id] = {
                    "channel": entity.get("channel"),
                    "duration": entity.get("duration"),
                    "agent_name": entity.get("agent", {}).get("name"),
                    "agent_id": entity.get("agent", {}).get("id")
                }
            
            return details_map

        except requests.exceptions.HTTPError as e:
            print(f"Analytics query failed: {e}")
            return {}
        except Exception as e:
            print(f"Error processing analytics data: {e}")
            return {}

analytics_service = AnalyticsService(auth)

Complete Working Example

This script combines authentication, quality data extraction, and analytics enrichment. It takes a list of conversation IDs, fetches their CSAT scores, and enriches them with agent and channel information.

import os
import time
import requests
import json
from typing import List, Dict, Optional
from dotenv import load_dotenv

# --- Authentication Class (From Step 1) ---
class GenesysAuth:
    def __init__(self):
        self.client_id = os.getenv("GENESYS_CLIENT_ID")
        self.client_secret = os.getenv("GENESYS_CLIENT_SECRET")
        self.env_url = os.getenv("GENESYS_ENV_URL", "https://api.mypurecloud.com")
        self.access_token = None
        self.token_expiry = 0

    def get_token(self) -> str:
        if self.access_token and time.time() < (self.token_expiry - 300):
            return self.access_token

        token_url = f"{self.env_url}/oauth/token"
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        try:
            response = requests.post(token_url, headers=headers, data=data)
            response.raise_for_status()
            token_data = response.json()
            self.access_token = token_data["access_token"]
            self.token_expiry = time.time() + token_data["expires_in"]
            return self.access_token
        except Exception as e:
            raise Exception(f"Auth Error: {e}")

# --- Service Classes (From Steps 2 & 3) ---
class CSATExtractor:
    def __init__(self):
        load_dotenv()
        self.auth = GenesysAuth()
        self.base_url = self.auth.env_url

    def get_survey_data(self, conversation_id: str) -> Optional[Dict]:
        """Fetches survey data from Quality API."""
        endpoint = f"{self.base_url}/api/v2/quality/conversations/{conversation_id}"
        token = self.auth.get_token()
        headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

        try:
            resp = requests.get(endpoint, headers=headers)
            if resp.status_code == 404:
                return None
            resp.raise_for_status()
            data = resp.json()
            survey = data.get("survey")
            if not survey:
                return None
            return {
                "score": survey.get("score"),
                "comments": survey.get("comments"),
                "survey_id": survey.get("id"),
                "timestamp": survey.get("timestamp")
            }
        except Exception as e:
            print(f"Error fetching survey for {conversation_id}: {e}")
            return None

    def enrich_with_analytics(self, conversation_ids: List[str]) -> Dict[str, Dict]:
        """Fetches metadata from Analytics API."""
        endpoint = f"{self.base_url}/api/v2/analytics/conversations/details/query"
        token = self.auth.get_token()
        headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
        
        # Analytics API has a limit on the number of IDs in a single query
        # We process in chunks of 50
        chunk_size = 50
        all_details = {}
        
        for i in range(0, len(conversation_ids), chunk_size):
            chunk_ids = conversation_ids[i : i + chunk_size]
            payload = {
                "groupBy": [],
                "interval": "PT1H",
                "dateFrom": "2020-01-01T00:00:00.000Z",
                "dateTo": "2025-12-31T23:59:59.999Z",
                "metrics": [],
                "filter": {
                    "type": "and",
                    "clauses": [
                        {
                            "dimension": "conversationId",
                            "type": "in",
                            "values": chunk_ids
                        }
                    ]
                },
                "size": 250
            }
            
            try:
                resp = requests.post(endpoint, headers=headers, json=payload)
                resp.raise_for_status()
                data = resp.json()
                for entity in data.get("entities", []):
                    all_details[entity["id"]] = {
                        "channel": entity.get("channel"),
                        "agent_name": entity.get("agent", {}).get("name"),
                        "duration_seconds": entity.get("duration", 0) / 1000.0
                    }
                time.sleep(0.5) # Rate limit courtesy
            except Exception as e:
                print(f"Analytics error for chunk starting at {i}: {e}")
                
        return all_details

    def run_extraction(self, conversation_ids: List[str]) -> List[Dict]:
        """Main execution flow."""
        print("1. Fetching Analytics Metadata...")
        metadata_map = self.enrich_with_analytics(conversation_ids)
        
        print("2. Fetching CSAT Survey Data...")
        final_results = []
        
        for conv_id in conversation_ids:
            time.sleep(0.2) # Rate limit for Quality API
            
            survey_data = self.get_survey_data(conv_id)
            meta = metadata_map.get(conv_id, {})
            
            result = {
                "conversation_id": conv_id,
                "channel": meta.get("channel", "Unknown"),
                "agent_name": meta.get("agent_name", "Unknown"),
                "duration_sec": meta.get("duration_seconds", 0),
                "has_survey": survey_data is not None
            }
            
            if survey_data:
                result.update(survey_data)
                
            final_results.append(result)
            
        return final_results

# --- Execution ---
if __name__ == "__main__":
    # Replace with actual Conversation IDs from your environment
    # You can get these from the Genesys Cloud UI or Analytics API
    SAMPLE_CONVERSATION_IDS = [
        "12345678-1234-1234-1234-123456789012",
        "87654321-4321-4321-4321-210987654321"
    ]
    
    extractor = CSATExtractor()
    results = extractor.run_extraction(SAMPLE_CONVERSATION_IDS)
    
    # Output results
    print("\n--- Extraction Results ---")
    for r in results:
        print(json.dumps(r, indent=2))

Common Errors & Debugging

Error: 404 Not Found on /api/v2/quality/conversations/{id}

• Cause: The conversation ID provided does not exist in the Quality database, or it is too old. Genesys Cloud retains quality data for a configurable period (default is often 90 days to 1 year depending on storage settings). Alternatively, the conversation was not eligible for quality review (e.g., system-generated messages).
• Fix: Verify the conversation ID exists in the Analytics API first. Ensure the conversation occurred within the retention window. Check if the conversation type (e.g., voice, chat) is enabled for quality monitoring in your Genesys Cloud instance.

Error: 403 Forbidden

• Cause: The OAuth token used does not have the quality:conversation:view scope.
• Fix: Update your OAuth Client in the Genesys Cloud Admin Console. Navigate to Admin > Security > OAuth Clients, edit your client, and add quality:conversation:view to the scopes. Re-authorize the client.

Error: 429 Too Many Requests

• Cause: You are sending requests faster than the Genesys Cloud API allows. The Quality API and Analytics API have separate rate limits.
• Fix: Implement exponential backoff. The RateLimitedQualityService example above includes a base delay. If you hit 429, parse the Retry-After header from the response and wait that many seconds before retrying.

# Example of handling 429 in requests
response = requests.get(url, headers=headers)
if response.status_code == 429:
    retry_after = int(response.headers.get("Retry-After", 5))
    print(f"Rate limited. Waiting {retry_after} seconds...")
    time.sleep(retry_after)
    # Retry logic here

Error: Survey Data is None but Conversation Exists

• Cause: The interaction did not trigger a CSAT survey. This happens if:
  1. The survey program is not configured for that specific skill or route.
  2. The customer did not complete the survey.
  3. The survey was completed but the data has not yet propagated to the Quality API (usually a few minutes delay).
• Fix: Check the Admin > Quality > Survey Programs configuration. Verify that the survey is active and targeted to the correct interactions.

Official References

• Genesys Cloud Quality API Documentation
• Genesys Cloud Analytics API Documentation
• OAuth 2.0 Client Credentials Grant Guide