Reuse Shared Flows in Genesys Cloud CX via the API

What You Will Build

• One sentence: The code creates a reusable Flow module and attaches it to multiple inbound call flows using the Genesys Cloud Platform API.
• One sentence: This uses the Genesys Cloud CX Flow API (/api/v2/flows) and Module API (/api/v2/flows/modules).
• One sentence: The programming language covered is Python using the genesyscloud SDK.

Prerequisites

• OAuth Client Type: Private Key or JWT.
• Required Scopes: flow:flow:write, flow:module:write, flow:flow:read, flow:module:read.
• SDK Version: genesyscloud Python SDK (v2.0.0+).
• Language/Runtime: Python 3.8+ with pip.
• External Dependencies: genesyscloud, pyyaml (for configuration).

Authentication Setup

The Genesys Cloud Python SDK handles OAuth token acquisition automatically when configured with a private key. You must store your private key securely. For this tutorial, we assume the private key is stored in a file named private_key.pem.

import os
from genesyscloud import PlatformClient
from genesyscloud.rest import ApiException

def init_platform_client() -> PlatformClient:
    """
    Initializes and authenticates the Genesys Cloud Platform Client.
    """
    # Load private key from file or environment variable
    private_key_path = os.getenv("GENESYS_PRIVATE_KEY_PATH", "private_key.pem")
    
    with open(private_key_path, "r") as f:
        private_key = f.read()

    # Initialize the client
    platform_client = PlatformClient()
    
    # Set credentials
    platform_client.set_credentials(
        api_key=None, # Not used for private key flow
        api_secret=None, # Not used for private key flow
        private_key=private_key,
        private_key_id=os.getenv("GENESYS_PRIVATE_KEY_ID"),
        client_id=os.getenv("GENESYS_CLIENT_ID"),
        client_secret=None, # Not used for private key flow
        org_id=os.getenv("GENESYS_ORG_ID")
    )
    
    return platform_client

Error Handling: If the private key is invalid, the set_credentials method may not throw immediately, but subsequent API calls will return a 401 Unauthorized error. Always verify the GENESYS_PRIVATE_KEY_ID matches the ID of the key uploaded in the Genesys Cloud Admin UI.

Implementation

Step 1: Create the Shared Module

A “Shared Flow” in Genesys Cloud is technically a Module. Modules are self-contained units of logic that can be invoked by other flows. They do not handle inbound events themselves; they are called by a parent flow.

We will create a simple module that collects a single digit input. This module will be reused by two different inbound flows.

OAuth Scope: flow:module:write

from genesyscloud import FlowApi
from genesyscloud.flow.models import (
    FlowModuleCreateRequest,
    FlowModuleDocument,
    FlowModuleDocumentContent,
    FlowModuleDocumentContentEntry,
    FlowModuleDocumentContentEntryType,
    FlowModuleDocumentContentEntryAction,
    FlowModuleDocumentContentEntryActionType
)
import uuid

def create_shared_module(api_client: FlowApi, module_name: str) -> str:
    """
    Creates a new Flow Module.
    Returns the ID of the created module.
    """
    # 1. Define the module structure
    module_id = str(uuid.uuid4())
    
    # Define the entry point of the module
    entry_action = FlowModuleDocumentContentEntryAction(
        action="collectInput",
        action_type="collectInput",
        parameters={
            "inputType": "DTMF",
            "maxDigits": 1,
            "timeout": 5000
        }
    )
    
    # Create the entry object
    entry = FlowModuleDocumentContentEntry(
        entry_type="action",
        action=entry_action
    )
    
    # Wrap in content
    content = FlowModuleDocumentContent(
        entries={
            "start": entry
        }
    )
    
    # Create the document
    document = FlowModuleDocument(
        content=content
    )
    
    # Create the module request body
    module_request = FlowModuleCreateRequest(
        name=module_name,
        description="Shared module for collecting a single digit.",
        document=document,
        type="module", # Critical: must be 'module' for reuse
        version=1
    )
    
    try:
        # 2. Call the API
        response = api_client.post_flow_module(
            body=module_request
        )
        
        print(f"Module created successfully. ID: {response.id}")
        return response.id
        
    except ApiException as e:
        if e.status == 409:
            print("Module with this name might already exist or conflict.")
        else:
            print(f"API Error: {e.status} - {e.body}")
        raise

Why this design: The type: "module" field is critical. If you omit this or set it to flow, the object will be created as a standalone flow that cannot be referenced by other flows as a sub-flow. Modules are explicitly designed for reuse.

Step 2: Create the Parent Inbound Flows

Next, we create two distinct inbound call flows that will invoke the shared module. To invoke a module, the parent flow must contain an invokeSubFlow action.

OAuth Scope: flow:flow:write

