Retrieving Voice Conversation Transcripts via Genesys Cloud Speech Analytics API

What You Will Build

• You will build a Python script that identifies voice conversations containing specific keywords using the Search API and then retrieves the full, timestamped text transcript for each match.
• This tutorial uses the Genesys Cloud Platform Client V2 SDK and the REST API endpoints for Speech Analytics (/api/v2/analytics/conversations/details/query and /api/v2/insights/conversations/{conversationId}).
• The programming language covered is Python 3.8+.

Prerequisites

• OAuth Client Type: You require a Service Account or OAuth2 Client Credentials grant.
• Required Scopes:
  • analytics:conversation:read (to search for conversations)
  • insights:conversation:read (to retrieve transcript details)
  • speech:conversation:read (optional, depending on specific transcript depth required)
• SDK Version: genesys-cloud-purecloud-sdk v2.0 or higher.
• Runtime: Python 3.8+
• Installation:

  pip install genesys-cloud-purecloud-sdk
  

Authentication Setup

Genesys Cloud APIs require an OAuth 2.0 Bearer token. For server-to-server integrations, the Client Credentials flow is the standard. You must generate a token using your Organization ID, Client ID, and Client Secret.

The following function handles token acquisition and basic caching to avoid unnecessary API calls within a short timeframe.

import os
import time
from genesyscloud.rest import Configuration
from genesyscloud.auth.api_client import ApiClient

class GenesysAuth:
    def __init__(self, org_id: str, client_id: str, client_secret: str):
        self.org_id = org_id
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_cache = None
        self.token_expiry = 0

    def get_token(self) -> str:
        """
        Retrieves an OAuth token. Caches it until 60 seconds before expiry.
        """
        if self.token_cache and time.time() < self.token_expiry - 60:
            return self.token_cache

        # Initialize the configuration with client credentials
        configuration = Configuration(
            client_id=self.client_id,
            client_secret=self.client_secret,
            org_id=self.org_id
        )

        api_client = ApiClient(configuration)
        
        try:
            # The SDK handles the POST to /oauth/token automatically
            token_response = api_client.auth_client.get_oauth_token()
            self.token_cache = token_response.access_token
            # Set expiry with a small buffer
            self.token_expiry = time.time() + token_response.expires_in
            return self.token_cache
        except Exception as e:
            raise RuntimeError(f"Failed to obtain OAuth token: {e}")

Implementation

Step 1: Search for Conversations with Specific Keywords

Before retrieving the full transcript, you must identify which conversations contain the data you need. The Genesys Cloud Analytics API allows you to query conversations based on speech analytics insights. We will search for conversations where the agent or caller said a specific keyword (e.g., “refund”).

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

The request body requires a query object. For speech analytics, you define the insights section.

from genesyscloud.analytics.api.conversations_api import ConversationsApi
from genesyscloud.analytics.model.conversation_query import ConversationQuery
from genesyscloud.analytics.model.conversation_query_filter import ConversationQueryFilter
from genesyscloud.analytics.model.conversation_query_filter_criteria import ConversationQueryFilterCriteria
from genesyscloud.analytics.model.conversation_query_filter_insights import ConversationQueryFilterInsights
from genesyscloud.analytics.model.conversation_query_filter_insights_criteria import ConversationQueryFilterInsightsCriteria
from genesyscloud.analytics.model.conversation_query_filter_insights_criteria_details import ConversationQueryFilterInsightsCriteriaDetails
from genesyscloud.analytics.model.conversation_query_filter_insights_criteria_details_phrase import ConversationQueryFilterInsightsCriteriaDetailsPhrase

