Reuse Flow Logic: Calling Shared Flows from Multiple Inbound Routes

What You Will Build

• One sentence: Create a reusable “shared” flow that handles common logic (such as authentication or menu routing) and invoke it from multiple distinct inbound call flows using the Genesys Cloud CX API.
• One sentence: This tutorial uses the Genesys Cloud CX PureCloud Platform Client SDK.
• One sentence: The code examples are provided in Python, utilizing the official genesyscloud SDK.

Prerequisites

• OAuth Client Type: Machine-to-Machine (M2M) client credentials.
• Required Scopes:
  • flow:flow:write (to create/update flows)
  • flow:flow:read (to list flows)
  • routing:route:write (to create/update routing strategies)
  • routing:route:read (to list routing strategies)
• SDK Version: genesyscloud Python SDK version 140.0.0 or later.
• Language/Runtime: Python 3.9+.
• External Dependencies:
  • genesyscloud
  • os (standard library)

Authentication Setup

Genesys Cloud CX uses OAuth 2.0 for authentication. For server-side integrations and bulk operations, the Client Credentials flow is the standard approach. You must store your API Key and Secret securely, preferably in environment variables.

The Genesys Cloud Python SDK handles token acquisition and refreshing automatically if configured correctly.

import os
from purecloudplatformclientv2 import Configuration, ApiClient

def get_api_client():
    """
    Configures and returns an authenticated ApiClient instance.
    """
    # Load credentials from environment variables
    api_key = os.getenv("GENESYS_API_KEY")
    api_secret = os.getenv("GENESYS_API_SECRET")
    environment = os.getenv("GENESYS_ENVIRONMENT", "mypurecloud.com")

    if not api_key or not api_secret:
        raise ValueError("GENESYS_API_KEY and GENESYS_API_SECRET must be set")

    # Configure the API client
    configuration = Configuration()
    configuration.host = f"https://{environment}"
    configuration.api_key['Authorization'] = f"{api_key}:{api_secret}"
    configuration.api_key_prefix['Authorization'] = 'Basic'

    # Initialize the ApiClient
    api_client = ApiClient(configuration)
    return api_client

This setup ensures that every subsequent API call includes a valid Authorization header. The SDK manages the underlying token exchange, so you do not need to manually handle POST /oauth/token requests.

Implementation

Step 1: Create the Shared Flow

A shared flow is simply a Flow resource in Genesys Cloud with a specific type. Unlike inbound or outbound flows, a shared flow cannot be directly routed to by customers. It serves as a library of logic.

We will create a simple shared flow that plays a greeting. In a production scenario, this might contain complex IVR menus or authentication logic.

Key Concept: The type field must be set to shared.

from purecloudplatformclientv2 import FlowEntityRequest, FlowAction, FlowTransition, FlowTransitionCondition
from purecloudplatformclientv2 import FlowStartNode, FlowPlayPrompt

def create_shared_flow(api_client, name: str, description: str) -> str:
    """
    Creates a new Shared Flow.
    Returns the ID of the created flow.
    """
    # 1. Define the Start Node
    # The start node is the entry point for any flow.
    start_node = FlowStartNode(
        id="start",
        name="Start",
        entry_transitions=[],
        actions=[]
    )

    # 2. Define a Play Prompt Action
    # This action plays a text-to-speech or media prompt.
    play_prompt_action = FlowPlayPrompt(
        text="Welcome to the shared module. This logic is reused.",
        language="en-US"
    )

    # 3. Define the Transition
    # After playing the prompt, we transition to the End node.
    transition = FlowTransition(
        condition=FlowTransitionCondition(
            type="always"
        ),
        target_id="end"
    )

    # 4. Construct the Flow Entity Request
    # Note: type='shared' is critical.
    flow_request = FlowEntityRequest(
        name=f"Shared Module - {name}",
        description=description,
        type="shared",
        start_node=start_node,
        actions=[play_prompt_action],
        transitions=[transition],
        end_nodes=[] # End nodes are implicit or defined separately in complex flows
    )

    # 5. API Call: POST /api/v2/flows
    flows_api = api_client.FlowsApi()
    try:
        response = flows_api.post_flow(body=flow_request)
        print(f"Created Shared Flow with ID: {response.id}")
        return response.id
    except Exception as e:
        print(f"Error creating shared flow: {e}")
        raise

Expected Response:
The API returns a Flow object. The most important field is id. You must capture this ID to reference the flow later.

{
  "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "name": "Shared Module - Auth Logic",
  "type": "shared",
  "status": "published",
  "version": 1
}

Error Handling:

• 409 Conflict: A flow with the same name already exists in this organization. Use unique names or check existing flows first.
• 400 Bad Request: The flow definition is invalid. Common errors include missing start_node or circular transitions. Check the errors array in the response body.

Step 2: Create an Inbound Flow that Calls the Shared Flow

Now we create an inbound call flow. This flow will receive a customer call and immediately delegate logic to the shared flow created in Step 1.

Key Concept: The RunSubflow action is used to invoke another flow. It requires the flow_id of the target shared flow.

