Mastering Shared Flow Reuse in Genesys Cloud via the API

What You Will Build

• Create a reusable Shared Flow that contains complex logic, such as authentication or data retrieval.
• Configure multiple distinct Inbound Call Flows to invoke this shared module using the API.
• Use the Genesys Cloud Python SDK to automate the creation of these flows and the linking of the shared flow references.

Prerequisites

• OAuth Client Type: Service Account or Client Credentials.
• Required Scopes: flow:sharedflow:read, flow:sharedflow:write, flow:call:read, flow:call:write, routing:skill:read, routing:queue:read.
• SDK Version: genesys-cloud-python-sdk v7.0.0 or higher.
• Language/Runtime: Python 3.8+.
• External Dependencies: pip install genesys-cloud-python-sdk.

Authentication Setup

Genesys Cloud uses OAuth 2.0 for API authentication. For server-side scripts and automation, the Client Credentials flow is the standard approach. This flow exchanges a client ID and client secret for an access token.

The following code demonstrates how to initialize the Genesys Cloud client with proper error handling for authentication failures.

import os
from purecloud_platform_client import (
    ApiClient,
    Configuration,
    ApiException
)

def get_purecloud_client(region: str = "us-east-1") -> ApiClient:
    """
    Initializes and returns a configured Genesys Cloud API Client.
    
    Args:
        region: The Genesys Cloud region (e.g., 'us-east-1', 'eu-west-1').
        
    Returns:
        ApiClient: A configured and authenticated API client instance.
    """
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    
    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET environment variables must be set.")

    # Define the base URL for the specific region
    base_url = f"https://api.{region}.mypurecloud.com"
    
    configuration = Configuration(
        host=base_url,
        oauth_client_id=client_id,
        oauth_client_secret=client_secret
    )

    # Initialize the API client
    client = ApiClient(configuration)
    
    # Attempt to get an access token to verify credentials immediately
    try:
        client.get_access_token()
        print("Authentication successful.")
    except ApiException as e:
        if e.status == 401:
            raise ConnectionError("Invalid Client ID or Secret.") from e
        elif e.status == 403:
            raise PermissionError("Insufficient permissions for the provided scopes.") from e
        else:
            raise ConnectionError(f"Authentication failed with status {e.status}: {e.body}") from e

    return client

Implementation

Step 1: Create the Shared Flow

A Shared Flow is a modular piece of logic that can be invoked by other flows. Unlike standard flows, it does not have a direct trigger (like an inbound call or web chat). It acts as a subroutine.

When creating a Shared Flow via the API, you must define:

1. Name and Description: Metadata for identification.
2. Type: Must be set to shared.
3. Initial Action: The first block executed when the flow is invoked.
4. Inputs/Outputs: Optional but recommended for passing data between the caller and the shared flow.

The following code creates a simple Shared Flow named “Shared Auth Logic”. This flow will simulate a basic authentication check.

from purecloud_platform_client import (
    FlowsApi,
    SharedFlow,
    SharedFlowDocument,
    SharedFlowInitialAction,
    SharedFlowAction,
    SharedFlowSetProperty,
    SharedFlowEnd
)

def create_shared_flow(client: ApiClient, name: str = "Shared Auth Logic") -> str:
    """
    Creates a new Shared Flow in Genesys Cloud.
    
    Args:
        client: The authenticated ApiClient.
        name: The name of the shared flow.
        
    Returns:
        str: The ID of the newly created shared flow.
    """
    flows_api = FlowsApi(client)

    # Define the document structure
    # For this example, we create a simple flow that sets a property 'auth_result' to 'success'
    # and then ends. This simulates a successful authentication step.
    
    # 1. Define the End Action
    end_action = SharedFlowEnd(
        end_type="normal",
        result=None
    )

    # 2. Define the Set Property Action (Simulating logic)
    set_prop_action = SharedFlowSetProperty(
        property_name="auth_result",
        property_value="success",
        next_action=end_action
    )

    # 3. Define the Initial Action
    initial_action = SharedFlowInitialAction(
        type="setProperty",
        property_name="auth_result",
        property_value="pending",
        next_action=set_prop_action
    )

    # 4. Assemble the Document
    document = SharedFlowDocument(
        initial_action=initial_action
    )

    # 5. Create the Shared Flow Object
    shared_flow = SharedFlow(
        name=name,
        description="A reusable module for authentication logic.",
        type="shared",
        document=document
    )

    try:
        response = flows_api.post_flow_shared_flow(body=shared_flow)
        print(f"Shared Flow created successfully. ID: {response.id}")
        return response.id
    except ApiException as e:
        print(f"Failed to create Shared Flow: {e.body}")
        raise

