How to Call a Shared Flow from Multiple Inbound Call Flows

How-To-Articles
1

How to Call a Shared Flow from Multiple Inbound Call Flows

What You Will Build

  • You will build a reusable Flow component that executes shared logic, such as authentication or queue routing, and invoke it from distinct inbound call flows.
  • This tutorial uses the Genesys Cloud Platform API v2 and the Python SDK (genesyscloud).
  • The primary programming language covered is Python, with reference to the underlying REST API structure for other languages.

Prerequisites

  • OAuth Client Type: Public or Confidential client with offline_access scope for long-lived refresh tokens.
  • Required Scopes:
    • flow:flow:read to retrieve Flow definitions.
    • flow:flow:write to update Flow definitions (if creating the subflow trigger via API).
    • flow:flow:execute is generally not required for definition updates but is relevant for testing execution.
  • SDK Version: genesyscloud >= 3.0.0.
  • Language/Runtime: Python 3.9+.
  • External Dependencies: genesyscloud (install via pip install genesyscloud).

Authentication Setup

Genesys Cloud uses OAuth 2.0. For automation scripts, the Client Credentials flow is standard if you have a confidential client, or the Authorization Code flow with PKCE for public clients. The Python SDK handles token caching and refreshing automatically when configured correctly.

import os
from genesyscloud import platform_builder

def get_platform_client():
    """
    Initializes the Genesys Cloud Platform Client.
    Uses environment variables for credentials.
    """
    # Load credentials from environment variables
    client_id = os.environ.get('GENESYS_CLIENT_ID')
    client_secret = os.environ.get('GENESYS_CLIENT_SECRET')
    env_name = os.environ.get('GENESYS_ENV', 'mypurecloud.com')
    
    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")

    # Build the platform client
    builder = platform_builder.PlatformBuilder()
    client = builder.build(
        client_id=client_id,
        client_secret=client_secret,
        env_name=env_name
    )
    
    return client

# Initialize the client
client = get_platform_client()

Critical Note: The platform_builder caches the access token. If the token expires during a long-running script, the SDK automatically fetches a new one using the refresh token or client credentials. You do not need to implement manual refresh logic.

Implementation

Step 1: Identify the Shared Flow ID

Before you can call a shared flow, you must know its UUID. Genesys Cloud Flows are identified by unique strings. You can find this ID in the Admin UI (under Flows) or by querying the API.

If you do not know the ID, query the flows by name. This is useful for scripts that need to be resilient to ID changes across environments (Dev vs. Prod).

from genesyscloud.flow import FlowsApi

def get_flow_by_name(client: FlowsApi, flow_name: str) -> str:
    """
    Retrieves the Flow ID by its name.
    Assumes the name is unique within the organization.
    """
    # Query flows with a limit of 100 and filter by name
    # The API does not support exact name filtering in the query parameters, 
    # so we fetch recent flows and filter client-side for simplicity.
    # In production, cache this ID or use a configuration file.
    
    try:
        response = client.post_flows_query(
            body={
                "pageSize": 100,
                "pageToken": None
            }
        )
        
        if response.entities and len(response.entities) > 0:
            for flow in response.entities:
                if flow.name == flow_name:
                    return flow.id
            
            raise ValueError(f"Flow with name '{flow_name}' not found in the first 100 results.")
        else:
            raise ValueError("No flows found.")
            
    except Exception as e:
        print(f"Error retrieving flow: {e}")
        raise

# Example usage
# shared_flow_id = get_flow_by_name(client.flows_api, "Global Authentication Subflow")
shared_flow_id = "a1b2c3d4-e5f6-7890-abcd-ef1234567890" # Replace with your actual ID

OAuth Scope Required: flow:flow:read

Step 2: Create the “Call Subflow” Trigger Definition

In Genesys Cloud, calling another flow is handled by the Call Subflow trigger. This trigger is defined within the JSON structure of the parent flow. You do not “invoke” it via a runtime API call; you configure it into the flow definition and then update the flow.

