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

What You Will Build

• A Python script that reads user data from a local CSV file and creates corresponding user accounts in Genesys Cloud.
• This implementation uses the official genesyscloud Python SDK (version 12+).
• The tutorial covers Python 3.8+ with standard library modules and the official SDK.

Prerequisites

• OAuth Client Type: Confidential Client (Client Credentials Flow) is recommended for server-side bulk operations.
• Required Scopes:
  • user:write (Required to create users)
  • user:read (Required to verify user existence or check status if needed)
  • routing:skillgroup:read (Optional, if assigning skill groups)
  • routing:team:read (Optional, if assigning teams)
• SDK Version: genesyscloud >= 12.0.0 (Latest stable release as of 2024).
• Language/Runtime: Python 3.8 or higher.
• External Dependencies:
  • genesyscloud: The official Genesys Cloud SDK.
  • python-dotenv: For secure environment variable management (optional but recommended).

Install the required packages:

pip install genesyscloud python-dotenv

Authentication Setup

The Genesys Cloud SDK handles OAuth token acquisition and refresh automatically once configured. You must provide your client ID, client secret, and environment (e.g., mypurecloud.com or usw2.pure.cloud).

Create a .env file in your project root to store credentials securely:

GENESYS_CLOUD_CLIENT_ID=your_client_id_here
GENESYS_CLOUD_CLIENT_SECRET=your_client_secret_here
GENESYS_CLOUD_ENVIRONMENT=mypurecloud.com

Initialize the SDK in your script. This configuration object is reused across all API calls.

import os
from dotenv import load_dotenv
from genesyscloud.platform.client import PlatformClient

# Load environment variables
load_dotenv()

def get_platform_client() -> PlatformClient:
    """
    Initializes and returns a configured Genesys Cloud PlatformClient.
    """
    # Retrieve credentials from environment variables
    client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
    environment = os.getenv("GENESYS_CLOUD_ENVIRONMENT", "mypurecloud.com")

    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLOUD_CLIENT_ID and GENESYS_CLOUD_CLIENT_SECRET must be set in environment.")

    # Configure the OAuth credentials
    # The SDK will automatically request a token and handle refreshes
    platform_client = PlatformClient(
        credentials={
            'client_id': client_id,
            'client_secret': client_secret,
            'environment': environment
        }
    )

    return platform_client

Implementation

Step 1: Define the CSV Structure and Parse Data

Before calling the API, you must structure your data. The Genesys Cloud User creation endpoint (POST /api/v2/users) expects a JSON payload with specific fields. Not all fields are mandatory, but a user generally requires at least a name and an email address.

Create a sample CSV file named users_to_create.csv:

first_name,last_name,email,phone_number,division_id
John,Doe,john.doe@example.com,+12025550101,
Jane,Smith,jane.smith@example.com,+12025550102,
Bob,Jones,bob.jones@example.com,+12025550103,

Note: division_id is optional. If omitted, the user is created in the default division. If you assign users to specific divisions, you must provide the valid UUID of that division.

Now, write the Python logic to parse this CSV. We will use the built-in csv module.

import csv
from typing import List, Dict, Any

def parse_csv_users(csv_filepath: str) -> List[Dict[str, Any]]:
    """
    Reads a CSV file and returns a list of dictionaries representing user data.
    
    Args:
        csv_filepath: Path to the CSV file.
        
    Returns:
        List of dictionaries with keys: first_name, last_name, email, phone_number, division_id
    """
    users_data = []
    
    with open(csv_filepath, mode='r', encoding='utf-8') as file:
        reader = csv.DictReader(file)
        for row in reader:
            # Basic validation: Email is required for user creation
            if not row.get('email'):
                print(f"Skipping row due to missing email: {row}")
                continue
            
            # Clean up whitespace
            cleaned_row = {k: v.strip() if v else v for k, v in row.items()}
            users_data.append(cleaned_row)
            
    return users_data

Step 2: Construct the User Creation Payload

The POST /api/v2/users endpoint requires a User object. In the Python SDK, this is represented by genesyscloud.models.user.User. You must map the CSV data to this model.