Step 2: Create Inbound Call Flows That Reference the Shared Flow

The core of this tutorial is linking the Shared Flow to Inbound Call Flows. In Genesys Cloud, you invoke a Shared Flow using the invokeSharedFlow action within any standard flow (Call, Chat, Webchat, etc.).

The invokeSharedFlow action requires:

1. SharedFlowId: The ID of the Shared Flow created in Step 1.
2. Input Properties: A map of properties to pass into the shared flow.
3. Output Properties: A map defining where to store the results returned by the shared flow.
4. Timeout: How long to wait for the shared flow to complete.

We will create two separate Inbound Call Flows: “Sales Call Flow” and “Support Call Flow”. Both will invoke the same “Shared Auth Logic” flow.

from purecloud_platform_client import (
    CallFlow,
    CallFlowDocument,
    CallFlowInitialAction,
    CallFlowInvokeSharedFlow,
    CallFlowPlayPrompt,
    CallFlowEnd
)
from datetime import datetime

def create_inbound_call_flow(client: ApiClient, name: str, shared_flow_id: str, queue_id: str) -> str:
    """
    Creates an Inbound Call Flow that invokes a specific Shared Flow.
    
    Args:
        client: The authenticated ApiClient.
        name: The name of the call flow.
        shared_flow_id: The ID of the Shared Flow to invoke.
        queue_id: The ID of the routing queue to assign to the flow (required for Call Flows).
        
    Returns:
        str: The ID of the newly created call flow.
    """
    flows_api = FlowsApi(client)

    # 1. Define the End Action
    end_action = CallFlowEnd(
        end_type="normal",
        result=None
    )

    # 2. Define a Play Prompt Action (Post-Shared Flow Logic)
    # This plays a message after the shared flow completes
    play_prompt_action = CallFlowPlayPrompt(
        prompt="You have been authenticated.",
        next_action=end_action
    )

    # 3. Define the Invoke Shared Flow Action
    # This is the critical step. It links the call flow to the shared module.
    invoke_shared_flow_action = CallFlowInvokeSharedFlow(
        shared_flow_id=shared_flow_id,
        input_properties={
            "user_id": "${user.id}" # Example: Passing a dynamic variable
        },
        output_properties={
            "auth_result": "shared_auth_result" # Store output in 'shared_auth_result'
        },
        timeout=30000, # 30 seconds
        next_action=play_prompt_action
    )

    # 4. Define the Initial Action
    # For simplicity, we start directly with the shared flow invocation.
    # In a real scenario, you might have a greeting first.
    initial_action = CallFlowInitialAction(
        type="invokeSharedFlow",
        shared_flow_id=shared_flow_id,
        input_properties={
            "user_id": "${user.id}"
        },
        output_properties={
            "auth_result": "shared_auth_result"
        },
        timeout=30000,
        next_action=play_prompt_action
    )

    # 5. Assemble the Document
    document = CallFlowDocument(
        initial_action=initial_action
    )

    # 6. Create the Call Flow Object
    # Note: Call flows require a queue for routing purposes
    call_flow = CallFlow(
        name=name,
        description=f"Inbound flow invoking shared auth logic for {name}.",
        type="call",
        document=document,
        queue_id=queue_id # Required for Call Flows
    )

    try:
        response = flows_api.post_flow_call_flow(body=call_flow)
        print(f"Call Flow '{name}' created successfully. ID: {response.id}")
        return response.id
    except ApiException as e:
        print(f"Failed to create Call Flow '{name}': {e.body}")
        raise

Step 3: Orchestrate the Creation Process

This step ties everything together. We will:

1. Authenticate.
2. Create the Shared Flow.
3. Retrieve an existing Queue ID (since Call Flows require a queue).
4. Create two distinct Call Flows that both reference the Shared Flow ID.

from purecloud_platform_client import RoutingApi

