Dynamically Adjusting Genesys Cloud Outbound Campaign Dial Rates with Python

What You Will Build

A Python daemon script that polls real-time Genesys Cloud Outbound campaign statistics, calculates live answer rates, and automatically adjusts the campaign dial rate using configurable throttling constraints via the Update Campaign endpoint. This tutorial covers the official purecloud_platform_client Python SDK, raw OAuth token flows, and production-ready rate limit handling. The solution is implemented in Python 3.9+.

Prerequisites

• OAuth 2.0 Client Credentials grant type with scopes: outbound:campaign:view and outbound:campaign:edit
• Genesys Cloud Python SDK version 13.0.0 or higher (pip install purecloud-platform-client)
• Python 3.9+ runtime
• External dependencies: requests (for token management), tenacity (for retry logic), pydantic (optional for config validation)
• A valid Genesys Cloud Outbound campaign ID
• Environment variables for credentials: GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET, GENESYS_BASE_URL, GENESYS_CAMPAIGN_ID

Authentication Setup

Genesys Cloud requires OAuth 2.0 Client Credentials for server-to-server integrations. The token endpoint issues a short-lived access token that must be cached and refreshed before expiration. The following implementation handles token acquisition, caching, and automatic refresh.

import os
import time
import logging
import requests
from typing import Optional

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

class GenesysAuthManager:
    def __init__(self, client_id: str, client_secret: str, base_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = f"{base_url}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0.0
        self.refresh_buffer = 60  # Refresh 60 seconds before expiration

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

        logger.info("Requesting new OAuth token...")
        payload = {
            "grant_type": "client_credentials",
            "scope": "outbound:campaign:view outbound:campaign:edit"
        }
        
        try:
            response = requests.post(
                self.token_url,
                data=payload,
                auth=(self.client_id, self.client_secret),
                timeout=10
            )
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
            logger.error(f"OAuth token request failed: {e.response.status_code} {e.response.text}")
            raise
        except requests.exceptions.RequestException as e:
            logger.error(f"Network error during OAuth request: {e}")
            raise

        token_data = response.json()
        self.access_token = token_data["access_token"]
        self.token_expiry = time.time() + token_data["expires_in"]
        logger.info("OAuth token acquired successfully.")
        return self.access_token

The get_access_token method checks the local cache before making network calls. The refresh_buffer prevents edge cases where the token expires mid-request. The scope string explicitly requests campaign read and write permissions.

Implementation

Step 1: Initialize SDK and Fetch Campaign Metadata

The Genesys Cloud Python SDK abstracts REST calls into typed models. You must configure the SDK with the active access token before issuing API calls. Fetching the existing campaign payload is required because the Update Campaign endpoint expects a complete or partial campaign object. Preserving existing configuration prevents accidental overwrites.

from purecloud_platform_client import PlatformClient, OutboundApi, Configuration
from purecloud_platform_client.rest import ApiException

def initialize_sdk(auth_manager: GenesysAuthManager) -> OutboundApi:
    config = Configuration(
        access_token=auth_manager.get_access_token(),
        host=auth_manager.token_url.replace("/oauth/token", "")
    )
    platform_client = PlatformClient(configuration=config)
    platform_client.set_oauth_client_credentials(
        client_id=auth_manager.client_id,
        client_secret=auth_manager.client_secret
    )
    return platform_client.outbound_api

def fetch_campaign_metadata(outbound_api: OutboundApi, campaign_id: str) -> dict:
    try:
        campaign = outbound_api.get_outbound_campaign(campaign_id=campaign_id)
        return {
            "id": campaign.id,
            "dial_rate": campaign.dial_rate,
            "name": campaign.name,
            "contact_list_id": campaign.contact_list.id,
            "script_id": campaign.script.id,
            "campaign_type": campaign.campaign_type
        }
    except ApiException as e:
        if e.status == 404:
            raise ValueError(f"Campaign {campaign_id} not found.")
        raise

The SDK automatically handles header injection and model serialization. The get_outbound_campaign call requires outbound:campaign:view. Storing the original dial_rate establishes the baseline for dynamic adjustments.

Step 2: Poll Real-Time Statistics and Calculate Answer Rates

Real-time campaign metrics are exposed via the /api/v2/outbound/campaigns/{campaignId}/realtime endpoint. The SDK method get_outbound_campaign_realtime returns a CampaignRealtimeStats object containing answer_rate, dial_rate, and call volume counters.

HTTP Request/Response Cycle:

GET /api/v2/outbound/campaigns/a1b2c3d4-e5f6-7890-abcd-ef1234567890/realtime HTTP/1.1
Host: api.mypurecloud.com
Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...
Accept: application/json

{
  "campaignId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "dialRate": 150,
  "answerRate": 0.12,
  "callsDialed": 12500,
  "callsAnswered": 1500,
  "avgAnswerTime": 3.2,
  "avgDialTime": 1.8,
  "avgCallTime": 45.5,
  "dispositionRate": 0.08
}

The answerRate field is a decimal representing the percentage of dialed calls that reached a live agent or customer. A value of 0.12 indicates a 12 percent answer rate. The polling loop must respect Genesys Cloud rate limits. The following implementation uses tenacity to handle 429 Too Many Requests responses with exponential backoff and Retry-After header compliance.

import time
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from purecloud_platform_client.rest import ApiException

def parse_retry_after(response_headers: dict) -> float:
    retry_after = response_headers.get("Retry-After")
    return int(retry_after) if retry_after and retry_after.isdigit() else 30

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    retry=retry_if_exception_type(ApiException),
    reraise=True
)
def poll_realtime_stats(outbound_api: OutboundApi, campaign_id: str) -> dict:
    try:
        stats = outbound_api.get_outbound_campaign_realtime(campaign_id=campaign_id)
        return {
            "answer_rate": stats.answer_rate,
            "dial_rate": stats.dial_rate,
            "calls_dialed": stats.calls_dialed,
            "calls_answered": stats.calls_answered
        }
    except ApiException as e:
        if e.status == 429:
            wait_time = parse_retry_after(e.headers) if e.headers else 30
            logger.warning(f"Rate limited. Retrying in {wait_time} seconds.")
            time.sleep(wait_time)
        raise

