Common Module Reuse — How to Call a Shared Flow from Multiple Inbound Call Flows

What You Will Build

• You will create a reusable “Shared Flow” in Genesys Cloud CX that handles common logic, such as authentication or menu routing, and invoke it from multiple distinct inbound call flows using the API.
• This tutorial uses the Genesys Cloud Platform API v2 and the Python SDK (genesys-cloud-sdk).
• The implementation covers creating a shared flow, updating existing inbound flows to reference it via the CallSubflow action, and verifying the linkage through API inspection.

Prerequisites

• OAuth Client Type: A Genesys Cloud OAuth client with the Client Credentials grant type.
• Required Scopes:
  • flow:write (to create and update flows)
  • flow:read (to inspect flows)
  • organization:read (optional, for debugging context)
• SDK Version: genesys-cloud-sdk version 130.0.0 or higher.
• Language/Runtime: Python 3.9+
• External Dependencies:

  pip install genesys-cloud-sdk
  

Authentication Setup

Genesys Cloud uses OAuth 2.0 for API authentication. For server-to-server integrations like this, the Client Credentials flow is the standard approach. You must obtain an access token before making any API calls.

import os
from purecloud_platform_client import (
    Configuration,
    PureCloudPlatformClientV2,
    ApiClient,
    OAuthClient,
    ApiException
)

def get_purecloud_client() -> PureCloudPlatformClientV2:
    """
    Initializes and authenticates the Genesys Cloud Platform Client.
    Uses environment variables for credentials.
    """
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    region = os.getenv("GENESYS_REGION", "mypurecloud.com")  # e.g., usw2.pure.cloud

    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set in environment.")

    # Configure OAuth
    oauth = OAuthClient(client_id=client_id, client_secret=client_secret)
    
    # Initialize the platform client
    client = PureCloudPlatformClientV2(oauth_client=oauth)
    
    # Force authentication to ensure token is valid
    try:
        client.auth_api.get_token()
    except ApiException as e:
        print(f"Authentication failed: {e.status} - {e.reason}")
        raise

    return client

# Initialize client
platform_client = get_purecloud_client()

Implementation

Step 1: Create a Reusable Shared Flow

A “Shared Flow” in Genesys Cloud is a standard flow with a specific type (shared). It does not have an associated application (like Telephony or Chat) directly attached in the same way an inbound flow does. Instead, it acts as a library of actions.

We will create a simple shared flow named Common_Menu_Logic. This flow will contain a single SetUserAttribute action to simulate processing a common task.

Required Scope: flow:write

from purecloud_platform_client import (
    FlowsApi,
    CreateFlowRequest,
    FlowAction,
    SetUserAttributeAction,
    FlowDocument
)

def create_shared_flow(client: PureCloudPlatformClientV2) -> str:
    """
    Creates a new Shared Flow and returns its ID.
    """
    flows_api = FlowsApi(client)

    # Define the action that will run inside the shared flow
    # In a real scenario, this might be a GatherInput, RouteToQueue, or API Call
    set_attr_action = SetUserAttributeAction(
        name="Set Common Attribute",
        id="action-1",
        type="SetUserAttribute",
        variable="CommonProcessed",
        value="True",
        on_error={"nextActionId": "end"},
        on_complete={"nextActionId": "end"}
    )

    # Define the end action
    end_action = FlowAction(
        name="End",
        id="end",
        type="End",
        on_error={},
        on_complete={}
    )

    # Construct the document with actions
    document = FlowDocument(
        actions=[set_attr_action, end_action],
        start_action_id="action-1"
    )

    # Prepare the create request
    create_req = CreateFlowRequest(
        name="Common_Menu_Logic",
        description="A shared flow for common menu processing",
        flow_type="shared",  # Critical: Must be 'shared' to be reusable
        document=document
    )

    try:
        response = flows_api.post_flow(body=create_req)
        print(f"Shared Flow Created: ID={response.id}, Name={response.name}")
        return response.id
    except ApiException as e:
        print(f"Failed to create shared flow: {e.status} - {e.body}")
        raise

# Create the shared flow
SHARED_FLOW_ID = create_shared_flow(platform_client)

Step 2: Create an Inbound Call Flow That Calls the Shared Flow

