Simulating Genesys Cloud Outbound Campaign Dial Plans with Python

What You Will Build

• This script clones an existing Genesys Cloud outbound campaign, applies configurable rate constraints to a test dataset, and mathematically projects contact dial volumes over a specified time window without placing a single call.
• The implementation uses the Genesys Cloud REST API for campaign management, list retrieval, and authentication.
• The tutorial covers Python with the requests library, type hints, and production-grade error handling.

Prerequisites

• OAuth Client Type: Service Account (Client Credentials Grant)
• Required Scopes: outbound:campaign:read, outbound:campaign:write, outbound:list:read
• API Version: Genesys Cloud v2 (/api/v2/)
• Runtime: Python 3.9+
• Dependencies: requests>=2.31.0, python-dateutil>=2.8.0 (install via pip install requests python-dateutil)

Authentication Setup

Genesys Cloud uses OAuth 2.0 for all API access. You must exchange your service account credentials for an access token before making outbound API calls. The following function handles token acquisition, caches it in memory, and implements automatic refresh logic when the token expires.

import requests
import time
import logging
from typing import Optional, Dict, Any

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

class GenesysAuth:
    def __init__(self, org_id: str, client_id: str, client_secret: str):
        self.org_id = org_id
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = f"https://{org_id}.mygenesys.com"
        self.access_token: Optional[str] = None
        self.token_expiry: Optional[float] = None

    def get_token(self) -> str:
        if self.access_token and self.token_expiry and time.time() < self.token_expiry:
            return self.access_token

        url = f"{self.base_url}/oauth/token"
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        response = requests.post(url, data=payload)
        response.raise_for_status()
        data = response.json()
        
        self.access_token = data["access_token"]
        self.token_expiry = time.time() + data["expires_in"] - 60  # Refresh 60 seconds early
        return self.access_token

    def get_headers(self) -> Dict[str, str]:
        return {
            "Authorization": f"Bearer {self.get_token()}",
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

The get_token method checks expiration before requesting a new token. The sixty-second buffer prevents race conditions where requests fail during token rollover. The get_headers method returns the standard authorization and content headers required by Genesys endpoints.

Implementation

Step 1: Fetch and Clone Campaign Configuration

You must retrieve the target campaign before cloning it. The /api/v2/outbound/campaigns endpoint supports pagination, which you must handle to locate the correct campaign by name or ID. After retrieval, you POST the configuration to the same endpoint with a modified name to create a simulation clone.

import requests
from typing import Dict, Any, List

def fetch_campaigns(auth: GenesysAuth, campaign_name: str) -> Dict[str, Any]:
    url = f"{auth.base_url}/api/v2/outbound/campaigns"
    params = {"pageSize": 25, "pageNumber": 1}
    
    while True:
        response = requests.get(url, headers=auth.get_headers(), params=params)
        response.raise_for_status()
        data = response.json()
        
        for campaign in data["entities"]:
            if campaign["name"] == campaign_name:
                return campaign
        
        if not data["nextPage"]:
            raise ValueError(f"Campaign '{campaign_name}' not found.")
        
        params["pageNumber"] += 1

def clone_campaign(auth: GenesysAuth, original: Dict[str, Any], suffix: str = "_SIMULATION") -> Dict[str, Any]:
    url = f"{auth.base_url}/api/v2/outbound/campaigns"
    
    # Remove immutable fields before cloning
    clone_config = {
        "name": f"{original['name']}{suffix}",
        "type": original["type"],
        "status": "inactive",
        "listIds": original["listIds"],
        "rate": original["rate"],
        "maxAttemptsPerDay": original["maxAttemptsPerDay"],
        "maxAttemptsPerWeek": original["maxAttemptsPerWeek"],
        "wrapUpTimeout": original["wrapUpTimeout"],
        "preDialDelay": original["preDialDelay"],
        "dropRate": original["dropRate"],
        "rulesetId": original.get("rulesetId"),
        "campaignId": original["id"]  # Reference original for tracking
    }
    
    response = requests.post(url, headers=auth.get_headers(), json=clone_config)
    if response.status_code == 409:
        raise RuntimeError("Campaign clone already exists. Use a different suffix.")
    response.raise_for_status()
    
    return response.json()

The fetch_campaigns function iterates through pages until it locates the campaign or exhausts the list. The clone_campaign function strips immutable identifiers, sets the status to inactive to prevent accidental dialing, and preserves all rate-related fields. The POST request requires the outbound:campaign:write scope.

Step 2: Apply Rate Constraints and Calculate Dial Plan Projection

Genesys dial plans enforce constraints through rate (calls per minute), maxAttemptsPerDay, maxAttemptsPerWeek, dropRate, wrapUpTimeout, and preDialDelay. You will fetch the associated contact list size, apply the constraints, and calculate how many contacts the dialer would attempt over a simulation window.

import math
from typing import Dict, Any

def get_list_contact_count(auth: GenesysAuth, list_id: str) -> int:
    url = f"{auth.base_url}/api/v2/outbound/lists/{list_id}"
    response = requests.get(url, headers=auth.get_headers())
    response.raise_for_status()
    return response.json()["contactCount"]

def calculate_projection(
    campaign: Dict[str, Any],
    list_size: int,
    simulation_hours: int = 8
) -> Dict[str, Any]:
    rate_per_minute = campaign.get("rate", 0)
    drop_rate = campaign.get("dropRate", 0.0)
    max_attempts_per_day = campaign.get("maxAttemptsPerDay", 0)
    
    total_minutes = simulation_hours * 60
    theoretical_calls = rate_per_minute * total_minutes
    
    # Apply drop rate (calls that fail to connect before agent answer)
    effective_calls = theoretical_calls * (1.0 - drop_rate)
    
    # Cap by daily attempt limits if simulation exceeds 24 hours
    if simulation_hours > 24:
        days = simulation_hours / 24.0
        daily_cap = max_attempts_per_day * days
        effective_calls = min(effective_calls, daily_cap)
    
    # Cap by actual list size
    projected_contacts = min(effective_calls, list_size)
    
    return {
        "simulationHours": simulation_hours,
        "listSize": list_size,
        "ratePerMinute": rate_per_minute,
        "dropRate": drop_rate,
        "maxAttemptsPerDay": max_attempts_per_day,
        "theoreticalCalls": theoretical_calls,
        "effectiveCallsAfterDrop": effective_calls,
        "projectedContactsDialed": projected_contacts,
        "utilizationPercentage": (projected_contacts / list_size * 100) if list_size > 0 else 0
    }

The calculate_projection function models the dialer behavior mathematically. The rate field dictates the maximum calls per minute. The dropRate reduces connected attempts. The maxAttemptsPerDay enforces regulatory or policy caps. The function returns a structured projection dictionary that maps directly to the campaign configuration.

Step 3: Generate Projected Contact Report

You will combine the cloned campaign metadata and the mathematical projection into a final JSON report. This step includes a retry wrapper for 429 rate limit responses, which frequently occur when polling campaign or list endpoints in rapid succession.

import time
import json
from typing import Callable, Any

def api_call_with_retry(func: Callable, max_retries: int = 3, backoff_factor: float = 1.5) -> Any:
    for attempt in range(max_retries):
        try:
            return func()
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429 and attempt < max_retries - 1:
                wait_time = backoff_factor ** (attempt + 1)
                logger.warning(f"Rate limited (429). Retrying in {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise

def generate_report(auth: GenesysAuth, campaign_id: str, projection: Dict[str, Any]) -> str:
    def fetch_campaign_details():
        url = f"{auth.base_url}/api/v2/outbound/campaigns/{campaign_id}"
        response = requests.get(url, headers=auth.get_headers())
        response.raise_for_status()
        return response.json()

    campaign_details = api_call_with_retry(fetch_campaign_details)
    
    report = {
        "reportType": "DialPlanSimulation",
        "generatedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "campaign": {
            "id": campaign_details["id"],
            "name": campaign_details["name"],
            "status": campaign_details["status"],
            "type": campaign_details["type"]
        },
        "dialPlanConstraints": {
            "rate": campaign_details.get("rate"),
            "maxAttemptsPerDay": campaign_details.get("maxAttemptsPerDay"),
            "maxAttemptsPerWeek": campaign_details.get("maxAttemptsPerWeek"),
            "wrapUpTimeout": campaign_details.get("wrapUpTimeout"),
            "preDialDelay": campaign_details.get("preDialDelay"),
            "dropRate": campaign_details.get("dropRate")
        },
        "projection": projection
    }
    
    return json.dumps(report, indent=2)

The api_call_with_retry function intercepts 429 responses and applies exponential backoff. The generate_report function fetches the live clone state, merges it with the projection data, and outputs a formatted JSON string. This approach guarantees you never block on transient rate limits while maintaining accurate campaign metadata in the report.

Complete Working Example

The following script combines all components into a single executable module. Replace the placeholder credentials before running.

import requests
import time
import json
import logging
from typing import Optional, Dict, Any, Callable

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

class GenesysAuth:
    def __init__(self, org_id: str, client_id: str, client_secret: str):
        self.org_id = org_id
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = f"https://{org_id}.mygenesys.com"
        self.access_token: Optional[str] = None
        self.token_expiry: Optional[float] = None

    def get_token(self) -> str:
        if self.access_token and self.token_expiry and time.time() < self.token_expiry:
            return self.access_token
        url = f"{self.base_url}/oauth/token"
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }
        response = requests.post(url, data=payload)
        response.raise_for_status()
        data = response.json()
        self.access_token = data["access_token"]
        self.token_expiry = time.time() + data["expires_in"] - 60
        return self.access_token

    def get_headers(self) -> Dict[str, str]:
        return {
            "Authorization": f"Bearer {self.get_token()}",
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

def api_call_with_retry(func: Callable, max_retries: int = 3, backoff_factor: float = 1.5) -> Any:
    for attempt in range(max_retries):
        try:
            return func()
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429 and attempt < max_retries - 1:
                wait_time = backoff_factor ** (attempt + 1)
                logger.warning(f"Rate limited (429). Retrying in {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise

def fetch_campaigns(auth: GenesysAuth, campaign_name: str) -> Dict[str, Any]:
    url = f"{auth.base_url}/api/v2/outbound/campaigns"
    params = {"pageSize": 25, "pageNumber": 1}
    while True:
        response = requests.get(url, headers=auth.get_headers(), params=params)
        response.raise_for_status()
        data = response.json()
        for campaign in data["entities"]:
            if campaign["name"] == campaign_name:
                return campaign
        if not data["nextPage"]:
            raise ValueError(f"Campaign '{campaign_name}' not found.")
        params["pageNumber"] += 1

def clone_campaign(auth: GenesysAuth, original: Dict[str, Any], suffix: str = "_SIMULATION") -> Dict[str, Any]:
    url = f"{auth.base_url}/api/v2/outbound/campaigns"
    clone_config = {
        "name": f"{original['name']}{suffix}",
        "type": original["type"],
        "status": "inactive",
        "listIds": original["listIds"],
        "rate": original["rate"],
        "maxAttemptsPerDay": original["maxAttemptsPerDay"],
        "maxAttemptsPerWeek": original["maxAttemptsPerWeek"],
        "wrapUpTimeout": original["wrapUpTimeout"],
        "preDialDelay": original["preDialDelay"],
        "dropRate": original["dropRate"],
        "rulesetId": original.get("rulesetId"),
        "campaignId": original["id"]
    }
    response = requests.post(url, headers=auth.get_headers(), json=clone_config)
    if response.status_code == 409:
        raise RuntimeError("Campaign clone already exists. Use a different suffix.")
    response.raise_for_status()
    return response.json()

def get_list_contact_count(auth: GenesysAuth, list_id: str) -> int:
    url = f"{auth.base_url}/api/v2/outbound/lists/{list_id}"
    response = requests.get(url, headers=auth.get_headers())
    response.raise_for_status()
    return response.json()["contactCount"]

def calculate_projection(campaign: Dict[str, Any], list_size: int, simulation_hours: int = 8) -> Dict[str, Any]:
    rate_per_minute = campaign.get("rate", 0)
    drop_rate = campaign.get("dropRate", 0.0)
    max_attempts_per_day = campaign.get("maxAttemptsPerDay", 0)
    total_minutes = simulation_hours * 60
    theoretical_calls = rate_per_minute * total_minutes
    effective_calls = theoretical_calls * (1.0 - drop_rate)
    if simulation_hours > 24:
        days = simulation_hours / 24.0
        daily_cap = max_attempts_per_day * days
        effective_calls = min(effective_calls, daily_cap)
    projected_contacts = min(effective_calls, list_size)
    return {
        "simulationHours": simulation_hours,
        "listSize": list_size,
        "ratePerMinute": rate_per_minute,
        "dropRate": drop_rate,
        "maxAttemptsPerDay": max_attempts_per_day,
        "theoreticalCalls": theoretical_calls,
        "effectiveCallsAfterDrop": effective_calls,
        "projectedContactsDialed": projected_contacts,
        "utilizationPercentage": (projected_contacts / list_size * 100) if list_size > 0 else 0
    }

def generate_report(auth: GenesysAuth, campaign_id: str, projection: Dict[str, Any]) -> str:
    def fetch_campaign_details():
        url = f"{auth.base_url}/api/v2/outbound/campaigns/{campaign_id}"
        response = requests.get(url, headers=auth.get_headers())
        response.raise_for_status()
        return response.json()
    campaign_details = api_call_with_retry(fetch_campaign_details)
    report = {
        "reportType": "DialPlanSimulation",
        "generatedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "campaign": {
            "id": campaign_details["id"],
            "name": campaign_details["name"],
            "status": campaign_details["status"],
            "type": campaign_details["type"]
        },
        "dialPlanConstraints": {
            "rate": campaign_details.get("rate"),
            "maxAttemptsPerDay": campaign_details.get("maxAttemptsPerDay"),
            "maxAttemptsPerWeek": campaign_details.get("maxAttemptsPerWeek"),
            "wrapUpTimeout": campaign_details.get("wrapUpTimeout"),
            "preDialDelay": campaign_details.get("preDialDelay"),
            "dropRate": campaign_details.get("dropRate")
        },
        "projection": projection
    }
    return json.dumps(report, indent=2)

if __name__ == "__main__":
    ORG_ID = "your-org-id"
    CLIENT_ID = "your-client-id"
    CLIENT_SECRET = "your-client-secret"
    TARGET_CAMPAIGN = "Production Outbound Q3"
    SIMULATION_HOURS = 8
    
    auth = GenesysAuth(ORG_ID, CLIENT_ID, CLIENT_SECRET)
    
    try:
        original = fetch_campaigns(auth, TARGET_CAMPAIGN)
        logger.info(f"Located campaign: {original['name']}")
        
        cloned = clone_campaign(auth, original)
        logger.info(f"Cloned campaign: {cloned['name']} (ID: {cloned['id']})")
        
        list_id = cloned["listIds"][0]
        list_size = get_list_contact_count(auth, list_id)
        logger.info(f"List size: {list_size}")
        
        projection = calculate_projection(cloned, list_size, SIMULATION_HOURS)
        report_json = generate_report(auth, cloned["id"], projection)
        
        print("\n=== SIMULATION REPORT ===")
        print(report_json)
        
    except Exception as e:
        logger.error(f"Simulation failed: {e}")
        raise

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token is expired, missing, or the service account lacks the outbound:campaign:read scope.
• Fix: Verify the client credentials match a service account with active status. Ensure the token refresh logic runs before each request. Check the scope assignment in the Genesys administration console under Security > Service Accounts.
• Code Fix: The GenesysAuth class automatically refreshes tokens. If you receive a 401, add time.sleep(1) before retrying to allow the background refresh to complete.

Error: 403 Forbidden

• Cause: The service account lacks outbound:campaign:write or outbound:list:read scopes, or the user role does not permit outbound campaign management.
• Fix: Assign the Outbound: Campaign Management role to the service account. Verify scope permissions match the exact strings required by the endpoint.
• Code Fix: Wrap the POST call in a try block that catches requests.exceptions.HTTPError with status 403 and logs the missing scope recommendation.

Error: 429 Too Many Requests

• Cause: You exceeded the Genesys rate limit for your organization tier. Outbound endpoints typically cap at 100 requests per minute per client.
• Fix: Implement exponential backoff. The api_call_with_retry function handles this automatically. Reduce concurrent threads if polling multiple lists.
• Code Fix: The retry wrapper sleeps for backoff_factor ** (attempt + 1) seconds. Increase backoff_factor to 3.0 if you operate in a high-tenant environment.

Error: 500 Internal Server Error

• Cause: Genesys backend transient failure or malformed JSON payload.
• Fix: Validate the clone configuration dictionary against the OpenAPI specification. Remove optional fields that contain None values before POSTing.
• Code Fix: Add a payload sanitizer that filters None values: clean_config = {k: v for k, v in clone_config.items() if v is not None}.

Official References

• Genesys Cloud OAuth Documentation
• Outbound Campaigns API Reference
• Outbound Lists API Reference
• Python Requests Library Documentation