from genesyscloud.flow.models import (
    FlowCreateRequest,
    FlowDocument,
    FlowDocumentContent,
    FlowDocumentContentEntry,
    FlowDocumentContentEntryAction,
    FlowDocumentContentEntryCondition,
    FlowDocumentContentEntryType,
    FlowDocumentContentEntryActionType
)

def create_inbound_flow(api_client: FlowApi, flow_name: str, module_id: str) -> str:
    """
    Creates an inbound flow that invokes a shared module.
    Returns the ID of the created flow.
    """
    # 1. Define the invokeSubFlow action
    # This action calls the shared module created in Step 1
    invoke_action = FlowDocumentContentEntryAction(
        action="invokeSubFlow",
        action_type="invokeSubFlow",
        parameters={
            "moduleId": module_id, # Reference the shared module ID
            "timeout": 30000
        }
    )
    
    # 2. Define the next step after the module returns
    # For simplicity, we just transfer to a queue or end
    end_action = FlowDocumentContentEntryAction(
        action="end",
        action_type="end"
    )
    
    # 3. Build the flow document structure
    # Start -> Invoke Module -> End
    invoke_entry = FlowDocumentContentEntry(
        entry_type="action",
        action=invoke_action,
        nexts={"default": "end_step"} # Default path after module returns
    )
    
    end_entry = FlowDocumentContentEntry(
        entry_type="action",
        action=end_action
    )
    
    content = FlowDocumentContent(
        entries={
            "start": invoke_entry,
            "end_step": end_entry
        }
    )
    
    document = FlowDocument(
        content=content
    )
    
    # 4. Create the flow request
    flow_request = FlowCreateRequest(
        name=flow_name,
        description=f"Inbound flow invoking shared module {module_id}",
        document=document,
        type="voice", # Must be 'voice' for inbound calls
        enabled=True,
        version=1
    )
    
    try:
        response = api_client.post_flow(
            body=flow_request
        )
        print(f"Flow '{flow_name}' created successfully. ID: {response.id}")
        return response.id
    except ApiException as e:
        print(f"Error creating flow: {e.body}")
        raise

Non-Obvious Parameter: The moduleId in the invokeSubFlow parameters must be the ID of the module, not the name. The API does not resolve names to IDs automatically in the payload. You must persist the module ID from Step 1.

Step 3: Verify and Link Flows

After creation, it is good practice to verify that the flows reference the module correctly. We can fetch the flow details to inspect the document.

OAuth Scope: flow:flow:read

def verify_flow_references_module(api_client: FlowApi, flow_id: str, expected_module_id: str) -> bool:
    """
    Fetches a flow and checks if it references the expected module ID.
    """
    try:
        # Fetch the flow
        flow = api_client.get_flow(
            flow_id=flow_id,
            expand=["document"] # Expand to get the full document structure
        )
        
        # Traverse the document to find invokeSubFlow actions
        content = flow.document.content
        entries = content.entries
        
        for entry_key, entry in entries.items():
            if entry.action and entry.action.action == "invokeSubFlow":
                referenced_module_id = entry.action.parameters.get("moduleId")
                if referenced_module_id == expected_module_id:
                    print(f"Flow {flow_id} correctly references module {expected_module_id}")
                    return True
        
        print(f"Flow {flow_id} does not reference module {expected_module_id}")
        return False
        
    except ApiException as e:
        print(f"Error verifying flow: {e.body}")
        return False

Complete Working Example

This script combines all steps: authentication, module creation, flow creation, and verification. It requires a .env file with GENESYS_PRIVATE_KEY_PATH, GENESYS_PRIVATE_KEY_ID, GENESYS_CLIENT_ID, and GENESYS_ORG_ID.

import os
from dotenv import load_dotenv
from genesyscloud import PlatformClient, FlowApi
from genesyscloud.rest import ApiException
import uuid
import sys

# Load environment variables
load_dotenv()

# Import models for module and flow creation
from genesyscloud.flow.models import (
    FlowModuleCreateRequest,
    FlowModuleDocument,
    FlowModuleDocumentContent,
    FlowModuleDocumentContentEntry,
    FlowModuleDocumentContentEntryAction,
    FlowCreateRequest,
    FlowDocument,
    FlowDocumentContent,
    FlowDocumentContentEntry,
    FlowDocumentContentEntryAction
)

def init_client():
    platform_client = PlatformClient()
    platform_client.set_credentials(
        private_key=open(os.getenv("GENESYS_PRIVATE_KEY_PATH")).read(),
        private_key_id=os.getenv("GENESYS_PRIVATE_KEY_ID"),
        client_id=os.getenv("GENESYS_CLIENT_ID"),
        org_id=os.getenv("GENESYS_ORG_ID")
    )
    return platform_client

