How to Call a Shared Flow from Multiple Inbound Call Flows via API

What You Will Build

• You will programmatically create an inbound call flow that invokes a pre-existing shared flow using the Genesys Cloud CX API.
• This tutorial uses the Genesys Cloud CX Python SDK (genesys-cloud-sdk) to manipulate Call Center routing configurations.
• The implementation covers Python, with conceptual mappings for JavaScript/TypeScript.

Prerequisites

• OAuth Client: A Genesys Cloud OAuth client with the admin or routing:callflow:write scope.
• SDK Version: genesys-cloud-sdk v5.0.0 or later (Python).
• Language/Runtime: Python 3.9+ or Node.js 18+ (if following JS examples).
• External Dependencies: pip install genesys-cloud-sdk.
• Existing Resources: You must have at least one existing Shared Flow in your Genesys Cloud organization to reference. You can find the ID by navigating to Admin > Routing > Flows > Shared Flows, or by querying the API: GET /api/v2/routing/flows?type=shared.

Authentication Setup

Genesys Cloud uses OAuth 2.0 for API authentication. For server-to-server integrations, the Client Credentials flow is the standard approach. The SDK handles token caching and refresh automatically, but you must initialize the PlatformClient with your credentials.

import os
from purecloudplatformclientv2 import PlatformClient
from purecloudplatformclientv2 import RoutingApi

def get_platform_client() -> PlatformClient:
    """
    Initializes and returns an authenticated PlatformClient instance.
    Uses environment variables for security.
    """
    client_id = os.environ.get("GENESYS_CLIENT_ID")
    client_secret = os.environ.get("GENESYS_CLIENT_SECRET")
    environment = os.environ.get("GENESYS_ENVIRONMENT", "mypurecloud.com") # e.g., usw2.pure.cloud

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

    platform_client = PlatformClient(client_id=client_id, client_secret=client_secret)
    platform_client.set_environment(environment)
    
    return platform_client

# Initialize the client
platform_client = get_platform_client()
routing_api = RoutingApi(platform_client)

Required Scope: routing:callflow:write

Implementation

Step 1: Identify the Target Shared Flow

Before creating the inbound flow, you must know the ID of the shared flow you intend to reuse. Shared flows are reusable logic blocks (e.g., authentication, menu routing, or post-call survey) that can be invoked by other flows.

We will query the system for a specific shared flow by name. This ensures the code is resilient to ID changes if you recreate the flow, though using a static ID is also valid in production if the flow is stable.

from purecloudplatformclientv2 import RoutingApi
from purecloudplatformclientv2.models import FlowQueryRequest

def find_shared_flow_by_name(api: RoutingApi, flow_name: str) -> str:
    """
    Queries Genesys Cloud for a shared flow by name and returns its ID.
    
    Args:
        api: The initialized RoutingApi instance.
        flow_name: The exact name of the shared flow.
        
    Returns:
        The UUID string of the shared flow.
        
    Raises:
        ValueError: If the flow is not found.
    """
    # Construct the query to filter by type=shared and name
    query_request = FlowQueryRequest(
        type="shared",
        name=flow_name
    )
    
    try:
        response = api.post_routing_flows_query(body=query_request)
        
        if not response.entities or len(response.entities) == 0:
            raise ValueError(f"Shared flow '{flow_name}' not found.")
            
        # Return the ID of the first matching flow
        return response.entities[0].id
        
    except Exception as e:
        print(f"Error querying flows: {e}")
        raise

# Example usage
SHARED_FLOW_ID = find_shared_flow_by_name(routing_api, "Global Authentication Menu")
print(f"Found Shared Flow ID: {SHARED_FLOW_ID}")

Expected Response: The post_routing_flows_query endpoint returns a FlowEntityListing object. The entities array contains Flow objects. Each Flow object has an id field (UUID).

Error Handling: If the flow name does not exist, the API returns an empty entities array. The code above raises a ValueError to halt execution, preventing the creation of an invalid inbound flow.

Step 2: Construct the Inbound Call Flow with Invoke Action

The core of this tutorial is constructing the JSON structure for an inbound call flow. In Genesys Cloud, flows are defined as a directed graph of nodes. To call a shared flow, you use the invoke action.

The invoke action requires:

1. flowId: The UUID of the shared flow.
2. onComplete: The node ID to transition to after the shared flow finishes.
3. onError: The node ID to transition to if the shared flow fails or times out.

We will build a simple inbound flow:

1. Start Node: Entry point.
2. Invoke Node: Calls the shared flow.
3. Queue Node: Handles the call after authentication (if successful).
4. Hangup Node: Handles errors or timeouts.

from purecloudplatformclientv2.models import (
    Flow,
    FlowNode,
    InvokeNode,
    QueueNode,
    HangupNode,
    FlowEdge,
    FlowEdgeType
)
import uuid

