How to Export All Architect Flows as JSON Using the CX as Code CLI

What You Will Build

• A script that authenticates to Genesys Cloud and exports every Architect flow in your organization as individual JSON files.
• This tutorial uses the Genesys Cloud CX as Code (formerly PureCloud as Code) Python SDK.
• The implementation is written in Python 3.9+ using the purecloud-platform-client-v2 package.

Prerequisites

• OAuth Client: A Genesys Cloud OAuth client with the read:Flow scope. You can create a client in the Admin Console under Platform Setup > Integrations > OAuth Clients.
• SDK Version: purecloud-platform-client-v2 version 140.0.0 or higher.
• Runtime: Python 3.9 or later.
• Dependencies:
  • purecloud-platform-client-v2
  • pathlib (standard library)
  • os (standard library)

Install the SDK via pip:

pip install purecloud-platform-client-v2

Authentication Setup

The CX as Code SDK requires an OAuth2 token to interact with the Genesys Cloud API. The standard pattern is to use a client credentials flow. You must store your Client ID and Client Secret securely. For this tutorial, we will use environment variables.

Set the following environment variables in your terminal or .env file:

export PURECLOUD_CLIENT_ID="your-client-id"
export PURECLOUD_CLIENT_SECRET="your-client-secret"
export PURECLOUD_REGION="mypurecloud.com" # Or your specific region, e.g., usw2.pure.cloud

The SDK provides a Configuration object that handles the token acquisition and refresh logic automatically when initialized with a ClientCredentials authenticator.

import os
from purecloud_platform_client_v2 import Configuration, ClientCredentials

def get_purecloud_configuration() -> Configuration:
    """
    Initializes the Genesys Cloud configuration with OAuth2 client credentials.
    """
    client_id = os.getenv("PURECLOUD_CLIENT_ID")
    client_secret = os.getenv("PURECLOUD_CLIENT_SECRET")
    region = os.getenv("PURECLOUD_REGION", "mypurecloud.com")

    if not client_id or not client_secret:
        raise ValueError("PURECLOUD_CLIENT_ID and PURECLOUD_CLIENT_SECRET environment variables are required.")

    # Define the base URL based on the region
    if region == "mypurecloud.com":
        host_url = "https://api.mypurecloud.com"
    else:
        host_url = f"https://api.{region}"

    config = Configuration(host=host_url)
    config.authenticator = ClientCredentials(client_id, client_secret)
    
    return config

Implementation

Step 1: Initialize the Flows API Client

To interact with Architect flows, you need an instance of the FlowsApi. This client wraps the REST endpoints for flow management. We will also set up a directory structure to store the exported JSON files.

from purecloud_platform_client_v2 import FlowsApi
from pathlib import Path

