Reuse Shared Flows in Genesys Cloud via API

How-To-Articles
1

Reuse Shared Flows in Genesys Cloud via API

What You Will Build

  • A Python script that programmatically updates an inbound IVR flow to invoke a shared sub-flow using the Genesys Cloud Platform SDK.
  • This solution uses the Genesys Cloud PureCloudPlatformClientV2 SDK to manage Flow definitions and Flow references.
  • The programming language covered is Python 3.9+.

Prerequisites

  • OAuth Client: A Genesys Cloud OAuth Client with client_credentials grant type.
  • Required Scopes: flow:flow:write (to update flows), flow:flow:read (to read existing flows).
  • SDK Version: genesys-cloud-purecloud-platform-client v138.0.0 or later.
  • Runtime: Python 3.9 or higher.
  • External Dependencies:
    • genesys-cloud-purecloud-platform-client
    • python-dotenv (for secure credential management)

Authentication Setup

Genesys Cloud APIs require OAuth 2.0 Bearer tokens. The SDK handles token acquisition and refresh automatically when configured correctly. You must store your Client ID and Client Secret securely. Never hardcode these values.

Create a .env file in your project root:

GENESYS_CLOUD_REGION=us-east-1 # or us-east-2, eu-west-1, etc.
GENESYS_CLOUD_CLIENT_ID=your_client_id_here
GENESYS_CLOUD_CLIENT_SECRET=your_client_secret_here

Initialize the API client in your Python script:

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

def get_flow_api_client():
    """
    Initializes and returns a configured FlowApi instance.
    """
    load_dotenv()

    # Load credentials
    client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
    region = os.getenv("GENESYS_CLOUD_REGION", "us-east-1")

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

    # Configure OAuth
    configuration = Configuration()
    configuration.client_id = client_id
    configuration.client_secret = client_secret
    
    # Set region if necessary (SDK defaults to us-east-1 if not specified)
    if region != "us-east-1":
        configuration.region = region

    # Create API client
    api_client = ApiClient(configuration)
    
    # Initialize Flow API
    flow_api = FlowApi(api_client)
    
    return flow_api

Implementation

Step 1: Retrieve the Target Inbound Flow

To update a flow, you must first retrieve its current definition. Genesys Cloud Flows are versioned. You must read the latest version to ensure you are not overwriting concurrent changes.

The endpoint used is GET /api/v2/flows/{flowId}.

def get_flow_definition(flow_api: FlowApi, flow_id: str) -> dict:
    """
    Retrieves the full definition of a flow.
    
    Args:
        flow_api: The initialized FlowApi client.
        flow_id: The UUID of the flow to retrieve.
        
    Returns:
        The flow definition dictionary.
    """
    try:
        # The SDK method gets_flow_by_id returns a Flow object
        # We convert it to a dict for easier manipulation
        flow_response = flow_api.get_flow_by_id(flow_id)
        
        # Convert SDK object to dictionary
        flow_dict = {
            "id": flow_response.id,
            "version": flow_response.version,
            "name": flow_response.name,
            "type": flow_response.type,
            "description": flow_response.description,
            "outcomes": flow_response.outcomes,
            "settings": flow_response.settings,
            "actions": flow_response.actions,
            "outcomesMap": flow_response.outcomes_map,
            "outcomesArray": flow_response.outcomes_array
        }
        
        return flow_dict

    except Exception as e:
        # Handle 404 (Flow not found) or 401/403 (Auth issues)
        if hasattr(e, 'status') and e.status == 404:
            raise ValueError(f"Flow with ID {flow_id} not found.")
        elif hasattr(e, 'status') and e.status in [401, 403]:
            raise PermissionError(f"Authentication or Authorization failed: {e.reason}")
        else:
            raise e

Step 2: Define the Shared Sub-Flow Reference

A shared flow is invoked using an invokeflow action. This action requires the ID of the shared flow and a mapping of inputs/outputs.

Assume you have a shared flow with ID shared-flow-uuid-123. You want to pass the caller’s phone number as an input named callerNumber and receive a result named routingDecision.

The critical structure for the action is:

{
  "type": "invokeflow",
  "name": "InvokeSharedRouting",
  "description": "Calls the shared routing logic",
  "inputs": {
    "callerNumber": "${caller.phoneNumber}"
  },
  "outputs": {
    "routingDecision": "sharedRoutingResult"
  },
  "flowId": "shared-flow-uuid-123",
  "onSuccess": "nextActionId",
  "onError": "errorActionId"
}

