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

What You Will Build

• One sentence: You will build a Python script that retrieves completed CSAT survey responses, extracts the associated interaction IDs, and joins them with conversation details to create a unified dataset for analysis.
• One sentence: This tutorial uses the Genesys Cloud CX Quality API (/api/v2/quality/evaluations) and the Conversations API (/api/v2/analytics/conversations/details/query).
• One sentence: The programming language covered is Python 3.9+ using the official genesyscloud SDK and requests library.

Prerequisites

• OAuth Client Type: Confidential Client (Client Credentials Grant).
• Required Scopes:
  • quality:evaluation:view (To retrieve survey evaluations)
  • analytics:conversation:query (To retrieve conversation details)
  • conversation:transcript:view (Optional, for deeper context)
• SDK Version: genesyscloud Python SDK v2.10.0+
• Runtime Requirements: Python 3.9 or higher.
• External Dependencies:

  pip install genesyscloud requests pandas
  

Authentication Setup

Genesys Cloud uses OAuth 2.0 for API authentication. For server-side integrations, the Client Credentials Grant is the standard flow. You must create a confidential client in the Genesys Cloud Admin Console with the scopes listed above.

The following code demonstrates how to initialize the PureCloudPlatformClientV2 using environment variables. This approach ensures credentials are not hardcoded.

import os
from purecloud_platform_client import (
    Configuration,
    PureCloudPlatformClientV2,
    OAuthApi,
    ApiClient
)

def get_auth_client() -> PureCloudPlatformClientV2:
    """
    Initializes and returns an authenticated Genesys Cloud client.
    """
    # Load credentials from environment variables
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    environment = os.getenv("GENESYS_ENVIRONMENT", "us-east-1") # e.g., us-east-1, eu-west-1

    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")

    # Create configuration
    config = Configuration(
        environment=environment,
        client_id=client_id,
        client_secret=client_secret
    )

    # Create the platform client
    client = PureCloudPlatformClientV2(config)
    
    # Initialize OAuth API to fetch token
    oauth_api = OAuthApi(client)
    
    try:
        # Request token
        token_response = oauth_api.post_oauth_token(
            grant_type="client_credentials",
            scope="quality:evaluation:view analytics:conversation:query"
        )
        print("Authentication successful.")
        return client
    except Exception as e:
        print(f"Authentication failed: {e}")
        raise

# Instantiate the client
client = get_auth_client()
quality_api = client.quality
analytics_api = client.analytics

Implementation

Step 1: Retrieving CSAT Survey Evaluations

In Genesys Cloud, CSAT surveys are stored as a specific type of evaluation. They are not stored in the standard “Evaluation” object used for QA scoring by agents. Instead, they are categorized with evaluation_type set to customer or agent depending on the survey direction, but typically CSAT is customer.

The endpoint /api/v2/quality/evaluations allows filtering by evaluation_type. We will query for all evaluations of type customer that have a status of completed.

OAuth Scope: quality:evaluation:view

from purecloud_platform_client import (
    QualityApi,
    CreateEvaluationQueryRequest,
    EvaluationQuery
)

def fetch_csat_evaluations(quality_api: QualityApi, page_size: int = 100) -> list:
    """
    Fetches all completed CSAT survey evaluations.
    Handles pagination automatically.
    """
    all_evaluations = []
    query_request = CreateEvaluationQueryRequest(
        query=EvaluationQuery(
            evaluation_type="customer",  # Critical: CSAT are customer evaluations
            status="completed"
        ),
        page_size=page_size
    )

    try:
        while True:
            response = quality_api.post_quality_evaluations_query(
                body=query_request
            )

            if not response.entities or len(response.entities) == 0:
                break

            all_evaluations.extend(response.entities)

            # Check for pagination
            if response.next_page_uri is None:
                break
            
            # SDK handles the next page URI internally if we pass it, 
            # but for clarity in this tutorial, we reconstruct the request 
            # or use the SDK's pagination helper if available. 
            # In the pure Python SDK, we often need to manage the cursor manually 
            # or use the response's next_page_token if exposed.
            # For simplicity, this example assumes a single large fetch or 
            # manual cursor management. 
            # NOTE: The post_quality_evaluations_query returns entities.
            # To paginate, you typically set 'page_token' in the next request.
            
            query_request.page_token = response.page_token # Assuming SDK supports this field

    except Exception as e:
        print(f"Error fetching evaluations: {e}")
        raise

    return all_evaluations

