Genesys Cloud: Reuse Shared Flows Across Multiple Inbound IVRs Using the Flow API

What You Will Build

This tutorial demonstrates how to architect a modular Genesys Cloud Voice deployment by creating a single “Shared Flow” that handles common logic, such as authentication or menu selection, and invoking it from multiple distinct inbound call flows.
You will use the Genesys Cloud Platform API (specifically the Flow API) to define these flows and the purecloud-platform-client-v2 Python SDK to manage the lifecycle of these objects programmatically.
The implementation covers Python for the SDK interaction and JSON for the Flow definition payloads.

Prerequisites

• OAuth Client: A Genesys Cloud OAuth client with the flow:view and flow:write scopes.
• SDK Version: purecloud-platform-client-v2 version 150.0.0 or higher.
• Runtime: Python 3.9 or higher.
• Dependencies: pip install purecloud-platform-client-v2 python-dotenv
• Genesys Cloud Org: A tenant with at least one configured User and a valid Voice license.

Authentication Setup

Genesys Cloud uses OAuth 2.0 for API access. For programmatic access via a script or backend service, the Client Credentials grant type is the standard. This flow does not require a user context, which is appropriate for infrastructure setup scripts.

Step 1: Install Dependencies and Configure Environment

Create a .env file in your project root with the following variables:

# .env
GENESYS_CLIENT_ID=your_client_id_here
GENESYS_CLIENT_SECRET=your_client_secret_here
GENESYS_REGION=us-east-1

Step 2: Initialize the API Client

The following Python code initializes the PlatformClient using the environment variables. It handles the token acquisition automatically upon the first API call.

import os
from dotenv import load_dotenv
from purecloudplatformclientv2 import (
    ApiClient,
    Configuration,
    FlowApi,
    ApiException
)

# Load environment variables
load_dotenv()

def get_flow_api_client():
    """
    Initializes and returns a configured FlowApi client.
    """
    # Load credentials from environment
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    region = os.getenv("GENESYS_REGION", "us-east-1")

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

    # Configure the API client
    configuration = Configuration()
    configuration.client_id = client_id
    configuration.client_secret = client_secret
    configuration.host = f"https://{region}.mypurecloud.com"
    
    # Initialize the API client
    api_client = ApiClient(configuration=configuration)
    
    # Return the FlowApi instance
    return FlowApi(api_client)

if __name__ == "__main__":
    try:
        api = get_flow_api_client()
        print("Authentication successful. Flow API client ready.")
    except Exception as e:
        print(f"Authentication failed: {e}")

OAuth Scope Requirement: flow:view is required to read flows. flow:write is required to create or update them.

Implementation

The core concept of reusing logic in Genesys Cloud is the Flow object. A Flow can be of type voice, chat, callback, etc. To reuse logic, you create a Flow of type voice that is not associated with any specific routing queue or IVR directly. Instead, you invoke this “Shared Flow” from other flows using the Play Prompt or Transfer logic, or more commonly, by using the Flow Action node type if available, but the most robust method for complex logic reuse is the Flow node within the Canvas editor, which maps to specific JSON structures in the API.

However, the most programmatic and clean way to achieve this via API is to use the flow action within a flow definition. This allows one flow to pause its execution, hand off control to another flow, and resume when the sub-flow completes.

Step 1: Create the Shared Flow (The “Library”)

First, we create a flow that performs a generic task, such as collecting a 4-digit PIN for authentication. This flow will accept an input variable and return a success/failure status.

Required Scope: flow:write

import json

