Reusable Flow Patterns: Calling Shared Flows from Multiple Inbound Call Flows

What You Will Build

• This tutorial demonstrates how to architect a Genesys Cloud CX contact center by extracting common logic (such as IVR menus, authentication, or after-call work) into a single, reusable “Shared Flow” and invoking it from multiple distinct inbound call flows.
• The solution utilizes the Genesys Cloud Platform API v2 and the Python SDK (genesyscloud).
• The programming language covered is Python 3.9+.

Prerequisites

• OAuth Client Type: Service Account or Client Credentials Grant.
• Required Scopes: flow:flow:read, flow:flow:write, routing:queue:read, routing:skill:read, user:read.
• SDK Version: genesyscloud >= 140.0.0.
• Runtime: Python 3.9 or higher.
• External Dependencies: genesyscloud, python-dotenv (for secure credential management).

pip install genesyscloud python-dotenv

Authentication Setup

Genesys Cloud uses OAuth 2.0 for API access. For server-side integrations and flow manipulation, the Client Credentials grant is the standard approach. You must generate a Service Account in the Genesys Cloud Admin Console with the necessary permissions.

Create a .env file in your project root:

GENESYS_CLOUD_REGION=us-east-1
GENESYS_CLOUD_CLIENT_ID=your_client_id_here
GENESYS_CLOUD_CLIENT_SECRET=your_client_secret_here

Initialize the SDK client in your script. This client handles token acquisition, caching, and automatic refresh.

import os
from dotenv import load_dotenv
from genesyscloud.platform.client import PlatformClient
from genesyscloud.auth.client_credentials.auth_client_credentials import AuthClientCredentials

load_dotenv()

def get_platform_client() -> PlatformClient:
    """
    Initializes and returns a configured Genesys Cloud Platform Client.
    """
    region = os.getenv("GENESYS_CLOUD_REGION")
    client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")

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

    # Set the base URL based on region
    base_url = f"https://{region}.mygen.com" if region != "us-east-1" else "https://api.mypurecloud.com"
    
    client = PlatformClient(base_url=base_url)
    auth = AuthClientCredentials(client_id, client_secret)
    client.auth = auth
    
    return client

Implementation

Step 1: Define the Shared Flow Structure

The core concept of reuse is creating a flow that acts as a “subroutine.” In Genesys Cloud, this is a Conversation Flow of type conversation. Unlike inbound routing flows, this flow does not start with a Start node connected to a queue. Instead, it starts with a Start node that defines the entry point, and it ends with a End node that defines the return mechanism.

Key architectural decision: The End node in a shared flow must use the Return action. This passes control back to the calling flow. You can also pass data back using returnVariables.

First, we identify or create the Shared Flow. For this tutorial, we assume you have already created a flow named “Shared IVR Menu” in the Admin Console. We will fetch its ID to use in our Python script.

from genesyscloud.flow.api_flow_api import ApiFlowApi

def get_shared_flow_id(client: PlatformClient, flow_name: str) -> str:
    """
    Fetches the ID of a flow by name.
    """
    flow_api = ApiFlowApi(client)
    
    # Search for flows by name
    # Note: In production, cache this ID. Do not query on every run.
    response = flow_api.get_flows(
        query=flow_name,
        type="conversation",
        expand=["nodes"]
    )

    if response.entities and len(response.entities) > 0:
        # Assume exact match for simplicity; in production, verify name equality
        return response.entities[0].id
    
    raise ValueError(f"Flow '{flow_name}' not found.")

Step 2: Create the Inbound Call Flow with Subflow Node

Now we construct the Inbound Call Flow. This flow will trigger when a call arrives at a specific queue. Instead of building the entire IVR logic here, we will insert a Subflow node.

The Subflow node requires:

1. flowId: The ID of the shared flow created in Step 1.
2. inputVariables: A map of variables to send TO the shared flow.
3. outputVariables: A map defining what variables the shared flow returns.

