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

What You Will Build

• A Python script that reads user data from a CSV file and creates corresponding agents in Genesys Cloud CX.
• This solution uses the genesyscloud Python SDK to handle authentication, validation, and API calls.
• The tutorial covers Python 3.8+ with standard library modules and the official SDK.

Prerequisites

• OAuth Client: A Genesys Cloud OAuth client with resource:users:write and resource:users:read scopes.
• SDK Version: genesyscloud>=2.100.0 (verify via pip show genesyscloud).
• Runtime: Python 3.8 or higher.
• Dependencies:

  pip install genesyscloud csv pathlib
  

• CSV Structure: A file named users.csv with columns: external_id, name, email, phone_number, division_id.

Authentication Setup

The Genesys Cloud Python SDK handles OAuth token management automatically when initialized with a client ID, client secret, and region. You must configure these values in your environment or directly in the script.

import os
from genesyscloud.platform_client_v2 import PlatformClientBuilder

def get_platform_client() -> PlatformClientBuilder:
    """
    Initializes and returns the configured PlatformClient.
    """
    # Load credentials from environment variables
    client_id = os.environ.get("GENESYS_CLIENT_ID")
    client_secret = os.environ.get("GENESYS_CLIENT_SECRET")
    region = os.environ.get("GENESYS_REGION", "us-east-1")

    if not client_id or not client_secret:
        raise EnvironmentError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")

    # Configure the platform client
    builder = PlatformClientBuilder(
        client_id=client_id,
        client_secret=client_secret,
        region=region
    )
    
    # Build the client instance
    client = builder.build()
    
    # Verify connectivity by fetching the current user info (optional but recommended for debugging)
    try:
        client.users.get_me()
        print("Authentication successful.")
    except Exception as e:
        print(f"Authentication failed: {e}")
        raise

    return client

Required Scopes:

• resource:users:write: Required to create new users.
• resource:users:read: Required to validate existing users or fetch division details if needed.

Implementation

Step 1: Parse and Validate CSV Data

Before making any API calls, you must parse the CSV file and structure the data into objects compatible with the SDK. The Genesys Cloud API requires a unique externalId for each user to prevent duplicates and allow for idempotent updates.

import csv
from pathlib import Path
from typing import List, Dict, Any

