How to Call a Shared Flow from Multiple Inbound Call Flows in Genesys Cloud CX

What You Will Build

• This tutorial demonstrates how to configure a reusable “Common Module” (a Shared Flow) and invoke it from multiple distinct inbound call flows using the Genesys Cloud CX API.
• The solution uses the Genesys Cloud CX Flow API (/api/v2/fm/flows) and the Python SDK (genesys-cloud-sdk-python) to programmatically create a shared flow and update existing flows to reference it.
• The programming language used is Python 3.9+.

Prerequisites

• OAuth Client: A Genesys Cloud CX OAuth Client with the private type.
• Required Scopes:
  • flow:flow:read (to inspect existing flows)
  • flow:flow:write (to create or update flows)
  • flow:flow:delete (optional, for cleanup)
  • organization:read (to resolve organization ID if not hardcoded)
• SDK Version: genesys-cloud-sdk-python version 139.0.0 or later.
• Runtime: Python 3.9 or later.
• Dependencies:

  pip install genesys-cloud-sdk-python
  

Authentication Setup

Genesys Cloud CX uses OAuth 2.0 for API authentication. The most robust method for server-to-server integration is the Client Credentials flow. The SDK handles token caching and refresh automatically if configured correctly.

import os
import sys
from purecloud_platform_client import Configuration, PureCloudAuthFlow, ApiClient, FlowApi

def get_purecloud_api_client() -> ApiClient:
    """
    Initialize and return a configured PureCloud API Client.
    Uses environment variables for credentials.
    """
    # Retrieve credentials from environment variables
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    env_url = os.getenv("GENESYS_ENV_URL", "https://api.mypurecloud.com")

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

    # Configure the SDK client
    config = Configuration(
        host=env_url,
        client_id=client_id,
        client_secret=client_secret,
        oauth_client_type="private",
        oauth_scopes=[
            "flow:flow:read",
            "flow:flow:write",
            "flow:flow:delete"
        ]
    )

    # Create the API client instance
    # The SDK manages the OAuth token lifecycle automatically
    api_client = ApiClient(configuration=config)
    return api_client

Implementation

Step 1: Create the Shared “Common Module” Flow

In Genesys Cloud CX, a “Common Module” is technically a Flow with the type set to common. This flow does not have an endpoint itself; it is designed to be called by other flows.

We will create a simple shared flow that performs two actions:

1. Adds a tag common-module-executed to the interaction.
2. Plays a brief audio message (using a standard system prompt or a custom recording ID).

Note: For this example, we will use a generic PlayPrompt action with a placeholder prompt ID. In production, you would replace PROMPT_ID with a valid prompt ID from your organization.

from purecloud_platform_client import FlowApi, Flow, FlowAction, FlowPlayPromptAction, FlowTagInteractionAction
from purecloud_platform_client import FlowPromptRef, FlowPromptTextRef

def create_shared_flow(api_client: ApiClient, org_id: str) -> str:
    """
    Creates a Common Type flow that tags the interaction.
    Returns the Flow ID.
    """
    flow_api = FlowApi(api_client)

    # Define the actions for the shared flow
    # Action 1: Tag the interaction
    tag_action = FlowTagInteractionAction(
        name="TagInteraction",
        tags=["common-module-executed", "shared-flow-test"]
    )

    # Action 2: Play a prompt (Text-to-Speech for simplicity)
    # In a real scenario, you might use a recording ID
    play_prompt_action = FlowPlayPromptAction(
        name="PlaySharedMessage",
        prompt=FlowPromptTextRef(
            text="This is a message from the shared common module.",
            locale="en-US"
        )
    )

    # Define the flow structure
    # A common flow typically has a single entry point and a sequence of actions
    flow_definition = Flow(
        name="Shared Common Module - Tag and Greet",
        description="A reusable common module that tags interactions and plays a greeting.",
        type="common",  # CRITICAL: Sets the flow type to Common Module
        active=True,
        actions=[
            tag_action,
            play_prompt_action
        ],
        # Common flows do not have endpoints, but they do have a definition structure
        # The SDK handles the internal node linking for simple linear flows
    )

    try:
        # Post the flow
        # Note: The SDK post_flow method requires a Flow object
        created_flow = flow_api.post_flow(body=flow_definition)
        print(f"Successfully created shared flow with ID: {created_flow.id}")
        return created_flow.id
    except Exception as e:
        print(f"Error creating shared flow: {e}")
        raise

Step 2: Identify Inbound Flows to Update

We need to find existing inbound call flows to which we will add the call to the shared module. In a production script, you would filter by name, tags, or specific criteria. For this tutorial, we will retrieve all active inbound flows and select the first one for demonstration.