def setup_export_directory(base_dir: str = "exported_flows") -> Path:
    """
    Creates the directory structure for storing exported flows.
    Returns the path to the output directory.
    """
    output_dir = Path(base_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    return output_dir

Step 2: List All Flows with Pagination

The get_flows endpoint returns a paginated list of flows. It does not return the full flow definition (the JSON graph), only metadata such as ID, name, and description. To export all flows, you must iterate through all pages.

The get_flows method accepts size and page parameters. The maximum size is typically 250. We will use a loop to fetch all pages until no more results are returned.

Required Scope: read:Flow

from purecloud_platform_client_v2.rest import ApiException

def list_all_flow_ids(flows_api: FlowsApi, page_size: int = 250) -> list[str]:
    """
    Fetches all flow IDs from the Genesys Cloud organization.
    Handles pagination automatically.
    """
    flow_ids = []
    page = 1
    
    while True:
        try:
            # Call the get_flows endpoint
            response = flows_api.get_flows(
                size=page_size,
                page=page
            )
            
            # Check if there are entities in the response
            if not response.entities:
                break
                
            # Extract IDs from the current page
            for flow in response.entities:
                flow_ids.append(flow.id)
                
            print(f"Fetched page {page}: {len(response.entities)} flows.")
            
            # Move to the next page
            page += 1
            
        except ApiException as e:
            print(f"Error fetching flows: {e.status} {e.reason}")
            if e.status == 401:
                print("Authentication failed. Check your Client ID and Secret.")
            elif e.status == 403:
                print("Forbidden. Ensure your OAuth client has the 'read:Flow' scope.")
            raise

    return flow_ids

Step 3: Export Individual Flow Definitions

Once you have the list of flow IDs, you must call get_flow for each ID. This endpoint returns the full flow definition, including the versioned graph, rules, and settings.

Important: The get_flow endpoint requires the flowId parameter. It returns a Flow object. The SDK serializes this object into Python dictionaries. We will convert this dictionary to a JSON string and write it to a file.

We must handle potential errors such as:

• 404 Not Found: The flow may have been deleted between the listing and the export.
• 429 Too Many Requests: Rate limiting. The SDK has built-in retry logic, but for bulk exports, you may want to add a small delay between requests.

import json
import time
from purecloud_platform_client_v2.rest import ApiException

def export_flows(flows_api: FlowsApi, flow_ids: list[str], output_dir: Path) -> None:
    """
    Iterates through flow IDs and exports each flow definition as a JSON file.
    """
    success_count = 0
    error_count = 0
    
    for flow_id in flow_ids:
        try:
            # Fetch the full flow definition
            flow_definition = flows_api.get_flow(flow_id=flow_id)
            
            # Get the flow name for the filename
            flow_name = flow_definition.name.replace(" ", "_")
            # Sanitize the filename to avoid invalid characters
            flow_name = "".join(c for c in flow_name if c.isalnum() or c in ('-', '_'))
            
            # Create the filename: {flow_id}_{flow_name}.json
            filename = f"{flow_id}_{flow_name}.json"
            file_path = output_dir / filename
            
            # Serialize the flow definition to JSON
            # The SDK object has a to_dict() method or can be serialized via json.dumps
            # We use json.dumps with default=str to handle any non-serializable objects
            flow_json = json.dumps(flow_definition.to_dict(), indent=2, default=str)
            
            # Write to file
            with open(file_path, 'w', encoding='utf-8') as f:
                f.write(flow_json)
                
            success_count += 1
            print(f"Exported: {filename}")
            
            # Rate limit mitigation: Sleep briefly between requests
            # This is optional but recommended for large orgs to avoid 429s
            time.sleep(0.1)
            
        except ApiException as e:
            error_count += 1
            print(f"Error exporting flow ID {flow_id}: {e.status} {e.reason}")
            if e.status == 404:
                print(f"  Flow {flow_id} not found. It may have been deleted.")
            elif e.status == 429:
                print("  Rate limited. Increasing sleep duration.")
                time.sleep(5) # Backoff strategy
        except Exception as e:
            error_count += 1
            print(f"Unexpected error exporting flow ID {flow_id}: {str(e)}")
            
    print(f"\nExport Complete. Success: {success_count}, Errors: {error_count}")

Step 4: Handle Versioning (Optional but Recommended)

Architect flows have versions. The get_flow endpoint by default returns the latest version. If you need to export a specific version, you can pass the version parameter to get_flow. For a full export, it is usually best to export the latest version unless you are doing an audit of historical versions.

The code above exports the latest version. If you want to export all versions of a flow, you would need to call get_flow_versions first, then iterate through those versions. For this tutorial, we stick to the latest version for simplicity and performance.

Complete Working Example

Here is the full, copy-pasteable script. Save this as export_flows.py.

import os
import json
import time
from pathlib import Path
from purecloud_platform_client_v2 import Configuration, ClientCredentials, FlowsApi
from purecloud_platform_client_v2.rest import ApiException

def get_purecloud_configuration() -> Configuration:
    """
    Initializes the Genesys Cloud configuration with OAuth2 client credentials.
    """
    client_id = os.getenv("PURECLOUD_CLIENT_ID")
    client_secret = os.getenv("PURECLOUD_CLIENT_SECRET")
    region = os.getenv("PURECLOUD_REGION", "mypurecloud.com")

    if not client_id or not client_secret:
        raise ValueError("PURECLOUD_CLIENT_ID and PURECLOUD_CLIENT_SECRET environment variables are required.")

    if region == "mypurecloud.com":
        host_url = "https://api.mypurecloud.com"
    else:
        host_url = f"https://api.{region}"

    config = Configuration(host=host_url)
    config.authenticator = ClientCredentials(client_id, client_secret)
    
    return config

def setup_export_directory(base_dir: str = "exported_flows") -> Path:
    """
    Creates the directory structure for storing exported flows.
    """
    output_dir = Path(base_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    return output_dir

def list_all_flow_ids(flows_api: FlowsApi, page_size: int = 250) -> list[str]:
    """
    Fetches all flow IDs from the Genesys Cloud organization.
    """
    flow_ids = []
    page = 1
    
    while True:
        try:
            response = flows_api.get_flows(
                size=page_size,
                page=page
            )
            
            if not response.entities:
                break
                
            for flow in response.entities:
                flow_ids.append(flow.id)
                
            print(f"Fetched page {page}: {len(response.entities)} flows.")
            page += 1
            
        except ApiException as e:
            print(f"Error fetching flows: {e.status} {e.reason}")
            if e.status == 401:
                print("Authentication failed. Check your Client ID and Secret.")
            elif e.status == 403:
                print("Forbidden. Ensure your OAuth client has the 'read:Flow' scope.")
            raise

    return flow_ids

def export_flows(flows_api: FlowsApi, flow_ids: list[str], output_dir: Path) -> None:
    """
    Iterates through flow IDs and exports each flow definition as a JSON file.
    """
    success_count = 0
    error_count = 0
    
    for flow_id in flow_ids:
        try:
            flow_definition = flows_api.get_flow(flow_id=flow_id)
            
            flow_name = flow_definition.name.replace(" ", "_")
            flow_name = "".join(c for c in flow_name if c.isalnum() or c in ('-', '_'))
            
            filename = f"{flow_id}_{flow_name}.json"
            file_path = output_dir / filename
            
            flow_json = json.dumps(flow_definition.to_dict(), indent=2, default=str)
            
            with open(file_path, 'w', encoding='utf-8') as f:
                f.write(flow_json)
                
            success_count += 1
            print(f"Exported: {filename}")
            
            # Rate limit mitigation
            time.sleep(0.1)
            
        except ApiException as e:
            error_count += 1
            print(f"Error exporting flow ID {flow_id}: {e.status} {e.reason}")
            if e.status == 404:
                print(f"  Flow {flow_id} not found.")
            elif e.status == 429:
                print("  Rate limited. Backing off.")
                time.sleep(5)
        except Exception as e:
            error_count += 1
            print(f"Unexpected error exporting flow ID {flow_id}: {str(e)}")
            
    print(f"\nExport Complete. Success: {success_count}, Errors: {error_count}")

def main():
    try:
        # 1. Setup Configuration
        config = get_purecloud_configuration()
        
        # 2. Setup API Client
        flows_api = FlowsApi(config)
        
        # 3. Setup Output Directory
        output_dir = setup_export_directory("architect_exports")
        
        # 4. List All Flows
        print("Fetching flow IDs...")
        flow_ids = list_all_flow_ids(flows_api)
        print(f"Total flows found: {len(flow_ids)}")
        
        if not flow_ids:
            print("No flows found. Exiting.")
            return
            
        # 5. Export Flows
        print("Exporting flow definitions...")
        export_flows(flows_api, flow_ids, output_dir)
        
    except Exception as e:
        print(f"Critical error: {str(e)}")
        raise

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

What causes it:
Your OAuth client does not have the required scope to read flows.

How to fix it:

1. Go to the Genesys Cloud Admin Console.
2. Navigate to Platform Setup > Integrations > OAuth Clients.
3. Click on your client.
4. Under Scopes, ensure read:Flow is checked.
5. Save the changes.
6. Your token will automatically update on the next request, or you may need to refresh your token cache if you are caching it externally.

Error: 429 Too Many Requests

What causes it:
You are hitting the rate limit for the get_flows or get_flow endpoints. Genesys Cloud has per-endpoint and per-organization rate limits.

How to fix it:
The script above includes a time.sleep(0.1) between flow exports. If you still encounter 429 errors, increase the sleep duration. For very large organizations (1000+ flows), consider implementing an exponential backoff strategy.

# Example of exponential backoff
import random

def sleep_with_backoff(attempt: int):
    delay = min(2 ** attempt, 60) + random.uniform(0, 1)
    time.sleep(delay)

Error: 401 Unauthorized

What causes it:
The OAuth token has expired or the Client ID/Secret is incorrect.

How to fix it:
The SDK’s ClientCredentials authenticator automatically refreshes the token. If you are seeing 401 errors, ensure your environment variables are set correctly. If you are using a cached token, ensure it has not expired. The script above initializes a fresh configuration each time, which handles token acquisition automatically.

Error: ModuleNotFoundError: No module named ‘purecloud_platform_client_v2’

What causes it:
The SDK is not installed in your Python environment.

How to fix it:
Run the following command in your terminal:

pip install purecloud_platform_client_v2

If you are using a virtual environment, ensure it is activated before running the script.

Official References

• Genesys Cloud CX as Code Python SDK Documentation
• Flows API Reference
• OAuth2 Client Credentials Flow