Critical fields for a basic user:

• email: Must be unique.
• first_name: Required.
• last_name: Required.
• phone_numbers: A list of phone number objects. Each phone number needs a type (e.g., “work”, “mobile”) and a display_number.

Here is the function to convert a dictionary from Step 1 into an SDK User object:

from genesyscloud.models.user import User
from genesyscloud.models.phone_number import PhoneNumber

def create_user_object(data: Dict[str, Any]) -> User:
    """
    Converts a dictionary of user data into a Genesys Cloud User object.
    
    Args:
        data: Dictionary containing first_name, last_name, email, etc.
        
    Returns:
        A configured User object ready for API submission.
    """
    # Initialize the User object
    user = User(
        first_name=data['first_name'],
        last_name=data['last_name'],
        email=data['email']
    )

    # Optional: Add phone number if provided
    if data.get('phone_number'):
        phone_number = PhoneNumber(
            type='work',  # Can be 'work', 'home', 'mobile', etc.
            display_number=data['phone_number']
        )
        user.phone_numbers = [phone_number]

    # Optional: Assign to a specific division
    # Note: If division_id is empty string, we leave it None so it uses default
    if data.get('division_id') and data['division_id'] != '':
        user.division_id = data['division_id']

    return user

Step 3: Execute Bulk Creation with Error Handling

Creating users one by one via API calls is inefficient and prone to rate limiting. However, the Genesys Cloud SDK does not have a native “bulk create users” endpoint in the same way it does for analytics queries. You must loop through the list and call post_users for each user.

To handle this robustly, you must implement:

1. Retry Logic: For transient network errors (5xx) or rate limits (429).
2. Error Tracking: Record which users succeeded and which failed, along with the error reason.
3. Duplicate Handling: If a user with the same email already exists, the API returns a 409 Conflict. We should catch this and log it as a skip rather than a critical failure.

import time
import logging
from genesyscloud.rest import ApiException
from typing import Tuple, List

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def create_single_user(platform_client: PlatformClient, user_obj: User) -> Tuple[bool, str]:
    """
    Attempts to create a single user.
    
    Returns:
        Tuple: (success: bool, message: str)
    """
    try:
        # Call the API
        # Scope required: user:write
        response = platform_client.users.post_users(body=user_obj)
        
        # If successful, response is a User object
        logger.info(f"Successfully created user: {response.email} (ID: {response.id})")
        return True, f"Created (ID: {response.id})"
        
    except ApiException as e:
        # Handle specific HTTP status codes
        if e.status == 409:
            # Conflict: User with this email likely already exists
            logger.warning(f"User already exists or conflict: {user_obj.email}. Skipping.")
            return False, "Duplicate/Conflict"
        elif e.status == 429:
            # Rate Limit: Should ideally be handled by a retry wrapper, but logged here for visibility
            logger.error(f"Rate limited while creating user: {user_obj.email}")
            return False, "Rate Limited"
        elif e.status == 400:
            # Bad Request: Invalid data format
            logger.error(f"Bad Request for user {user_obj.email}: {e.body}")
            return False, f"Bad Request: {e.body}"
        else:
            logger.error(f"Unexpected error creating user {user_obj.email}: {e.status} - {e.reason}")
            return False, f"API Error: {e.status}"
    except Exception as e:
        logger.error(f"Unexpected error: {str(e)}")
        return False, f"Unexpected Error: {str(e)}"

