Bulk-Create Genesys Cloud Users from CSV Using the Python Platform SDK

What You Will Build

• You will build a Python script that reads a CSV file containing user details and creates corresponding users in Genesys Cloud CX.
• You will use the genesyscloud Python SDK (version 130+) to handle authentication and API calls.
• You will implement robust error handling, rate-limit retries, and duplicate detection to ensure production-ready reliability.

Prerequisites

OAuth Client Configuration

You must have a Genesys Cloud OAuth client with the following permissions. The Confidential Client type is recommended for backend scripts.

Required Scopes:

• user:write - Required to create new user records.
• user:read - Required to check for existing users before creation (optional but recommended).
• routing:queue:write - Required if you are assigning users to specific routing queues during creation.
• routing:skill:read - Required if you are assigning specific skills or proficiencies.

Environment Setup

• Python Version: 3.8 or higher.
• IDE: VS Code, PyCharm, or any editor with Python support.
• Dependencies:
  • genesyscloud: The official Genesys Cloud Python SDK.
  • pandas: For efficient CSV parsing and data manipulation.
  • requests: Included in the SDK dependencies, but useful for debugging if needed.

Install the dependencies via pip:

pip install genesyscloud pandas

CSV File Structure

Your input CSV file (users.csv) must contain the following columns at a minimum. This tutorial assumes a standard structure:

email,first_name,last_name,username,division_id,queue_id
jsmith@example.com,John,Smith,jsmith,12345678-1234-1234-1234-123456789012,87654321-4321-4321-4321-210987654321
jdoe@example.com,Jane,Doe,jdoe,12345678-1234-1234-1234-123456789012,87654321-4321-4321-4321-210987654321

Note:

• division_id: The UUID of the division where the user will be created. If omitted, the default division is used.
• queue_id: The UUID of the routing queue to assign the user to. This is optional for user creation but common in bulk onboarding.

Authentication Setup

The Genesys Cloud Python SDK uses a credential object to manage OAuth tokens. For a backend script, the Confidential Client flow is the standard. This flow exchanges a client ID and secret for an access token.

The SDK handles token refresh automatically if you use the genesyscloud.auth module correctly. However, for a simple batch script, initializing the client once and reusing it is sufficient.

Step 1: Initialize the API Client

Create a Python file named bulk_create_users.py. Start by importing the necessary modules and defining your credentials.

import os
import sys
import csv
import time
import logging
import pandas as pd
from genesyscloud.auth import OAuthClient
from genesyscloud.platform_client import PlatformClient
from genesyscloud.user_api import UserApi
from genesyscloud.routing_api import RoutingApi
from genesyscloud.models import CreateUserRequest, UserPresenceConfiguration
from genesyscloud.rest import ApiException

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Configuration
CLIENT_ID = os.environ.get('GENESYS_CLIENT_ID')
CLIENT_SECRET = os.environ.get('GENESYS_CLIENT_SECRET')
ENVIRONMENT = os.environ.get('GENESYS_ENVIRONMENT', 'mypurecloud.com')
CSV_FILE_PATH = 'users.csv'

def init_genesys_client():
    """
    Initializes and returns the Genesys Cloud Platform Client.
    Raises an exception if authentication fails.
    """
    if not CLIENT_ID or not CLIENT_SECRET:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET environment variables are required.")

    # Create OAuth client
    oauth = OAuthClient(client_id=CLIENT_ID, client_secret=CLIENT_SECRET, environment=ENVIRONMENT)
    
    try:
        # Authenticate using the confidential client flow
        oauth.authenticate()
    except Exception as e:
        logger.error(f"Authentication failed: {e}")
        raise

    # Create the platform client
    platform_client = PlatformClient(oauth)
    return platform_client

Why this works:

• The OAuthClient manages the token lifecycle.
• PlatformClient wraps the underlying HTTP calls and provides access to all API modules (User, Routing, etc.).
• Using environment variables ensures credentials are not hardcoded in the script.

Implementation

Step 2: Load and Validate CSV Data

Before making any API calls, load the CSV data into a structured format. Pandas is ideal for this because it handles missing values and type conversions efficiently.