In the SDK, you construct this as a dictionary or use the specific model classes. Using dictionaries is often simpler for bulk updates.

Step 3: Insert the Action into the Flow Definition

You must insert the invokeflow action into the actions list of the retrieved flow. Ensure you assign a unique ID to the new action and update the outcomesMap if necessary (though invokeflow typically relies on onSuccess/onError transitions).

def add_invoke_flow_action(flow_dict: dict, shared_flow_id: str, next_action_id: str, error_action_id: str) -> dict:
    """
    Adds an invokeflow action to the flow definition.
    
    Args:
        flow_dict: The current flow definition dictionary.
        shared_flow_id: The ID of the shared flow to invoke.
        next_action_id: The ID of the action to run on success.
        error_action_id: The ID of the action to run on error.
        
    Returns:
        The updated flow definition dictionary.
    """
    import uuid
    
    # Generate a unique ID for the new action
    new_action_id = str(uuid.uuid4())
    
    # Define the invokeflow action
    invoke_action = {
        "id": new_action_id,
        "type": "invokeflow",
        "name": "InvokeSharedLogic",
        "description": "Delegates to shared flow for common logic",
        "inputs": {
            "callerNumber": "${caller.phoneNumber}",
            "language": "${caller.language}"
        },
        "outputs": {
            "result": "sharedFlowResult"
        },
        "flowId": shared_flow_id,
        "onSuccess": next_action_id,
        "onError": error_action_id
    }
    
    # Append the action to the existing actions list
    if "actions" not in flow_dict:
        flow_dict["actions"] = []
        
    flow_dict["actions"].append(invoke_action)
    
    # Note: You must also ensure that the previous action in the flow 
    # points to this new action's ID as its 'next' step.
    # For this example, we assume the first action is the entry point 
    # and we are inserting this as the second action.
    
    if len(flow_dict["actions"]) > 1:
        # Update the first action to point to our new invoke action
        first_action = flow_dict["actions"][0]
        if "next" in first_action:
            first_action["next"] = new_action_id
        else:
            first_action["next"] = new_action_id

    return flow_dict

Step 4: Update the Flow

To save the changes, you must send a PUT request to /api/v2/flows/{flowId}. The request body must include the version number from the retrieved flow. Genesys Cloud uses optimistic locking; if the version number does not match the current server version, the update will fail with a 409 Conflict.

def update_flow_definition(flow_api: FlowApi, flow_id: str, flow_dict: dict) -> str:
    """
    Updates the flow definition in Genesys Cloud.
    
    Args:
        flow_api: The initialized FlowApi client.
        flow_id: The UUID of the flow to update.
        flow_dict: The updated flow definition dictionary.
        
    Returns:
        The ID of the updated flow.
    """
    # Construct the request body
    # The SDK expects a Flow object or a dict that maps to Flow fields
    # We must ensure 'version' is included for optimistic locking
    
    try:
        # Update the flow
        # The SDK method update_flow takes flow_id and the flow object/dict
        response = flow_api.update_flow(flow_id=flow_id, body=flow_dict)
        
        return response.id

    except Exception as e:
        if hasattr(e, 'status'):
            if e.status == 409:
                raise ConflictError("Flow version conflict. Another user may have modified the flow. Please retry.")
            elif e.status == 422:
                raise ValueError(f"Unprocessable Entity. Invalid flow definition: {e.reason}")
        raise e

Complete Working Example

This script retrieves an inbound flow, inserts an invokeflow action to call a shared flow, and saves the changes.

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

def get_flow_api_client():
    load_dotenv()
    client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
    region = os.getenv("GENESYS_CLOUD_REGION", "us-east-1")

    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLOUD_CLIENT_ID and GENESYS_CLOUD_CLIENT_SECRET must be set.")

    configuration = Configuration()
    configuration.client_id = client_id
    configuration.client_secret = client_secret
    if region != "us-east-1":
        configuration.region = region

    api_client = ApiClient(configuration)
    return FlowApi(api_client)