def parse_users_csv(filepath: str) -> List[Dict[str, Any]]:
    """
    Reads a CSV file and returns a list of dictionaries representing user data.
    
    Args:
        filepath: Path to the CSV file.
        
    Returns:
        List of dictionaries with keys: external_id, name, email, phone_number, division_id
    """
    users = []
    path = Path(filepath)
    
    if not path.exists():
        raise FileNotFoundError(f"CSV file not found: {filepath}")

    with open(path, mode='r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        
        # Validate required columns
        required_columns = {'external_id', 'name', 'email', 'phone_number', 'division_id'}
        if not required_columns.issubset(set(reader.fieldnames or [])):
            missing = required_columns - set(reader.fieldnames or [])
            raise ValueError(f"CSV missing required columns: {missing}")

        for row in reader:
            # Basic validation
            if not row.get('external_id') or not row.get('email'):
                print(f"Skipping invalid row: {row}")
                continue
            
            users.append({
                'external_id': row['external_id'].strip(),
                'name': row['name'].strip(),
                'email': row['email'].strip(),
                'phone_number': row.get('phone_number', '').strip(),
                'division_id': row.get('division_id', '').strip()
            })
            
    return users

Step 2: Construct User Creation Payloads

The Genesys Cloud API expects a specific JSON structure for creating users. You must map the CSV data to the UserCreateRequest model provided by the SDK. Key fields include name, email, externalId, and phoneNumbers.

from genesyscloud.models import UserCreateRequest, PhoneNumber

def create_user_payload(user_data: Dict[str, Any]) -> UserCreateRequest:
    """
    Converts a dictionary of user data into a UserCreateRequest object.
    
    Args:
        user_data: Dictionary containing user details.
        
    Returns:
        UserCreateRequest object ready for the API.
    """
    # Initialize phone numbers list
    phone_numbers = []
    if user_data.get('phone_number'):
        phone_numbers.append(PhoneNumber(
            type="work",
            value=user_data['phone_number'],
            primary=True
        ))

    # Construct the request object
    request = UserCreateRequest(
        name=user_data['name'],
        email=user_data['email'],
        external_id=user_data['external_id'],
        phone_numbers=phone_numbers if phone_numbers else None,
        division_id=user_data.get('division_id') # Optional: assigns to specific division
    )
    
    return request

Critical Parameter Notes:

• external_id: This is the most important field for bulk operations. If you attempt to create a user with an external_id that already exists, the API will return a 409 Conflict. You must handle this gracefully.
• division_id: If omitted, the user is created in the default division. Ensure the division_id provided is valid and accessible by the OAuth client.

Step 3: Execute Bulk Creation with Error Handling

The Genesys Cloud API does not have a single endpoint for bulk user creation. You must iterate through the list and call POST /api/v2/users for each user. To avoid rate limiting (429 Too Many Requests), you should implement a delay between requests or use the SDK’s built-in retry logic.

import time
from genesyscloud.api import UsersApi
from genesyscloud.rest import ApiException

def bulk_create_users(client: PlatformClientBuilder, users_data: List[Dict[str, Any]], delay: float = 0.5) -> List[Dict[str, Any]]:
    """
    Creates users in Genesys Cloud based on the provided data.
    
    Args:
        client: The configured PlatformClient.
        users_data: List of user dictionaries from CSV.
        delay: Seconds to wait between API calls to avoid rate limiting.
        
    Returns:
        List of results containing success/failure status for each user.
    """
    users_api = UsersApi(client)
    results = []
    
    print(f"Starting bulk creation for {len(users_data)} users...")
    
    for i, user_data in enumerate(users_data):
        try:
            # Construct the payload
            payload = create_user_payload(user_data)
            
            # Make the API call
            response = users_api.post_users(body=payload)
            
            results.append({
                'status': 'success',
                'external_id': user_data['external_id'],
                'genesys_id': response.id,
                'name': user_data['name']
            })
            print(f"[{i+1}/{len(users_data)}] Created: {user_data['name']} (ID: {response.id})")
            
        except ApiException as e:
            if e.status == 409:
                # Conflict: User with this external_id already exists
                results.append({
                    'status': 'exists',
                    'external_id': user_data['external_id'],
                    'name': user_data['name'],
                    'error': 'User already exists'
                })
                print(f"[{i+1}/{len(users_data)}] Exists: {user_data['name']}")
            elif e.status == 429:
                # Rate limit exceeded: Wait longer
                wait_time = 5
                print(f"[{i+1}/{len(users_data)}] Rate limited. Waiting {wait_time}s...")
                time.sleep(wait_time)
                # Retry this user by re-adding to the front of the queue (simplified for this example)
                users_data.insert(0, user_data)
                continue
            else:
                # Other errors (400, 401, 500)
                results.append({
                    'status': 'error',
                    'external_id': user_data['external_id'],
                    'name': user_data['name'],
                    'error': str(e.body)
                })
                print(f"[{i+1}/{len(users_data)}] Error: {user_data['name']} - {e.status}")
                
        except Exception as e:
            results.append({
                'status': 'error',
                'external_id': user_data['external_id'],
                'name': user_data['name'],
                'error': str(e)
            })
            print(f"[{i+1}/{len(users_data)}] Unexpected Error: {user_data['name']} - {e}")

        # Respect rate limits
        if i < len(users_data) - 1:
            time.sleep(delay)
            
    return results

Rate Limiting Strategy:

• The Genesys Cloud API enforces rate limits per OAuth client.
• The default delay of 0.5 seconds between requests is a safe starting point for moderate volumes (up to 100 users).
• For larger volumes, monitor the Retry-After header in 429 responses and adjust the delay dynamically.

Step 4: Process and Report Results

After the bulk creation process completes, you should summarize the results to identify which users were created, which already existed, and which failed.

def print_results(results: List[Dict[str, Any]]):
    """
    Prints a summary of the bulk creation results.
    """
    success_count = sum(1 for r in results if r['status'] == 'success')
    exists_count = sum(1 for r in results if r['status'] == 'exists')
    error_count = sum(1 for r in results if r['status'] == 'error')
    
    print("\n--- Bulk Creation Summary ---")
    print(f"Total Processed: {len(results)}")
    print(f"Successfully Created: {success_count}")
    print(f"Already Existed: {exists_count}")
    print(f"Errors: {error_count}")
    
    if error_count > 0:
        print("\n--- Failed Users ---")
        for r in results:
            if r['status'] == 'error':
                print(f"External ID: {r['external_id']} | Name: {r['name']} | Error: {r['error']}")

Complete Working Example

import os
import csv
import time
from pathlib import Path
from typing import List, Dict, Any

# Genesys Cloud SDK Imports
from genesyscloud.platform_client_v2 import PlatformClientBuilder
from genesyscloud.api import UsersApi
from genesyscloud.rest import ApiException
from genesyscloud.models import UserCreateRequest, PhoneNumber

def get_platform_client() -> PlatformClientBuilder:
    """Initializes and returns the configured PlatformClient."""
    client_id = os.environ.get("GENESYS_CLIENT_ID")
    client_secret = os.environ.get("GENESYS_CLIENT_SECRET")
    region = os.environ.get("GENESYS_REGION", "us-east-1")

    if not client_id or not client_secret:
        raise EnvironmentError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")

    builder = PlatformClientBuilder(
        client_id=client_id,
        client_secret=client_secret,
        region=region
    )
    
    client = builder.build()
    
    try:
        client.users.get_me()
        print("Authentication successful.")
    except Exception as e:
        print(f"Authentication failed: {e}")
        raise

    return client

def parse_users_csv(filepath: str) -> List[Dict[str, Any]]:
    """Reads a CSV file and returns a list of dictionaries representing user data."""
    users = []
    path = Path(filepath)
    
    if not path.exists():
        raise FileNotFoundError(f"CSV file not found: {filepath}")

    with open(path, mode='r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        
        required_columns = {'external_id', 'name', 'email', 'phone_number', 'division_id'}
        if not required_columns.issubset(set(reader.fieldnames or [])):
            missing = required_columns - set(reader.fieldnames or [])
            raise ValueError(f"CSV missing required columns: {missing}")

        for row in reader:
            if not row.get('external_id') or not row.get('email'):
                print(f"Skipping invalid row: {row}")
                continue
            
            users.append({
                'external_id': row['external_id'].strip(),
                'name': row['name'].strip(),
                'email': row['email'].strip(),
                'phone_number': row.get('phone_number', '').strip(),
                'division_id': row.get('division_id', '').strip()
            })
            
    return users

def create_user_payload(user_data: Dict[str, Any]) -> UserCreateRequest:
    """Converts a dictionary of user data into a UserCreateRequest object."""
    phone_numbers = []
    if user_data.get('phone_number'):
        phone_numbers.append(PhoneNumber(
            type="work",
            value=user_data['phone_number'],
            primary=True
        ))

    request = UserCreateRequest(
        name=user_data['name'],
        email=user_data['email'],
        external_id=user_data['external_id'],
        phone_numbers=phone_numbers if phone_numbers else None,
        division_id=user_data.get('division_id')
    )
    
    return request

def bulk_create_users(client: PlatformClientBuilder, users_data: List[Dict[str, Any]], delay: float = 0.5) -> List[Dict[str, Any]]:
    """Creates users in Genesys Cloud based on the provided data."""
    users_api = UsersApi(client)
    results = []
    
    print(f"Starting bulk creation for {len(users_data)} users...")
    
    for i, user_data in enumerate(users_data):
        try:
            payload = create_user_payload(user_data)
            response = users_api.post_users(body=payload)
            
            results.append({
                'status': 'success',
                'external_id': user_data['external_id'],
                'genesys_id': response.id,
                'name': user_data['name']
            })
            print(f"[{i+1}/{len(users_data)}] Created: {user_data['name']} (ID: {response.id})")
            
        except ApiException as e:
            if e.status == 409:
                results.append({
                    'status': 'exists',
                    'external_id': user_data['external_id'],
                    'name': user_data['name'],
                    'error': 'User already exists'
                })
                print(f"[{i+1}/{len(users_data)}] Exists: {user_data['name']}")
            elif e.status == 429:
                wait_time = 5
                print(f"[{i+1}/{len(users_data)}] Rate limited. Waiting {wait_time}s...")
                time.sleep(wait_time)
                users_data.insert(0, user_data)
                continue
            else:
                results.append({
                    'status': 'error',
                    'external_id': user_data['external_id'],
                    'name': user_data['name'],
                    'error': str(e.body)
                })
                print(f"[{i+1}/{len(users_data)}] Error: {user_data['name']} - {e.status}")
                
        except Exception as e:
            results.append({
                'status': 'error',
                'external_id': user_data['external_id'],
                'name': user_data['name'],
                'error': str(e)
            })
            print(f"[{i+1}/{len(users_data)}] Unexpected Error: {user_data['name']} - {e}")

        if i < len(users_data) - 1:
            time.sleep(delay)
            
    return results

def print_results(results: List[Dict[str, Any]]):
    """Prints a summary of the bulk creation results."""
    success_count = sum(1 for r in results if r['status'] == 'success')
    exists_count = sum(1 for r in results if r['status'] == 'exists')
    error_count = sum(1 for r in results if r['status'] == 'error')
    
    print("\n--- Bulk Creation Summary ---")
    print(f"Total Processed: {len(results)}")
    print(f"Successfully Created: {success_count}")
    print(f"Already Existed: {exists_count}")
    print(f"Errors: {error_count}")
    
    if error_count > 0:
        print("\n--- Failed Users ---")
        for r in results:
            if r['status'] == 'error':
                print(f"External ID: {r['external_id']} | Name: {r['name']} | Error: {r['error']}")

if __name__ == "__main__":
    try:
        # 1. Authenticate
        client = get_platform_client()
        
        # 2. Parse CSV
        csv_path = "users.csv"
        users_data = parse_users_csv(csv_path)
        print(f"Parsed {len(users_data)} users from CSV.")
        
        # 3. Bulk Create
        results = bulk_create_users(client, users_data, delay=0.5)
        
        # 4. Report Results
        print_results(results)
        
    except Exception as e:
        print(f"Fatal error: {e}")

Common Errors & Debugging

Error: 409 Conflict

• Cause: The external_id provided in the CSV already exists in Genesys Cloud.
• Fix: The script above handles this by logging the user as “exists”. If you want to update existing users instead, you must first retrieve the user by external_id using GET /api/v2/users?externalId={id} and then call PUT /api/v2/users/{id}.

Error: 400 Bad Request

• Cause: Invalid data format. Common issues include:
  • Email address is not valid.
  • Phone number format is incorrect (E.164 format is recommended).
  • division_id is invalid or the OAuth client lacks access to that division.
• Fix: Validate email and phone formats in the parse_users_csv function before sending to the API. Ensure the division_id is correct.

Error: 401 Unauthorized

• Cause: OAuth token is expired or invalid.
• Fix: The SDK handles token refresh automatically. If this error persists, verify that the GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET are correct and that the OAuth client is active in Genesys Cloud.

Error: 429 Too Many Requests

• Cause: The API rate limit for the OAuth client has been exceeded.
• Fix: The script above implements a simple backoff strategy. For higher throughput, consider implementing exponential backoff or reducing the number of concurrent requests if using threading.

Official References

• Genesys Cloud Users API Documentation
• Genesys Cloud Python SDK Reference
• OAuth 2.0 Authorization Guide