def get_active_inbound_flows(api_client: ApiClient, limit: int = 10) -> list:
    """
    Retrieves a list of active inbound flows.
    """
    flow_api = FlowApi(api_client)
    
    try:
        # Query for active inbound flows
        # The API returns a paged response
        response = flow_api.get_flows(
            type="inbound",
            active="true",
            page_size=limit,
            page_number=1
        )
        
        if response.entities:
            return response.entities
        else:
            print("No active inbound flows found.")
            return []
    except Exception as e:
        print(f"Error retrieving inbound flows: {e}")
        return []

Step 3: Update Inbound Flow to Call the Shared Module

This is the core integration step. We must modify an existing inbound flow to include a CallFlow action that points to the shared flow ID created in Step 1.

Important: Modifying a flow requires updating the entire flow definition. You cannot patch individual actions via the REST API directly in all SDK versions; it is safer to retrieve the flow, modify the object in memory, and post the updated object back.

Logic:

1. Retrieve the inbound flow.
2. Create a CallFlow action referencing the shared flow ID.
3. Insert this action at the beginning of the flow’s action list (or at a specific node if the flow has complex branching).
4. Post the updated flow back to Genesys.

from purecloud_platform_client import FlowCallFlowAction

def add_shared_flow_call_to_inbound_flow(api_client: ApiClient, inbound_flow_id: str, shared_flow_id: str) -> bool:
    """
    Updates an inbound flow to call the shared common module at the start.
    """
    flow_api = FlowApi(api_client)

    try:
        # 1. Get the existing inbound flow
        existing_flow = flow_api.get_flow(flow_id=inbound_flow_id)
        
        # 2. Create the CallFlow action
        call_shared_action = FlowCallFlowAction(
            name="CallSharedCommonModule",
            flow=FlowPromptRef( # Note: SDK typing for flow ref often uses PromptRef structure for ID/Version
                id=shared_flow_id,
                version=existing_flow.version # It is best practice to use the current version of the target flow if known, or None for latest
            )
        )

        # 3. Insert the new action at the beginning of the list
        # Note: This assumes a linear flow structure for simplicity.
        # In complex flows with nodes/edges, you would need to manipulate the 'nodes' and 'edges' arrays.
        # For this tutorial, we assume the 'actions' list is used (simplified linear flow).
        
        if not hasattr(existing_flow, 'actions') or existing_flow.actions is None:
            print("Warning: Inbound flow does not have a simple actions list. Skipping update.")
            return False

        # Prepend the call action
        existing_flow.actions.insert(0, call_shared_action)

        # 4. Post the updated flow
        # The SDK handles the PUT request with the updated version
        updated_flow = flow_api.put_flow(
            flow_id=inbound_flow_id,
            body=existing_flow
        )

        print(f"Successfully updated inbound flow '{updated_flow.name}' to call shared flow.")
        return True

    except Exception as e:
        print(f"Error updating inbound flow {inbound_flow_id}: {e}")
        return False

Complete Working Example

The following script combines all steps. It authenticates, creates a shared common module, finds an active inbound flow, and updates that inbound flow to call the shared module.

import os
import sys
from purecloud_platform_client import Configuration, ApiClient, FlowApi
from purecloud_platform_client import Flow, FlowAction, FlowPlayPromptAction, FlowTagInteractionAction
from purecloud_platform_client import FlowPromptTextRef, FlowCallFlowAction, FlowPromptRef

def get_purecloud_api_client() -> ApiClient:
    """Initialize and return a configured PureCloud API Client."""
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    env_url = os.getenv("GENESYS_ENV_URL", "https://api.mypurecloud.com")

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

    config = Configuration(
        host=env_url,
        client_id=client_id,
        client_secret=client_secret,
        oauth_client_type="private",
        oauth_scopes=[
            "flow:flow:read",
            "flow:flow:write"
        ]
    )
    return ApiClient(configuration=config)

def create_shared_flow(api_client: ApiClient) -> str:
    """Creates a Common Type flow that tags the interaction."""
    flow_api = FlowApi(api_client)
    
    # Define actions
    tag_action = FlowTagInteractionAction(
        name="TagInteraction",
        tags=["common-module-executed"]
    )
    
    play_prompt_action = FlowPlayPromptAction(
        name="PlaySharedMessage",
        prompt=FlowPromptTextRef(
            text="This is a message from the shared common module.",
            locale="en-US"
        )
    )

    # Define flow
    flow_definition = Flow(
        name="Shared Common Module - Tag and Greet",
        description="A reusable common module.",
        type="common",
        active=True,
        actions=[tag_action, play_prompt_action]
    )

    try:
        created_flow = flow_api.post_flow(body=flow_definition)
        return created_flow.id
    except Exception as e:
        print(f"Error creating shared flow: {e}")
        sys.exit(1)

