Programmatically Start and Stop Call Recordings in Genesys Cloud

What You Will Build

• This tutorial demonstrates how to initiate and terminate conversation recordings programmatically using the Genesys Cloud CX Recording API.
• The implementation uses the PureCloudPlatformClientV2 Python SDK to interact with the /api/v2/recording/conversations endpoints.
• The code is written in Python 3.9+ and handles OAuth2 authentication, recording lifecycle management, and error retries.

Prerequisites

• OAuth Client Type: Confidential Client (Client Credentials Flow).
• Required Scopes:
  • recording:conversation:read
  • recording:conversation:write
  • conversation:read
• SDK Version: genesys-cloud-purecloud-platform-client v100+
• Runtime: Python 3.9 or higher
• Dependencies:
  • pip install genesys-cloud-purecloud-platform-client
  • pip install python-dotenv (for secure credential management)

Authentication Setup

Genesys Cloud uses OAuth 2.0 for API authentication. For server-to-server integrations, the Client Credentials flow is the standard approach. The SDK handles token refresh automatically, but you must configure the client ID and secret correctly.

Create a .env file in your project root:

GENESYS_CLIENT_ID=your_client_id_here
GENESYS_CLIENT_SECRET=your_client_secret_here
GENESYS_ENVIRONMENT=us-east-1 # or your specific region, e.g., eu-west-1

Initialize the SDK with these credentials:

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

load_dotenv()

def get_genesys_api_client():
    """
    Configures and returns a Genesys Cloud API client instance.
    """
    config = Configuration()
    config.client_id = os.getenv("GENESYS_CLIENT_ID")
    config.client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    
    # Set the base path based on environment
    env = os.getenv("GENESYS_ENVIRONMENT", "us-east-1")
    if env == "us-east-1":
        config.host = "https://api.mypurecloud.com"
    else:
        config.host = f"https://{env}.mypurecloud.com"

    api_client = ApiClient(config)
    return api_client

# Initialize the client
api_client = get_genesys_api_client()

Implementation

Step 1: Identify the Conversation

Before starting a recording, you must identify the specific conversation to record. In Genesys Cloud, every interaction (call, chat, message) is assigned a unique conversationId.

If you do not have the conversationId beforehand, you can query recent conversations. However, for real-time control, you typically receive this ID via the Conversations WebSocket or by looking up the conversation by participant ID.

Here is how to retrieve a conversation ID by searching for a specific user or queue. For this tutorial, we assume you have obtained a valid conversationId from an active call.

from genesyscloud import ConversationsApi

def get_active_conversation(api_client, user_id=None, queue_id=None):
    """
    Retrieves an active conversation ID.
    Note: In production, you often get this from the Conversations WebSocket.
    """
    conversations_api = ConversationsApi(api_client)
    
    # Search for active conversations
    # This is a simplified lookup. Real-time apps should use the WebSocket stream.
    try:
        # Get all active conversations for a user (if user_id provided)
        if user_id:
            resp = conversations_api.get_conversations_user(user_id, expand="participants")
        else:
            # Fallback: Get recent conversations (requires broader scopes)
            # This endpoint is for demonstration; in reality, you track IDs via events.
            resp = conversations_api.get_conversations(expand="participants", conversation_type="call")
        
        if resp.entities:
            # Return the first active call conversation
            for conv in resp.entities:
                if conv.conversation_type == "call":
                    return conv.id
        return None
    except Exception as e:
        print(f"Error retrieving conversation: {e}")
        return None

# Example usage (replace with your actual user ID or logic)
# conversation_id = get_active_conversation(api_client, user_id="your_user_id")
# For this tutorial, we will use a placeholder variable.
CONVERSATION_ID = "your_active_conversation_id_here"

Step 2: Start the Recording

To start a recording, you must send a POST request to /api/v2/recording/conversations. The body must contain the conversationId and optionally a recordingType.

Important: The conversation must be in a state where recording is allowed (e.g., not muted by policy, active participants).

from genesyscloud import RecordingsApi
from genesyscloud.models import ConversationRecordingRequest

def start_recording(api_client, conversation_id):
    """
    Initiates a recording for a specific conversation.
    
    Args:
        api_client: The configured Genesys API client.
        conversation_id (str): The ID of the conversation to record.
        
    Returns:
        dict: The response object containing recording details.
    """
    recordings_api = RecordingsApi(api_client)
    
    # Define the request body
    # recording_type can be "default", "agent", "customer", or "both"
    # "default" uses the organization's default recording settings
    recording_request = ConversationRecordingRequest(
        recording_type="default"
    )
    
    try:
        # Execute the API call
        # Endpoint: POST /api/v2/recording/conversations/{conversationId}
        response = recordings_api.post_recording_conversations(
            conversation_id=conversation_id,
            body=recording_request
        )
        
        print(f"Recording started successfully.")
        print(f"Conversation ID: {conversation_id}")
        print(f"Response Status: {response.status_code}")
        
        return response
        
    except Exception as e:
        # Handle specific HTTP errors
        if hasattr(e, 'body'):
            print(f"API Error Response: {e.body}")
        else:
            print(f"An error occurred: {e}")
        raise