def get_first_queue_id(client: ApiClient) -> str:
    """
    Retrieves the ID of the first available routing queue.
    
    Args:
        client: The authenticated ApiClient.
        
    Returns:
        str: The ID of the first queue.
    """
    routing_api = RoutingApi(client)
    try:
        # Get queues with a small limit
        response = routing_api.get_routing_queues(
            limit=1,
            expand=None
        )
        if response.entities and len(response.entities) > 0:
            return response.entities[0].id
        else:
            raise ValueError("No queues found. Please create a queue in Genesys Cloud first.")
    except ApiException as e:
        print(f"Failed to retrieve queues: {e.body}")
        raise

def main():
    """
    Main execution function.
    """
    try:
        # 1. Authenticate
        client = get_purecloud_client("us-east-1")
        
        # 2. Create Shared Flow
        shared_flow_id = create_shared_flow(client, name="Shared Auth Logic")
        
        # 3. Get a Queue ID for the Call Flows
        queue_id = get_first_queue_id(client)
        print(f"Using Queue ID: {queue_id}")
        
        # 4. Create Inbound Call Flows referencing the Shared Flow
        call_flow_1_id = create_inbound_call_flow(
            client, 
            name="Sales Call Flow", 
            shared_flow_id=shared_flow_id, 
            queue_id=queue_id
        )
        
        call_flow_2_id = create_inbound_call_flow(
            client, 
            name="Support Call Flow", 
            shared_flow_id=shared_flow_id, 
            queue_id=queue_id
        )
        
        print("\n--- Summary ---")
        print(f"Shared Flow ID: {shared_flow_id}")
        print(f"Sales Call Flow ID: {call_flow_1_id}")
        print(f"Support Call Flow ID: {call_flow_2_id}")
        print("\nBoth Call Flows now invoke the same Shared Flow logic.")

    except Exception as e:
        print(f"An error occurred: {e}")
        raise

if __name__ == "__main__":
    main()

Complete Working Example

The following script combines all previous steps into a single, runnable Python file. Save this as create_shared_flow_reuse.py. Ensure you have set the environment variables GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET.

import os
from purecloud_platform_client import (
    ApiClient,
    Configuration,
    ApiException,
    FlowsApi,
    SharedFlow,
    SharedFlowDocument,
    SharedFlowInitialAction,
    SharedFlowSetProperty,
    SharedFlowEnd,
    CallFlow,
    CallFlowDocument,
    CallFlowInitialAction,
    CallFlowPlayPrompt,
    CallFlowEnd,
    RoutingApi
)

def get_purecloud_client(region: str = "us-east-1") -> ApiClient:
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    
    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET environment variables must be set.")

    base_url = f"https://api.{region}.mypurecloud.com"
    configuration = Configuration(
        host=base_url,
        oauth_client_id=client_id,
        oauth_client_secret=client_secret
    )
    client = ApiClient(configuration)
    
    try:
        client.get_access_token()
    except ApiException as e:
        if e.status == 401:
            raise ConnectionError("Invalid Client ID or Secret.") from e
        elif e.status == 403:
            raise PermissionError("Insufficient permissions for the provided scopes.") from e
        else:
            raise ConnectionError(f"Authentication failed with status {e.status}: {e.body}") from e

    return client

def create_shared_flow(client: ApiClient, name: str = "Shared Auth Logic") -> str:
    flows_api = FlowsApi(client)
    
    end_action = SharedFlowEnd(end_type="normal")
    set_prop_action = SharedFlowSetProperty(
        property_name="auth_result",
        property_value="success",
        next_action=end_action
    )
    initial_action = SharedFlowInitialAction(
        type="setProperty",
        property_name="auth_result",
        property_value="pending",
        next_action=set_prop_action
    )
    document = SharedFlowDocument(initial_action=initial_action)
    
    shared_flow = SharedFlow(
        name=name,
        description="A reusable module for authentication logic.",
        type="shared",
        document=document
    )

    try:
        response = flows_api.post_flow_shared_flow(body=shared_flow)
        print(f"Shared Flow created successfully. ID: {response.id}")
        return response.id
    except ApiException as e:
        print(f"Failed to create Shared Flow: {e.body}")
        raise

def get_first_queue_id(client: ApiClient) -> str:
    routing_api = RoutingApi(client)
    try:
        response = routing_api.get_routing_queues(limit=1)
        if response.entities and len(response.entities) > 0:
            return response.entities[0].id
        else:
            raise ValueError("No queues found. Please create a queue in Genesys Cloud first.")
    except ApiException as e:
        print(f"Failed to retrieve queues: {e.body}")
        raise