def load_csv_data(file_path: str) -> pd.DataFrame:
    """
    Loads the CSV file and validates required columns.
    Returns a DataFrame with cleaned data.
    """
    try:
        df = pd.read_csv(file_path)
    except FileNotFoundError:
        logger.error(f"CSV file not found: {file_path}")
        sys.exit(1)
    except pd.errors.EmptyDataError:
        logger.error("CSV file is empty.")
        sys.exit(1)

    # Required columns
    required_columns = ['email', 'first_name', 'last_name', 'username']
    missing_columns = [col for col in required_columns if col not in df.columns]
    
    if missing_columns:
        logger.error(f"Missing required columns in CSV: {missing_columns}")
        sys.exit(1)

    # Clean data: strip whitespace and convert to lowercase for emails/usernames
    df['email'] = df['email'].str.strip().str.lower()
    df['username'] = df['username'].str.strip().str.lower()
    df['first_name'] = df['first_name'].str.strip()
    df['last_name'] = df['last_name'].str.strip()

    # Drop rows with missing essential data
    df.dropna(subset=['email', 'first_name', 'last_name', 'username'], inplace=True)
    
    logger.info(f"Loaded {len(df)} valid user records from {file_path}")
    return df

Key Validation Points:

• Uniqueness: Genesys Cloud requires unique username and email per organization. The script below will check for duplicates before creation.
• Data Cleaning: Whitespace in emails or usernames often causes API rejections. Stripping them ensures consistency.

Step 3: Create Users with Error Handling and Retries

The core logic involves iterating through the DataFrame and calling the UserApi. Genesys Cloud APIs return 429 Too Many Requests if you exceed rate limits. The SDK does not automatically retry 429s in all versions, so explicit retry logic is recommended for bulk operations.

Additionally, creating a user does not automatically assign them to a queue. You must make a separate API call to RoutingApi to assign the user to a queue if required.

def create_user(client: PlatformClient, user_data: dict) -> str:
    """
    Creates a single user in Genesys Cloud.
    Returns the user ID if successful, None otherwise.
    """
    user_api = client.user_api
    routing_api = client.routing_api

    email = user_data['email']
    first_name = user_data['first_name']
    last_name = user_data['last_name']
    username = user_data['username']
    division_id = user_data.get('division_id')
    queue_id = user_data.get('queue_id')

    # 1. Construct the CreateUserRequest object
    # The SDK model enforces required fields
    create_request = CreateUserRequest(
        email=email,
        name=f"{first_name} {last_name}",
        username=username,
        division_id=division_id,
        presence_configuration=UserPresenceConfiguration(
            presence_state_id="available" # Set default presence to available
        )
    )

    try:
        # 2. Create the user
        # Max retries for 429 errors
        max_retries = 3
        retry_count = 0
        
        while retry_count <= max_retries:
            try:
                response = user_api.post_users(body=create_request)
                user_id = response.id
                logger.info(f"User created successfully: {email} (ID: {user_id})")
                
                # 3. Assign to Queue if provided
                if queue_id:
                    try:
                        # Create a new member object for the queue
                        from genesyscloud.models import QueueMember
                        queue_member = QueueMember(
                            user_id=user_id,
                            wrap_up_code="default" # Optional: set default wrap-up code
                        )
                        routing_api.post_routing_queues_members(queue_id=queue_id, body=queue_member)
                        logger.info(f"User {email} assigned to queue {queue_id}")
                    except ApiException as routing_e:
                        if routing_e.status == 409:
                            logger.warning(f"User {email} already in queue {queue_id} or duplicate member error.")
                        else:
                            logger.error(f"Failed to assign user {email} to queue: {routing_e}")
                
                return user_id

            except ApiException as e:
                if e.status == 429:
                    retry_count += 1
                    wait_time = 2 ** retry_count  # Exponential backoff: 2s, 4s, 8s
                    logger.warning(f"Rate limit hit for {email}. Retrying in {wait_time}s...")
                    time.sleep(wait_time)
                elif e.status == 409:
                    # 409 Conflict usually means the user already exists (duplicate username/email)
                    logger.warning(f"User {email} already exists (Conflict 409). Skipping.")
                    return None
                else:
                    logger.error(f"Failed to create user {email}: {e.status} - {e.reason}")
                    return None
        
        # If we exit the loop after max retries
        logger.error(f"Failed to create user {email} after {max_retries} retries due to rate limiting.")
        return None

    except Exception as e:
        logger.error(f"Unexpected error creating user {email}: {e}")
        return None