def get_first_active_inbound_flow(api_client: ApiClient) -> str:
    """Retrieves the ID of the first active inbound flow."""
    flow_api = FlowApi(api_client)
    try:
        response = flow_api.get_flows(type="inbound", active="true", page_size=1)
        if response.entities and len(response.entities) > 0:
            return response.entities[0].id
        else:
            print("No active inbound flows found.")
            sys.exit(1)
    except Exception as e:
        print(f"Error retrieving flows: {e}")
        sys.exit(1)

def update_inbound_flow_with_shared_call(api_client: ApiClient, inbound_flow_id: str, shared_flow_id: str) -> None:
    """Updates an inbound flow to call the shared common module."""
    flow_api = FlowApi(api_client)
    
    try:
        # Get existing flow
        existing_flow = flow_api.get_flow(flow_id=inbound_flow_id)
        
        # Create CallFlow action
        call_shared_action = FlowCallFlowAction(
            name="CallSharedCommonModule",
            flow=FlowPromptRef(
                id=shared_flow_id,
                version=None # Use None to always call the latest active version
            )
        )

        # Check if flow has actions list (linear flow assumption)
        if not hasattr(existing_flow, 'actions') or existing_flow.actions is None:
            print("This flow does not use a simple actions list. Manual node manipulation required.")
            return

        # Insert at the beginning
        existing_flow.actions.insert(0, call_shared_action)

        # Update flow
        flow_api.put_flow(flow_id=inbound_flow_id, body=existing_flow)
        print(f"Successfully updated flow '{existing_flow.name}' (ID: {inbound_flow_id})")
        
    except Exception as e:
        print(f"Error updating flow: {e}")
        raise

def main():
    # 1. Authenticate
    print("Authenticating...")
    api_client = get_purecloud_api_client()
    
    # 2. Create Shared Flow
    print("Creating shared common module...")
    shared_flow_id = create_shared_flow(api_client)
    print(f"Shared Flow ID: {shared_flow_id}")
    
    # 3. Get Target Inbound Flow
    print("Finding active inbound flow...")
    inbound_flow_id = get_first_active_inbound_flow(api_client)
    print(f"Target Inbound Flow ID: {inbound_flow_id}")
    
    # 4. Update Inbound Flow
    print("Updating inbound flow to call shared module...")
    update_inbound_flow_with_shared_call(api_client, inbound_flow_id, shared_flow_id)
    
    print("Done.")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden or Insufficient Permissions

Cause: The OAuth client lacks the required scopes.
Fix: Ensure your OAuth client has flow:flow:write and flow:flow:read. If you are using the SDK, verify the oauth_scopes list in the Configuration object.

Code Check:

oauth_scopes=[
    "flow:flow:read",
    "flow:flow:write"
]

Error: 409 Conflict or Version Mismatch

Cause: When updating a flow via put_flow, the SDK sends the current version of the flow. If another user or script modified the flow between your get_flow and put_flow calls, the version number will be stale.
Fix: Implement a retry loop that re-fetches the flow before attempting the update again.

Code Fix:

import time

def update_with_retry(api_client, inbound_flow_id, shared_flow_id, max_retries=3):
    for attempt in range(max_retries):
        try:
            update_inbound_flow_with_shared_call(api_client, inbound_flow_id, shared_flow_id)
            return
        except Exception as e:
            if "version" in str(e).lower() and attempt < max_retries - 1:
                print(f"Version conflict. Retrying ({attempt + 1}/{max_retries})...")
                time.sleep(1)
                continue
            raise e

Error: 500 Internal Server Error or Invalid Flow Definition

Cause: The flow definition is structurally invalid. Common issues include:

• Missing type field (must be common for shared modules).
• Referencing a non-existent prompt ID.
• Circular references in complex node-based flows.

Fix: Validate the JSON structure before posting. Use the Genesys Cloud CX Flow Builder UI to export a working flow as JSON to compare against your SDK-generated object. Ensure the FlowPlayPromptAction uses a valid FlowPromptRef or FlowPromptTextRef.

Error: CallFlow Action Not Executing

Cause: The inbound flow is not configured to wait for the common module to complete, or the common module has an error.
Fix:

1. Check the Interaction Tags in the Genesys Cloud CX Admin Console. If common-module-executed is present, the call succeeded.
2. If the tag is missing, check the Flow Diagnostics for the inbound flow.
3. Ensure the common module is active=True. If a common module is deactivated, calls to it may fail or be skipped depending on the inbound flow’s error handling configuration.

Official References

• Genesys Cloud CX Flow API Documentation
• Python SDK Reference: FlowApi
• Understanding Common Modules in Genesys Cloud CX
• OAuth Scopes Reference