We will use the SDK to construct the JSON body for a new flow. The structure is complex, so we build it programmatically.

from genesyscloud.flow.model.flow import Flow
from genesyscloud.flow.model.flow_node import FlowNode
from genesyscloud.flow.model.flow_node_start import FlowNodeStart
from genesyscloud.flow.model.flow_node_subflow import FlowNodeSubflow
from genesyscloud.flow.model.flow_node_end import FlowNodeEnd
from genesyscloud.flow.model.flow_node_transition import FlowNodeTransition
from genesyscloud.flow.model.flow_node_condition import FlowNodeCondition

def create_inbound_flow_with_subflow(client: PlatformClient, shared_flow_id: str, queue_id: str) -> str:
    """
    Creates a new Inbound Call Flow that delegates logic to a Shared Flow.
    Returns the ID of the newly created flow.
    """
    flow_api = ApiFlowApi(client)

    # 1. Define the Start Node
    # Inbound flows start with a 'start' node.
    start_node = FlowNodeStart(
        id="start",
        type="start",
        name="Start",
        transitions=[
            FlowNodeTransition(
                id="start-to-subflow",
                target="subflow-node"
            )
        ]
    )

    # 2. Define the Subflow Node
    # This is the core reuse mechanism.
    subflow_node = FlowNodeSubflow(
        id="subflow-node",
        type="subflow",
        name="Call Shared IVR",
        flow_id=shared_flow_id,
        # Pass the caller's phone number to the shared flow
        input_variables={
            "callerPhone": "${contact.attributes.phoneNumber}"
        },
        # Define what we expect back from the shared flow
        output_variables={
            "selectedOption": "selectedOption",
            "authResult": "authResult"
        },
        transitions=[
            # On success, go to a queue
            FlowNodeTransition(
                id="subflow-to-queue",
                target="queue-node",
                conditions=[
                    FlowNodeCondition(
                        type="value",
                        value="${selectedOption}",
                        operator="equals",
                        comparison_value="1"
                    )
                ]
            ),
            # On hangup or error, end the call
            FlowNodeTransition(
                id="subflow-to-end",
                target="end-node"
            )
        ]
    )

    # 3. Define the Queue Node (Destination)
    # In a real scenario, this would be a Queue node or another Subflow.
    # For this example, we just show the structure of a transition target.
    # Note: A full Queue node requires more complex JSON. We will use a simple End node for brevity 
    # to demonstrate the flow structure, but in production, replace with QueueNode.
    queue_node = FlowNodeEnd(
        id="queue-node",
        type="end",
        name="Route to Queue (Placeholder)",
        action="hangup" # Placeholder for actual queue routing
    )

    # 4. Define the End Node
    end_node = FlowNodeEnd(
        id="end-node",
        type="end",
        name="End Call",
        action="hangup"
    )

    # 5. Assemble the Flow Object
    new_flow = Flow(
        name="Inbound Flow with Shared Subflow",
        type="call",
        description="Inbound flow that reuses a shared IVR logic.",
        nodes=[
            start_node,
            subflow_node,
            queue_node,
            end_node
        ],
        # Associate with a queue to make it active for inbound calls
        queue_ids=[queue_id],
        enabled=True
    )

    # 6. Post the Flow
    response = flow_api.post_flows(body=new_flow)
    print(f"Created Inbound Flow ID: {response.id}")
    return response.id

Step 3: Handling Data Exchange and Error States

The power of subflows lies in the data contract. The input_variables and output_variables maps must align between the caller and the callee.

Critical Detail: Variable names in Genesys Cloud are case-sensitive. If the Shared Flow returns a variable named selectedOption, the calling flow must reference it exactly as selectedOption in its output_variables map.

Let us look at how to update the Shared Flow to ensure it returns the correct data. We will fetch the Shared Flow, modify its End node to include a returnVariables map, and save it.

from genesyscloud.flow.model.flow_node_end import FlowNodeEnd