from purecloudplatformclientv2 import FlowRunSubflow

def create_inbound_flow_with_shared_call(api_client, shared_flow_id: str, name: str) -> str:
    """
    Creates an Inbound Flow that calls the provided Shared Flow.
    Returns the ID of the created inbound flow.
    """
    # 1. Define the Start Node
    start_node = FlowStartNode(
        id="start",
        name="Start",
        entry_transitions=[],
        actions=[]
    )

    # 2. Define the Run Subflow Action
    # This is the core mechanism for reuse.
    run_subflow_action = FlowRunSubflow(
        flow_id=shared_flow_id,
        # Optional: Pass variables to the shared flow
        # variables={
        #     "customer_id": "{{contact.attributes.customerId}}"
        # }
    )

    # 3. Define Transition to End
    transition = FlowTransition(
        condition=FlowTransitionCondition(
            type="always"
        ),
        target_id="end"
    )

    # 4. Construct the Inbound Flow Request
    # Note: type='inbound'
    inbound_flow_request = FlowEntityRequest(
        name=f"Inbound Caller - {name}",
        description="Inbound flow that reuses shared logic",
        type="inbound",
        start_node=start_node,
        actions=[run_subflow_action],
        transitions=[transition],
        end_nodes=[]
    )

    # 5. API Call: POST /api/v2/flows
    flows_api = api_client.FlowsApi()
    try:
        response = flows_api.post_flow(body=inbound_flow_request)
        print(f"Created Inbound Flow with ID: {response.id}")
        return response.id
    except Exception as e:
        print(f"Error creating inbound flow: {e}")
        raise

Why This Works:
When a customer calls a queue associated with this inbound flow, the Genesys platform executes the RunSubflow action. It suspends the current flow context, executes the shared flow, and then resumes the inbound flow after the shared flow completes. This allows you to update the shared flow once, and the change propagates to all inbound flows that reference it.

Step 3: Create a Second Inbound Flow Reusing the Same Shared Flow

To demonstrate reuse, we create a second inbound flow. It does not need to redefine the logic. It simply references the same shared_flow_id.

def create_second_inbound_flow(api_client, shared_flow_id: str) -> str:
    """
    Creates a second Inbound Flow reusing the same Shared Flow.
    """
    # Reuse the same structure as the first inbound flow
    start_node = FlowStartNode(
        id="start",
        name="Start",
        entry_transitions=[],
        actions=[]
    )

    run_subflow_action = FlowRunSubflow(
        flow_id=shared_flow_id
    )

    transition = FlowTransition(
        condition=FlowTransitionCondition(
            type="always"
        ),
        target_id="end"
    )

    inbound_flow_request = FlowEntityRequest(
        name="Inbound Caller - Secondary Channel",
        description="Second inbound flow reusing the same shared logic",
        type="inbound",
        start_node=start_node,
        actions=[run_subflow_action],
        transitions=[transition],
        end_nodes=[]
    )

    flows_api = api_client.FlowsApi()
    try:
        response = flows_api.post_flow(body=inbound_flow_request)
        print(f"Created Second Inbound Flow with ID: {response.id}")
        return response.id
    except Exception as e:
        print(f"Error creating second inbound flow: {e}")
        raise

Step 4: Verify the Flow Definitions

It is critical to verify that the flows were created correctly and that the references are valid. You can retrieve the flow details using the get_flow method.

def verify_flow(api_client, flow_id: str):
    """
    Retrieves and prints the details of a flow to verify the shared flow reference.
    """
    flows_api = api_client.FlowsApi()
    try:
        response = flows_api.get_flow(flow_id=flow_id)
        print(f"Flow Name: {response.name}")
        print(f"Flow Type: {response.type}")
        
        # Inspect the start node actions
        if response.start_node and response.start_node.actions:
            for action in response.start_node.actions:
                if hasattr(action, 'flow_id'):
                    print(f"  - Calls Shared Flow ID: {action.flow_id}")
        
        return response
    except Exception as e:
        print(f"Error verifying flow {flow_id}: {e}")
        raise

Complete Working Example

This script combines all steps into a single executable workflow. It creates a shared flow, two inbound flows that reference it, and verifies the connections.

import os
import sys
from purecloudplatformclientv2 import (
    Configuration, ApiClient, FlowEntityRequest, FlowStartNode, 
    FlowPlayPrompt, FlowRunSubflow, FlowTransition, FlowTransitionCondition
)

def get_api_client():
    api_key = os.getenv("GENESYS_API_KEY")
    api_secret = os.getenv("GENESYS_API_SECRET")
    environment = os.getenv("GENESYS_ENVIRONMENT", "mypurecloud.com")

    if not api_key or not api_secret:
        raise ValueError("GENESYS_API_KEY and GENESYS_API_SECRET must be set")

    configuration = Configuration()
    configuration.host = f"https://{environment}"
    configuration.api_key['Authorization'] = f"{api_key}:{api_secret}"
    configuration.api_key_prefix['Authorization'] = 'Basic'
    return ApiClient(configuration)

