How to use the Platform SDK for Python to bulk-create users from a CSV file
What You Will Build
You will build a Python script that reads employee data from a CSV file and creates corresponding user accounts in Genesys Cloud using the official genesys-cloud-sdk-python.
The script handles authentication via OAuth client credentials, processes the CSV data, constructs the required UserPost payloads, and submits them to the /api/v2/users endpoint.
The tutorial uses Python 3.9+ and the genesys-cloud-sdk-python package.
Prerequisites
- A Genesys Cloud organization with an API user or OAuth Client configured.
- OAuth Client Type: Confidential Client (Client Credentials Grant).
- Required OAuth Scope:
user:write(anduser:readif you wish to verify creation). - Python 3.9 or higher installed.
- The
genesys-cloud-sdk-pythonlibrary installed via pip. - A CSV file (
users.csv) with columns:email,name,division_id.
Install the SDK:
pip install genesys-cloud-sdk-python
Authentication Setup
The Genesys Cloud SDK simplifies authentication by handling the OAuth token exchange and refresh logic internally when you use the PureCloudPlatformClientV2 with a Configuration object. You do not need to manually manage HTTP requests for token retrieval.
You must initialize the platform client with your OAuth client ID, client secret, and environment. For most production integrations, you will use my.genesys.cloud (US) or au.my.genesys.cloud (Australia).
import os
from purecloudplatformclientv2 import Configuration, PureCloudPlatformClientV2
def initialize_platform_client() -> PureCloudPlatformClientV2:
"""
Initializes and returns a configured PureCloudPlatformClientV2 instance.
Uses environment variables for sensitive credentials.
"""
# 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("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET environment variables are required.")
# Configure the client
configuration = Configuration()
configuration.client_id = client_id
configuration.client_secret = client_secret
# Set the host based on your region. Default is US.
# For Australia, use: configuration.host = "https://au.my.genesys.cloud"
configuration.host = "https://api.mypurecloud.com"
# Initialize the platform client
pure_cloud_client = PureCloudPlatformClientV2(configuration)
return pure_cloud_client
Important Security Note: Never hardcode client IDs or secrets in your source code. Use environment variables or a secure secret management service. The SDK caches the access token and automatically requests a new one when the current token expires, so you only need to initialize the client once per script execution.
Implementation
Step 1: Parse CSV and Validate Data
Before interacting with the API, you must load the CSV data and validate the structure. The Genesys Cloud User API requires specific fields for a minimal user creation. The absolute minimum required fields are email and name. However, in most multi-division organizations, you must also specify the division_id. If the division is omitted, the user is assigned to the default division, which may not be desired.
You will use the standard csv module to parse the file. This avoids external dependencies like pandas for this specific task, keeping the script lightweight.
import csv
from typing import List, Dict
def load_users_from_csv(file_path: str) -> List[Dict[str, str]]:
"""
Reads a CSV file and returns a list of dictionaries representing user data.
Expected CSV columns: email, name, division_id
"""
users_data = []
try:
with open(file_path, mode='r', encoding='utf-8') as csvfile:
reader = csv.DictReader(csvfile)
# Validate that required columns exist
required_columns = {'email', 'name', 'division_id'}
if not required_columns.issubset(set(reader.fieldnames or [])):
missing = required_columns - set(reader.fieldnames or [])
raise ValueError(f"CSV is missing required columns: {missing}")
for row in reader:
# Basic validation
if not row.get('email') or not row.get('name'):
print(f"Warning: Skipping row with missing email or name: {row}")
continue
users_data.append({
'email': row['email'].strip(),
'name': row['name'].strip(),
'division_id': row['division_id'].strip()
})
except FileNotFoundError:
raise FileNotFoundError(f"The file {file_path} was not found.")
except Exception as e:
raise RuntimeError(f"Error reading CSV: {e}")
return users_data
Step 2: Map Data to SDK Objects
The Genesys Cloud Python SDK uses strongly typed classes for request bodies. To create a user, you must instantiate the UserPost class. You cannot simply pass a raw dictionary to the create_user method; the SDK expects the specific model object.
You need to import UserPost from the purecloudplatformclientv2.models module.
from purecloudplatformclientv2.models import UserPost
def create_user_post_object(user_data: Dict[str, str]) -> UserPost:
"""
Converts a dictionary from the CSV into a UserPost SDK object.
"""
# The UserPost constructor accepts keyword arguments matching the JSON schema
user_post = UserPost(
email=user_data['email'],
name=user_data['name'],
division_id=user_data['division_id']
)
return user_post
Note on Division IDs: The division_id must be a valid UUID string from your Genesys Cloud organization. You can find these via the Admin UI (Organization → Divisions) or by querying the API (GET /api/v2/organizations/divisions). If you provide an invalid division ID, the API will return a 400 Bad Request.
Step 3: Implement Bulk Creation with Error Handling
The Genesys Cloud API does not have a single “bulk create users” endpoint. You must call POST /api/v2/users individually for each user. This approach allows for granular error handling. If one user fails to create (e.g., duplicate email), the others can still succeed.
You must handle specific HTTP status codes:
- 400 Bad Request: Usually indicates invalid data (bad email format, missing required field, invalid division ID).
- 409 Conflict: Indicates a user with that email already exists.
- 429 Too Many Requests: Rate limiting. The SDK does not automatically retry 429s in all methods, so you should implement a backoff strategy or process users sequentially with delays if creating large volumes.
- 401/403: Authentication or permission issues.
For this tutorial, we will process users sequentially. For very large datasets (1000+ users), you would implement threading or async processing, but that introduces complexity with rate limits. Sequential processing is safer for initial implementations.
import time
import logging
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def bulk_create_users(client: PureCloudPlatformClientV2, users_data: List[Dict[str, str]]) -> List[Dict]:
"""
Iterates through user data and creates users in Genesys Cloud.
Returns a list of results containing success/failure status.
"""
results = []
# Initialize the Users API client
users_api = client.users_api
for i, user_info in enumerate(users_data):
logger.info(f"Processing user {i+1}/{len(users_data)}: {user_info['email']}")
try:
# Create the SDK object
user_post = create_user_post_object(user_info)
# Call the API
# The create_user method returns a User object on success
created_user = users_api.post_users(body=user_post)
results.append({
'email': user_info['email'],
'status': 'success',
'user_id': created_user.id,
'name': created_user.name
})
logger.info(f"Successfully created user: {created_user.name} (ID: {created_user.id})")
except Exception as e:
# Capture the error for reporting
results.append({
'email': user_info['email'],
'status': 'failed',
'error': str(e)
})
logger.error(f"Failed to create user {user_info['email']}: {e}")
# Small delay to help avoid rate limiting (429)
# Genesys Cloud allows ~100 requests per second per client, but
# user creation is heavier. 100ms delay is safe for <100 users.
time.sleep(0.1)
return results
Step 4: Handle Rate Limiting (Advanced)
If you are creating more than 100 users, you should implement exponential backoff for 429 errors. The SDK raises a ApiException for HTTP errors. You can inspect the status code within the exception.
Here is an enhanced version of the creation loop with retry logic:
from purecloudplatformclientv2.rest import ApiException
def create_user_with_retry(client: PureCloudPlatformClientV2, user_post: UserPost, max_retries: int = 3) -> any:
"""
Attempts to create a user with exponential backoff on 429 errors.
"""
users_api = client.users_api
for attempt in range(1, max_retries + 1):
try:
return users_api.post_users(body=user_post)
except ApiException as e:
if e.status == 429:
wait_time = 2 ** attempt # 2, 4, 8 seconds
logger.warning(f"Rate limited (429). Retrying in {wait_time} seconds... (Attempt {attempt}/{max_retries})")
time.sleep(wait_time)
else:
# Re-raise other exceptions immediately
raise e
raise Exception("Max retries exceeded for 429 Too Many Requests")
Complete Working Example
The following script combines all steps into a single executable file. Save this as bulk_create_users.py.
import os
import csv
import time
import logging
from typing import List, Dict
from purecloudplatformclientv2 import Configuration, PureCloudPlatformClientV2
from purecloudplatformclientv2.models import UserPost
from purecloudplatformclientv2.rest import ApiException
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def initialize_platform_client() -> PureCloudPlatformClientV2:
"""
Initializes and returns a configured PureCloudPlatformClientV2 instance.
"""
client_id = os.getenv("GENESYS_CLIENT_ID")
client_secret = os.getenv("GENESYS_CLIENT_SECRET")
if not client_id or not client_secret:
raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET environment variables are required.")
configuration = Configuration()
configuration.client_id = client_id
configuration.client_secret = client_secret
configuration.host = "https://api.mypurecloud.com" # Change for other regions
return PureCloudPlatformClientV2(configuration)
def load_users_from_csv(file_path: str) -> List[Dict[str, str]]:
"""
Reads a CSV file and returns a list of dictionaries representing user data.
Expected CSV columns: email, name, division_id
"""
users_data = []
try:
with open(file_path, mode='r', encoding='utf-8') as csvfile:
reader = csv.DictReader(csvfile)
required_columns = {'email', 'name', 'division_id'}
if not required_columns.issubset(set(reader.fieldnames or [])):
missing = required_columns - set(reader.fieldnames or [])
raise ValueError(f"CSV is missing required columns: {missing}")
for row in reader:
if not row.get('email') or not row.get('name'):
logger.warning(f"Skipping row with missing email or name: {row}")
continue
users_data.append({
'email': row['email'].strip(),
'name': row['name'].strip(),
'division_id': row['division_id'].strip()
})
except FileNotFoundError:
raise FileNotFoundError(f"The file {file_path} was not found.")
except Exception as e:
raise RuntimeError(f"Error reading CSV: {e}")
return users_data
def create_user_post_object(user_data: Dict[str, str]) -> UserPost:
"""
Converts a dictionary from the CSV into a UserPost SDK object.
"""
return UserPost(
email=user_data['email'],
name=user_data['name'],
division_id=user_data['division_id']
)
def bulk_create_users(client: PureCloudPlatformClientV2, users_data: List[Dict[str, str]]) -> List[Dict]:
"""
Iterates through user data and creates users in Genesys Cloud.
"""
results = []
users_api = client.users_api
for i, user_info in enumerate(users_data):
logger.info(f"Processing user {i+1}/{len(users_data)}: {user_info['email']}")
try:
user_post = create_user_post_object(user_info)
# Simple retry logic for 429s
max_retries = 3
created_user = None
for attempt in range(1, max_retries + 1):
try:
created_user = users_api.post_users(body=user_post)
break
except ApiException as e:
if e.status == 429:
wait_time = 2 ** attempt
logger.warning(f"Rate limited (429). Retrying in {wait_time}s...")
time.sleep(wait_time)
else:
raise e
if created_user:
results.append({
'email': user_info['email'],
'status': 'success',
'user_id': created_user.id,
'name': created_user.name
})
logger.info(f"Success: Created {created_user.name} (ID: {created_user.id})")
else:
results.append({
'email': user_info['email'],
'status': 'failed',
'error': 'Max retries exceeded'
})
except Exception as e:
results.append({
'email': user_info['email'],
'status': 'failed',
'error': str(e)
})
logger.error(f"Failed to create user {user_info['email']}: {e}")
# Standard delay between requests
time.sleep(0.1)
return results
def save_results_to_csv(results: List[Dict], output_file: str):
"""
Saves the results of the bulk operation to a CSV file.
"""
if not results:
logger.warning("No results to save.")
return
with open(output_file, mode='w', newline='', encoding='utf-8') as csvfile:
fieldnames = ['email', 'status', 'user_id', 'name', 'error']
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
for row in results:
# Ensure all fields are present
writer.writerow({
'email': row.get('email', ''),
'status': row.get('status', ''),
'user_id': row.get('user_id', ''),
'name': row.get('name', ''),
'error': row.get('error', '')
})
logger.info(f"Results saved to {output_file}")
def main():
csv_file_path = os.getenv("CSV_INPUT_PATH", "users.csv")
output_file_path = os.getenv("CSV_OUTPUT_PATH", "creation_results.csv")
logger.info("Starting Bulk User Creation Script")
try:
# 1. Initialize Client
client = initialize_platform_client()
logger.info("Platform client initialized successfully.")
# 2. Load Data
users_data = load_users_from_csv(csv_file_path)
if not users_data:
logger.warning("No users found in CSV. Exiting.")
return
logger.info(f"Loaded {len(users_data)} users from CSV.")
# 3. Create Users
results = bulk_create_users(client, users_data)
# 4. Save Results
save_results_to_csv(results, output_file_path)
# 5. Summary
success_count = sum(1 for r in results if r['status'] == 'success')
fail_count = sum(1 for r in results if r['status'] == 'failed')
logger.info(f"Completed. Success: {success_count}, Failed: {fail_count}")
except Exception as e:
logger.error(f"Critical error: {e}")
raise
if __name__ == "__main__":
main()
Common Errors & Debugging
Error: 400 Bad Request - Invalid Division ID
Cause: The division_id provided in the CSV is not a valid UUID or does not exist in your organization.
Fix: Verify the division ID by running the following code snippet:
from purecloudplatformclientv2 import OrganizationsApi
orgs_api = client.organizations_api
division_id = "your-division-uuid-here"
try:
division = orgs_api.get_organization_division(division_id)
print(f"Division found: {division.name}")
except ApiException as e:
print(f"Division not found or invalid: {e.status} {e.reason}")
Error: 409 Conflict - Email Already Exists
Cause: You attempted to create a user with an email address that is already associated with an existing user in Genesys Cloud.
Fix: The API prevents duplicate emails. You must either skip the user, update the existing user, or use a different email. To check if a user exists before creating:
users_api = client.users_api
try:
# Search for user by email
search_result = users_api.post_users_search(body={"query": f"email:{user_email}"})
if search_result.total > 0:
print(f"User already exists: {search_result.users[0].name}")
except ApiException as e:
print(f"Search failed: {e}")
Error: 401 Unauthorized
Cause: The OAuth client ID or secret is incorrect, or the token has expired and could not be refreshed.
Fix: Ensure your environment variables GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET are correct. Verify that the OAuth client is active in the Admin UI (Admin → Integrations → OAuth).
Error: 403 Forbidden
Cause: The OAuth client does not have the user:write scope.
Fix: Go to Admin → Integrations → OAuth, select your client, and ensure user:write is checked under Scopes. Save the changes. Note that changes to scopes may require re-authentication or generating a new client secret.