How to use the Platform SDK for Python to bulk-create users from a CSV file

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 (and user:read if you wish to verify creation).
  • Python 3.9 or higher installed.
  • The genesys-cloud-sdk-python library 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.

Official References