The retry decorator catches ApiException, extracts the Retry-After header when present, and delays subsequent requests. The function returns a flattened dictionary for straightforward mathematical operations.

Step 3: Apply Throttling Constraints and Update Campaign

Throttling logic compares the live answer rate against a target threshold. If the answer rate falls below the target, the dial rate decreases. If the answer rate exceeds the target by a margin, the dial rate increases. Constraints prevent aggressive oscillation.

HTTP Request/Response Cycle for Update:

PUT /api/v2/outbound/campaigns/a1b2c3d4-e5f6-7890-abcd-ef1234567890 HTTP/1.1
Host: api.mypurecloud.com
Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...
Content-Type: application/json

{
  "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "name": "Q3 Outbound Campaign",
  "contactList": { "id": "cl-12345" },
  "script": { "id": "sc-67890" },
  "campaignType": "preview",
  "dialRate": 100
}

{
  "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "name": "Q3 Outbound Campaign",
  "contactList": { "id": "cl-12345" },
  "script": { "id": "sc-67890" },
  "campaignType": "preview",
  "dialRate": 100,
  "createdBy": { "id": "..." },
  "updatedBy": { "id": "..." }
}

The update payload must include all required campaign fields. The following function applies the throttling algorithm and issues the PUT request.

from purecloud_platform_client.models import OutboundCampaign, OutboundContactList, OutboundScript

def calculate_adjusted_dial_rate(
    current_rate: int,
    current_answer_rate: float,
    target_answer_rate: float,
    min_rate: int,
    max_rate: int,
    step: int
) -> int:
    if current_answer_rate < target_answer_rate:
        new_rate = current_rate - step
    elif current_answer_rate > target_answer_rate + 0.05:
        new_rate = current_rate + step
    else:
        return current_rate
    
    return max(min_rate, min(max_rate, new_rate))

def update_campaign_dial_rate(
    outbound_api: OutboundApi,
    campaign_id: str,
    new_dial_rate: int,
    metadata: dict
) -> bool:
    campaign = OutboundCampaign(
        id=campaign_id,
        name=metadata["name"],
        contact_list=OutboundContactList(id=metadata["contact_list_id"]),
        script=OutboundScript(id=metadata["script_id"]),
        campaign_type=metadata["campaign_type"],
        dial_rate=new_dial_rate
    )
    
    try:
        outbound_api.put_outbound_campaign(campaign_id=campaign_id, body=campaign)
        logger.info(f"Campaign dial rate updated to {new_dial_rate} cpm.")
        return True
    except ApiException as e:
        if e.status == 400:
            logger.error(f"Validation error: {e.body}")
        elif e.status == 403:
            logger.error("Insufficient permissions for campaign update.")
        raise

The calculate_adjusted_dial_rate function enforces floor and ceiling constraints. The update_campaign_dial_rate function reconstructs the campaign object using SDK models. The put_outbound_campaign method requires outbound:campaign:edit.