def search_conversations_by_keyword(auth: GenesysAuth, keyword: str, start_time: str, end_time: str) -> list:
    """
    Searches for voice conversations containing a specific keyword.
    
    Args:
        auth: GenesysAuth instance
        keyword: The word or phrase to search for
        start_time: ISO 8601 start datetime (e.g., "2023-10-01T00:00:00Z")
        end_time: ISO 8601 end datetime (e.g., "2023-10-02T00:00:00Z")
    """
    # Initialize API client with the current token
    configuration = Configuration(
        client_id=auth.client_id,
        client_secret=auth.client_secret,
        org_id=auth.org_id
    )
    api_client = ApiClient(configuration)
    conversations_api = ConversationsApi(api_client)

    # Construct the filter criteria
    # We are looking for an exact phrase match in the transcript
    phrase_filter = ConversationQueryFilterInsightsCriteriaDetailsPhrase(
        phrase=keyword,
        match="exact" # Options: "exact", "partial", "regex"
    )

    criteria_details = ConversationQueryFilterInsightsCriteriaDetails(
        phrases=[phrase_filter]
    )

    insights_criteria = ConversationQueryFilterInsightsCriteria(
        details=criteria_details
    )

    insights_filter = ConversationQueryFilterInsights(
        criteria=insights_criteria
    )

    # Define the time window
    filter_obj = ConversationQueryFilter(
        start_time=start_time,
        end_time=end_time,
        insights=insights_filter
    )

    # Construct the main query
    query = ConversationQuery(
        filter=filter_obj,
        size=25 # Limit results per page for testing
    )

    try:
        # Execute the query
        response = conversations_api.post_analytics_conversations_details_query(body=query)
        
        # Extract conversation IDs
        conversation_ids = [conv.id for conv in response.entities] if response.entities else []
        return conversation_ids

    except Exception as e:
        print(f"Error searching conversations: {e}")
        return []

Step 2: Retrieve the Full Transcript for a Conversation

Once you have the conversationId, you can retrieve the detailed transcript. The Insights API provides the full breakdown of who spoke when and what they said.

Endpoint: GET /api/v2/insights/conversations/{conversationId}
Scope: insights:conversation:read

The response contains a transcript array. Each element represents a segment of speech with a timestamp, participant (agent or caller), and text.

from genesyscloud.insights.api.conversations_api import ConversationsApi as InsightsConversationsApi
from genesyscloud.rest import ApiException

def get_conversation_transcript(auth: GenesysAuth, conversation_id: str) -> dict:
    """
    Retrieves the full transcript for a specific conversation ID.
    """
    configuration = Configuration(
        client_id=auth.client_id,
        client_secret=auth.client_secret,
        org_id=auth.org_id
    )
    api_client = ApiClient(configuration)
    insights_api = InsightsConversationsApi(api_client)

    try:
        # Retrieve the conversation details
        # The SDK method maps to GET /api/v2/insights/conversations/{conversationId}
        response = insights_api.get_insights_conversations(conversation_id)
        
        # Check if transcript data exists
        if not response.transcript:
            print(f"No transcript data available for conversation {conversation_id}")
            return {}

        # Parse the transcript segments
        transcript_data = []
        for segment in response.transcript:
            transcript_data.append({
                "timestamp": segment.timestamp,
                "participant_id": segment.participant.id,
                "participant_name": segment.participant.name,
                "participant_type": segment.participant.type, # e.g., "agent", "customer"
                "text": segment.text,
                "confidence": segment.confidence # Sentiment or speech confidence if available
            })
            
        return {
            "conversation_id": conversation_id,
            "transcript": transcript_data
        }

    except ApiException as e:
        if e.status == 404:
            print(f"Conversation {conversation_id} not found.")
        elif e.status == 403:
            print(f"Forbidden: Check if you have 'insights:conversation:read' scope.")
        else:
            print(f"API Error {e.status}: {e.reason}")
        return {}
    except Exception as e:
        print(f"Unexpected error retrieving transcript: {e}")
        return {}

Step 3: Handling Pagination and Rate Limits

The Search API (/api/v2/analytics/conversations/details/query) returns a nextPageToken if there are more results. You must handle pagination to ensure you retrieve all matching conversations. Additionally, Genesys Cloud enforces rate limits (typically 429 Too Many Requests). You should implement exponential backoff.

import time
import random