def bulk_create_users(platform_client: PlatformClient, users_data: List[Dict[str, Any]], max_retries: int = 3) -> Dict[str, List[str]]:
    """
    Iterates through a list of user data dictionaries and creates them in Genesys Cloud.
    
    Args:
        platform_client: The authenticated SDK client.
        users_data: List of user data dictionaries from CSV.
        max_retries: Number of times to retry on 429/5xx errors.
        
    Returns:
        Dictionary with keys 'created', 'failed', 'skipped' containing lists of emails.
    """
    results = {
        'created': [],
        'failed': [],
        'skipped': []
    }

    for i, data in enumerate(users_data):
        email = data.get('email', 'Unknown')
        logger.info(f"Processing user {i+1}/{len(users_data)}: {email}")
        
        success = False
        last_error = ""
        
        for attempt in range(max_retries):
            try:
                user_obj = create_user_object(data)
                success, message = create_single_user(platform_client, user_obj)
                
                if success:
                    results['created'].append(email)
                    break
                elif "Duplicate" in message or "Conflict" in message:
                    results['skipped'].append(email)
                    break
                elif "Rate Limited" in message or "500" in message:
                    # Wait before retrying (Exponential backoff)
                    wait_time = 2 ** attempt
                    logger.warning(f"Retrying in {wait_time} seconds... (Attempt {attempt+1}/{max_retries})")
                    time.sleep(wait_time)
                    continue
                else:
                    # Non-retryable error (e.g., 400 Bad Request)
                    results['failed'].append(f"{email}: {message}")
                    break
                    
            except Exception as e:
                logger.error(f"Unexpected error during processing of {email}: {e}")
                results['failed'].append(f"{email}: Unexpected Error")
                break
        
        if not success and email not in results['skipped']:
            # If loop finished without success/skip, and not already added to failed
            if email not in [f.split(':')[0] for f in results['failed']]:
                results['failed'].append(f"{email}: Max retries exceeded")

    return results

Complete Working Example

This is the full, copy-pasteable script. Save this as bulk_create_users.py.

import os
import csv
import time
import logging
from typing import List, Dict, Any, Tuple

from dotenv import load_dotenv
from genesyscloud.platform.client import PlatformClient
from genesyscloud.models.user import User
from genesyscloud.models.phone_number import PhoneNumber
from genesyscloud.rest import ApiException

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

def get_platform_client() -> PlatformClient:
    """Initializes the Genesys Cloud SDK client."""
    load_dotenv()
    
    client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
    environment = os.getenv("GENESYS_CLOUD_ENVIRONMENT", "mypurecloud.com")

    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLOUD_CLIENT_ID and GENESYS_CLOUD_CLIENT_SECRET must be set.")

    return PlatformClient(
        credentials={
            'client_id': client_id,
            'client_secret': client_secret,
            'environment': environment
        }
    )

def parse_csv_users(csv_filepath: str) -> List[Dict[str, Any]]:
    """Parses CSV file into list of user data dictionaries."""
    users_data = []
    with open(csv_filepath, mode='r', encoding='utf-8') as file:
        reader = csv.DictReader(file)
        for row in reader:
            if not row.get('email'):
                logger.warning(f"Skipping row due to missing email: {row}")
                continue
            # Clean whitespace
            cleaned_row = {k: v.strip() if v else v for k, v in row.items()}
            users_data.append(cleaned_row)
    return users_data

def create_user_object(data: Dict[str, Any]) -> User:
    """Maps CSV data to Genesys Cloud User model."""
    user = User(
        first_name=data['first_name'],
        last_name=data['last_name'],
        email=data['email']
    )

    if data.get('phone_number'):
        phone_number = PhoneNumber(
            type='work',
            display_number=data['phone_number']
        )
        user.phone_numbers = [phone_number]

    if data.get('division_id') and data['division_id'] != '':
        user.division_id = data['division_id']

    return user

def create_single_user(platform_client: PlatformClient, user_obj: User) -> Tuple[bool, str]:
    """Attempts to create a single user via API."""
    try:
        response = platform_client.users.post_users(body=user_obj)
        logger.info(f"Created user: {response.email} (ID: {response.id})")
        return True, f"Created (ID: {response.id})"
    except ApiException as e:
        if e.status == 409:
            logger.warning(f"Conflict (User exists): {user_obj.email}")
            return False, "Duplicate"
        elif e.status == 429:
            logger.warning(f"Rate Limited: {user_obj.email}")
            return False, "Rate Limited"
        elif e.status == 400:
            logger.error(f"Bad Request: {user_obj.email} - {e.body}")
            return False, f"Bad Request: {e.body}"
        else:
            logger.error(f"API Error {e.status}: {user_obj.email}")
            return False, f"API Error: {e.status}"
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        return False, f"Unexpected Error: {str(e)}"