def create_inbound_call_flow(client: ApiClient, name: str, shared_flow_id: str, queue_id: str) -> str:
    flows_api = FlowsApi(client)

    end_action = CallFlowEnd(end_type="normal")
    play_prompt_action = CallFlowPlayPrompt(
        prompt="You have been authenticated.",
        next_action=end_action
    )
    
    initial_action = CallFlowInitialAction(
        type="invokeSharedFlow",
        shared_flow_id=shared_flow_id,
        input_properties={"user_id": "${user.id}"},
        output_properties={"auth_result": "shared_auth_result"},
        timeout=30000,
        next_action=play_prompt_action
    )

    document = CallFlowDocument(initial_action=initial_action)
    
    call_flow = CallFlow(
        name=name,
        description=f"Inbound flow invoking shared auth logic for {name}.",
        type="call",
        document=document,
        queue_id=queue_id
    )

    try:
        response = flows_api.post_flow_call_flow(body=call_flow)
        print(f"Call Flow '{name}' created successfully. ID: {response.id}")
        return response.id
    except ApiException as e:
        print(f"Failed to create Call Flow '{name}': {e.body}")
        raise

def main():
    try:
        client = get_purecloud_client("us-east-1")
        shared_flow_id = create_shared_flow(client, name="Shared Auth Logic")
        queue_id = get_first_queue_id(client)
        print(f"Using Queue ID: {queue_id}")
        
        call_flow_1_id = create_inbound_call_flow(client, "Sales Call Flow", shared_flow_id, queue_id)
        call_flow_2_id = create_inbound_call_flow(client, "Support Call Flow", shared_flow_id, queue_id)
        
        print("\n--- Summary ---")
        print(f"Shared Flow ID: {shared_flow_id}")
        print(f"Sales Call Flow ID: {call_flow_1_id}")
        print(f"Support Call Flow ID: {call_flow_2_id}")
        print("\nBoth Call Flows now invoke the same Shared Flow logic.")
    except Exception as e:
        print(f"An error occurred: {e}")
        raise

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 400 Bad Request - “Invalid Flow Document”

• What causes it: The JSON structure of the flow document does not match the Genesys Cloud schema. This often happens when next_action links are missing or circular, or when required fields like type are omitted.
• How to fix it: Ensure every action (except the final End action) has a next_action property pointing to another action. Verify that the type field matches the action class (e.g., invokeSharedFlow for CallFlowInvokeSharedFlow).
• Code showing the fix:

  # Incorrect: Missing next_action
  invoke_action = CallFlowInvokeSharedFlow(
      shared_flow_id=shared_flow_id,
      # next_action is missing
  )
  
  # Correct: Explicitly linking to the next action
  invoke_action = CallFlowInvokeSharedFlow(
      shared_flow_id=shared_flow_id,
      next_action=play_prompt_action # Must point to a defined action object
  )
  

Error: 403 Forbidden - “Insufficient Permissions”

• What causes it: The OAuth token used does not have the required scopes for Flows.
• How to fix it: Ensure your Service Account or Client Credentials have the flow:sharedflow:write and flow:call:write scopes assigned in the Genesys Cloud Admin Console under Security > OAuth > Applications.
• Code showing the fix:
  No code change is required, but you must verify the scopes in the Admin Console. The error message will explicitly state which scope is missing.

Error: 404 Not Found - “Queue Not Found”

• What causes it: The Call Flow creation requires a valid queue_id. If the retrieved queue ID is invalid or the queue was deleted, the API returns 404.
• How to fix it: Verify that at least one Routing Queue exists in your Genesys Cloud organization. Use the get_first_queue_id function to dynamically retrieve a valid ID, or hardcode a known valid Queue ID for testing.

Error: 409 Conflict - “Flow Name Already Exists”

• What causes it: You are trying to create a flow with a name that is already taken in the organization.
• How to fix it: Append a timestamp or unique suffix to the flow name during creation, or use the get_flow API to check if a flow exists before creating it.
• Code showing the fix:

  from datetime import datetime
  
  unique_name = f"Shared Auth Logic - {datetime.now().strftime('%Y%m%d%H%M%S')}"
  shared_flow = SharedFlow(name=unique_name, ...)
  

Official References

• Genesys Cloud Flows API Documentation
• Shared Flow Object Reference
• Call Flow Object Reference
• Genesys Cloud Python SDK Reference