Why this structure is critical:

• 409 Conflict Handling: Genesys Cloud returns 409 if a user with the same username or email already exists. Ignoring this will crash the script or require manual cleanup. By returning None, you can track skipped users.
• 429 Rate Limiting: The Genesys Cloud API has strict rate limits (e.g., 10 requests per second for some endpoints). The exponential backoff strategy prevents the script from being blocked entirely.
• Queue Assignment: User creation and queue assignment are separate operations. You cannot assign a user to a queue in the CreateUserRequest body. You must call POST /api/v2/routing/queues/{queueId}/members after the user is created.

Step 4: Process Results and Generate Report

After processing all users, generate a summary report. This helps administrators verify the outcome and identify failed entries.

def process_users(client: PlatformClient, df: pd.DataFrame) -> pd.DataFrame:
    """
    Iterates through the DataFrame and creates users.
    Returns a DataFrame with results.
    """
    results = []
    
    for index, row in df.iterrows():
        logger.info(f"Processing user {index + 1}/{len(df)}: {row['email']}")
        
        user_id = create_user(client, row.to_dict())
        
        results.append({
            'email': row['email'],
            'username': row['username'],
            'user_id': user_id,
            'status': 'Created' if user_id else 'Skipped/Error',
            'error': None if user_id else 'Exists or Failed'
        })
        
        # Optional: Small delay between requests to avoid triggering rate limits
        # This is a softer approach than exponential backoff for normal operation
        time.sleep(0.2)

    return pd.DataFrame(results)

Complete Working Example

Below is the complete, copy-pasteable script. Save this as bulk_create_users.py.

import os
import sys
import csv
import time
import logging
import pandas as pd
from genesyscloud.auth import OAuthClient
from genesyscloud.platform_client import PlatformClient
from genesyscloud.user_api import UserApi
from genesyscloud.routing_api import RoutingApi
from genesyscloud.models import CreateUserRequest, UserPresenceConfiguration, QueueMember
from genesyscloud.rest import ApiException

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Configuration
CLIENT_ID = os.environ.get('GENESYS_CLIENT_ID')
CLIENT_SECRET = os.environ.get('GENESYS_CLIENT_SECRET')
ENVIRONMENT = os.environ.get('GENESYS_ENVIRONMENT', 'mypurecloud.com')
CSV_FILE_PATH = 'users.csv'

def init_genesys_client():
    """Initializes and returns the Genesys Cloud Platform Client."""
    if not CLIENT_ID or not CLIENT_SECRET:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET environment variables are required.")

    oauth = OAuthClient(client_id=CLIENT_ID, client_secret=CLIENT_SECRET, environment=ENVIRONMENT)
    
    try:
        oauth.authenticate()
    except Exception as e:
        logger.error(f"Authentication failed: {e}")
        raise

    return PlatformClient(oauth)

def load_csv_data(file_path: str) -> pd.DataFrame:
    """Loads the CSV file and validates required columns."""
    try:
        df = pd.read_csv(file_path)
    except FileNotFoundError:
        logger.error(f"CSV file not found: {file_path}")
        sys.exit(1)
    except pd.errors.EmptyDataError:
        logger.error("CSV file is empty.")
        sys.exit(1)

    required_columns = ['email', 'first_name', 'last_name', 'username']
    missing_columns = [col for col in required_columns if col not in df.columns]
    
    if missing_columns:
        logger.error(f"Missing required columns in CSV: {missing_columns}")
        sys.exit(1)

    df['email'] = df['email'].str.strip().str.lower()
    df['username'] = df['username'].str.strip().str.lower()
    df['first_name'] = df['first_name'].str.strip()
    df['last_name'] = df['last_name'].str.strip()
    df.dropna(subset=['email', 'first_name', 'last_name', 'username'], inplace=True)
    
    logger.info(f"Loaded {len(df)} valid user records from {file_path}")
    return df