The callsubflow trigger requires:

  1. flowId: The UUID of the target shared flow.
  2. arguments: (Optional) A map of variables to pass to the subflow.
  3. returnVariables: (Optional) Variables to map from the subflow’s return values back to the parent flow.

Here is how to construct the trigger object using the Python SDK models.

from genesyscloud.flow.models import Trigger, TriggerProperties

def create_call_subflow_trigger(flow_id: str, trigger_id: str) -> Trigger:
    """
    Creates a Trigger object for calling a subflow.
    
    Args:
        flow_id: The UUID of the shared flow to call.
        trigger_id: A unique ID for this trigger within the parent flow.
    
    Returns:
        A configured Trigger object.
    """
    
    # Define the trigger properties
    trigger_properties = TriggerProperties(
        type='callsubflow',
        flowId=flow_id,
        # Optional: Pass variables to the subflow
        # arguments={
        #     "caller_phone": "${callerPhoneNumber}",
        #     "intent": "authentication"
        # },
        # Optional: Map return values from subflow
        # returnVariables={
        #     "auth_result": "authenticationStatus"
        # }
    )
    
    # Create the Trigger wrapper
    trigger = Trigger(
        id=trigger_id,
        type='callsubflow',
        properties=trigger_properties
    )
    
    return trigger

# Create the trigger instance
subflow_trigger = create_call_subflow_trigger(
    flow_id=shared_flow_id, 
    trigger_id="trigger_call_global_auth"
)

Why this design? Genesys Cloud Flows are event-driven graphs. A “Call Subflow” is not a function call in the traditional programming sense; it is a transition in the state machine that hands control over to another graph. The properties object dictates the handoff contract.

Step 3: Update the Parent Flow to Include the Trigger

To make the shared flow callable from a specific parent flow, you must update the parent flow’s definition to include the new trigger and ensure the flow’s outcomes or edges point to it.

For this tutorial, we will assume a simple linear flow:

  1. Start
  2. Call Subflow (The trigger we just created)
  3. End

We need to retrieve the current flow, inject the trigger, define the edges, and save it.

from genesyscloud.flow.models import Flow, Edge

def update_parent_flow_with_subflow(client: FlowsApi, parent_flow_id: str, subflow_trigger: Trigger, subflow_flow_id: str):
    """
    Updates a parent flow to call a subflow.
    
    Args:
        client: The Genesys Cloud Platform Client.
        parent_flow_id: The UUID of the flow being modified.
        subflow_trigger: The Trigger object created in Step 2.
        subflow_flow_id: The ID of the shared flow (for validation).
    """
    
    # 1. Get the current flow definition
    try:
        response = client.get_flow(parent_flow_id)
        current_flow = response
    except Exception as e:
        print(f"Failed to get parent flow: {e}")
        return

    # 2. Prepare the new triggers list
    # We must preserve existing triggers if any, or create a new list.
    new_triggers = []
    if current_flow.triggers:
        new_triggers = current_flow.triggers
    
    # Check if trigger already exists to avoid duplication
    existing_trigger_ids = [t.id for t in new_triggers]
    if subflow_trigger.id not in existing_trigger_ids:
        new_triggers.append(subflow_trigger)
    else:
        print("Trigger already exists. Updating properties if necessary.")
        # Find and replace if needed
        for i, t in enumerate(new_triggers):
            if t.id == subflow_trigger.id:
                new_triggers[i] = subflow_trigger
                break

    # 3. Define Edges
    # In Genesys Cloud, edges connect triggers.
    # A simple flow often has a "start" virtual trigger pointing to the first real trigger.
    # However, the API allows defining explicit edges between triggers.
    
    # For this example, we assume the flow starts with our subflow call.
    # We need to ensure there is an edge from the "start" to our subflow trigger.
    # Note: The "start" node is implicit in some UI views but explicit in API edges as 'start'.
    
    new_edges = []
    if current_flow.edges:
        new_edges = current_flow.edges
    
    # Remove any existing edges pointing to or from the subflow trigger to avoid conflicts
    new_edges = [e for e in new_edges if e.fromId != subflow_trigger.id and e.toId != subflow_trigger.id]
    
    # Add edge from Start to Subflow Trigger
    start_edge = Edge(
        fromId='start',
        toId=subflow_trigger.id,
        conditions=[] # No condition, always execute
    )
    new_edges.append(start_edge)
    
    # Add edge from Subflow Trigger to End (or another trigger)
    # Assuming we just want to end after the subflow for this demo
    end_edge = Edge(
        fromId=subflow_trigger.id,
        toId='end',
        conditions=[]
    )
    new_edges.append(end_edge)

    # 4. Update the Flow object
    current_flow.triggers = new_triggers
    current_flow.edges = new_edges
    
    # Ensure the flow is enabled if it was enabled
    # current_flow.enabled = True 

    # 5. Save the flow
    try:
        # Put the updated flow back
        client.put_flow(
            flow_id=parent_flow_id,
            body=current_flow
        )
        print(f"Successfully updated parent flow {parent_flow_id} to call subflow {subflow_flow_id}.")
        
    except Exception as e:
        print(f"Failed to update parent flow: {e}")
        # Handle 409 Conflict (version mismatch) or 400 Bad Request (invalid JSON)
        if hasattr(e, 'body'):
            print(f"Error Details: {e.body}")