# Execute fetch
csat_evals = fetch_csat_evaluations(quality_api)
print(f"Fetched {len(csat_evals)} CSAT evaluations.")

Key Parameter Explanation:

• evaluation_type="customer": This is the most critical filter. QA scores are agent. CSAT surveys are customer.
• status="completed": Surveys can be submitted (partial) or completed. Only completed surveys have valid scores.

Step 2: Extracting Interaction IDs and Formatting for Analytics

Each evaluation object contains an external_id or a reference to the conversation. In Genesys Cloud, the link between a survey and the conversation is often found in the external_id field of the evaluation, which usually contains the conversationId or interactionId. However, the most reliable way to join them is via the conversation_id field if present, or by parsing the external_id.

For CSAT, the external_id often maps to the conversationId. We will extract these IDs to query the Analytics API.

def extract_conversation_ids(evaluations: list) -> list:
    """
    Extracts conversation IDs from CSAT evaluations.
    """
    conversation_ids = []
    
    for eval_obj in evaluations:
        # The external_id often holds the conversation ID for CSAT
        # However, the structure can vary. Let's inspect the object.
        # Typically: eval_obj.external_id is the conversation ID.
        if eval_obj.external_id:
            conversation_ids.append(eval_obj.external_id)
        elif hasattr(eval_obj, 'conversation_id') and eval_obj.conversation_id:
            conversation_ids.append(eval_obj.conversation_id)
            
    return list(set(conversation_ids)) # Remove duplicates

conv_ids = extract_conversation_ids(csat_evals)
print(f"Found {len(conv_ids)} unique conversation IDs to analyze.")

Step 3: Querying Conversation Details via Analytics API

Now that we have the list of conversation IDs, we need to pull the actual interaction data (timestamp, channel, duration) to enrich our CSAT data. We use the /api/v2/analytics/conversations/details/query endpoint.

This endpoint accepts a conversationIds filter. Note that this filter has a limit (usually 100 IDs per request). We must batch the requests.

OAuth Scope: analytics:conversation:query

from purecloud_platform_client import (
    AnalyticsApi,
    PostConversationsDetailsQueryRequest,
    ConversationDetailRequest
)

def fetch_conversation_details(analytics_api: AnalyticsApi, conv_ids: list, batch_size: int = 100) -> list:
    """
    Fetches detailed conversation data for a list of conversation IDs.
    Batches requests to avoid payload size limits.
    """
    all_conversations = []
    
    # Split IDs into batches
    batches = [conv_ids[i:i + batch_size] for i in range(0, len(conv_ids), batch_size)]

    for batch in batches:
        try:
            # Construct the query
            # We need to specify the time range. 
            # Since we don't have the date from the ID, we fetch a wide range 
            # or rely on the ID filter being sufficient if the ID is unique.
            # The Analytics API requires a time range. We will use a reasonable default 
            # (last 90 days) assuming recent surveys.
            
            from datetime import datetime, timedelta
            end_date = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
            start_date = (datetime.utcnow() - timedelta(days=90)).strftime("%Y-%m-%dT%H:%M:%SZ")

            query_body = PostConversationsDetailsQueryRequest(
                body=ConversationDetailRequest(
                    interval=f"{start_date}/{end_date}",
                    entity={"conversationIds": batch},
                    view="conversation",
                    size=1000 # Max size per page
                )
            )

            response = analytics_api.post_analytics_conversations_details_query(
                body=query_body
            )

            if response.entities and len(response.entities) > 0:
                all_conversations.extend(response.entities)

        except Exception as e:
            print(f"Error fetching conversation batch: {e}")
            # Implement retry logic here for 429s if necessary
            continue

    return all_conversations

conv_details = fetch_conversation_details(analytics_api, conv_ids)
print(f"Fetched details for {len(conv_details)} conversations.")