Complete Working Example

The following script combines authentication, polling, throttling logic, and update execution into a single runnable module. Save as campaign_throttler.py and execute with python campaign_throttler.py.

import os
import time
import logging
import requests
from typing import Optional
from purecloud_platform_client import PlatformClient, OutboundApi, Configuration
from purecloud_platform_client.rest import ApiException
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from purecloud_platform_client.models import OutboundCampaign, OutboundContactList, OutboundScript

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

class GenesysAuthManager:
    def __init__(self, client_id: str, client_secret: str, base_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = f"{base_url}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0.0
        self.refresh_buffer = 60

    def get_access_token(self) -> str:
        if self.access_token and time.time() < self.token_expiry - self.refresh_buffer:
            return self.access_token
        payload = {"grant_type": "client_credentials", "scope": "outbound:campaign:view outbound:campaign:edit"}
        try:
            response = requests.post(self.token_url, data=payload, auth=(self.client_id, self.client_secret), timeout=10)
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
            logger.error(f"OAuth token request failed: {e.response.status_code} {e.response.text}")
            raise
        token_data = response.json()
        self.access_token = token_data["access_token"]
        self.token_expiry = time.time() + token_data["expires_in"]
        return self.access_token

def initialize_sdk(auth_manager: GenesysAuthManager) -> OutboundApi:
    config = Configuration(
        access_token=auth_manager.get_access_token(),
        host=auth_manager.token_url.replace("/oauth/token", "")
    )
    platform_client = PlatformClient(configuration=config)
    platform_client.set_oauth_client_credentials(
        client_id=auth_manager.client_id,
        client_secret=auth_manager.client_secret
    )
    return platform_client.outbound_api

def fetch_campaign_metadata(outbound_api: OutboundApi, campaign_id: str) -> dict:
    try:
        campaign = outbound_api.get_outbound_campaign(campaign_id=campaign_id)
        return {
            "id": campaign.id,
            "dial_rate": campaign.dial_rate,
            "name": campaign.name,
            "contact_list_id": campaign.contact_list.id,
            "script_id": campaign.script.id,
            "campaign_type": campaign.campaign_type
        }
    except ApiException as e:
        if e.status == 404:
            raise ValueError(f"Campaign {campaign_id} not found.")
        raise

def parse_retry_after(response_headers: dict) -> float:
    retry_after = response_headers.get("Retry-After")
    return int(retry_after) if retry_after and retry_after.isdigit() else 30

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    retry=retry_if_exception_type(ApiException),
    reraise=True
)
def poll_realtime_stats(outbound_api: OutboundApi, campaign_id: str) -> dict:
    try:
        stats = outbound_api.get_outbound_campaign_realtime(campaign_id=campaign_id)
        return {
            "answer_rate": stats.answer_rate,
            "dial_rate": stats.dial_rate,
            "calls_dialed": stats.calls_dialed,
            "calls_answered": stats.calls_answered
        }
    except ApiException as e:
        if e.status == 429:
            wait_time = parse_retry_after(e.headers) if e.headers else 30
            logger.warning(f"Rate limited. Retrying in {wait_time} seconds.")
            time.sleep(wait_time)
        raise

def calculate_adjusted_dial_rate(
    current_rate: int,
    current_answer_rate: float,
    target_answer_rate: float,
    min_rate: int,
    max_rate: int,
    step: int
) -> int:
    if current_answer_rate < target_answer_rate:
        new_rate = current_rate - step
    elif current_answer_rate > target_answer_rate + 0.05:
        new_rate = current_rate + step
    else:
        return current_rate
    return max(min_rate, min(max_rate, new_rate))

def update_campaign_dial_rate(
    outbound_api: OutboundApi,
    campaign_id: str,
    new_dial_rate: int,
    metadata: dict
) -> bool:
    campaign = OutboundCampaign(
        id=campaign_id,
        name=metadata["name"],
        contact_list=OutboundContactList(id=metadata["contact_list_id"]),
        script=OutboundScript(id=metadata["script_id"]),
        campaign_type=metadata["campaign_type"],
        dial_rate=new_dial_rate
    )
    try:
        outbound_api.put_outbound_campaign(campaign_id=campaign_id, body=campaign)
        logger.info(f"Campaign dial rate updated to {new_dial_rate} cpm.")
        return True
    except ApiException as e:
        if e.status == 400:
            logger.error(f"Validation error: {e.body}")
        elif e.status == 403:
            logger.error("Insufficient permissions for campaign update.")
        raise