# Start the recording
# start_recording(api_client, CONVERSATION_ID)

Step 3: Stop the Recording

Stopping a recording is done via a DELETE request to /api/v2/recording/conversations/{conversationId}. This action is immediate. The recording file will be finalized and processed according to your organization’s retention policies.

def stop_recording(api_client, conversation_id):
    """
    Stops an active recording for a specific conversation.
    
    Args:
        api_client: The configured Genesys API client.
        conversation_id (str): The ID of the conversation to stop recording.
    """
    recordings_api = RecordingsApi(api_client)
    
    try:
        # Endpoint: DELETE /api/v2/recording/conversations/{conversationId}
        response = recordings_api.delete_recording_conversations(
            conversation_id=conversation_id
        )
        
        # DELETE requests typically return 204 No Content on success
        print(f"Recording stopped successfully for conversation: {conversation_id}")
        return response
        
    except Exception as e:
        if hasattr(e, 'body'):
            print(f"API Error Response: {e.body}")
        else:
            print(f"An error occurred: {e}")
        raise

# Stop the recording
# stop_recording(api_client, CONVERSATION_ID)

Step 4: Verify Recording Status and Retrieve Media

After stopping, the recording is not immediately available as a downloadable file. It enters a processing state. You should poll the GET /api/v2/recording/conversations/{conversationId} endpoint to check the status.

import time

def wait_for_recording_completion(api_client, conversation_id, timeout_seconds=120, poll_interval=5):
    """
    Polls for the recording status until it is completed or fails.
    
    Args:
        api_client: The configured Genesys API client.
        conversation_id (str): The ID of the conversation.
        timeout_seconds (int): Max time to wait.
        poll_interval (int): Seconds between polls.
        
    Returns:
        object: The recording entity if successful, else None.
    """
    recordings_api = RecordingsApi(api_client)
    start_time = time.time()
    
    while time.time() - start_time < timeout_seconds:
        try:
            # Endpoint: GET /api/v2/recording/conversations/{conversationId}
            response = recordings_api.get_recording_conversations(conversation_id)
            
            # Check the status of the recording
            # The response contains a 'recordings' array or direct entity depending on version
            # In V2, the response body is the Recording entity
            
            status = response.status # e.g., 'queued', 'processing', 'completed', 'failed'
            
            if status == "completed":
                print("Recording processing completed.")
                return response
            elif status in ["failed", "error"]:
                print(f"Recording failed with status: {status}")
                return None
            else:
                print(f"Recording status: {status}. Waiting...")
                
        except Exception as e:
            print(f"Error checking status: {e}")
            return None
            
        time.sleep(poll_interval)
        
    print("Timeout waiting for recording completion.")
    return None

# Example usage
# completed_recording = wait_for_recording_completion(api_client, CONVERSATION_ID)

Once the status is completed, you can download the media. The response object from get_recording_conversations contains a media object with URLs.

import requests

def download_recording(api_client, conversation_id):
    """
    Downloads the final recording file.
    """
    recordings_api = RecordingsApi(api_client)
    
    # First, get the recording details
    recording_entity = recordings_api.get_recording_conversations(conversation_id)
    
    if recording_entity.status != "completed":
        raise Exception("Recording is not yet completed.")
    
    # Extract the media URL
    # The structure varies slightly by conversation type (call vs chat)
    if recording_entity.media and recording_entity.media.transcripts:
        # For calls, audio is in recording_entity.media.recordings
        for rec in recording_entity.media.recordings:
            if rec.media_url:
                # The media_url is a presigned URL that requires the Bearer token
                headers = {
                    "Authorization": f"Bearer {api_client.configuration.access_token}"
                }
                
                print(f"Downloading recording from: {rec.media_url}")
                resp = requests.get(rec.media_url, headers=headers)
                
                if resp.status_code == 200:
                    filename = f"recording_{conversation_id}.wav"
                    with open(filename, "wb") as f:
                        f.write(resp.content)
                    print(f"Saved to {filename}")
                else:
                    print(f"Failed to download: {resp.status_code}")
                return
    else:
        print("No media found in the recording entity.")

# download_recording(api_client, CONVERSATION_ID)

Complete Working Example

The following script combines all steps into a single executable module. It starts a recording, waits for it to be ready, stops it, waits for processing, and then downloads the file.

import os
import time
import requests
from dotenv import load_dotenv
from genesyscloud import Configuration, ApiClient, RecordingsApi
from genesyscloud.models import ConversationRecordingRequest

load_dotenv()