Now we create an inbound telephony flow. Instead of duplicating the logic, we will use the CallSubflow action to invoke the Common_Menu_Logic shared flow we just created.

Key Concept: The CallSubflow action requires the flowId of the target shared flow. It also requires on_complete and on_error handlers to define where the execution continues after the shared flow returns.

Required Scope: flow:write

from purecloud_platform_client import (
    CreateFlowRequest,
    FlowAction,
    CallSubflowAction,
    SetUserAttributeAction,
    EndAction,
    FlowDocument
)

def create_inbound_flow_with_subflow(client: PureCloudPlatformClientV2, shared_flow_id: str) -> str:
    """
    Creates an inbound telephony flow that calls the specified shared flow.
    """
    flows_api = FlowsApi(client)

    # Action 1: Call the shared flow
    call_subflow = CallSubflowAction(
        name="Call Common Menu",
        id="subflow-call",
        type="CallSubflow",
        flow_id=shared_flow_id,  # Reference the shared flow ID
        on_complete={"nextActionId": "post-subflow-action"},
        on_error={"nextActionId": "error-handler"}
    )

    # Action 2: Post-subflow logic (runs after shared flow completes)
    post_action = SetUserAttributeAction(
        name="Log Completion",
        id="post-subflow-action",
        type="SetUserAttribute",
        variable="SubflowExecuted",
        value="Yes",
        on_error={"nextActionId": "end"},
        on_complete={"nextActionId": "end"}
    )

    # Action 3: Error Handler
    error_handler = SetUserAttributeAction(
        name="Handle Subflow Error",
        id="error-handler",
        type="SetUserAttribute",
        variable="SubflowError",
        value="True",
        on_error={"nextActionId": "end"},
        on_complete={"nextActionId": "end"}
    )

    # Action 4: End
    end_action = EndAction(
        name="End",
        id="end"
    )

    # Construct document
    document = FlowDocument(
        actions=[call_subflow, post_action, error_handler, end_action],
        start_action_id="subflow-call"
    )

    create_req = CreateFlowRequest(
        name="Inbound Flow A",
        description="Inbound flow using shared logic",
        flow_type="telephony",
        document=document,
        application={
            "id": None, # For new flows, application ID is often set via environment or explicitly if known. 
                        # For this tutorial, we assume default environment handling or that the user knows their app ID.
                        # In production, you typically query the Application ID first.
        }
    )
    
    # Note: For Telephony flows, you usually need to specify the application. 
    # To keep this script runnable without complex app-lookup logic, we will assume 
    # the flow is created in a draft state or associated with a known app ID in production.
    # Here we omit application ID for brevity, but in real code, you must provide it.
    
    try:
        response = flows_api.post_flow(body=create_req)
        print(f"Inbound Flow Created: ID={response.id}")
        return response.id
    except ApiException as e:
        print(f"Failed to create inbound flow: {e.status} - {e.body}")
        raise

# Create the first inbound flow
INBOUND_FLOW_A_ID = create_inbound_flow_with_subflow(platform_client, SHARED_FLOW_ID)

Step 3: Reuse the Same Shared Flow in a Second Inbound Flow

To demonstrate true reuse, we create a second inbound flow (Inbound Flow B) that calls the same SHARED_FLOW_ID. This confirms that multiple parent flows can reference a single child flow.

def create_second_inbound_flow(client: PureCloudPlatformClientV2, shared_flow_id: str) -> str:
    """
    Creates a second inbound flow referencing the same shared flow.
    """
    flows_api = FlowsApi(client)

    # Action: Call the same shared flow
    call_subflow = CallSubflowAction(
        name="Call Common Menu",
        id="subflow-call",
        type="CallSubflow",
        flow_id=shared_flow_id,
        on_complete={"nextActionId": "end"},
        on_error={"nextActionId": "end"}
    )

    end_action = EndAction(
        name="End",
        id="end"
    )

    document = FlowDocument(
        actions=[call_subflow, end_action],
        start_action_id="subflow-call"
    )

    create_req = CreateFlowRequest(
        name="Inbound Flow B",
        description="Second inbound flow using same shared logic",
        flow_type="telephony",
        document=document
    )

    try:
        response = flows_api.post_flow(body=create_req)
        print(f"Second Inbound Flow Created: ID={response.id}")
        return response.id
    except ApiException as e:
        print(f"Failed to create second inbound flow: {e.status} - {e.body}")
        raise