# Usage
# update_parent_flow_with_subflow(client.flows_api, "parent-flow-uuid", subflow_trigger, shared_flow_id)

OAuth Scope Required: flow:flow:write

Error Handling Note: If you receive a 409 Conflict, it means the flow was modified by another user since you fetched it. You must implement an optimistic locking retry mechanism: fetch again, re-apply changes, and try to save.

Step 4: Handling Variables and Return Values

The power of subflows lies in data exchange. When you define the callsubflow trigger, you map variables.

Passing Data In:
In the arguments map of the trigger properties, you use ${variableName} syntax. This resolves to the value of variableName in the parent flow context at the time of execution.

Receiving Data Out:
In the returnVariables map, you define {parent_var_name: subflow_var_name}. When the subflow completes (exits via an “End” or “Return” trigger), Genesys Cloud copies the value of subflow_var_name into parent_var_name.

Here is how to programmatically set these mappings for a robust integration.

from genesyscloud.flow.models import TriggerProperties

def create_advanced_subflow_trigger(flow_id: str, trigger_id: str):
    """
    Creates a trigger with explicit variable passing.
    """
    
    # Define inputs to the subflow
    arguments = {
        "input_phone_number": "${callerPhoneNumber}",
        "input_language": "${language}"
    }
    
    # Define outputs from the subflow
    # The subflow must set these variables before exiting
    return_variables = {
        "auth_success": "authenticationResult",
        "customer_id": "resolvedCustomerId"
    }
    
    trigger_props = TriggerProperties(
        type='callsubflow',
        flowId=flow_id,
        arguments=arguments,
        returnVariables=return_variables
    )
    
    return Trigger(
        id=trigger_id,
        type='callsubflow',
        properties=trigger_props
    )

Best Practice: Ensure the variable names in returnVariables match exactly the variable names set in the shared flow. Genesys Cloud variable names are case-sensitive.

Complete Working Example

This script combines all steps into a single executable module. It assumes you have two flows: one “Shared” flow and one “Parent” flow. It updates the Parent flow to call the Shared flow.

import os
import sys
from genesyscloud import platform_builder
from genesyscloud.flow import FlowsApi
from genesyscloud.flow.models import Trigger, TriggerProperties, Edge, Flow