def get_all_conversation_ids(auth: GenesysAuth, keyword: str, start_time: str, end_time: str) -> list:
    """
    Paginates through all conversations matching the keyword.
    """
    all_ids = []
    next_page_token = None
    max_retries = 5

    while True:
        # Re-initialize API client for each request to ensure fresh token if needed
        configuration = Configuration(
            client_id=auth.client_id,
            client_secret=auth.client_secret,
            org_id=auth.org_id
        )
        api_client = ApiClient(configuration)
        conversations_api = ConversationsApi(api_client)

        # Construct query (same as Step 1)
        phrase_filter = ConversationQueryFilterInsightsCriteriaDetailsPhrase(
            phrase=keyword,
            match="exact"
        )
        criteria_details = ConversationQueryFilterInsightsCriteriaDetails(phrases=[phrase_filter])
        insights_criteria = ConversationQueryFilterInsightsCriteria(details=criteria_details)
        insights_filter = ConversationQueryFilterInsights(criteria=insights_criteria)
        
        filter_obj = ConversationQueryFilter(
            start_time=start_time,
            end_time=end_time,
            insights=insights_filter
        )
        
        query = ConversationQuery(
            filter=filter_obj,
            size=100, # Max page size
            page_token=next_page_token
        )

        retries = 0
        while retries < max_retries:
            try:
                response = conversations_api.post_analytics_conversations_details_query(body=query)
                
                if response.entities:
                    all_ids.extend([conv.id for conv in response.entities])
                
                # Check for next page
                if response.next_page_token:
                    next_page_token = response.next_page_token
                else:
                    break # No more pages

                # Respect rate limits: small delay between requests
                time.sleep(0.5)
                break # Success, exit retry loop

            except ApiException as e:
                if e.status == 429:
                    # Exponential backoff with jitter
                    wait_time = (2 ** retries) + random.uniform(0, 1)
                    print(f"Rate limited (429). Waiting {wait_time:.2f} seconds...")
                    time.sleep(wait_time)
                    retries += 1
                else:
                    raise e

        if retries == max_retries:
            print("Max retries reached due to rate limiting.")
            break

    return all_ids

Complete Working Example

The following script combines authentication, pagination, and transcript retrieval into a single runnable module.

import os
import sys
from datetime import datetime, timezone

# Import SDK modules
from genesyscloud.rest import Configuration
from genesyscloud.auth.api_client import ApiClient
from genesyscloud.analytics.api.conversations_api import ConversationsApi
from genesyscloud.analytics.model.conversation_query import ConversationQuery
from genesyscloud.analytics.model.conversation_query_filter import ConversationQueryFilter
from genesyscloud.analytics.model.conversation_query_filter_insights import ConversationQueryFilterInsights
from genesyscloud.analytics.model.conversation_query_filter_insights_criteria import ConversationQueryFilterInsightsCriteria
from genesyscloud.analytics.model.conversation_query_filter_insights_criteria_details import ConversationQueryFilterInsightsCriteriaDetails
from genesyscloud.analytics.model.conversation_query_filter_insights_criteria_details_phrase import ConversationQueryFilterInsightsCriteriaDetailsPhrase
from genesyscloud.insights.api.conversations_api import ConversationsApi as InsightsConversationsApi
from genesyscloud.rest import ApiException

# --- Authentication Helper ---
class GenesysAuth:
    def __init__(self, org_id: str, client_id: str, client_secret: str):
        self.org_id = org_id
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_cache = None
        self.token_expiry = 0

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

        configuration = Configuration(
            client_id=self.client_id,
            client_secret=self.client_secret,
            org_id=self.org_id
        )
        api_client = ApiClient(configuration)
        
        try:
            token_response = api_client.auth_client.get_oauth_token()
            self.token_cache = token_response.access_token
            self.token_expiry = time.time() + token_response.expires_in
            return self.token_cache
        except Exception as e:
            raise RuntimeError(f"Failed to obtain OAuth token: {e}")

# --- Core Logic ---

