How to Programmatically Start and Stop Call Recordings via the Genesys Cloud Recording API
What You Will Build
- A Python script that initiates a recording on an active conversation using the Conversation ID and terminates it on demand.
- This tutorial uses the Genesys Cloud Platform API v2 (
/api/v2/conversations/recordings) and the official Python SDK. - The implementation is written in Python 3.9+ using the
genesyscloudpackage andrequestsfor raw HTTP verification.
Prerequisites
- OAuth Client Type: A Service Account (Confidential Client) or a User Account (Public Client) with valid credentials.
- Required OAuth Scopes:
conversation:recording:write(Required to start/stop recordings)conversation:read(Required to verify conversation state)
- SDK Version:
genesyscloud>= 2.0.0 - Runtime: Python 3.9 or higher.
- External Dependencies:
genesyscloud(Official SDK)requests(For raw API examples)
Authentication Setup
Genesys Cloud uses OAuth 2.0 for authentication. For server-side integrations (like this one), the Client Credentials Grant flow is the standard approach. This flow requires a Client ID and Client Secret associated with a Service Account.
The token expires after a short duration (typically 1 hour). In production, you must implement token caching and automatic refresh logic. For this tutorial, we will use the SDK’s built-in token handling, which manages caching if configured correctly, but we will also show the raw HTTP request for clarity.
Raw HTTP Authentication Flow
import requests
import json
from typing import Optional
class GenesysAuth:
def __init__(self, client_id: str, client_secret: str, base_url: str = "https://api.mypurecloud.com"):
self.client_id = client_id
self.client_secret = client_secret
self.base_url = base_url
self.token_url = f"{base_url}/oauth/token"
self.access_token: Optional[str] = None
self.expires_in: int = 0
def get_access_token(self) -> str:
"""
Retrieves an OAuth2 access token using Client Credentials flow.
"""
# Check if we have a cached token (simplified for tutorial)
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,
"scope": "conversation:recording:write conversation:read"
}
try:
response = requests.post(self.token_url, 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 ValueError("Invalid Client ID or Secret. Check your Service Account permissions.")
elif response.status_code == 403:
raise ValueError("Forbidden. The Service Account lacks the required scopes.")
else:
raise RuntimeError(f"Authentication failed: {e}")
except requests.exceptions.RequestException as e:
raise RuntimeError(f"Network error during authentication: {e}")
# Usage Example
# auth = GenesysAuth(client_id="YOUR_CLIENT_ID", client_secret="YOUR_CLIENT_SECRET")
# token = auth.get_access_token()
# print(f"Token: {token[:10]}...")
Critical Note on Scopes: If you receive a 403 Forbidden error when attempting to start a recording, verify that the Service Account has the conversation:recording:write scope. This scope is not included in the default admin role for all service accounts and must be explicitly assigned in the Admin Console under Admin > Platform Services > Integrations > OAuth Clients.
Implementation
Step 1: Verify Conversation State Before Recording
Before sending a command to start a recording, you must ensure the conversation exists and is in an active state. Attempting to record a disconnected or queued conversation will result in a 400 Bad Request or 404 Not Found.
We will use the GET /api/v2/conversations/{conversationId} endpoint.
Python SDK Implementation
from purecloudplatformclientv2 import (
ApiClient,
Configuration,
ConversationApi,
ApiException
)
def get_conversation_status(api_client: ApiClient, conversation_id: str) -> dict:
"""
Retrieves the current state of a conversation.
Args:
api_client: Initialized PureCloudPlatformClientV2 API client.
conversation_id: The UUID of the conversation.
Returns:
Dictionary containing conversation details.
"""
conversation_api = ConversationApi(api_client)
try:
# Fetch conversation details
response = conversation_api.get_conversation(
conversation_id=conversation_id
)
return {
"id": response.id,
"state": response.state, # 'connected', 'ringing', 'queued', 'disconnected'
"type": response.type, # 'voice', 'chat', etc.
"participants": response.participants
}
except ApiException as e:
if e.status == 404:
raise ValueError(f"Conversation {conversation_id} not found. It may have ended.")
elif e.status == 403:
raise PermissionError("Access denied. Check OAuth scopes.")
else:
raise RuntimeError(f"Error fetching conversation: {e.body}")
Realistic Response Body
{
"id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
"type": "voice",
"state": "connected",
"participants": [
{
"id": "participant-id-1",
"externalContact": {
"name": "John Doe",
"phone": "+15550199"
},
"state": "connected"
},
{
"id": "participant-id-2",
"user": {
"id": "agent-id-123",
"name": "Jane Agent"
},
"state": "connected"
}
]
}
Step 2: Start the Recording
To start a recording, you send a POST request to /api/v2/conversations/{conversationId}/recordings.
The body of this request can be empty for a default recording, or you can include a recording object to specify metadata such as name or tags.
Python SDK Implementation
from purecloudplatformclientv2.models import ConversationRecording
def start_recording(api_client: ApiClient, conversation_id: str, recording_name: str = "Auto-Recorded Call") -> str:
"""
Initiates a recording on an active conversation.
Args:
api_client: Initialized PureCloudPlatformClientV2 API client.
conversation_id: The UUID of the conversation.
recording_name: Optional name for the recording.
Returns:
The ID of the newly created recording.
"""
conversation_api = ConversationApi(api_client)
# Construct the recording body
# Note: The SDK requires the body to be a ConversationRecording object
# even if minimal fields are used.
body = ConversationRecording()
body.name = recording_name
try:
# Post the recording start command
response = conversation_api.post_conversation_recording(
conversation_id=conversation_id,
body=body
)
print(f"Recording started successfully. Recording ID: {response.id}")
return response.id
except ApiException as e:
if e.status == 400:
raise ValueError(f"Bad Request: {e.body}. Ensure conversation is 'connected'.")
elif e.status == 409:
raise RuntimeError("Conflict: A recording is already in progress for this conversation.")
elif e.status == 429:
raise RuntimeError("Rate Limited: Too many requests. Implement exponential backoff.")
else:
raise RuntimeError(f"Failed to start recording: {e.body}")
Raw HTTP Equivalent (for debugging)
If you need to bypass the SDK for debugging, here is the raw requests call:
def start_recording_raw(access_token: str, conversation_id: str, base_url: str = "https://api.mypurecloud.com"):
url = f"{base_url}/api/v2/conversations/{conversation_id}/recordings"
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
}
# Minimal body. Genesys Cloud generates the rest.
payload = {
"name": "Manual API Recording"
}
response = requests.post(url, headers=headers, json=payload)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"Failed to start recording: {response.status_code} - {response.text}")
Step 3: Stop the Recording
To stop a recording, you send a DELETE request to /api/v2/conversations/{conversationId}/recordings/{recordingId}.
You must know the recordingId. This is returned in the response of the POST call in Step 2. If you did not capture it, you must list existing recordings for the conversation using GET /api/v2/conversations/{conversationId}/recordings.
Python SDK Implementation
def stop_recording(api_client: ApiClient, conversation_id: str, recording_id: str) -> bool:
"""
Stops an active recording.
Args:
api_client: Initialized PureCloudPlatformClientV2 API client.
conversation_id: The UUID of the conversation.
recording_id: The UUID of the recording to stop.
Returns:
True if successful.
"""
conversation_api = ConversationApi(api_client)
try:
# Delete the recording resource to stop it
# Note: This does not delete the audio file, it just stops the capture.
conversation_api.delete_conversation_recording(
conversation_id=conversation_id,
recording_id=recording_id
)
print(f"Recording {recording_id} stopped successfully.")
return True
except ApiException as e:
if e.status == 404:
raise ValueError(f"Recording {recording_id} not found. It may have already stopped.")
elif e.status == 400:
raise ValueError(f"Bad Request: {e.body}. Recording might not be active.")
else:
raise RuntimeError(f"Failed to stop recording: {e.body}")
Step 4: Handling Edge Cases and Pagination
If you need to find a recording ID because you missed it during creation, you must query the conversation’s recordings. This endpoint supports pagination.
from purecloudplatformclientv2.models import ConversationRecordingList
def get_active_recording_id(api_client: ApiClient, conversation_id: str) -> Optional[str]:
"""
Finds the ID of an active recording for a conversation.
Args:
api_client: Initialized PureCloudPlatformClientV2 API client.
conversation_id: The UUID of the conversation.
Returns:
The recording ID if one is active, else None.
"""
conversation_api = ConversationApi(api_client)
try:
# Fetch recordings for the conversation
# Limit to 1 for efficiency if we only need the latest
response = conversation_api.get_conversation_recordings(
conversation_id=conversation_id,
limit=10,
expand=["recordings"]
)
if response.recordings:
for rec in response.recordings:
# Check if the recording is still active
# The API does not explicitly mark 'active' in the list response,
# but a recording present in the list and not yet processed is usually active.
# However, the most reliable way is to check the 'state' if available,
# or assume the latest one is active if the conversation is connected.
if rec.state == "active":
return rec.id
return None
except ApiException as e:
raise RuntimeError(f"Error fetching recordings: {e.body}")
Complete Working Example
This script combines authentication, validation, starting, and stopping logic into a single runnable module.
import os
import sys
from purecloudplatformclientv2 import (
ApiClient,
Configuration,
ConversationApi,
ApiException,
ConversationRecording
)
# Configuration
CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")
BASE_URL = os.getenv("GENESYS_BASE_URL", "https://api.mypurecloud.com")
if not CLIENT_ID or not CLIENT_SECRET:
raise EnvironmentError("Missing GENESYS_CLIENT_ID or GENESYS_CLIENT_SECRET environment variables.")
def initialize_api_client() -> ApiClient:
"""
Initializes the Genesys Cloud API Client with OAuth credentials.
"""
configuration = Configuration(
host=BASE_URL,
client_id=CLIENT_ID,
client_secret=CLIENT_SECRET,
scope=["conversation:recording:write", "conversation:read"]
)
# The SDK handles token acquisition automatically when the first API call is made
return ApiClient(configuration)
def manage_recording(conversation_id: str, action: str = "start", recording_id: str = None):
"""
Main logic to start or stop a recording.
Args:
conversation_id: UUID of the conversation.
action: 'start' or 'stop'.
recording_id: Required if action is 'stop'.
"""
api_client = initialize_api_client()
conversation_api = ConversationApi(api_client)
try:
# Step 1: Verify Conversation
print(f"Checking conversation {conversation_id}...")
conv_response = conversation_api.get_conversation(conversation_id)
if conv_response.state != "connected":
print(f"Warning: Conversation is in state '{conv_response.state}'. Recording may fail.")
if action == "start":
print("Starting recording...")
body = ConversationRecording()
body.name = "Programmatic Recording via API"
response = conversation_api.post_conversation_recording(
conversation_id=conversation_id,
body=body
)
print(f"Success: Recording started. ID: {response.id}")
return response.id
elif action == "stop":
if not recording_id:
raise ValueError("recording_id is required for 'stop' action.")
print(f"Stopping recording {recording_id}...")
conversation_api.delete_conversation_recording(
conversation_id=conversation_id,
recording_id=recording_id
)
print("Success: Recording stopped.")
return True
except ApiException as e:
print(f"API Error {e.status}: {e.body}")
sys.exit(1)
except Exception as e:
print(f"Unexpected Error: {str(e)}")
sys.exit(1)
if __name__ == "__main__":
# Example Usage
# Replace with a real active conversation ID
TARGET_CONVERSATION_ID = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
# Start
rec_id = manage_recording(TARGET_CONVERSATION_ID, action="start")
# Stop (Simulated delay in real scenario)
if rec_id:
manage_recording(TARGET_CONVERSATION_ID, action="stop", recording_id=rec_id)
Common Errors & Debugging
Error: 403 Forbidden
- Cause: The Service Account used for authentication does not have the
conversation:recording:writescope. - Fix: Go to Admin > Platform Services > Integrations > OAuth Clients. Edit your client. Under Scopes, ensure
conversation:recording:writeis checked. Save and regenerate the token.
Error: 409 Conflict
- Cause: A recording is already active for this conversation. Genesys Cloud does not support multiple simultaneous recordings on the same conversation by default.
- Fix: Check if a recording is already running using
GET /api/v2/conversations/{conversationId}/recordings. If one exists, stop it first or ignore the start command.
Error: 400 Bad Request
- Cause: The conversation is not in the
connectedstate. It might bequeued,ringing, ordisconnected. - Fix: Implement a polling mechanism to wait for the conversation state to become
connectedbefore issuing thePOSTcommand.
Error: 429 Too Many Requests
- Cause: You have exceeded the API rate limit for your tenant.
- Fix: Implement exponential backoff. Wait 1 second, then 2, then 4, etc., before retrying the request. Check the
Retry-Afterheader in the response for the exact wait time.
import time
def retry_with_backoff(func, *args, max_retries=3, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except ApiException as e:
if e.status == 429:
wait_time = 2 ** attempt
print(f"Rate limited. Retrying in {wait_time} seconds...")
time.sleep(wait_time)
else:
raise
raise RuntimeError("Max retries exceeded.")