def create_shared_flow(api_client, name: str) -> str:
    start_node = FlowStartNode(id="start", name="Start", entry_transitions=[], actions=[])
    play_prompt = FlowPlayPrompt(text="Executing shared logic.", language="en-US")
    transition = FlowTransition(condition=FlowTransitionCondition(type="always"), target_id="end")
    
    flow_request = FlowEntityRequest(
        name=f"Shared - {name}",
        description="A reusable shared flow",
        type="shared",
        start_node=start_node,
        actions=[play_prompt],
        transitions=[transition],
        end_nodes=[]
    )
    
    flows_api = api_client.FlowsApi()
    response = flows_api.post_flow(body=flow_request)
    return response.id

def create_inbound_flow(api_client, shared_flow_id: str, name: str) -> str:
    start_node = FlowStartNode(id="start", name="Start", entry_transitions=[], actions=[])
    run_subflow = FlowRunSubflow(flow_id=shared_flow_id)
    transition = FlowTransition(condition=FlowTransitionCondition(type="always"), target_id="end")
    
    flow_request = FlowEntityRequest(
        name=f"Inbound - {name}",
        description="Inbound flow calling shared flow",
        type="inbound",
        start_node=start_node,
        actions=[run_subflow],
        transitions=[transition],
        end_nodes=[]
    )
    
    flows_api = api_client.FlowsApi()
    response = flows_api.post_flow(body=flow_request)
    return response.id

def main():
    try:
        print("Initializing API Client...")
        api_client = get_api_client()
        
        print("Step 1: Creating Shared Flow...")
        shared_flow_id = create_shared_flow(api_client, "CommonAuth")
        print(f"Shared Flow ID: {shared_flow_id}")
        
        print("Step 2: Creating First Inbound Flow...")
        inbound_flow_id_1 = create_inbound_flow(api_client, shared_flow_id, "SalesLine")
        print(f"Inbound Flow 1 ID: {inbound_flow_id_1}")
        
        print("Step 3: Creating Second Inbound Flow...")
        inbound_flow_id_2 = create_inbound_flow(api_client, shared_flow_id, "SupportLine")
        print(f"Inbound Flow 2 ID: {inbound_flow_id_2}")
        
        print("Step 4: Verifying Flows...")
        # Verify that both inbound flows point to the same shared flow
        flows_api = api_client.FlowsApi()
        flow1 = flows_api.get_flow(flow_id=inbound_flow_id_1)
        flow2 = flows_api.get_flow(flow_id=inbound_flow_id_2)
        
        print(f"Flow 1 ('{flow1.name}') references Shared Flow: {flow1.start_node.actions[0].flow_id}")
        print(f"Flow 2 ('{flow2.name}') references Shared Flow: {flow2.start_node.actions[0].flow_id}")
        
        if flow1.start_node.actions[0].flow_id == flow2.start_node.actions[0].flow_id:
            print("SUCCESS: Both inbound flows correctly reference the same shared flow.")
        else:
            print("ERROR: Flows do not reference the same shared flow.")
            
    except Exception as e:
        print(f"An error occurred: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 400 Bad Request - “Invalid Flow Definition”

What causes it:
The flow structure is malformed. Common issues include:

• Missing start_node.
• Transitions pointing to non-existent node IDs.
• Circular dependencies (though less common in simple examples).
• Using a shared flow type for an inbound route.

How to fix it:
Ensure the FlowEntityRequest object has a valid start_node with an id that matches the target_id in any transitions. Verify that the type is correct (shared for libraries, inbound for customer-facing flows).

Code showing the fix:

# Ensure the start node ID matches the transition target
start_node = FlowStartNode(id="start", ...)
transition = FlowTransition(target_id="start", ...) # Correct
# transition = FlowTransition(target_id="nonexistent", ...) # Causes 400

Error: 403 Forbidden - “Insufficient Permissions”

What causes it:
The OAuth token does not have the required scopes. Specifically, flow:flow:write is needed to create flows.

How to fix it:
Check your API Key settings in the Genesys Cloud Admin console. Navigate to Admin > Security > API Keys. Ensure the key has the flow:flow:write and flow:flow:read scopes granted.

Error: 404 Not Found - “Flow ID Not Found”

What causes it:
The shared_flow_id passed to the FlowRunSubflow action is invalid or refers to a flow that does not exist.

How to fix it:
Verify the shared_flow_id returned from Step 1. Ensure you are using the correct organization ID if operating in a multi-tenant environment (though the API usually scopes to the current org automatically).

Error: Flow Logic Not Executing

What causes it:
The inbound flow is not associated with a Queue or Routing Strategy. Creating the flow resource does not automatically route calls to it.

How to fix it:
You must create a Routing Strategy or assign the flow to a Queue. This is outside the scope of the flow creation API but is a necessary next step. Use the routing:route:write scope to update the queue’s flow reference.

Official References

• Genesys Cloud CX Flows API Documentation
• Genesys Cloud CX Flow Object Reference
• Python SDK Reference: PureCloudPlatformClientV2
• Best Practices for Shared Flows