def create_shared_auth_flow(api_client: FlowApi, flow_name: str = "Shared Auth Flow"):
    """
    Creates a reusable voice flow that collects a PIN.
    """
    # Define the flow body
    # Note: This is a simplified representation of the Canvas JSON structure.
    # In production, you would likely load this from a .json file.
    flow_body = {
        "name": flow_name,
        "type": "voice",
        "enabled": True,
        "version": 1,
        "body": {
            "type": "flow",
            "startNode": "start",
            "nodes": {
                "start": {
                    "type": "start",
                    "name": "Start",
                    "outgoingTransitions": {
                        "default": "collectPin"
                    }
                },
                "collectPin": {
                    "type": "collectinput",
                    "name": "Collect PIN",
                    "prompt": "Please enter your 4 digit PIN followed by the pound sign.",
                    "maxDigits": 4,
                    "outgoingTransitions": {
                        "input": "validatePin"
                    }
                },
                "validatePin": {
                    "type": "condition",
                    "name": "Validate PIN",
                    "conditions": [
                        {
                            "expression": "${input.pin} == '1234'",
                            "transition": "success"
                        },
                        {
                            "expression": "${input.pin} != '1234'",
                            "transition": "failure"
                        }
                    ]
                },
                "success": {
                    "type": "end",
                    "name": "Success",
                    "outgoingTransitions": {}
                },
                "failure": {
                    "type": "end",
                    "name": "Failure",
                    "outgoingTransitions": {}
                }
            }
        }
    }

    try:
        # Post the flow
        response = api_client.post_flow(body=flow_body)
        print(f"Created Shared Flow with ID: {response.id}")
        return response
    except ApiException as e:
        print(f"Failed to create flow: {e.status} - {e.reason}")
        raise

if __name__ == "__main__":
    api = get_flow_api_client()
    # shared_flow = create_shared_auth_flow(api)

Expected Response:
The API returns a Flow object containing the id, name, version, and body. You must store this id to reference it in other flows.

Step 2: Create Inbound Flows That Invoke the Shared Flow

Now, we create two different inbound flows (e.g., “Sales IVR” and “Support IVR”) that both invoke the Shared Flow created in Step 1.

In Genesys Cloud, you invoke another flow using the flow node type in the JSON body. This node allows you to pass variables into the sub-flow and retrieve variables from it.

Required Scope: flow:write

def create_inbound_flow_with_shared_logic(api_client: FlowApi, flow_name: str, shared_flow_id: str):
    """
    Creates an inbound voice flow that invokes a shared flow.
    """
    flow_body = {
        "name": flow_name,
        "type": "voice",
        "enabled": True,
        "version": 1,
        "body": {
            "type": "flow",
            "startNode": "start",
            "nodes": {
                "start": {
                    "type": "start",
                    "name": "Start",
                    "outgoingTransitions": {
                        "default": "invokeSharedAuth"
                    }
                },
                "invokeSharedAuth": {
                    "type": "flow",
                    "name": "Invoke Shared Auth",
                    "flowId": shared_flow_id,
                    # Pass variables into the shared flow if needed
                    "inputVariables": {
                        "context": "${conversation.context}"
                    },
                    # Define what happens based on the sub-flow's exit
                    "outgoingTransitions": {
                        "success": "routeToQueue",
                        "failure": "routeToOperator"
                    }
                },
                "routeToQueue": {
                    "type": "route",
                    "name": "Route to Queue",
                    "queueId": "your_queue_id_here", # Replace with a real queue ID
                    "outgoingTransitions": {
                        "default": "end"
                    }
                },
                "routeToOperator": {
                    "type": "route",
                    "name": "Route to Operator",
                    "queueId": "your_operator_queue_id_here", # Replace with a real queue ID
                    "outgoingTransitions": {
                        "default": "end"
                    }
                },
                "end": {
                    "type": "end",
                    "name": "End",
                    "outgoingTransitions": {}
                }
            }
        }
    }

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

if __name__ == "__main__":
    api = get_flow_api_client()
    # Assuming you have the shared_flow_id from Step 1
    # shared_flow_id = "your-shared-flow-id-from-step-1"
    # create_inbound_flow_with_shared_logic(api, "Sales IVR", shared_flow_id)
    # create_inbound_flow_with_shared_logic(api, "Support IVR", shared_flow_id)

Key Parameter Explanation:

• flowId: This is the critical field. It references the ID of the flow created in Step 1.
• inputVariables: A map of variables passed from the parent flow to the child flow. The key is the variable name in the child flow, and the value is an expression referencing the parent flow’s variables.
• outgoingTransitions: These keys (success, failure) must match the name of the end nodes in the shared flow. When the shared flow hits an end node named “success”, the parent flow transitions to the “routeToQueue” node.

Step 3: Updating the Shared Flow Without Breaking Inbound Flows

A major benefit of this architecture is that you can update the shared flow without touching the inbound flows. However, you must manage Versions.

When you update a flow, you increment the version number. Genesys Cloud supports multiple versions of the same flow. Existing inbound flows will continue to use the version they were configured with unless you explicitly update them.

To update the shared flow:

def update_shared_flow(api_client: FlowApi, flow_id: str, new_version: int):
    """
    Updates an existing flow with a new version.
    """
    # Get the current flow to preserve existing structure
    current_flow = api_client.get_flow(flow_id=flow_id)
    
    # Increment version
    current_flow.version = new_version
    
    # Modify the body as needed (e.g., change the PIN to '5678')
    # This is a simplified example; in reality, you would patch the JSON body
    # For brevity, we assume the body is loaded from a file or previous state
    
    try:
        response = api_client.put_flow(flow_id=flow_id, body=current_flow)
        print(f"Updated Flow {flow_id} to version {new_version}")
        return response
    except ApiException as e:
        print(f"Failed to update flow: {e.status} - {e.reason}")
        raise

if __name__ == "__main__":
    api = get_flow_api_client()
    # update_shared_flow(api, "your-shared-flow-id", 2)

Important Note on Versioning:
If you update the shared flow to version 2, the inbound flows will not automatically switch to version 2. They will continue to use version 1. To switch an inbound flow to use the new version of the shared flow, you must update the inbound flow’s JSON body to reference the new flowId if you created a new flow, or more commonly, since the flowId remains the same for the same logical flow, you must ensure the inbound flow is configured to use the latest version or a specific version.

In the Genesys Cloud API, the flowId reference in the parent flow typically points to the latest version of the child flow by default. However, for strict control, you can pin to a specific version. The API does not expose a direct version field in the flow node reference in all SDK versions, so it is best practice to test thoroughly after updating the shared flow.

Complete Working Example

The following script combines all steps into a single executable module. It creates a shared flow, creates two inbound flows that invoke it, and demonstrates error handling.

import os
import sys
import json
from dotenv import load_dotenv
from purecloudplatformclientv2 import (
    ApiClient,
    Configuration,
    FlowApi,
    ApiException
)

# Load environment variables
load_dotenv()

def get_flow_api_client():
    """Initializes and returns a configured FlowApi client."""
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    region = os.getenv("GENESYS_REGION", "us-east-1")

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

    configuration = Configuration()
    configuration.client_id = client_id
    configuration.client_secret = client_secret
    configuration.host = f"https://{region}.mypurecloud.com"
    
    api_client = ApiClient(configuration=configuration)
    return FlowApi(api_client)

def create_shared_auth_flow(api_client: FlowApi, flow_name: str = "Shared Auth Flow"):
    """Creates a reusable voice flow that collects a PIN."""
    flow_body = {
        "name": flow_name,
        "type": "voice",
        "enabled": True,
        "version": 1,
        "body": {
            "type": "flow",
            "startNode": "start",
            "nodes": {
                "start": {
                    "type": "start",
                    "name": "Start",
                    "outgoingTransitions": {"default": "collectPin"}
                },
                "collectPin": {
                    "type": "collectinput",
                    "name": "Collect PIN",
                    "prompt": "Please enter your 4 digit PIN followed by the pound sign.",
                    "maxDigits": 4,
                    "outgoingTransitions": {"input": "validatePin"}
                },
                "validatePin": {
                    "type": "condition",
                    "name": "Validate PIN",
                    "conditions": [
                        {"expression": "${input.pin} == '1234'", "transition": "success"},
                        {"expression": "${input.pin} != '1234'", "transition": "failure"}
                    ]
                },
                "success": {"type": "end", "name": "Success", "outgoingTransitions": {}},
                "failure": {"type": "end", "name": "Failure", "outgoingTransitions": {}}
            }
        }
    }

    try:
        response = api_client.post_flow(body=flow_body)
        print(f"[SUCCESS] Created Shared Flow with ID: {response.id}")
        return response.id
    except ApiException as e:
        print(f"[ERROR] Failed to create flow: {e.status} - {e.reason}")
        raise

