Master Flow Reuse: Calling Shared Flows from Multiple Inbound Call Flows

What You Will Build

• One sentence: The code constructs a Genesys Cloud CX configuration where a single “Common” flow is invoked by multiple distinct inbound call flows using the Subflow node.
• One sentence: This tutorial uses the Genesys Cloud CX REST API v2 via the Python SDK (purecloud-platform-client-v2).
• One sentence: The programming language covered is Python 3.9+.

Prerequisites

• OAuth Client Type: Service Account with offline_access scope.
• Required Scopes:
  • flow:flow:read (to read existing flows for reference, if needed)
  • flow:flow:write (to create or update flows)
  • flow:flow:admin (optional, but recommended for managing flow versions)
• SDK Version: purecloud-platform-client-v2 >= 149.0.0 (Ensure you are on a recent version to support the latest Subflow node properties).
• Language/Runtime: Python 3.9 or higher.
• External Dependencies:

  pip install purecloud-platform-client-v2
  

Authentication Setup

Genesys Cloud uses OAuth 2.0. For server-to-server integrations, you must use the Service Account flow. The following code demonstrates how to initialize the SDK client with a valid access token.

Note: In production, implement token caching. The AccessToken object returned by the SDK handles refresh tokens automatically if the offline_access scope is granted during client creation.

import os
from purecloud_platform_client import Configuration, ApiClient, FlowApi, AuthenticationApi
from purecloud_platform_client.rest import ApiException

def get_api_client() -> ApiClient:
    """
    Initializes and returns a configured ApiClient instance.
    Assumes environment variables are set:
    - GENESYS_CLOUD_ENVIRONMENT (e.g., 'mypurecloud.com')
    - GENESYS_CLOUD_CLIENT_ID
    - GENESYS_CLOUD_CLIENT_SECRET
    """
    environment = os.getenv("GENESYS_CLOUD_ENVIRONMENT")
    client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")

    if not all([environment, client_id, client_secret]):
        raise ValueError("Missing required environment variables for Genesys Cloud authentication.")

    # Configure the SDK
    configuration = Configuration(
        base_url=f"https://{environment}",
        client_id=client_id,
        client_secret=client_secret,
        scopes=["flow:flow:write", "flow:flow:read"]
    )

    api_client = ApiClient(configuration)
    
    # Validate connection by fetching current user info (optional but good for debugging)
    try:
        auth_api = AuthenticationApi(api_client)
        # This forces a token fetch if none exists
        auth_api.post_oauth2token() 
    except ApiException as e:
        print(f"Authentication failed: {e.status} {e.reason}")
        raise

    return api_client

Implementation

Step 1: Create the “Common” Shared Flow

Before you can call a shared flow, it must exist. We will create a simple flow named “Common Error Handler” that plays an error message and disconnects. This flow will serve as the target for our Subflow calls.

Key Concept: A shared flow is just a standard flow with the type set to generic or call. The distinction lies in how it is invoked. We will use type: generic for this example as it is agnostic of interaction type.

from purecloud_platform_client import (
    FlowApi,
    Flow,
    FlowType,
    FlowNode,
    PlayMessageNode,
    DisconnectNode,
    FlowNodeOut,
    FlowNodeIn
)

