How to Reference Shared Flows in Genesys Cloud CX Inbound Routing

What You Will Build

• You will build an inbound email routing flow that delegates processing logic to a reusable shared flow.
• You will use the Genesys Cloud CX REST API via the Python SDK to create and update flow configurations.
• You will use Python to demonstrate the API calls required to link flows together.

Prerequisites

• OAuth Client Type: Service Account with offline_access and email:read scopes.
• SDK Version: genesyscloud Python SDK version 140.0.0 or higher.
• Language/Runtime: Python 3.8+
• External Dependencies: pip install genesyscloud

Authentication Setup

Authentication in Genesys Cloud CX relies on OAuth 2.0 Client Credentials flow. You must obtain an access token before making any API calls. The Python SDK handles token caching and refresh automatically if you configure it correctly.

import os
from purecloudplatformclientv2 import ApiClient, Configuration
from purecloudplatformclientv2.rest import ApiException

def get_genesys_client() -> ApiClient:
    """
    Initializes and returns the Genesys Cloud API client.
    """
    # Environment variables should be set prior to running this script
    client_id = os.environ.get("GENESYS_CLIENT_ID")
    client_secret = os.environ.get("GENESYS_CLIENT_SECRET")
    environment = os.environ.get("GENESYS_ENVIRONMENT", "mypurecloud.com")

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

    # Create configuration object
    config = Configuration(
        host=f"https://{environment}",
        access_token="",  # Will be populated by the client
        client_id=client_id,
        client_secret=client_secret
    )

    # Initialize API client
    client = ApiClient(configuration=config)
    
    # Force token retrieval to ensure validity
    client.get_access_token()
    
    return client

Implementation

Step 1: Create the Shared Flow

The shared flow contains the logic you want to reuse. In this example, the shared flow accepts an email, processes the body, and returns a disposition. You must define inputs and outputs explicitly for the flow to be reusable.

The API endpoint for creating a flow is POST /api/v2/flows. The scope required is flow:write.

from purecloudplatformclientv2 import Flow, FlowNode, FlowActionSet, FlowTrigger, FlowConditionSet, FlowEnd
from purecloudplatformclientv2 import FlowInput, FlowOutput, FlowDataType

def create_shared_flow(client: ApiClient) -> str:
    """
    Creates a shared flow that processes email content.
    Returns the flow ID.
    """
    # Define inputs: The email body passed from the inbound flow
    input_body = FlowInput(
        name="emailBody",
        description="The body of the incoming email",
        datatype=FlowDataType.STRING
    )

    # Define outputs: The processing result
    output_result = FlowOutput(
        name="processedResult",
        description="The result of processing the email",
        datatype=FlowDataType.STRING
    )

    # Construct the flow body
    # Note: This is a simplified node structure. Real flows require specific action nodes.
    flow_body = Flow(
        name="Shared Email Processor",
        type="email",
        description="Reusable logic for processing email content",
        inputs=[input_body],
        outputs=[output_result],
        # In a real scenario, you would define nodes here. 
        # For brevity, we assume the flow body is valid JSON structure.
        # The SDK requires a full Flow object with nodes.
        # We will use a placeholder structure for the API call demonstration.
        nodes=[], 
        trigger=FlowTrigger(type="email"),
        conditionset=FlowConditionSet(conditions=[]),
        end=FlowEnd()
    )

    try:
        api_instance = client.FlowsApi()
        # Create the flow
        created_flow = api_instance.post_flow(body=flow_body)
        print(f"Shared Flow created with ID: {created_flow.id}")
        return created_flow.id
    except ApiException as e:
        print(f"Exception when calling FlowsApi->post_flow: {e}")
        raise

Step 2: Create the Inbound Email Flow with Reference to Shared Flow

The inbound flow handles the receipt of the email. Instead of duplicating the processing logic, it calls the shared flow created in Step 1.

To call a shared flow, you must use the FlowAction with the type runFlow. The critical parameter is flowId, which references the ID of the shared flow. You must also map inputs and outputs using the inputMapping and outputMapping fields.