def main():
    # 1. Authentication
    client_id = os.environ.get('GENESYS_CLIENT_ID')
    client_secret = os.environ.get('GENESYS_CLIENT_SECRET')
    env_name = os.environ.get('GENESYS_ENV', 'mypurecloud.com')
    
    if not client_id or not client_secret:
        print("Error: GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET environment variables are required.")
        sys.exit(1)

    builder = platform_builder.PlatformBuilder()
    client = builder.build(
        client_id=client_id,
        client_secret=client_secret,
        env_name=env_name
    )
    
    flows_api = client.flows_api

    # 2. Configuration
    SHARED_FLOW_NAME = "Global Auth Subflow"
    PARENT_FLOW_ID = "your-parent-flow-uuid-here" # Replace with actual ID
    TRIGGER_ID = "trigger_call_global_auth"

    try:
        # 3. Find Shared Flow ID
        print(f"Searching for shared flow: {SHARED_FLOW_NAME}")
        response = flows_api.post_flows_query(body={"pageSize": 100})
        shared_flow_id = None
        
        if response.entities:
            for flow in response.entities:
                if flow.name == SHARED_FLOW_NAME:
                    shared_flow_id = flow.id
                    break
        
        if not shared_flow_id:
            print(f"Error: Could not find flow named '{SHARED_FLOW_NAME}'.")
            sys.exit(1)
            
        print(f"Found shared flow ID: {shared_flow_id}")

        # 4. Create Subflow Trigger
        subflow_trigger = Trigger(
            id=TRIGGER_ID,
            type='callsubflow',
            properties=TriggerProperties(
                type='callsubflow',
                flowId=shared_flow_id,
                arguments={
                    "caller_phone": "${callerPhoneNumber}"
                },
                returnVariables={
                    "auth_status": "authenticationResult"
                }
            )
        )

        # 5. Update Parent Flow
        print(f"Updating parent flow: {PARENT_FLOW_ID}")
        
        # Fetch current flow
        current_flow = flows_api.get_flow(PARENT_FLOW_ID).body
        
        # Prepare triggers
        new_triggers = current_flow.triggers or []
        existing_ids = [t.id for t in new_triggers]
        
        if TRIGGER_ID not in existing_ids:
            new_triggers.append(subflow_trigger)
        else:
            # Replace existing trigger
            for i, t in enumerate(new_triggers):
                if t.id == TRIGGER_ID:
                    new_triggers[i] = subflow_trigger
                    break
        
        # Prepare edges
        new_edges = current_flow.edges or []
        # Clean up old edges related to this trigger
        new_edges = [e for e in new_edges if e.fromId != TRIGGER_ID and e.toId != TRIGGER_ID]
        
        # Add Start -> Subflow Edge
        new_edges.append(Edge(fromId='start', toId=TRIGGER_ID, conditions=[]))
        
        # Add Subflow -> End Edge
        new_edges.append(Edge(fromId=TRIGGER_ID, toId='end', conditions=[]))
        
        # Update flow object
        current_flow.triggers = new_triggers
        current_flow.edges = new_edges
        
        # Save
        flows_api.put_flow(flow_id=PARENT_FLOW_ID, body=current_flow)
        print("Success: Parent flow updated to call shared subflow.")

    except Exception as e:
        print(f"An error occurred: {e}")
        if hasattr(e, 'body'):
            print(f"API Error Body: {e.body}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 409 Conflict

  • What causes it: The flow was modified by another user or process after you fetched it but before you saved it. Genesys Cloud uses optimistic locking.
  • How to fix it: Implement a retry loop. Fetch the flow again, re-apply your changes (merging carefully with new changes if any), and attempt the put_flow again.
  • Code Fix:
    retries = 3
for i in range(retries):
    try:
        current_flow = flows_api.get_flow(PARENT_FLOW_ID).body
        # ... apply changes ...
        flows_api.put_flow(flow_id=PARENT_FLOW_ID, body=current_flow)
        break
    except Exception as e:
        if "Conflict" in str(e) or "409" in str(e):
            if i < retries - 1:
                continue
        raise

Error: 400 Bad Request - Invalid Edge Configuration

  • What causes it: The edges do not form a valid graph. Common issues include circular dependencies, missing “start” or “end” nodes, or edges pointing to non-existent triggers.
  • How to fix it: Verify that every trigger has an incoming edge (except those reachable from ‘start’) and an outgoing edge (except those leading to ‘end’). Ensure fromId and toId match existing trigger IDs or the literals ‘start’/‘end’.

Error: Variable Mapping Not Working

  • What causes it: Typo in variable names or case sensitivity mismatch.
  • How to fix it: Use the Genesys Cloud Flow UI to inspect the variable names in both the parent and subflow. Ensure the returnVariables map in the API payload matches exactly. Test with a simple string variable first.

Official References