def update_shared_flow_return_logic(client: PlatformClient, flow_id: str) -> None:
    """
    Updates the Shared Flow's End node to return specific variables.
    """
    flow_api = ApiFlowApi(client)

    # Get the current flow
    flow_response = flow_api.get_flow(flow_id=flow_id)
    
    # Find the End node (assuming there is one named 'end')
    end_node = None
    for node in flow_response.nodes:
        if node.type == "end" and node.id == "end":
            end_node = node
            break

    if not end_node:
        raise ValueError("End node with ID 'end' not found in the shared flow.")

    # Update the End node to return variables
    # The key is the variable name in the calling flow context
    # The value is the variable name in the shared flow context
    updated_end_node = FlowNodeEnd(
        id="end",
        type="end",
        name="Return to Caller",
        action="return",
        return_variables={
            "selectedOption": "${local.selectedOption}",
            "authResult": "${local.authResult}"
        },
        transitions=[] # End nodes do not have transitions
    )

    # Replace the node in the list
    new_nodes = []
    for node in flow_response.nodes:
        if node.id == "end":
            new_nodes.append(updated_end_node)
        else:
            new_nodes.append(node)

    # Update the flow object
    flow_response.nodes = new_nodes

    # Put the updated flow
    flow_api.put_flow(flow_id=flow_id, body=flow_response)
    print("Shared Flow updated with return logic.")

Complete Working Example

Below is a complete, runnable Python script that ties these concepts together. It authenticates, finds a shared flow, creates an inbound flow that calls it, and updates the shared flow to return data.

Prerequisite: You must have a Queue ID and a pre-existing Conversation Flow named “Shared IVR Menu” in your Genesys Cloud org.

import os
import sys
from dotenv import load_dotenv
from genesyscloud.platform.client import PlatformClient
from genesyscloud.auth.client_credentials.auth_client_credentials import AuthClientCredentials
from genesyscloud.flow.api_flow_api import ApiFlowApi
from genesyscloud.flow.model.flow import Flow
from genesyscloud.flow.model.flow_node_start import FlowNodeStart
from genesyscloud.flow.model.flow_node_subflow import FlowNodeSubflow
from genesyscloud.flow.model.flow_node_end import FlowNodeEnd
from genesyscloud.flow.model.flow_node_transition import FlowNodeTransition
from genesyscloud.flow.model.flow_node_condition import FlowNodeCondition

load_dotenv()

def init_client() -> PlatformClient:
    region = os.getenv("GENESYS_CLOUD_REGION", "us-east-1")
    client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
    
    if not client_id or not client_secret:
        raise EnvironmentError("Client ID and Secret must be set in .env")

    base_url = f"https://{region}.mygen.com" if region != "us-east-1" else "https://api.mypurecloud.com"
    client = PlatformClient(base_url=base_url)
    client.auth = AuthClientCredentials(client_id, client_secret)
    return client

def get_flow_by_name(client: PlatformClient, name: str) -> str:
    api = ApiFlowApi(client)
    response = api.get_flows(query=name, type="conversation")
    if not response.entities:
        raise ValueError(f"Flow '{name}' not found.")
    return response.entities[0].id