def main():
    # Configuration
    INBOUND_FLOW_ID = os.getenv("INBOUND_FLOW_ID", "your-inbound-flow-uuid")
    SHARED_FLOW_ID = os.getenv("SHARED_FLOW_ID", "your-shared-flow-uuid")
    NEXT_ACTION_ID = "existing-action-id-after-invoke" # Replace with actual ID from your flow
    ERROR_ACTION_ID = "error-handling-action-id"       # Replace with actual ID

    if INBOUND_FLOW_ID == "your-inbound-flow-uuid" or SHARED_FLOW_ID == "your-shared-flow-uuid":
        print("Error: Please set INBOUND_FLOW_ID and SHARED_FLOW_ID in your .env file.")
        sys.exit(1)

    try:
        # 1. Initialize Client
        flow_api = get_flow_api_client()
        print(f"Authenticating...")

        # 2. Get Current Flow
        print(f"Retrieving flow {INBOUND_FLOW_ID}...")
        flow_response = flow_api.get_flow_by_id(INBOUND_FLOW_ID)
        
        # Convert to dict for manipulation
        flow_dict = {
            "id": flow_response.id,
            "version": flow_response.version,
            "name": flow_response.name,
            "type": flow_response.type,
            "description": flow_response.description,
            "outcomes": flow_response.outcomes,
            "settings": flow_response.settings,
            "actions": flow_response.actions,
            "outcomesMap": flow_response.outcomes_map,
            "outcomesArray": flow_response.outcomes_array
        }

        # 3. Prepare InvokeFlow Action
        new_action_id = str(uuid.uuid4())
        invoke_action = {
            "id": new_action_id,
            "type": "invokeflow",
            "name": "CallSharedFlow",
            "description": "Invokes the shared business logic flow",
            "inputs": {
                "callerNumber": "${caller.phoneNumber}"
            },
            "outputs": {
                "result": "sharedResult"
            },
            "flowId": SHARED_FLOW_ID,
            "onSuccess": NEXT_ACTION_ID,
            "onError": ERROR_ACTION_ID
        }

        # 4. Insert Action into Flow
        # For simplicity, we append to the end and assume the previous action 
        # will be manually updated or this is a new flow. 
        # In production, you must update the 'next' pointer of the preceding action.
        
        if not flow_dict.get("actions"):
            flow_dict["actions"] = []
            
        # Example: Insert at index 1 (after the first action)
        if len(flow_dict["actions"]) > 0:
            # Update the last action's 'next' to point to our new action
            last_action = flow_dict["actions"][-1]
            last_action["next"] = new_action_id
            
        flow_dict["actions"].append(invoke_action)

        # 5. Save Flow
        print(f"Updating flow {INBOUND_FLOW_ID}...")
        update_response = flow_api.update_flow(INBOUND_FLOW_ID, body=flow_dict)
        
        print(f"Success! Flow updated. New Version: {update_response.version}")

    except Exception as e:
        print(f"Error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 409 Conflict (Version Mismatch)

What causes it: You retrieved the flow, modified it, and attempted to save it, but another user or process updated the flow in between. The version number in your request body does not match the current server version.

How to fix it: Implement a retry mechanism that re-fetches the flow, re-applies your changes, and retries the update.

def update_flow_with_retry(flow_api: FlowApi, flow_id: str, flow_dict: dict, max_retries: int = 3) -> str:
    for attempt in range(max_retries):
        try:
            response = flow_api.update_flow(flow_id=flow_id, body=flow_dict)
            return response.id
        except Exception as e:
            if hasattr(e, 'status') and e.status == 409:
                # Re-fetch the flow to get the latest version
                fresh_flow = flow_api.get_flow_by_id(flow_id)
                # Re-apply your changes to the fresh flow data
                flow_dict["version"] = fresh_flow.version
                # Re-apply other modifications here
                continue
            else:
                raise e
    raise Exception("Max retries exceeded due to version conflicts.")

Error: 422 Unprocessable Entity (Invalid Flow Definition)

What causes it: The flow definition JSON is invalid. Common causes include:

  • Missing next pointer in an action.
  • Referencing a non-existent action ID in onSuccess or onError.
  • Circular references in the flow logic.

How to fix it: Validate the flow structure before sending. Ensure every action has a valid next, onSuccess, or onError pointer that references an existing action ID in the same flow.

Error: 403 Forbidden (Insufficient Scopes)

What causes it: The OAuth token does not have the flow:flow:write scope.

How to fix it: Regenerate the OAuth token using a client that has the flow:flow:write scope assigned in the Genesys Cloud Admin Console under Users > Applications > OAuth.

Official References