def build_inbound_flow_with_invoke(shared_flow_id: str, queue_id: str) -> Flow:
    """
    Constructs a Flow object that invokes a shared flow and then queues the call.
    
    Args:
        shared_flow_id: UUID of the shared flow to invoke.
        queue_id: UUID of the queue to place the call into after invocation.
        
    Returns:
        A fully configured Flow object ready for API submission.
    """
    
    # Generate unique node IDs for this flow structure
    start_node_id = str(uuid.uuid4())
    invoke_node_id = str(uuid.uuid4())
    queue_node_id = str(uuid.uuid4())
    hangup_node_id = str(uuid.uuid4())

    # 1. Define the Nodes
    
    # Start Node: No action, just a pass-through to the invoke node
    start_node = FlowNode(
        id=start_node_id,
        name="Start",
        type="start"
    )

    # Invoke Node: Calls the shared flow
    invoke_node = FlowNode(
        id=invoke_node_id,
        name="Invoke Shared Auth Flow",
        type="invoke",
        action=InvokeNode(
            flow_id=shared_flow_id,
            # Optional: Pass variables to the shared flow
            # variables=[{"name": "callerId", "value": "${CallerId}"}]
        )
    )

    # Queue Node: Places the call in a queue if auth succeeded
    queue_node = FlowNode(
        id=queue_node_id,
        name="Queue to Support",
        type="queue",
        action=QueueNode(
            queue_id=queue_id,
            # Optional: Set skills, labels, etc.
        )
    )

    # Hangup Node: Ends the call if auth failed or timed out
    hangup_node = FlowNode(
        id=hangup_node_id,
        name="Hangup on Error",
        type="hangup",
        action=HangupNode(
            reason="Caller disconnected or authentication failed"
        )
    )

    # 2. Define the Edges (Connections)
    
    edges = [
        # Start -> Invoke
        FlowEdge(
            id=str(uuid.uuid4()),
            from_node_id=start_node_id,
            to_node_id=invoke_node_id,
            type=FlowEdgeType.TRANSITION
        ),
        # Invoke -> Queue (on success)
        FlowEdge(
            id=str(uuid.uuid4()),
            from_node_id=invoke_node_id,
            to_node_id=queue_node_id,
            type=FlowEdgeType.ON_COMPLETE
        ),
        # Invoke -> Hangup (on error/timeout)
        FlowEdge(
            id=str(uuid.uuid4()),
            from_node_id=invoke_node_id,
            to_node_id=hangup_node_id,
            type=FlowEdgeType.ON_ERROR
        )
    ]

    # 3. Assemble the Flow Object
    
    flow = Flow(
        name="Inbound Flow - Uses Shared Auth",
        description="Inbound flow that invokes a global authentication shared flow before queuing.",
        type="inbound",
        nodes=[start_node, invoke_node, queue_node, hangup_node],
        edges=edges,
        start_node_id=start_node_id
    )
    
    return flow

Key Configuration Details:

• type="invoke": This tells the engine to pause the current flow and execute the target flow.
• onComplete Edge: This is the critical path for success. When the shared flow reaches its end node (or a return action), execution resumes here.
• onError Edge: This handles scenarios where the shared flow is disabled, times out, or encounters an internal error. It is crucial for robustness.
• Variables: You can pass data into the shared flow using the variables array in the InvokeNode. These variables become available within the shared flow context.

Step 3: Create the Flow via API

Once the Flow object is constructed, you send it to the post_routing_flow endpoint.

def create_inbound_flow(api: RoutingApi, flow_obj: Flow) -> str:
    """
    Creates the flow in Genesys Cloud and returns the new flow ID.
    
    Args:
        api: The initialized RoutingApi instance.
        flow_obj: The Flow object constructed in Step 2.
        
    Returns:
        The ID of the newly created flow.
    """
    try:
        response = api.post_routing_flow(body=flow_obj)
        print(f"Successfully created flow: {response.name} (ID: {response.id})")
        return response.id
        
    except Exception as e:
        # Handle specific API errors
        if hasattr(e, 'status') and e.status == 409:
            print("Conflict: A flow with this name or configuration may already exist.")
        elif hasattr(e, 'status') and e.status == 400:
            print(f"Bad Request: Invalid flow configuration. Details: {e.body}")
        else:
            print(f"Unexpected error creating flow: {e}")
        raise

Expected Response: A 201 Created status with the full Flow object, including the newly assigned id.

Complete Working Example

This script combines all steps into a single executable module. It assumes you have environment variables set for authentication and that you have a valid Queue ID and Shared Flow Name.

import os
import sys
from purecloudplatformclientv2 import PlatformClient, RoutingApi
from purecloudplatformclientv2.models import (
    Flow, FlowNode, InvokeNode, QueueNode, HangupNode, 
    FlowEdge, FlowEdgeType, FlowQueryRequest
)
import uuid