Edge Case Handling:

• Time Range Limitation: The Analytics API requires a time interval. If the survey is older than your query range, it will not return. To fix this, you must parse the external_id or fetch the evaluation’s completed_date from the Quality API and use that to set the interval dynamically. For this tutorial, we assume a 90-day window.
• Batch Size: The conversationIds filter can handle up to 100 IDs. Sending more will result in a 400 Bad Request.

Step 4: Merging Data and Calculating Metrics

Finally, we merge the survey scores with the conversation metadata. We will create a DataFrame for easy analysis.

import pandas as pd

def merge_csat_and_conversations(evals: list, convs: list) -> pd.DataFrame:
    """
    Merges CSAT evaluation data with conversation details.
    """
    # Create a map of conversation_id -> conversation_details
    conv_map = {}
    for conv in convs:
        conv_map[conv.id] = conv

    merged_data = []

    for eval_obj in evals:
        conv_id = eval_obj.external_id
        conv_data = conv_map.get(conv_id)

        if conv_data:
            # Extract score. Note: CSAT scores are often in 'scores' or 'result'
            # Structure varies slightly by survey template. 
            # Commonly: eval_obj.scores[0].score or eval_obj.result
            
            score = 0
            if eval_obj.scores and len(eval_obj.scores) > 0:
                score = eval_obj.scores[0].score
            
            merged_data.append({
                "conversation_id": conv_id,
                "csat_score": score,
                "survey_completed_at": eval_obj.completed_date,
                "conversation_start": conv_data.start_time,
                "conversation_end": conv_data.end_time,
                "channel": conv_data.channel,
                "duration_seconds": conv_data.duration_seconds
            })

    df = pd.DataFrame(merged_data)
    return df

# Merge and display
result_df = merge_csat_and_conversations(csat_evals, conv_details)
print(result_df.head())
print(f"Average CSAT Score: {result_df['csat_score'].mean():.2f}")

Complete Working Example

Below is the full, copy-pasteable script. Ensure you have set the environment variables GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET.

import os
import pandas as pd
from datetime import datetime, timedelta
from purecloud_platform_client import (
    Configuration,
    PureCloudPlatformClientV2,
    OAuthApi,
    QualityApi,
    AnalyticsApi,
    CreateEvaluationQueryRequest,
    EvaluationQuery,
    PostConversationsDetailsQueryRequest,
    ConversationDetailRequest
)

def get_auth_client() -> PureCloudPlatformClientV2:
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    environment = os.getenv("GENESYS_ENVIRONMENT", "us-east-1")

    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")

    config = Configuration(environment=environment, client_id=client_id, client_secret=client_secret)
    client = PureCloudPlatformClientV2(config)
    oauth_api = OAuthApi(client)
    
    try:
        oauth_api.post_oauth_token(grant_type="client_credentials", scope="quality:evaluation:view analytics:conversation:query")
        return client
    except Exception as e:
        raise Exception(f"Auth failed: {e}")

def fetch_csat_evaluations(quality_api: QualityApi) -> list:
    all_evaluations = []
    query_request = CreateEvaluationQueryRequest(
        query=EvaluationQuery(evaluation_type="customer", status="completed"),
        page_size=100
    )

    while True:
        response = quality_api.post_quality_evaluations_query(body=query_request)
        if not response.entities or len(response.entities) == 0:
            break
        all_evaluations.extend(response.entities)
        
        if response.next_page_uri is None:
            break
        
        # Note: In actual SDK usage, you might need to handle pagination differently 
        # depending on the specific SDK version's support for next_page_uri in the request.
        # This loop assumes standard pagination behavior.
        if not hasattr(query_request, 'page_token') or query_request.page_token == response.page_token:
            break # Prevent infinite loop if token doesn't update
            
        query_request.page_token = response.page_token

    return all_evaluations

def extract_conversation_ids(evaluations: list) -> list:
    conversation_ids = []
    for eval_obj in evaluations:
        if eval_obj.external_id:
            conversation_ids.append(eval_obj.external_id)
        elif hasattr(eval_obj, 'conversation_id') and eval_obj.conversation_id:
            conversation_ids.append(eval_obj.conversation_id)
    return list(set(conversation_ids))