def create_common_flow(api_client: ApiClient, org_id: str) -> str:
    """
    Creates a generic flow that plays an error message and disconnects.
    Returns the Flow ID.
    """
    flow_api = FlowApi(api_client)

    # 1. Define the Play Message Node
    play_node = PlayMessageNode(
        name="Play Error Message",
        message_type="text",
        message_text="An error has occurred. Please try again later.",
        # In production, use 'media' and reference a valid Media ID
        # message_type="media",
        # message_id="VALID_MEDIA_ID"
    )

    # 2. Define the Disconnect Node
    disconnect_node = DisconnectNode(
        name="Disconnect",
        disposition="busy"
    )

    # 3. Define the Start Node (Implicit in flow creation, but we define connections)
    # The start node is always "Start". We need to connect it to our first node.
    
    # 4. Assemble the Flow Object
    flow = Flow(
        name="Common Error Handler",
        description="Shared flow for handling generic errors.",
        type="generic",
        version=1,
        nodes={
            "Start": FlowNode(
                name="Start",
                node_type="Start",
                out={
                    "default": FlowNodeOut(
                        node_id="PlayError",
                        transition_type="conditional"
                    )
                }
            ),
            "PlayError": FlowNode(
                name="Play Error Message",
                node_type="PlayMessage",
                node_data=play_node,
                out={
                    "completed": FlowNodeOut(
                        node_id="Disconnect",
                        transition_type="conditional"
                    ),
                    "error": FlowNodeOut(
                        node_id="Disconnect",
                        transition_type="conditional"
                    )
                }
            ),
            "Disconnect": FlowNode(
                name="Disconnect",
                node_type="Disconnect",
                node_data=disconnect_node
            )
        }
    )

    try:
        # Create the flow
        # Note: post_flow requires the body to be a Flow object
        response = flow_api.post_flow(body=flow)
        print(f"Created Common Flow with ID: {response.id}")
        return response.id
    except ApiException as e:
        print(f"Failed to create flow: {e.status} {e.reason}")
        raise

Step 2: Create the “Subflow” Node Configuration

The core of module reuse is the Subflow node. This node pauses the current flow, executes the target flow, and then resumes the current flow.

Critical Parameters:

• flow_id: The ID of the target flow (created in Step 1).
• input_mapping: A dictionary mapping variables from the calling flow to the target flow.
• output_mapping: A dictionary mapping variables from the target flow back to the calling flow.
• on_error: Defines behavior if the subflow fails (e.g., disconnect, return, or fallback).

from purecloud_platform_client import (
    SubflowNode,
    FlowNodeOut,
    FlowNodeIn
)

def create_subflow_node(target_flow_id: str) -> dict:
    """
    Returns the node configuration for a Subflow node.
    """
    subflow_node_data = SubflowNode(
        flow_id=target_flow_id,
        # Example: Pass the caller's phone number to the subflow as 'caller_number'
        input_mapping={
            "caller_number": "${interaction.customer.phone}"
        },
        # Example: Capture the result status from the subflow
        output_mapping={
            "subflow_result": "${subflow.output.result}"
        },
        on_error="return" # Return to the calling flow on error, do not disconnect
    )

    return {
        "node_type": "Subflow",
        "node_data": subflow_node_data,
        "out": {
            "completed": FlowNodeOut(
                node_id="End",
                transition_type="conditional"
            ),
            "error": FlowNodeOut(
                node_id="End",
                transition_type="conditional"
            )
        }
    }

Step 3: Create Multiple Inbound Call Flows Using the Shared Module

Now we create two distinct inbound call flows (InboundFlowA and InboundFlowB). Both will include a Subflow node pointing to the Common Error Handler created in Step 1.

Scenario:

• InboundFlowA: Plays “Welcome to Service A”, then calls the Common Error Handler, then disconnects.
• InboundFlowB: Plays “Welcome to Service B”, then calls the Common Error Handler, then disconnects.

This demonstrates how to reuse the same logic without duplicating the error handling code.

from purecloud_platform_client import (
    FlowApi,
    Flow,
    PlayMessageNode,
    DisconnectNode,
    FlowNode
)

def create_inbound_flow(api_client: ApiClient, flow_name: str, welcome_message: str, common_flow_id: str) -> str:
    """
    Creates an inbound call flow that plays a welcome message and then invokes the common subflow.
    """
    flow_api = FlowApi(api_client)

    # 1. Welcome Node
    welcome_node = PlayMessageNode(
        name="Welcome",
        message_type="text",
        message_text=welcome_message
    )

    # 2. Subflow Node (Reusing the logic from Step 2)
    subflow_config = create_subflow_node(common_flow_id)
    
    # 3. Disconnect Node
    disconnect_node = DisconnectNode(
        name="Disconnect",
        disposition="normal"
    )

    # 4. Assemble the Flow
    flow = Flow(
        name=flow_name,
        description=f"Inbound flow for {flow_name} using common subflow.",
        type="call",
        version=1,
        nodes={
            "Start": FlowNode(
                name="Start",
                node_type="Start",
                out={
                    "default": FlowNodeOut(
                        node_id="Welcome",
                        transition_type="conditional"
                    )
                }
            ),
            "Welcome": FlowNode(
                name="Welcome",
                node_type="PlayMessage",
                node_data=welcome_node,
                out={
                    "completed": FlowNodeOut(
                        node_id="CallCommon",
                        transition_type="conditional"
                    ),
                    "error": FlowNodeOut(
                        node_id="CallCommon",
                        transition_type="conditional"
                    )
                }
            ),
            "CallCommon": FlowNode(
                name="Invoke Common Error Handler",
                **subflow_config
            ),
            "End": FlowNode(
                name="Disconnect",
                node_type="Disconnect",
                node_data=disconnect_node
            )
        }
    )

    try:
        response = flow_api.post_flow(body=flow)
        print(f"Created Inbound Flow '{flow_name}' with ID: {response.id}")
        return response.id
    except ApiException as e:
        print(f"Failed to create flow '{flow_name}': {e.status} {e.reason}")
        raise