def create_shared_module(flow_api: FlowApi, name: str) -> str:
    # Define module structure
    entry_action = FlowModuleDocumentContentEntryAction(
        action="collectInput",
        action_type="collectInput",
        parameters={"inputType": "DTMF", "maxDigits": 1, "timeout": 5000}
    )
    entry = FlowModuleDocumentContentEntry(entry_type="action", action=entry_action)
    content = FlowModuleDocumentContent(entries={"start": entry})
    document = FlowModuleDocument(content=content)
    
    request = FlowModuleCreateRequest(
        name=name,
        description="Shared DTMF collector",
        document=document,
        type="module",
        version=1
    )
    
    try:
        resp = flow_api.post_flow_module(body=request)
        return resp.id
    except ApiException as e:
        print(f"Failed to create module: {e.body}")
        sys.exit(1)

def create_inbound_flow(flow_api: FlowApi, name: str, module_id: str) -> str:
    # Define invoke action
    invoke_action = FlowDocumentContentEntryAction(
        action="invokeSubFlow",
        action_type="invokeSubFlow",
        parameters={"moduleId": module_id, "timeout": 30000}
    )
    
    end_action = FlowDocumentContentEntryAction(action="end", action_type="end")
    
    invoke_entry = FlowDocumentContentEntry(
        entry_type="action",
        action=invoke_action,
        nexts={"default": "end_step"}
    )
    
    end_entry = FlowDocumentContentEntry(entry_type="action", action=end_action)
    
    content = FlowDocumentContent(entries={"start": invoke_entry, "end_step": end_entry})
    document = FlowDocument(content=content)
    
    request = FlowCreateRequest(
        name=name,
        description=f"Flow invoking module {module_id}",
        document=document,
        type="voice",
        enabled=True,
        version=1
    )
    
    try:
        resp = flow_api.post_flow(body=request)
        return resp.id
    except ApiException as e:
        print(f"Failed to create flow: {e.body}")
        sys.exit(1)

def main():
    print("Initializing Genesys Cloud Client...")
    client = init_client()
    flow_api = FlowApi(client)
    
    # Step 1: Create Shared Module
    module_name = "SharedDTMFCollector"
    print(f"Creating shared module: {module_name}")
    module_id = create_shared_module(flow_api, module_name)
    print(f"Module ID: {module_id}")
    
    # Step 2: Create Multiple Inbound Flows Reusing the Module
    flow_ids = []
    for i in range(1, 3):
        flow_name = f"InboundFlow_{i}"
        print(f"Creating flow: {flow_name}")
        flow_id = create_inbound_flow(flow_api, flow_name, module_id)
        flow_ids.append(flow_id)
        
    # Step 3: Verify References
    for fid in flow_ids:
        print(f"Verifying flow {fid}...")
        try:
            flow = flow_api.get_flow(flow_id=fid, expand=["document"])
            content = flow.document.content
            # Simple check if moduleId is present in the JSON dump of the document
            doc_str = str(content.entries)
            if module_id in doc_str:
                print(f"Success: Flow {fid} references module {module_id}")
            else:
                print(f"Warning: Flow {fid} may not reference module correctly")
        except ApiException as e:
            print(f"Verification failed for {fid}: {e.body}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 400 Bad Request - “Invalid Flow Document”

What causes it: The JSON structure of the FlowDocument or FlowModuleDocument does not match the schema. Common issues include missing nexts keys, incorrect action_type values, or referencing a non-existent moduleId.

How to fix it:

1. Ensure all action_type strings match the Flow Action Reference.
2. Verify that every nexts key in an entry corresponds to a valid key in the entries dictionary, except for terminal actions like end.
3. Check that the moduleId in invokeSubFlow is a valid UUID of a module in the same organization.

Code Fix:

# Ensure the 'nexts' dictionary keys match the entry IDs
invoke_entry = FlowDocumentContentEntry(
    entry_type="action",
    action=invoke_action,
    nexts={"default": "end_step"} # "end_step" must exist in content.entries
)

Error: 403 Forbidden - “Insufficient Permissions”

What causes it: The OAuth token used does not have the flow:flow:write or flow:module:write scopes.

How to fix it:

1. Go to the Genesys Cloud Admin UI → Security → OAuth.
2. Edit the client associated with your private key.
3. Ensure flow:flow:write and flow:module:write are checked.
4. Re-generate the private key if necessary and update your environment variables.

Error: 409 Conflict - “Resource Already Exists”

What causes it: You attempted to create a module or flow with a name that already exists in the organization. Genesys Cloud enforces unique names for flows and modules within an org.

How to fix it:

1. Append a unique suffix (e.g., timestamp or UUID) to the name during creation.
2. Or, first search for existing flows/modules and update them instead of creating new ones.

Code Fix:

import time
unique_name = f"SharedModule_{int(time.time())}"
module_id = create_shared_module(flow_api, unique_name)

Official References

• Genesys Cloud Flow API Documentation
• Genesys Cloud Flow Module API
• Flow Actions Reference
• Python SDK Reference