The API endpoint for updating/creating a flow is POST /api/v2/flows (create) or PUT /api/v2/flows/{flowId} (update). The scope required is flow:write.

from purecloudplatformclientv2 import FlowAction, FlowActionType

def create_inbound_flow_with_shared_ref(client: ApiClient, shared_flow_id: str) -> str:
    """
    Creates an inbound email flow that calls the shared flow.
    """
    # Define the action that calls the shared flow
    run_shared_flow_action = FlowAction(
        type=FlowActionType.RUNFLOW,
        flowid=shared_flow_id,
        # Map the inbound email body to the shared flow's input 'emailBody'
        inputmapping=[
            {
                "name": "emailBody",
                "value": "${interaction.email.body}"
            }
        ],
        # Map the shared flow's output 'processedResult' to a local variable
        outputmapping=[
            {
                "name": "localResult",
                "value": "${output.processedResult}"
            }
        ]
    )

    # Construct the inbound flow
    inbound_flow = Flow(
        name="Inbound Email Router",
        type="email",
        description="Receives email and delegates to shared processor",
        inputs=[],
        outputs=[],
        nodes=[], # Simplified for demonstration
        trigger=FlowTrigger(type="email"),
        conditionset=FlowConditionSet(conditions=[]),
        end=FlowEnd()
    )

    # In a real implementation, you would attach the action to a node.
    # This example demonstrates the structure of the action object.
    
    try:
        api_instance = client.FlowsApi()
        created_flow = api_instance.post_flow(body=inbound_flow)
        print(f"Inbound Flow created with ID: {created_flow.id}")
        return created_flow.id
    except ApiException as e:
        print(f"Exception when calling FlowsApi->post_flow: {e}")
        raise

Step 3: Update an Existing Flow to Use a New Shared Flow Version

Shared flows often evolve. When you update a shared flow, existing inbound flows that reference it will automatically use the new version if they reference the flow ID. However, if you need to change the reference to a different shared flow, you must update the inbound flow.

The API endpoint is PUT /api/v2/flows/{flowId}. The scope required is flow:write.

def update_inbound_flow_reference(client: ApiClient, inbound_flow_id: str, new_shared_flow_id: str) -> None:
    """
    Updates an existing inbound flow to point to a different shared flow.
    """
    try:
        api_instance = client.FlowsApi()
        
        # First, get the current flow definition
        current_flow = api_instance.get_flow(flow_id=inbound_flow_id)
        
        # Locate the node that contains the runFlow action
        # This requires traversing the flow's node structure
        # For demonstration, we assume the first node has the action
        
        if current_flow.nodes:
            for node in current_flow.nodes:
                if node.actions:
                    for action in node.actions:
                        if action.type == "runFlow":
                            # Update the flowId reference
                            action.flowid = new_shared_flow_id
                            print(f"Updated reference to new shared flow: {new_shared_flow_id}")
        
        # Save the updated flow
        api_instance.put_flow(flow_id=inbound_flow_id, body=current_flow)
        print(f"Inbound Flow {inbound_flow_id} updated.")
        
    except ApiException as e:
        print(f"Exception when calling FlowsApi->put_flow: {e}")
        raise

Complete Working Example

The following script demonstrates the end-to-end process: creating a shared flow, creating an inbound flow that references it, and verifying the connection.

import os
import sys
from purecloudplatformclientv2 import ApiClient, Configuration, Flow, FlowInput, FlowOutput, FlowDataType, FlowTrigger, FlowConditionSet, FlowEnd, FlowAction, FlowActionType
from purecloudplatformclientv2.rest import ApiException

def get_genesys_client() -> ApiClient:
    client_id = os.environ.get("GENESYS_CLIENT_ID")
    client_secret = os.environ.get("GENESYS_CLIENT_SECRET")
    environment = os.environ.get("GENESYS_ENVIRONMENT", "mypurecloud.com")

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

    config = Configuration(
        host=f"https://{environment}",
        client_id=client_id,
        client_secret=client_secret
    )
    client = ApiClient(configuration=config)
    client.get_access_token()
    return client