Step 4: Execute the Orchestration

This function ties everything together. It creates the common flow first, then creates two distinct inbound flows that reference the common flow ID.

def main():
    # Initialize API Client
    api_client = get_api_client()
    
    # Get Organization ID (usually required for some contexts, though flow creation is tenant-scoped)
    # For simplicity, we assume the default tenant context.
    
    try:
        # Step 1: Create the Common Shared Flow
        common_flow_id = create_common_flow(api_client, org_id="default")
        
        # Step 2: Create Inbound Flow A
        flow_a_id = create_inbound_flow(
            api_client=api_client,
            flow_name="Inbound Service A",
            welcome_message="Welcome to Service A. You are being transferred to the common handler.",
            common_flow_id=common_flow_id
        )
        
        # Step 3: Create Inbound Flow B
        flow_b_id = create_inbound_flow(
            api_client=api_client,
            flow_name="Inbound Service B",
            welcome_message="Welcome to Service B. You are being transferred to the common handler.",
            common_flow_id=common_flow_id
        )
        
        print(f"\n--- Summary ---")
        print(f"Common Flow ID: {common_flow_id}")
        print(f"Inbound Flow A ID: {flow_a_id}")
        print(f"Inbound Flow B ID: {flow_b_id}")
        
    except Exception as e:
        print(f"Fatal error: {e}")
        raise

Complete Working Example

import os
import sys
from purecloud_platform_client import (
    Configuration, ApiClient, FlowApi, AuthenticationApi,
    Flow, FlowNode, PlayMessageNode, DisconnectNode, FlowNodeOut,
    SubflowNode, ApiException
)

def get_api_client() -> ApiClient:
    environment = os.getenv("GENESYS_CLOUD_ENVIRONMENT")
    client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")

    if not all([environment, client_id, client_secret]):
        raise ValueError("Missing required environment variables.")

    configuration = Configuration(
        base_url=f"https://{environment}",
        client_id=client_id,
        client_secret=client_secret,
        scopes=["flow:flow:write", "flow:flow:read"]
    )

    api_client = ApiClient(configuration)
    
    try:
        auth_api = AuthenticationApi(api_client)
        auth_api.post_oauth2token() 
    except ApiException as e:
        raise Exception(f"Authentication failed: {e.status} {e.reason}")

    return api_client

def create_common_flow(api_client: ApiClient) -> str:
    flow_api = FlowApi(api_client)

    play_node = PlayMessageNode(
        name="Play Error",
        message_type="text",
        message_text="System error. Please retry."
    )

    disconnect_node = DisconnectNode(
        name="Disconnect",
        disposition="busy"
    )

    flow = Flow(
        name="Common Error Handler",
        type="generic",
        version=1,
        nodes={
            "Start": FlowNode(
                name="Start",
                node_type="Start",
                out={"default": FlowNodeOut(node_id="PlayError", transition_type="conditional")}
            ),
            "PlayError": FlowNode(
                name="Play Error",
                node_type="PlayMessage",
                node_data=play_node,
                out={
                    "completed": FlowNodeOut(node_id="Disconnect", transition_type="conditional"),
                    "error": FlowNodeOut(node_id="Disconnect", transition_type="conditional")
                }
            ),
            "Disconnect": FlowNode(
                name="Disconnect",
                node_type="Disconnect",
                node_data=disconnect_node
            )
        }
    )

    try:
        response = flow_api.post_flow(body=flow)
        return response.id
    except ApiException as e:
        raise Exception(f"Create common flow failed: {e.reason}")

