Programmatically Start and Stop Call Recordings in Genesys Cloud CX

What You Will Build

• You will build a script that programmatically initiates a recording for an active conversation and subsequently stops it using specific recording identifiers.
• This tutorial uses the Genesys Cloud CX Recording API (/api/v2/recording) and the Python SDK (genesyscloud).
• The implementation is provided in Python 3.10+ using the requests library for raw HTTP calls and the official Genesys Cloud Python SDK for SDK-based interactions.

Prerequisites

• OAuth Client: A Genesys Cloud OAuth Client configured with the Client Credentials grant type.
• Required Scopes:
  • recording:view (to query existing recordings)
  • recording:update (to start and stop recordings)
  • conversation:read (optional, to verify conversation state)
• SDK Version: genesyscloud >= 7.0.0 (or compatible version supporting PureCloudPlatformClientV2).
• Runtime: Python 3.10 or higher.
• Dependencies:

  pip install genesyscloud requests python-dotenv
  

Authentication Setup

Genesys Cloud uses OAuth 2.0 for authentication. For server-to-server integrations, the Client Credentials flow is standard. You must obtain an access token before making any API calls. The token expires after 3600 seconds (1 hour), so your production code should implement token caching and refresh logic.

Below is a robust authentication helper using the requests library.

import requests
import json
import os
from typing import Optional