def bulk_create_users(platform_client: PlatformClient, users_data: List[Dict[str, Any]], max_retries: int = 3) -> Dict[str, List[str]]:
    """Executes the bulk creation loop with retry logic."""
    results = {'created': [], 'failed': [], 'skipped': []}

    for i, data in enumerate(users_data):
        email = data.get('email', 'Unknown')
        logger.info(f"[{i+1}/{len(users_data)}] Processing: {email}")
        
        success = False
        for attempt in range(max_retries):
            try:
                user_obj = create_user_object(data)
                success, message = create_single_user(platform_client, user_obj)
                
                if success:
                    results['created'].append(email)
                    break
                elif "Duplicate" in message:
                    results['skipped'].append(email)
                    break
                elif "Rate Limited" in message:
                    wait_time = 2 ** attempt
                    logger.info(f"Rate limited. Waiting {wait_time}s before retry...")
                    time.sleep(wait_time)
                    continue
                else:
                    results['failed'].append(f"{email}: {message}")
                    break
            except Exception as e:
                logger.error(f"Error processing {email}: {e}")
                results['failed'].append(f"{email}: System Error")
                break
        
        if not success and email not in results['skipped']:
            if not any(email in f for f in results['failed']):
                results['failed'].append(f"{email}: Max retries exceeded")

    return results

def main():
    # Configuration
    CSV_FILE = 'users_to_create.csv'
    
    # 1. Initialize Client
    try:
        client = get_platform_client()
    except Exception as e:
        logger.error(f"Failed to initialize client: {e}")
        return

    # 2. Parse CSV
    try:
        users_data = parse_csv_users(CSV_FILE)
        logger.info(f"Parsed {len(users_data)} users from CSV.")
    except FileNotFoundError:
        logger.error(f"CSV file '{CSV_FILE}' not found.")
        return
    except Exception as e:
        logger.error(f"Error parsing CSV: {e}")
        return

    if not users_data:
        logger.warning("No users to process.")
        return

    # 3. Bulk Create
    logger.info("Starting bulk user creation...")
    results = bulk_create_users(client, users_data)

    # 4. Report Results
    logger.info("="*30)
    logger.info("SUMMARY")
    logger.info("="*30)
    logger.info(f"Created: {len(results['created'])}")
    logger.info(f"Skipped (Duplicates): {len(results['skipped'])}")
    logger.info(f"Failed: {len(results['failed'])}")
    
    if results['failed']:
        logger.warning("Failed Users:")
        for fail in results['failed']:
            logger.warning(f"  - {fail}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 409 Conflict

• Cause: A user with the specified email address already exists in the Genesys Cloud organization. Email addresses must be unique per organization.
• Fix: The script above catches this and adds the user to the skipped list. If you want to update existing users, you must first search for the user by email using GET /api/v2/users?email={email} and then use PUT /api/v2/users/{id} instead of POST.

Error: 400 Bad Request

• Cause: The payload does not meet validation rules. Common issues include:
  • Missing first_name or last_name.
  • Invalid email format.
  • Invalid phone number format (must include country code, e.g., +1...).
  • Invalid division_id (UUID format incorrect or division does not exist).
• Fix: Check the e.body in the ApiException catch block. It usually contains a detailed JSON error message from Genesys Cloud indicating exactly which field failed validation.

Error: 429 Too Many Requests

• Cause: You are hitting the API rate limit. Genesys Cloud enforces limits on the number of requests per minute per client ID.
• Fix: The script implements exponential backoff. If you are creating thousands of users, consider increasing the max_retries or adding a small fixed delay (e.g., time.sleep(1)) between each user creation to stay well under the limit.

Error: 401 Unauthorized

• Cause: The OAuth token is invalid or expired.
• Fix: Ensure your CLIENT_ID and CLIENT_SECRET are correct. The SDK handles refresh, but if the client credentials are revoked or incorrect, initialization will fail. Check your .env file.

Official References

• Genesys Cloud Python SDK Documentation
• Users API Reference: POST /api/v2/users
• Genesys Cloud OAuth Documentation