def main():
    # 1. Setup Authentication
    if not os.environ.get("GENESYS_CLIENT_ID") or not os.environ.get("GENESYS_CLIENT_SECRET"):
        print("Error: GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET environment variables are required.")
        sys.exit(1)

    platform_client = PlatformClient(
        client_id=os.environ.get("GENESYS_CLIENT_ID"),
        client_secret=os.environ.get("GENESYS_CLIENT_SECRET")
    )
    platform_client.set_environment(os.environ.get("GENESYS_ENVIRONMENT", "mypurecloud.com"))
    
    routing_api = RoutingApi(platform_client)

    # 2. Configuration Inputs
    SHARED_FLOW_NAME = "Global Authentication Menu" # Must exist
    QUEUE_ID = os.environ.get("GENESYS_QUEUE_ID")    # Must exist
    
    if not QUEUE_ID:
        print("Error: GENESYS_QUEUE_ID environment variable is required.")
        sys.exit(1)

    try:
        # 3. Find Shared Flow ID
        print(f"Looking for shared flow: '{SHARED_FLOW_NAME}'...")
        shared_flow_id = find_shared_flow_by_name(routing_api, SHARED_FLOW_NAME)
        print(f"Found Shared Flow ID: {shared_flow_id}")

        # 4. Build Inbound Flow
        print("Constructing inbound flow structure...")
        inbound_flow = build_inbound_flow_with_invoke(shared_flow_id, QUEUE_ID)

        # 5. Create Flow
        print("Creating inbound flow in Genesys Cloud...")
        new_flow_id = create_inbound_flow(routing_api, inbound_flow)
        
        print(f"Success! Inbound Flow created with ID: {new_flow_id}")
        
    except ValueError as ve:
        print(f"Configuration Error: {ve}")
    except Exception as e:
        print(f"Execution Error: {e}")

def find_shared_flow_by_name(api: RoutingApi, flow_name: str) -> str:
    query_request = FlowQueryRequest(type="shared", name=flow_name)
    response = api.post_routing_flows_query(body=query_request)
    
    if not response.entities or len(response.entities) == 0:
        raise ValueError(f"Shared flow '{flow_name}' not found.")
        
    return response.entities[0].id

def build_inbound_flow_with_invoke(shared_flow_id: str, queue_id: str) -> Flow:
    start_node_id = str(uuid.uuid4())
    invoke_node_id = str(uuid.uuid4())
    queue_node_id = str(uuid.uuid4())
    hangup_node_id = str(uuid.uuid4())

    start_node = FlowNode(id=start_node_id, name="Start", type="start")
    
    invoke_node = FlowNode(
        id=invoke_node_id,
        name="Invoke Shared Auth",
        type="invoke",
        action=InvokeNode(flow_id=shared_flow_id)
    )
    
    queue_node = FlowNode(
        id=queue_node_id,
        name="Queue Call",
        type="queue",
        action=QueueNode(queue_id=queue_id)
    )
    
    hangup_node = FlowNode(
        id=hangup_node_id,
        name="Hangup",
        type="hangup",
        action=HangupNode(reason="Auth failed or timeout")
    )

    edges = [
        FlowEdge(id=str(uuid.uuid4()), from_node_id=start_node_id, to_node_id=invoke_node_id, type=FlowEdgeType.TRANSITION),
        FlowEdge(id=str(uuid.uuid4()), from_node_id=invoke_node_id, to_node_id=queue_node_id, type=FlowEdgeType.ON_COMPLETE),
        FlowEdge(id=str(uuid.uuid4()), from_node_id=invoke_node_id, to_node_id=hangup_node_id, type=FlowEdgeType.ON_ERROR)
    ]

    return Flow(
        name="Inbound - Shared Auth Integration",
        type="inbound",
        nodes=[start_node, invoke_node, queue_node, hangup_node],
        edges=edges,
        start_node_id=start_node_id
    )

def create_inbound_flow(api: RoutingApi, flow_obj: Flow) -> str:
    response = api.post_routing_flow(body=flow_obj)
    return response.id

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 400 Bad Request - Invalid Flow Graph

• Cause: The flow graph is not connected, contains cycles, or has missing edge types. Specifically, an invoke node must have both onComplete and onError edges defined if you want explicit handling.
• Fix: Ensure every node (except the start node) has an incoming edge. Ensure the invoke node has edges for both success and failure paths. Check that from_node_id and to_node_id in edges match actual node IDs.

Error: 404 Not Found - Shared Flow ID

• Cause: The shared_flow_id passed to the InvokeNode does not exist or is not of type shared.
• Fix: Verify the ID using the find_shared_flow_by_name function. Ensure the flow is enabled in the Genesys Cloud UI. Disabled flows cannot be invoked.

Error: 429 Too Many Requests

• Cause: You are creating flows too quickly or exceeding the API rate limit.
• Fix: Implement exponential backoff. The Genesys Cloud Python SDK does not automatically retry all errors, so you should wrap the post_routing_flow call in a retry loop if you are creating multiple flows in a batch.

Error: Variable Scope Issues

• Cause: You are trying to pass a variable to the shared flow that does not exist in the inbound context, or the shared flow expects a variable that was not passed.
• Fix: Use the variables array in the InvokeNode action to explicitly map inbound variables to shared flow variables. For example: variables=[{"name": "shared_var_name", "value": "${InboundVarName}"}].

Official References

• Genesys Cloud CX API: Post Routing Flow
• Genesys Cloud CX API: Flow Node Types
• Python SDK Reference: RoutingApi
• Best Practices: Shared Flows