def create_user(client: PlatformClient, user_data: dict) -> str:
    """Creates a single user in Genesys Cloud."""
    user_api = client.user_api
    routing_api = client.routing_api

    email = user_data['email']
    first_name = user_data['first_name']
    last_name = user_data['last_name']
    username = user_data['username']
    division_id = user_data.get('division_id')
    queue_id = user_data.get('queue_id')

    create_request = CreateUserRequest(
        email=email,
        name=f"{first_name} {last_name}",
        username=username,
        division_id=division_id,
        presence_configuration=UserPresenceConfiguration(
            presence_state_id="available"
        )
    )

    max_retries = 3
    retry_count = 0
    
    while retry_count <= max_retries:
        try:
            response = user_api.post_users(body=create_request)
            user_id = response.id
            logger.info(f"User created successfully: {email} (ID: {user_id})")
            
            if queue_id:
                try:
                    queue_member = QueueMember(
                        user_id=user_id,
                        wrap_up_code="default"
                    )
                    routing_api.post_routing_queues_members(queue_id=queue_id, body=queue_member)
                    logger.info(f"User {email} assigned to queue {queue_id}")
                except ApiException as routing_e:
                    if routing_e.status == 409:
                        logger.warning(f"User {email} already in queue {queue_id}.")
                    else:
                        logger.error(f"Failed to assign user {email} to queue: {routing_e}")
            
            return user_id

        except ApiException as e:
            if e.status == 429:
                retry_count += 1
                wait_time = 2 ** retry_count
                logger.warning(f"Rate limit hit for {email}. Retrying in {wait_time}s...")
                time.sleep(wait_time)
            elif e.status == 409:
                logger.warning(f"User {email} already exists (Conflict 409). Skipping.")
                return None
            else:
                logger.error(f"Failed to create user {email}: {e.status} - {e.reason}")
                return None
    
    logger.error(f"Failed to create user {email} after {max_retries} retries.")
    return None

def process_users(client: PlatformClient, df: pd.DataFrame) -> pd.DataFrame:
    """Iterates through the DataFrame and creates users."""
    results = []
    
    for index, row in df.iterrows():
        logger.info(f"Processing user {index + 1}/{len(df)}: {row['email']}")
        
        user_id = create_user(client, row.to_dict())
        
        results.append({
            'email': row['email'],
            'username': row['username'],
            'user_id': user_id,
            'status': 'Created' if user_id else 'Skipped/Error',
            'error': None if user_id else 'Exists or Failed'
        })
        
        time.sleep(0.2)

    return pd.DataFrame(results)

def main():
    """Main execution function."""
    try:
        client = init_genesys_client()
        logger.info("Authentication successful.")
        
        df_users = load_csv_data(CSV_FILE_PATH)
        
        if df_users.empty:
            logger.warning("No users to process.")
            return

        results_df = process_users(client, df_users)
        
        # Save results to a new CSV file
        output_file = 'user_creation_results.csv'
        results_df.to_csv(output_file, index=False)
        logger.info(f"Results saved to {output_file}")
        
        # Summary
        created_count = len(results_df[results_df['status'] == 'Created'])
        failed_count = len(results_df[results_df['status'] == 'Skipped/Error'])
        logger.info(f"Summary: {created_count} Created, {failed_count} Skipped/Error.")

    except Exception as e:
        logger.error(f"Fatal error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token is invalid, expired, or the client ID/secret is incorrect.
• Fix: Verify your environment variables. Ensure the OAuth client is active in the Genesys Cloud Admin Portal. Check that the client has the user:write scope.

Error: 403 Forbidden

• Cause: The OAuth client lacks the required permissions.
• Fix: In the Genesys Cloud Admin Portal, navigate to Integrations > OAuth Clients. Edit your client and ensure the User permission set includes Write.

Error: 409 Conflict

• Cause: A user with the same username or email already exists in the organization.
• Fix: The script handles this by logging a warning and skipping the user. If you want to update existing users instead, you must first query the user by email (GET /api/v2/users?email={email}) and then use PUT /api/v2/users/{userId}.

Error: 429 Too Many Requests

• Cause: You exceeded the API rate limit.
• Fix: The script implements exponential backoff. If you still see failures, increase the time.sleep(0.2) delay between requests in the process_users function. Genesys Cloud typically allows 10-20 requests per second for most endpoints, but this varies by tenant load.

Error: 400 Bad Request

• Cause: Invalid data in the CSV (e.g., invalid email format, missing required fields).
• Fix: Check the CSV for malformed emails or empty required columns. Ensure division_id and queue_id are valid UUIDs if provided.

Official References

• Genesys Cloud Python SDK Documentation
• User API Reference
• Routing API Reference
• OAuth 2.0 Flow Documentation