def search_and_retrieve_transcripts(org_id: str, client_id: str, client_secret: str, keyword: str, days_back: int = 7):
    """
    Main function to search for conversations and print transcripts.
    """
    auth = GenesysAuth(org_id, client_id, client_secret)
    
    # Calculate date range
    end_time = datetime.now(timezone.utc).isoformat()
    start_time = (datetime.now(timezone.utc) - timedelta(days=days_back)).isoformat()
    
    print(f"Searching for conversations with keyword '{keyword}' from {start_time} to {end_time}...")

    # 1. Get all matching conversation IDs with pagination
    conversation_ids = []
    next_page_token = None
    max_retries = 5
    
    while True:
        configuration = Configuration(
            client_id=client_id,
            client_secret=client_secret,
            org_id=org_id
        )
        api_client = ApiClient(configuration)
        conv_api = ConversationsApi(api_client)

        # Build Query
        phrase = ConversationQueryFilterInsightsCriteriaDetailsPhrase(phrase=keyword, match="exact")
        details = ConversationQueryFilterInsightsCriteriaDetails(phrases=[phrase])
        criteria = ConversationQueryFilterInsightsCriteria(details=details)
        insights = ConversationQueryFilterInsights(criteria=criteria)
        
        filter_obj = ConversationQueryFilter(start_time=start_time, end_time=end_time, insights=insights)
        query = ConversationQuery(filter=filter_obj, size=100, page_token=next_page_token)

        retries = 0
        while retries < max_retries:
            try:
                response = conv_api.post_analytics_conversations_details_query(body=query)
                
                if response.entities:
                    conversation_ids.extend([c.id for c in response.entities])
                
                if response.next_page_token:
                    next_page_token = response.next_page_token
                else:
                    break
                
                time.sleep(0.5) # Rate limit courtesy
                break

            except ApiException as e:
                if e.status == 429:
                    wait = (2 ** retries) + random.uniform(0, 1)
                    print(f"429 Rate Limit. Retrying in {wait:.2f}s...")
                    time.sleep(wait)
                    retries += 1
                else:
                    raise e
        
        if retries == max_retries:
            print("Max retries exceeded.")
            break

    print(f"Found {len(conversation_ids)} conversations.")

    # 2. Retrieve Transcript for each conversation
    for conv_id in conversation_ids:
        print(f"\n--- Transcript for Conversation: {conv_id} ---")
        
        configuration = Configuration(
            client_id=client_id,
            client_secret=client_secret,
            org_id=org_id
        )
        api_client = ApiClient(configuration)
        insights_api = InsightsConversationsApi(api_client)

        try:
            response = insights_api.get_insights_conversations(conv_id)
            
            if not response.transcript:
                print("No transcript data available.")
                continue

            for segment in response.transcript:
                # Format timestamp for readability
                ts = segment.timestamp.strftime("%H:%M:%S") if segment.timestamp else "Unknown"
                speaker = segment.participant.name or segment.participant.type
                text = segment.text or "[Silence/Non-speech]"
                
                print(f"[{ts}] {speaker}: {text}")

        except ApiException as e:
            print(f"Error retrieving transcript for {conv_id}: {e.status} {e.reason}")
        except Exception as e:
            print(f"Unexpected error: {e}")

if __name__ == "__main__":
    # Configuration from Environment Variables
    ORG_ID = os.getenv("GENESYS_ORG_ID")
    CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
    CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")
    KEYWORD = os.getenv("SEARCH_KEYWORD", "refund")

    if not all([ORG_ID, CLIENT_ID, CLIENT_SECRET]):
        print("Error: Missing environment variables GENESYS_ORG_ID, GENESYS_CLIENT_ID, or GENESYS_CLIENT_SECRET")
        sys.exit(1)

    # Import timedelta here for the main block
    from datetime import timedelta
    import time
    import random

    search_and_retrieve_transcripts(ORG_ID, CLIENT_ID, CLIENT_SECRET, KEYWORD)

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token is invalid, expired, or the Client ID/Secret is incorrect.
• Fix: Verify your credentials in the Genesys Cloud Admin Console. Ensure your code refreshes the token before it expires. The GenesysAuth class above handles refresh, but ensure you are using the latest token for each API call if caching is disabled.

Error: 403 Forbidden

• Cause: The OAuth client does not have the required scopes.
• Fix: Go to Admin > Security > OAuth Clients. Edit your client and ensure analytics:conversation:read and insights:conversation:read are checked. Save the changes. Note that scope changes may take up to 15 minutes to propagate.

Error: 404 Not Found

• Cause: The conversationId does not exist, or the conversation has not yet been processed by the speech analytics engine.
• Fix: Speech analytics processing is not instantaneous. It may take 1-2 hours after the conversation ends for the transcript to be available. If you are testing with a live call, wait. If you are testing with historical data, ensure the date range includes conversations that are already processed.

Error: Empty Transcript Array

• Cause: The conversation is voice, but speech analytics is not enabled for that specific workflow or queue, or the recording failed.
• Fix: Verify that “Speech Analytics” is enabled for the relevant Queue or Workflow in the Genesys Cloud Admin Console. Check the “Recordings” section to ensure the audio file was successfully captured.

Official References

• Genesys Cloud Conversations API Documentation
• Genesys Cloud Insights API Documentation
• Python SDK Reference