# Create the second inbound flow
INBOUND_FLOW_B_ID = create_second_inbound_flow(platform_client, SHARED_FLOW_ID)

Step 4: Verify the Linkage via API

It is crucial to verify that the flows are correctly linked. We will fetch the flow definition of Inbound Flow A and inspect the CallSubflow action to ensure it points to SHARED_FLOW_ID.

Required Scope: flow:read

def verify_flow_linkage(client: PureCloudPlatformClientV2, flow_id: str) -> bool:
    """
    Retrieves a flow and verifies if it contains a CallSubflow action pointing to a shared flow.
    """
    flows_api = FlowsApi(client)
    
    try:
        # Expand 'document' to get the full flow definition
        response = flows_api.get_flow(id=flow_id, expand=["document"])
        
        if not response.document or not response.document.actions:
            print(f"Flow {flow_id} has no document or actions.")
            return False

        # Iterate through actions to find CallSubflow
        for action in response.document.actions:
            if action.type == "CallSubflow":
                call_subflow = action  # Type hinting helps here if using mypy
                if hasattr(call_subflow, 'flow_id') and call_subflow.flow_id:
                    print(f"Verified: Flow {flow_id} calls shared flow ID: {call_subflow.flow_id}")
                    return True
        
        print(f"Flow {flow_id} does not contain a CallSubflow action.")
        return False

    except ApiException as e:
        print(f"Failed to verify flow: {e.status} - {e.reason}")
        return False

# Verify both inbound flows
verify_flow_linkage(platform_client, INBOUND_FLOW_A_ID)
verify_flow_linkage(platform_client, INBOUND_FLOW_B_ID)

Complete Working Example

Below is the consolidated script. Save this as shared_flow_reuse.py. Ensure your environment variables are set.

import os
import sys
from purecloud_platform_client import (
    Configuration,
    PureCloudPlatformClientV2,
    ApiClient,
    OAuthClient,
    ApiException,
    FlowsApi,
    CreateFlowRequest,
    FlowAction,
    SetUserAttributeAction,
    CallSubflowAction,
    EndAction,
    FlowDocument
)

def get_purecloud_client() -> PureCloudPlatformClientV2:
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    
    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set in environment.")

    oauth = OAuthClient(client_id=client_id, client_secret=client_secret)
    client = PureCloudPlatformClientV2(oauth_client=oauth)
    
    try:
        client.auth_api.get_token()
    except ApiException as e:
        print(f"Authentication failed: {e.status} - {e.reason}")
        sys.exit(1)
        
    return client

def create_shared_flow(client: PureCloudPlatformClientV2) -> str:
    flows_api = FlowsApi(client)
    
    set_attr_action = SetUserAttributeAction(
        name="Set Common Attribute",
        id="action-1",
        type="SetUserAttribute",
        variable="CommonProcessed",
        value="True",
        on_error={"nextActionId": "end"},
        on_complete={"nextActionId": "end"}
    )

    end_action = EndAction(name="End", id="end")

    document = FlowDocument(
        actions=[set_attr_action, end_action],
        start_action_id="action-1"
    )

    create_req = CreateFlowRequest(
        name="Common_Menu_Logic",
        description="A shared flow for common menu processing",
        flow_type="shared",
        document=document
    )

    try:
        response = flows_api.post_flow(body=create_req)
        print(f"[OK] Shared Flow Created: ID={response.id}")
        return response.id
    except ApiException as e:
        print(f"[ERROR] Failed to create shared flow: {e.status} - {e.body}")
        sys.exit(1)