def main():
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    base_url = os.getenv("GENESYS_BASE_URL", "https://api.mypurecloud.com")
    campaign_id = os.getenv("GENESYS_CAMPAIGN_ID")
    
    if not all([client_id, client_secret, campaign_id]):
        raise EnvironmentError("Missing required environment variables.")

    auth_manager = GenesysAuthManager(client_id, client_secret, base_url)
    outbound_api = initialize_sdk(auth_manager)
    metadata = fetch_campaign_metadata(outbound_api, campaign_id)

    TARGET_ANSWER_RATE = 0.20
    MIN_DIAL_RATE = 50
    MAX_DIAL_RATE = 500
    DIAL_RATE_STEP = 25
    POLL_INTERVAL = 30
    UPDATE_COOLDOWN = 60
    last_update_time = 0

    logger.info(f"Starting throttler for campaign {campaign_id}. Target answer rate: {TARGET_ANSWER_RATE*100}%")
    
    while True:
        try:
            stats = poll_realtime_stats(outbound_api, campaign_id)
            current_rate = stats["dial_rate"]
            current_answer_rate = stats["answer_rate"]
            
            logger.info(f"Current dial rate: {current_rate} | Answer rate: {current_answer_rate:.2%} | Dialed: {stats['calls_dialed']}")
            
            new_rate = calculate_adjusted_dial_rate(
                current_rate, current_answer_rate, TARGET_ANSWER_RATE,
                MIN_DIAL_RATE, MAX_DIAL_RATE, DIAL_RATE_STEP
            )
            
            if new_rate != current_rate:
                if time.time() - last_update_time >= UPDATE_COOLDOWN:
                    update_campaign_dial_rate(outbound_api, campaign_id, new_rate, metadata)
                    last_update_time = time.time()
                else:
                    logger.info("Update skipped due to cooldown period.")
            else:
                logger.info("Dial rate stable. No adjustment required.")
                
        except Exception as e:
            logger.error(f"Polling loop error: {e}")
            time.sleep(15)
            
        time.sleep(POLL_INTERVAL)

if __name__ == "__main__":
    main()

The script runs an infinite polling loop with a configurable interval. The UPDATE_COOLDOWN variable prevents rapid API calls when the answer rate fluctuates near the threshold. The loop catches unexpected exceptions, logs them, and continues execution without terminating.

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token has expired or the client credentials are invalid.
• Fix: Verify GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET match a registered OAuth 2.0 client in the Genesys Cloud admin console. Ensure the client type is set to confidential. Check that the token refresh buffer in GenesysAuthManager accounts for clock skew.
• Code Fix: The get_access_token method automatically refreshes tokens. If persistent, validate the client secret against the Genesys Cloud developer portal.

Error: 403 Forbidden

• Cause: The OAuth client lacks the required scopes or the associated user does not have the Outbound: Campaign: Edit permission.
• Fix: Navigate to the OAuth client configuration and append outbound:campaign:edit to the scopes list. Assign a service account to the client and grant it the Outbound Campaign Manager role.
• Code Fix: The scope string in the authentication payload must include both outbound:campaign:view and outbound:campaign:edit.

Error: 429 Too Many Requests

• Cause: The polling interval exceeds Genesys Cloud rate limits for the Outbound API.
• Fix: The tenacity decorator implements exponential backoff. The parse_retry_after function respects the Retry-After header. Increase POLL_INTERVAL to 60 seconds if rate limits persist during high-traffic periods.
• Code Fix: Monitor the Retry-After header value. The retry logic automatically delays subsequent requests. Do not remove the decorator in production environments.

Error: 400 Bad Request

• Cause: The campaign update payload contains invalid field values or missing required properties.
• Fix: Verify that contact_list_id and script_id match active resources in Genesys Cloud. Ensure dial_rate falls within the platform-enforced minimum and maximum for the specific campaign type.
• Code Fix: The OutboundCampaign model enforces type validation. Print e.body from the ApiException to inspect the exact validation error returned by the server.

Error: SDK Model Serialization Failure

• Cause: Passing raw dictionaries to SDK methods instead of typed model instances.
• Fix: Use OutboundCampaign, OutboundContactList, and OutboundScript constructors. The SDK requires explicit model instantiation for PUT and POST operations.
• Code Fix: The update_campaign_dial_rate function demonstrates correct model construction. Avoid using dict payloads with put_outbound_campaign.

Official References

• Genesys Cloud Outbound Campaign API Documentation
• Campaign Real-Time Stats Endpoint
• Update Campaign Endpoint
• PureCloud Platform Client Python SDK Reference
• OAuth 2.0 Client Credentials Flow