def create_reusable_flow_architecture(client: PlatformClient, queue_id: str):
    # 1. Locate the Shared Flow
    shared_flow_name = "Shared IVR Menu"
    try:
        shared_flow_id = get_flow_by_name(client, shared_flow_name)
        print(f"Found Shared Flow ID: {shared_flow_id}")
    except ValueError as e:
        print(f"Error: {e}")
        return

    # 2. Update Shared Flow to Return Data
    # In a real CI/CD pipeline, this would be an idempotent check
    print(f"Ensuring Shared Flow '{shared_flow_name}' has correct return logic...")
    # Note: The update logic from Step 3 would go here. 
    # For safety in this example, we assume it is already configured 
    # or you uncomment the update function call below.
    # update_shared_flow_return_logic(client, shared_flow_id)

    # 3. Create Inbound Flow
    print("Creating Inbound Flow with Subflow node...")
    
    start_node = FlowNodeStart(
        id="start",
        type="start",
        name="Start",
        transitions=[FlowNodeTransition(id="s2sf", target="subflow")]
    )

    subflow_node = FlowNodeSubflow(
        id="subflow",
        type="subflow",
        name="Execute Shared IVR",
        flow_id=shared_flow_id,
        input_variables={
            "incomingCallerId": "${contact.attributes.phoneNumber}"
        },
        output_variables={
            "menuSelection": "selectedOption",
            "authStatus": "authResult"
        },
        transitions=[
            FlowNodeTransition(
                id="sf2q",
                target="queue",
                conditions=[
                    FlowNodeCondition(
                        type="value",
                        value="${menuSelection}",
                        operator="equals",
                        comparison_value="1"
                    )
                ]
            ),
            FlowNodeTransition(id="sf2e", target="end")
        ]
    )

    # Placeholder Queue Node - In production, use FlowNodeQueue
    queue_node = FlowNodeEnd(
        id="queue",
        type="end",
        name="Queue Entry (Demo)",
        action="hangup"
    )

    end_node = FlowNodeEnd(
        id="end",
        type="end",
        name="Goodbye",
        action="hangup"
    )

    inbound_flow = Flow(
        name="Inbound - Reuse Shared IVR",
        type="call",
        description="Demonstrates subflow reuse pattern.",
        nodes=[start_node, subflow_node, queue_node, end_node],
        queue_ids=[queue_id],
        enabled=True
    )

    try:
        api = ApiFlowApi(client)
        result = api.post_flows(body=inbound_flow)
        print(f"Success! Created Inbound Flow ID: {result.id}")
        print(f"Flow URL: https://{os.getenv('GENESYS_CLOUD_REGION')}.mypurecloud.com/admin/#/admin/flows/view/{result.id}")
    except Exception as e:
        print(f"Failed to create flow: {e}")

if __name__ == "__main__":
    if not os.getenv("GENESYS_CLOUD_QUEUE_ID"):
        print("Error: Set GENESYS_CLOUD_QUEUE_ID in .env")
        sys.exit(1)

    client = init_client()
    create_reusable_flow_architecture(client, os.getenv("GENESYS_CLOUD_QUEUE_ID"))

Common Errors & Debugging

Error: 400 Bad Request - Invalid Flow Structure

• What causes it: The subflow node references a flowId that does not exist, is not of type conversation, or is disabled. Alternatively, the input_variables or output_variables keys do not match variable names defined in the respective flows.
• How to fix it: Verify the Shared Flow ID is correct. Ensure the Shared Flow is enabled. Check that the variable names in the output_variables map of the Subflow node exactly match the keys in the return_variables map of the Shared Flow’s End node.

Error: 403 Forbidden - Insufficient Scopes

• What causes it: The OAuth token lacks the flow:flow:write scope.
• How to fix it: Update your Service Account in the Genesys Cloud Admin Console. Navigate to Admin > Security > Service Accounts, select your account, and ensure “Flow Management” permissions are granted. Regenerate the client secret if necessary.

Error: Variable Not Found at Runtime

• What causes it: The Shared Flow tries to use a variable (e.g., ${callerPhone}) that was not passed in via the Subflow node’s input_variables.
• How to fix it: Inspect the Debug logs in Genesys Cloud. When testing the flow, use the “Test” feature in the Flow Designer. Check the “Variables” tab to see what values were passed into the Subflow. Ensure the key in the calling flow’s input_variables matches the variable name expected in the Shared Flow’s nodes.

Error: Infinite Loop or Hang

• What causes it: The Shared Flow does not have an End node with action="return", or the calling flow has no transition for the default case after the Subflow.
• How to fix it: Ensure the Shared Flow explicitly ends with a Return action. In the calling flow, ensure the Subflow node has a default transition (one without conditions) that handles cases where the Shared Flow returns unexpected values or errors.

Official References

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