def get_genesys_api_client():
    config = Configuration()
    config.client_id = os.getenv("GENESYS_CLIENT_ID")
    config.client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    
    env = os.getenv("GENESYS_ENVIRONMENT", "us-east-1")
    if env == "us-east-1":
        config.host = "https://api.mypurecloud.com"
    else:
        config.host = f"https://{env}.mypurecloud.com"

    return ApiClient(config)

def start_recording(api_client, conversation_id):
    recordings_api = RecordingsApi(api_client)
    recording_request = ConversationRecordingRequest(recording_type="default")
    
    try:
        response = recordings_api.post_recording_conversations(
            conversation_id=conversation_id,
            body=recording_request
        )
        print(f"Start Recording Response: {response.status_code}")
        return True
    except Exception as e:
        print(f"Failed to start recording: {e}")
        return False

def stop_recording(api_client, conversation_id):
    recordings_api = RecordingsApi(api_client)
    try:
        response = recordings_api.delete_recording_conversations(conversation_id=conversation_id)
        print(f"Stop Recording Response: {response.status_code}")
        return True
    except Exception as e:
        print(f"Failed to stop recording: {e}")
        return False

def wait_for_status(api_client, conversation_id, target_status, timeout=60, interval=5):
    recordings_api = RecordingsApi(api_client)
    start_time = time.time()
    
    while time.time() - start_time < timeout:
        try:
            entity = recordings_api.get_recording_conversations(conversation_id)
            if entity.status == target_status:
                return entity
            elif entity.status in ["failed", "error"]:
                print(f"Recording failed: {entity.status}")
                return None
        except Exception as e:
            print(f"Error polling status: {e}")
            return None
        time.sleep(interval)
    return None

def download_recording(api_client, conversation_id):
    recordings_api = RecordingsApi(api_client)
    entity = recordings_api.get_recording_conversations(conversation_id)
    
    if not entity.media or not entity.media.recordings:
        print("No media available.")
        return

    for rec in entity.media.recordings:
        if rec.media_url:
            headers = {"Authorization": f"Bearer {api_client.configuration.access_token}"}
            resp = requests.get(rec.media_url, headers=headers)
            if resp.status_code == 200:
                filename = f"recording_{conversation_id}.wav"
                with open(filename, "wb") as f:
                    f.write(resp.content)
                print(f"Downloaded: {filename}")
            else:
                print(f"Download failed: {resp.status_code}")

def main():
    if not os.getenv("GENESYS_CLIENT_ID"):
        print("Missing credentials in .env file.")
        return

    api_client = get_genesys_api_client()
    conversation_id = os.getenv("TARGET_CONVERSATION_ID")
    
    if not conversation_id:
        print("Please set TARGET_CONVERSATION_ID in .env")
        return

    print(f"Starting process for conversation: {conversation_id}")
    
    # 1. Start Recording
    if not start_recording(api_client, conversation_id):
        return
    
    # Simulate a short call duration for demonstration
    # In production, you would trigger stop via another event
    print("Recording active for 10 seconds...")
    time.sleep(10)
    
    # 2. Stop Recording
    if not stop_recording(api_client, conversation_id):
        return
    
    # 3. Wait for Completion
    print("Waiting for recording to process...")
    completed_entity = wait_for_status(api_client, conversation_id, "completed", timeout=120)
    
    if completed_entity:
        # 4. Download
        download_recording(api_client, conversation_id)
    else:
        print("Recording did not complete in time.")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

Cause: The OAuth token lacks the recording:conversation:write scope, or the user associated with the client does not have the necessary permissions in the Genesys Cloud Admin console.
Fix:

1. Verify the OAuth client has recording:conversation:write scope.
2. Check the user’s role in Genesys Cloud Admin. The role must have “View recordings” and “Manage recordings” permissions.

Error: 404 Not Found

Cause: The conversationId is invalid, expired, or the recording was already stopped/deleted.
Fix:

1. Ensure the conversation is currently active.
2. Check if a recording is already active for this conversation. You cannot start a second recording on the same conversation ID simultaneously.

Error: 409 Conflict

Cause: A recording is already in progress for this conversation.
Fix:

1. Call GET /api/v2/recording/conversations/{conversationId} to check the current status.
2. If the status is recording, do not attempt to start again.

Error: 429 Too Many Requests

Cause: Rate limiting. The Recording API has strict rate limits to protect infrastructure.
Fix:
Implement exponential backoff. The Python SDK does not automatically retry 429s for all methods, so wrap API calls in a retry loop.

import time

def retry_api_call(func, args=(), kwargs={}, max_retries=3, backoff_factor=1):
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            if hasattr(e, 'status') and e.status == 429:
                wait_time = backoff_factor * (2 ** attempt)
                print(f"Rate limited. Retrying in {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise
    raise Exception("Max retries exceeded")

Official References

• Recording API Documentation
• Python SDK Reference
• OAuth 2.0 Guide