def create_subflow_node(target_flow_id: str) -> dict:
    subflow_node_data = SubflowNode(
        flow_id=target_flow_id,
        input_mapping={},
        output_mapping={},
        on_error="return"
    )
    return {
        "node_type": "Subflow",
        "node_data": subflow_node_data,
        "out": {
            "completed": FlowNodeOut(node_id="End", transition_type="conditional"),
            "error": FlowNodeOut(node_id="End", transition_type="conditional")
        }
    }

def create_inbound_flow(api_client: ApiClient, flow_name: str, welcome_msg: str, common_id: str) -> str:
    flow_api = FlowApi(api_client)
    
    welcome_node = PlayMessageNode(
        name="Welcome",
        message_type="text",
        message_text=welcome_msg
    )
    
    subflow_config = create_subflow_node(common_id)
    
    disconnect_node = DisconnectNode(
        name="Disconnect",
        disposition="normal"
    )

    flow = Flow(
        name=flow_name,
        type="call",
        version=1,
        nodes={
            "Start": FlowNode(
                name="Start",
                node_type="Start",
                out={"default": FlowNodeOut(node_id="Welcome", transition_type="conditional")}
            ),
            "Welcome": FlowNode(
                name="Welcome",
                node_type="PlayMessage",
                node_data=welcome_node,
                out={
                    "completed": FlowNodeOut(node_id="CallCommon", transition_type="conditional"),
                    "error": FlowNodeOut(node_id="CallCommon", transition_type="conditional")
                }
            ),
            "CallCommon": FlowNode(
                name="Call Common",
                **subflow_config
            ),
            "End": FlowNode(
                name="End",
                node_type="Disconnect",
                node_data=disconnect_node
            )
        }
    )

    try:
        response = flow_api.post_flow(body=flow)
        return response.id
    except ApiException as e:
        raise Exception(f"Create inbound flow failed: {e.reason}")

if __name__ == "__main__":
    try:
        api_client = get_api_client()
        common_id = create_common_flow(api_client)
        print(f"Common Flow ID: {common_id}")
        
        id_a = create_inbound_flow(api_client, "Flow A", "Welcome to A", common_id)
        print(f"Flow A ID: {id_a}")
        
        id_b = create_inbound_flow(api_client, "Flow B", "Welcome to B", common_id)
        print(f"Flow B ID: {id_b}")
        
    except Exception as e:
        print(f"Error: {e}")
        sys.exit(1)

Common Errors & Debugging

Error: 400 Bad Request - Invalid Node Structure

Cause: The Subflow node requires a valid flow_id that exists in the same organization. Additionally, the out object must define transitions for completed and error. If you omit error in the out object, the flow validation will fail.

Fix: Ensure the SubflowNode is initialized with a valid ID and that the parent FlowNode includes the out dictionary with both completed and error keys.

# Correct structure
"out": {
    "completed": FlowNodeOut(node_id="NextNode", transition_type="conditional"),
    "error": FlowNodeOut(node_id="ErrorNode", transition_type="conditional")
}

Error: 403 Forbidden - Insufficient Scope

Cause: The OAuth token does not include flow:flow:write.

Fix: Update your Service Account configuration in the Genesys Cloud Admin Portal. Navigate to Security > OAuth 2.0 > Service Accounts. Edit your client and ensure flow:flow:write is checked in the Scopes section. Regenerate the client secret if necessary.

Error: 422 Unprocessable Entity - Flow Validation Error

Cause: The flow contains circular references or disconnected nodes. For example, if the Start node does not have an out definition, the flow is invalid.

Fix: Verify that every node (except terminal nodes like Disconnect) has an out object. Ensure that the node_id references in FlowNodeOut match the keys in the nodes dictionary exactly.

# Check this mapping
nodes={
    "Start": FlowNode(..., out={"default": FlowNodeOut(node_id="Welcome", ...)}),
    "Welcome": FlowNode(...), # Key "Welcome" must exist
    # ...
}

Official References

• Genesys Cloud Flow API Documentation
• Subflow Node Reference
• Python SDK Documentation