class GenesysAuth:
    def __init__(self, client_id: str, client_secret: str, environment: str = "mypurecloud.com"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = f"https://api.{environment}"
        self.token_endpoint = f"{self.base_url}/oauth/token"
        self.access_token: Optional[str] = None
        self.expires_in: int = 0

    def get_access_token(self) -> str:
        """
        Retrieves an OAuth access token.
        In a production environment, cache this token and check expiration before requesting a new one.
        """
        if self.access_token and self._is_token_valid():
            return self.access_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(self.token_endpoint, headers=headers, data=data)
            response.raise_for_status()
            token_data = response.json()
            self.access_token = token_data["access_token"]
            self.expires_in = token_data.get("expires_in", 3600)
            return self.access_token
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                raise Exception("Authentication failed. Check your Client ID and Secret.") from e
            raise

    def _is_token_valid(self) -> bool:
        # Simple check. In production, compare current time against token issuance time + expires_in
        # For this tutorial, we assume a single run or implement a simple time check
        import time
        # This is a simplified check. Real implementations should store token issuance timestamp.
        # Here we just return True if token exists, assuming short-lived script execution.
        return self.access_token is not None

# Usage Example
# auth = GenesysAuth(os.getenv("GENESYS_CLIENT_ID"), os.getenv("GENESYS_CLIENT_SECRET"))
# token = auth.get_access_token()

Implementation

Step 1: Identify the Active Conversation and Recording

To start a recording, you need a target. While you can start a recording on a conversation if you already know the conversationId, it is often useful to query active conversations to find the one you want to record. Conversely, to stop a recording, you need the recordingId.

First, let us query for active conversations. This requires the conversation:read scope.

Endpoint: GET /api/v2/conversations

import requests
from typing import Dict, Any

def get_active_conversations(auth: GenesysAuth) -> Dict[str, Any]:
    """
    Retrieves a list of active conversations.
    Scope Required: conversation:read
    """
    url = f"{auth.base_url}/api/v2/conversations"
    headers = {
        "Authorization": f"Bearer {auth.get_access_token()}",
        "Content-Type": "application/json"
    }

    # Filter for active conversations only
    params = {
        "filter": "status=active"
    }

    try:
        response = requests.get(url, headers=headers, params=params)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        print(f"HTTP Error: {e.response.status_code} - {e.response.text}")
        raise

# Example Usage
# conversations = get_active_conversations(auth)
# if conversations.get("entities"):
#     target_conversation_id = conversations["entities"][0]["id"]
#     print(f"Target Conversation ID: {target_conversation_id}")

Step 2: Start a Recording Programmatically

Once you have the conversationId, you can initiate a recording. The Genesys Cloud Recording API allows you to start a recording by POSTing to the /api/v2/recording endpoint.

Critical Parameters:

• conversationId: The ID of the conversation to record.
• recordingType: Usually "conversation" for standard call recording. Other types include "interaction" or "analytics".
• recordingSettings: Optional. You can specify specific audio channels or metadata here.

Endpoint: POST /api/v2/recording

def start_recording(auth: GenesysAuth, conversation_id: str) -> Dict[str, Any]:
    """
    Starts a recording for a specific conversation.
    Scope Required: recording:update
    """
    url = f"{auth.base_url}/api/v2/recording"
    headers = {
        "Authorization": f"Bearer {auth.get_access_token()}",
        "Content-Type": "application/json"
    }

    payload = {
        "conversationId": conversation_id,
        "recordingType": "conversation",
        # Optional: Add custom metadata if your org supports it
        # "recordingSettings": {
        #     "metadata": {
        #         "source": "api-triggered"
        #     }
        # }
    }

    try:
        response = requests.post(url, headers=headers, json=payload)
        
        # Check for success (201 Created)
        if response.status_code == 201:
            recording_data = response.json()
            recording_id = recording_data.get("id")
            print(f"Recording started successfully. Recording ID: {recording_id}")
            return recording_data
        elif response.status_code == 409:
            # Conflict: A recording might already be active for this conversation
            print(f"Conflict: {response.json().get('message', 'A recording is already active for this conversation.')}")
            return None
        else:
            response.raise_for_status()
            
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 403:
            print("Error: Insufficient permissions. Ensure the OAuth client has 'recording:update' scope.")
        elif e.response.status_code == 404:
            print(f"Error: Conversation {conversation_id} not found or not active.")
        else:
            print(f"HTTP Error: {e.response.status_code} - {e.response.text}")
        raise

# Example Usage
# recording = start_recording(auth, target_conversation_id)
# if recording:
#     recording_id = recording["id"]

Step 3: Stop a Recording Programmatically

To stop a recording, you must use the recordingId obtained during the start phase. You cannot stop a recording using only the conversationId. You must perform a PATCH or PUT operation depending on the desired behavior, but the standard approach to explicitly stop a recording is to update its status or use the specific stop endpoint if available.

In Genesys Cloud, the standard way to stop a recording is to update the recording entity. However, a more direct method often used is to query the recording by conversation ID to get the current recordingId if you did not save it, and then issue a stop command.

Actually, the most reliable API pattern for stopping a specific recording instance is to use the PATCH /api/v2/recording/{recordingId} endpoint with a status of stopped.

Endpoint: PATCH /api/v2/recording/{recordingId}

def stop_recording(auth: GenesysAuth, recording_id: str) -> Dict[str, Any]:
    """
    Stops an active recording by its ID.
    Scope Required: recording:update
    """
    url = f"{auth.base_url}/api/v2/recording/{recording_id}"
    headers = {
        "Authorization": f"Bearer {auth.get_access_token()}",
        "Content-Type": "application/json"
    }

    payload = {
        "status": "stopped"
    }

    try:
        response = requests.patch(url, headers=headers, json=payload)
        
        if response.status_code == 200:
            recording_data = response.json()
            print(f"Recording stopped successfully. Final Status: {recording_data.get('status')}")
            return recording_data
        else:
            response.raise_for_status()
            
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 404:
            print(f"Error: Recording {recording_id} not found.")
        elif e.response.status_code == 409:
            print(f"Error: {e.response.json().get('message', 'Recording is already stopped.')}")
        else:
            print(f"HTTP Error: {e.response.status_code} - {e.response.text}")
        raise

# Example Usage
# stop_recording(auth, recording_id)

Alternative: Find and Stop Recording by Conversation ID

If you do not have the recordingId saved, you can query recordings by conversation ID.

Endpoint: GET /api/v2/recording/conversations/{conversationId}/recordings

def stop_recording_by_conversation(auth: GenesysAuth, conversation_id: str) -> None:
    """
    Finds the active recording for a conversation and stops it.
    Scope Required: recording:update, recording:view
    """
    # Step 1: Get recordings for the conversation
    query_url = f"{auth.base_url}/api/v2/recording/conversations/{conversation_id}/recordings"
    headers = {
        "Authorization": f"Bearer {auth.get_access_token()}"
    }

    try:
        response = requests.get(query_url, headers=headers)
        response.raise_for_status()
        recordings = response.json().get("entities", [])

        if not recordings:
            print(f"No recordings found for conversation {conversation_id}.")
            return

        # Find the first active recording
        active_recording = next((r for r in recordings if r.get("status") == "active"), None)

        if active_recording:
            recording_id = active_recording["id"]
            print(f"Found active recording ID: {recording_id}")
            stop_recording(auth, recording_id)
        else:
            print(f"No active recordings found for conversation {conversation_id}.")

    except requests.exceptions.HTTPError as e:
        print(f"Error querying recordings: {e.response.text}")
        raise

# Example Usage
# stop_recording_by_conversation(auth, target_conversation_id)

Complete Working Example

This script demonstrates the full lifecycle: Authenticate, find an active conversation, start a recording, wait (simulated), and stop the recording.

import os
import time
import requests
from typing import Optional, Dict, Any

# --- Authentication Class (from Step 1) ---
class GenesysAuth:
    def __init__(self, client_id: str, client_secret: str, environment: str = "mypurecloud.com"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = f"https://api.{environment}"
        self.token_endpoint = f"{self.base_url}/oauth/token"
        self.access_token: Optional[str] = None

    def get_access_token(self) -> str:
        if self.access_token:
            return self.access_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(self.token_endpoint, headers=headers, data=data)
            response.raise_for_status()
            self.access_token = response.json()["access_token"]
            return self.access_token
        except Exception as e:
            raise Exception(f"Auth failed: {e}")

# --- API Functions ---

def get_active_conversation(auth: GenesysAuth) -> Optional[str]:
    """Returns the ID of the first active conversation."""
    url = f"{auth.base_url}/api/v2/conversations"
    headers = {"Authorization": f"Bearer {auth.get_access_token()}"}
    params = {"filter": "status=active"}
    
    try:
        resp = requests.get(url, headers=headers, params=params)
        resp.raise_for_status()
        entities = resp.json().get("entities", [])
        if entities:
            return entities[0]["id"]
        return None
    except Exception as e:
        print(f"Error getting conversations: {e}")
        return None

def start_recording(auth: GenesysAuth, conversation_id: str) -> Optional[str]:
    """Starts a recording and returns the recording ID."""
    url = f"{auth.base_url}/api/v2/recording"
    headers = {
        "Authorization": f"Bearer {auth.get_access_token()}",
        "Content-Type": "application/json"
    }
    payload = {
        "conversationId": conversation_id,
        "recordingType": "conversation"
    }
    
    try:
        resp = requests.post(url, headers=headers, json=payload)
        if resp.status_code == 201:
            return resp.json().get("id")
        elif resp.status_code == 409:
            print("Recording already active for this conversation.")
            return None
        else:
            resp.raise_for_status()
    except Exception as e:
        print(f"Error starting recording: {e}")
        return None

def stop_recording(auth: GenesysAuth, recording_id: str) -> bool:
    """Stops a recording by ID."""
    url = f"{auth.base_url}/api/v2/recording/{recording_id}"
    headers = {
        "Authorization": f"Bearer {auth.get_access_token()}",
        "Content-Type": "application/json"
    }
    payload = {"status": "stopped"}
    
    try:
        resp = requests.patch(url, headers=headers, json=payload)
        if resp.status_code == 200:
            print("Recording stopped successfully.")
            return True
        else:
            resp.raise_for_status()
    except Exception as e:
        print(f"Error stopping recording: {e}")
        return False

# --- Main Execution ---

def main():
    # Load credentials from environment variables
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")

    if not client_id or not client_secret:
        raise ValueError("Set GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET environment variables.")

    # 1. Authenticate
    print("Authenticating...")
    auth = GenesysAuth(client_id, client_secret)
    token = auth.get_access_token()
    print("Authentication successful.")

    # 2. Find an active conversation
    print("Searching for active conversations...")
    conv_id = get_active_conversation(auth)
    
    if not conv_id:
        print("No active conversations found. Start a call to test this script.")
        return

    print(f"Found active conversation: {conv_id}")

    # 3. Start Recording
    print("Starting recording...")
    recording_id = start_recording(auth, conv_id)
    
    if not recording_id:
        print("Failed to start recording or recording already exists.")
        return

    print(f"Recording started with ID: {recording_id}")

    # 4. Wait for a few seconds (Simulate call duration)
    print("Waiting 10 seconds...")
    time.sleep(10)

    # 5. Stop Recording
    print("Stopping recording...")
    success = stop_recording(auth, recording_id)
    
    if success:
        print("Process complete.")
    else:
        print("Failed to stop recording.")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token is invalid, expired, or missing.
• Fix: Ensure your client_id and client_secret are correct. Verify that the token refresh logic is working if the script runs for longer than an hour. Check that the OAuth client has the correct grant type (Client Credentials).

Error: 403 Forbidden

• Cause: The OAuth client lacks the required scopes.
• Fix: Go to the Genesys Cloud Admin Console > Platform Settings > OAuth Clients. Edit your client and ensure recording:update and recording:view are checked. Also, ensure the user associated with the client (if using Authorization Code flow) has the necessary security roles (e.g., Recording Admin).

Error: 409 Conflict

• Cause: Attempting to start a recording when one is already active for that conversation.
• Fix: Check the response body for the existing recordingId. If you need to stop the existing recording, use that ID. If you want to start a new one, you must stop the old one first.

Error: 404 Not Found

• Cause: The conversationId does not exist, is not active, or the recordingId is invalid.
• Fix: Verify that the conversation is indeed active (status=active). Recordings cannot be started on ended or queued conversations. Ensure the recordingId used for stopping is the one returned by the start API.

Error: 429 Too Many Requests

• Cause: You have exceeded the API rate limit.
• Fix: Implement exponential backoff. Genesys Cloud returns a Retry-After header in 429 responses. Parse this header and wait before retrying the request.

Official References

• Recording API Documentation
• Conversations API Documentation
• OAuth 2.0 Documentation
• Python SDK Documentation