def fetch_conversation_details(analytics_api: AnalyticsApi, conv_ids: list) -> list:
    all_conversations = []
    batch_size = 100
    batches = [conv_ids[i:i + batch_size] for i in range(0, len(conv_ids), batch_size)]

    end_date = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
    start_date = (datetime.utcnow() - timedelta(days=90)).strftime("%Y-%m-%dT%H:%M:%SZ")

    for batch in batches:
        try:
            query_body = PostConversationsDetailsQueryRequest(
                body=ConversationDetailRequest(
                    interval=f"{start_date}/{end_date}",
                    entity={"conversationIds": batch},
                    view="conversation",
                    size=1000
                )
            )
            response = analytics_api.post_analytics_conversations_details_query(body=query_body)
            if response.entities:
                all_conversations.extend(response.entities)
        except Exception as e:
            print(f"Error fetching batch: {e}")
    return all_conversations

def main():
    client = get_auth_client()
    quality_api = client.quality
    analytics_api = client.analytics

    print("Fetching CSAT evaluations...")
    csat_evals = fetch_csat_evaluations(quality_api)
    print(f"Found {len(csat_evals)} evaluations.")

    if not csat_evals:
        print("No CSAT evaluations found in the last 90 days.")
        return

    print("Extracting Conversation IDs...")
    conv_ids = extract_conversation_ids(csat_evals)
    print(f"Found {len(conv_ids)} unique conversations.")

    print("Fetching Conversation Details...")
    conv_details = fetch_conversation_details(analytics_api, conv_ids)
    print(f"Retrieved details for {len(conv_details)} conversations.")

    print("Merging Data...")
    conv_map = {conv.id: conv for conv in conv_details}
    merged_data = []

    for eval_obj in csat_evals:
        conv_id = eval_obj.external_id
        conv_data = conv_map.get(conv_id)
        
        if conv_data:
            score = 0
            if eval_obj.scores and len(eval_obj.scores) > 0:
                score = eval_obj.scores[0].score
            
            merged_data.append({
                "conversation_id": conv_id,
                "csat_score": score,
                "survey_completed_at": eval_obj.completed_date,
                "conversation_start": conv_data.start_time,
                "channel": conv_data.channel,
                "duration_seconds": conv_data.duration_seconds
            })

    df = pd.DataFrame(merged_data)
    
    if not df.empty:
        print("Results:")
        print(df.head())
        print(f"Average CSAT: {df['csat_score'].mean():.2f}")
        df.to_csv("csat_analysis.csv", index=False)
        print("Saved to csat_analysis.csv")
    else:
        print("No matching conversation details found.")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token has expired or the client credentials are incorrect.
• Fix: Ensure GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET are correct. The script requests a new token on initialization. If running long processes, implement token refresh logic.

Error: 403 Forbidden

• Cause: The OAuth client does not have the required scopes.
• Fix: Go to Admin > Platform Services > API Access > OAuth 2.0 Clients. Edit your client and ensure quality:evaluation:view and analytics:conversation:query are checked.

Error: 429 Too Many Requests

• Cause: You are hitting the API rate limit. The Analytics API has strict rate limits.
• Fix: Implement exponential backoff. In the fetch_conversation_details function, catch the 429 exception, wait for a few seconds, and retry.

import time

# Inside fetch_conversation_details loop
except Exception as e:
    if "429" in str(e):
        wait_time = 5
        print(f"Rate limited. Waiting {wait_time} seconds...")
        time.sleep(wait_time)
        continue # Retry the batch
    else:
        raise

Error: Empty Results from Analytics API

• Cause: The time interval in the Analytics query does not overlap with the conversation dates.
• Fix: The Analytics API requires a time range. If you query for conversations from 2022 but set the interval to “Last 90 Days”, you get nothing. Adjust the start_date and end_date in the ConversationDetailRequest to cover the date range of your evaluations.

Official References

• Quality Evaluations API Documentation
• Analytics Conversations Details API Documentation
• Python SDK Reference