def create_inbound_flow(client: PureCloudPlatformClientV2, shared_flow_id: str, flow_name: str) -> str:
    flows_api = FlowsApi(client)
    
    call_subflow = CallSubflowAction(
        name="Call Common Menu",
        id="subflow-call",
        type="CallSubflow",
        flow_id=shared_flow_id,
        on_complete={"nextActionId": "end"},
        on_error={"nextActionId": "end"}
    )

    end_action = EndAction(name="End", id="end")

    document = FlowDocument(
        actions=[call_subflow, end_action],
        start_action_id="subflow-call"
    )

    create_req = CreateFlowRequest(
        name=flow_name,
        description=f"Inbound flow calling shared flow {shared_flow_id}",
        flow_type="telephony",
        document=document
    )

    try:
        response = flows_api.post_flow(body=create_req)
        print(f"[OK] Inbound Flow '{flow_name}' Created: ID={response.id}")
        return response.id
    except ApiException as e:
        print(f"[ERROR] Failed to create inbound flow: {e.status} - {e.body}")
        sys.exit(1)

def verify_flow_linkage(client: PureCloudPlatformClientV2, flow_id: str) -> None:
    flows_api = FlowsApi(client)
    
    try:
        response = flows_api.get_flow(id=flow_id, expand=["document"])
        
        if not response.document or not response.document.actions:
            print(f"[WARN] Flow {flow_id} has no document or actions.")
            return

        for action in response.document.actions:
            if action.type == "CallSubflow":
                if hasattr(action, 'flow_id') and action.flow_id:
                    print(f"[VERIFIED] Flow {flow_id} correctly links to Shared Flow: {action.flow_id}")
                    return
        
        print(f"[WARN] Flow {flow_id} does not contain a CallSubflow action.")

    except ApiException as e:
        print(f"[ERROR] Failed to verify flow: {e.status} - {e.reason}")

def main():
    print("--- Genesys Cloud Shared Flow Reuse Tutorial ---")
    
    # 1. Authenticate
    client = get_purecloud_client()
    
    # 2. Create Shared Flow
    shared_flow_id = create_shared_flow(client)
    
    # 3. Create Multiple Inbound Flows referencing the Shared Flow
    flow_a_id = create_inbound_flow(client, shared_flow_id, "Inbound Flow A")
    flow_b_id = create_inbound_flow(client, shared_flow_id, "Inbound Flow B")
    
    # 4. Verify Linkage
    print("\n--- Verification ---")
    verify_flow_linkage(client, flow_a_id)
    verify_flow_linkage(client, flow_b_id)
    
    print("\n--- Tutorial Complete ---")
    print(f"Shared Flow ID: {shared_flow_id}")
    print(f"Inbound Flow A ID: {flow_a_id}")
    print(f"Inbound Flow B ID: {flow_b_id}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 400 Bad Request - “Flow type must be ‘shared’ for reusable flows”

• What causes it: You attempted to create a flow with flow_type="telephony" or flow_type="webchat" and then tried to reference it as a shared flow, or you tried to set flow_type="shared" but included application-specific properties that are invalid for shared flows.
• How to fix it: Ensure the CreateFlowRequest for the reusable logic explicitly sets flow_type="shared". Do not attach an application object to shared flows.

Error: 400 Bad Request - “Invalid flow ID in CallSubflow action”

• What causes it: The flow_id parameter in the CallSubflow action points to a non-existent flow, or a flow that is not of type shared.
• How to fix it: Verify that the SHARED_FLOW_ID used in create_inbound_flow matches the ID returned from create_shared_flow. Ensure the target flow is actually published (or at least saved) before referencing it.

Error: 403 Forbidden - “Insufficient permissions”

• What causes it: The OAuth client lacks the flow:write or flow:read scopes.
• How to fix it: Go to the Genesys Cloud Admin Console → Platform → OAuth Clients. Edit your client and ensure the following scopes are checked:
  • flow:write
  • flow:read
  • flow:delete (if you plan to clean up)

Error: 429 Too Many Requests

• What causes it: You are creating or querying flows too rapidly. Genesys Cloud APIs have rate limits.
• How to fix it: Implement exponential backoff. In Python, you can use a library like tenacity or manually sleep between requests.

import time

def safe_api_call(func, *args, retries=3, backoff_factor=2):
    for attempt in range(retries):
        try:
            return func(*args)
        except ApiException as e:
            if e.status == 429:
                wait_time = backoff_factor ** attempt
                print(f"Rate limited. Waiting {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise
    raise Exception("Max retries exceeded")

Official References

• Genesys Cloud Flows API Documentation
• Create Flow Endpoint
• CallSubflow Action Documentation
• Python SDK Reference