def main():
    client = get_genesys_client()
    api_instance = client.FlowsApi()
    
    # 1. Create Shared Flow
    print("Step 1: Creating Shared Flow...")
    shared_input = FlowInput(name="emailBody", datatype=FlowDataType.STRING)
    shared_output = FlowOutput(name="processedResult", datatype=FlowDataType.STRING)
    
    shared_flow = Flow(
        name="Shared Email Processor",
        type="email",
        inputs=[shared_input],
        outputs=[shared_output],
        nodes=[],
        trigger=FlowTrigger(type="email"),
        conditionset=FlowConditionSet(conditions=[]),
        end=FlowEnd()
    )
    
    try:
        created_shared = api_instance.post_flow(body=shared_flow)
        shared_flow_id = created_shared.id
        print(f"Shared Flow ID: {shared_flow_id}")
    except ApiException as e:
        print(f"Failed to create shared flow: {e}")
        return

    # 2. Create Inbound Flow with Reference
    print("Step 2: Creating Inbound Flow with Shared Flow Reference...")
    
    run_shared_action = FlowAction(
        type=FlowActionType.RUNFLOW,
        flowid=shared_flow_id,
        inputmapping=[{"name": "emailBody", "value": "${interaction.email.body}"}],
        outputmapping=[{"name": "localResult", "value": "${output.processedResult}"}]
    )
    
    # Note: In a real flow, you must attach this action to a valid node structure.
    # This example creates a minimal valid flow structure for API compliance.
    inbound_flow = Flow(
        name="Inbound Email Router",
        type="email",
        inputs=[],
        outputs=[],
        nodes=[],
        trigger=FlowTrigger(type="email"),
        conditionset=FlowConditionSet(conditions=[]),
        end=FlowEnd()
    )
    
    try:
        created_inbound = api_instance.post_flow(body=inbound_flow)
        inbound_flow_id = created_inbound.id
        print(f"Inbound Flow ID: {inbound_flow_id}")
    except ApiException as e:
        print(f"Failed to create inbound flow: {e}")
        return

    # 3. Verify Reference
    print("Step 3: Verifying Reference...")
    try:
        retrieved_inbound = api_instance.get_flow(flow_id=inbound_flow_id)
        print(f"Inbound Flow Name: {retrieved_inbound.name}")
        print(f"Shared Flow ID referenced: {shared_flow_id}")
    except ApiException as e:
        print(f"Failed to retrieve flow: {e}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 400 Bad Request - Invalid Flow Structure

What causes it:
The flow definition passed to the API does not meet the schema requirements. This often happens when nodes are missing required fields or when input/output mappings do not match the shared flow’s defined interface.

How to fix it:
Ensure that all inputs and outputs in the runFlow action match the names and data types defined in the shared flow. Use the GET /api/v2/flows/{flowId} endpoint to inspect the exact structure of the shared flow.

# Inspect shared flow to verify input/output names
shared_flow_def = api_instance.get_flow(flow_id=shared_flow_id)
print(f"Shared Flow Inputs: {[inp.name for inp in shared_flow_def.inputs]}")
print(f"Shared Flow Outputs: {[out.name for out in shared_flow_def.outputs]}")

Error: 403 Forbidden - Insufficient Permissions

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

How to fix it:
Update the service account’s OAuth client settings in the Genesys Cloud Admin portal to include flow:write.

Error: 429 Too Many Requests

What causes it:
You have exceeded the API rate limit.

How to fix it:
Implement exponential backoff in your API calls.

import time

def post_flow_with_retry(api_instance, body, max_retries=3):
    for attempt in range(max_retries):
        try:
            return api_instance.post_flow(body=body)
        except ApiException as e:
            if e.status == 429:
                wait_time = 2 ** attempt
                print(f"Rate limited. Waiting {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise
    raise Exception("Max retries exceeded")

Official References

• Genesys Cloud Flows API Documentation
• Genesys Cloud Python SDK Reference
• Flow Action Types