def create_inbound_flow(api_client: FlowApi, flow_name: str, shared_flow_id: str, queue_id: str):
    """Creates an inbound voice flow that invokes a shared flow."""
    flow_body = {
        "name": flow_name,
        "type": "voice",
        "enabled": True,
        "version": 1,
        "body": {
            "type": "flow",
            "startNode": "start",
            "nodes": {
                "start": {
                    "type": "start",
                    "name": "Start",
                    "outgoingTransitions": {"default": "invokeSharedAuth"}
                },
                "invokeSharedAuth": {
                    "type": "flow",
                    "name": "Invoke Shared Auth",
                    "flowId": shared_flow_id,
                    "inputVariables": {"context": "${conversation.context}"},
                    "outgoingTransitions": {
                        "success": "routeToQueue",
                        "failure": "routeToOperator"
                    }
                },
                "routeToQueue": {
                    "type": "route",
                    "name": "Route to Queue",
                    "queueId": queue_id,
                    "outgoingTransitions": {"default": "end"}
                },
                "routeToOperator": {
                    "type": "route",
                    "name": "Route to Operator",
                    "queueId": queue_id, # Using same queue for demo, typically different
                    "outgoingTransitions": {"default": "end"}
                },
                "end": {"type": "end", "name": "End", "outgoingTransitions": {}}
            }
        }
    }

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

def main():
    try:
        # 1. Initialize Client
        api = get_flow_api_client()
        
        # 2. Define Constants
        SHARED_FLOW_NAME = "Shared Auth Flow"
        SALES_QUEUE_ID = "your_sales_queue_id" # Replace with real ID
        SUPPORT_QUEUE_ID = "your_support_queue_id" # Replace with real ID
        
        # 3. Create Shared Flow
        print("--- Step 1: Creating Shared Flow ---")
        shared_flow_id = create_shared_auth_flow(api, SHARED_FLOW_NAME)
        
        # 4. Create Inbound Flows
        print("\n--- Step 2: Creating Inbound Flows ---")
        create_inbound_flow(api, "Sales IVR", shared_flow_id, SALES_QUEUE_ID)
        create_inbound_flow(api, "Support IVR", shared_flow_id, SUPPORT_QUEUE_ID)
        
        print("\n--- Deployment Complete ---")
        
    except Exception as e:
        print(f"Fatal Error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 400 Bad Request - “Invalid Flow Body”

Cause: The JSON structure of the flow body does not match the Genesys Cloud schema. Common issues include missing outgoingTransitions for a node, or mismatched node names in conditions.
Fix: Validate your JSON against the Flow API Schema. Ensure every condition in a condition node references a valid node name in the nodes object.

Error: 403 Forbidden - “Insufficient Permissions”

Cause: The OAuth client lacks the flow:write scope.
Fix: Go to the Genesys Cloud Admin portal, navigate to Security > OAuth, find your client, and ensure flow:write is checked.

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

Cause: The flowId referenced in the parent flow does not exist or has been deleted.
Fix: Verify the shared_flow_id returned in Step 1 is correctly passed to Step 2.

Error: 429 Too Many Requests

Cause: You are creating flows too rapidly. Genesys Cloud enforces rate limits.
Fix: Implement exponential backoff in your retry logic.

import time

def post_flow_with_retry(api_client: FlowApi, body: dict, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            return api_client.post_flow(body=body)
        except ApiException as e:
            if e.status == 429:
                wait_time = 2 ** attempt
                print(f"Rate limited. Retrying in {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise

Official References

• Flow API Documentation
• Flow Node Types Reference
• Python SDK Reference