Genesys Cloud: Implementing Shared Flow Logic via API and SDK

What You Will Build

• You will programmatically create a shared flow module and link it to multiple inbound call flows to centralize business logic.
• This tutorial uses the Genesys Cloud PureCloud Platform Client v2 API and SDKs.
• The primary language covered is Python, with supplementary examples in JavaScript/TypeScript.

Prerequisites

• OAuth Client Type: Application (Confidential) or Service Account.
• Required Scopes:
  • flow:flow:write (to create/update flows)
  • flow:flow:read (to retrieve flow definitions)
  • flow:flowmodule:write (to create shared modules)
  • flow:flowmodule:read (to read shared modules)
• SDK Version: genesyscloud Python SDK >= 1.0.0 or @genesyscloud/purecloud-platform-client-v2 JS SDK >= 1.0.0.
• Runtime: Python 3.8+ or Node.js 16+.
• Dependencies:
  • Python: pip install genesyscloud python-dotenv
  • JavaScript: npm install @genesyscloud/purecloud-platform-client-v2 dotenv

Authentication Setup

Genesys Cloud APIs require OAuth 2.0 Bearer tokens. For server-side integrations, the Client Credentials flow is the standard. You must cache the token and handle expiration.

Python Authentication Helper

import os
import time
import requests
from typing import Dict, Optional

class GenesysAuth:
    def __init__(self, client_id: str, client_secret: str, org_id: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.org_id = org_id
        self.token_url = f"https://{org_id}.mypurecloud.com/oauth/token"
        self._token_data: Optional[Dict] = None
        self._expires_at: float = 0

    def get_access_token(self) -> str:
        """
        Returns a valid OAuth access token.
        Handles caching and automatic refresh.
        """
        if self._token_data and time.time() < self._expires_at:
            return self._token_data['access_token']

        payload = {
            'grant_type': 'client_credentials',
            'client_id': self.client_id,
            'client_secret': self.client_secret,
            'scope': 'flow:flow:write flow:flow:read flow:flowmodule:write flow:flowmodule:read'
        }

        try:
            response = requests.post(self.token_url, data=payload)
            response.raise_for_status()
            self._token_data = response.json()
            # Subtract 60 seconds to ensure token does not expire during API call
            self._expires_at = time.time() + self._token_data['expires_in'] - 60
            return self._token_data['access_token']
        except requests.exceptions.HTTPError as e:
            raise Exception(f"Authentication failed: {e.response.text}") from e

JavaScript Authentication Helper

const axios = require('axios');
require('dotenv').config();

class GenesysAuth {
    constructor() {
        this.clientId = process.env.GENESYS_CLIENT_ID;
        this.clientSecret = process.env.GENESYS_CLIENT_SECRET;
        this.orgId = process.env.GENESYS_ORG_ID;
        this.tokenUrl = `https://${this.orgId}.mypurecloud.com/oauth/token`;
        this.tokenData = null;
        this.expiresAt = 0;
    }

    async getAccessToken() {
        if (this.tokenData && Date.now() < this.expiresAt) {
            return this.tokenData.access_token;
        }

        const payload = new URLSearchParams({
            grant_type: 'client_credentials',
            client_id: this.clientId,
            client_secret: this.clientSecret,
            scope: 'flow:flow:write flow:flow:read flow:flowmodule:write flow:flowmodule:read'
        });

        try {
            const response = await axios.post(this.tokenUrl, payload, {
                headers: { 'Content-Type': 'application/x-www-form-urlencoded' }
            });
            this.tokenData = response.data;
            this.expiresAt = Date.now() + (this.tokenData.expires_in * 1000) - 60000;
            return this.tokenData.access_token;
        } catch (error) {
            throw new Error(`Authentication failed: ${error.response?.data || error.message}`);
        }
    }
}

Implementation

Shared flows in Genesys Cloud are technically Flow Modules. A Flow Module is a reusable block of flow logic that can be invoked by multiple flows. This decouples common logic (e.g., IVR menus, authentication, data lookups) from the specific entry points of your call flows.

Step 1: Create a Shared Flow Module

To create a shared module, you must define a flow definition that acts as a “library.” The key distinction is that this flow will be referenced by other flows using the Flow Module node type.

API Endpoint: POST /api/v2/flows/flowmodules
Scope: flow:flowmodule:write

Python Implementation

from genesyscloud import PlatformClient
from genesyscloud.flows.model import FlowModule, FlowModuleDefinition

def create_shared_module(client: PlatformClient, name: str, description: str) -> str:
    """
    Creates a new Flow Module.
    Returns the ID of the created module.
    """
    try:
        # Define the module metadata
        module = FlowModule(
            name=name,
            description=description,
            # Initially empty definition; we will add nodes in Step 2
            definition=FlowModuleDefinition(
                nodes=[],
                edges=[]
            )
        )
        
        # Create the module via the SDK
        # Note: The SDK method is create_flow_flowmodule
        response = client.flow.create_flow_flowmodule(body=module)
        
        if response.id:
            print(f"Created Flow Module: {response.id}")
            return response.id
        else:
            raise Exception("Flow Module creation did not return an ID.")
            
    except Exception as e:
        print(f"Error creating Flow Module: {e}")
        raise

Key Parameters Explanation

• name: Must be unique within the organization. Use a naming convention like shared_auth_logic_v1.
• description: Critical for maintainability. Describe the input/output contract of the module.
• definition: Initially empty. In Genesys Cloud, you typically create the shell first, then update it with nodes, or provide the full definition at creation. For complex modules, creating then updating is safer to avoid large payload timeouts.

Step 2: Define the Module Logic (Nodes and Edges)

A Flow Module is essentially a mini-flow. You must add nodes (e.g., Set Data, Play Prompt, HTTP Request) and edges (connections) to it.

API Endpoint: PUT /api/v2/flows/flowmodules/{flowModuleId}
Scope: flow:flowmodule:write

Python Implementation: Adding a “Set Data” Node

This example adds a node that sets a variable auth_status to pending. This simulates the start of an authentication process.

from genesyscloud.flows.model import (
    FlowModuleDefinition,
    FlowModuleNode,
    FlowModuleEdge,
    FlowModuleSetDataNode
)
import uuid

def update_module_with_logic(client: PlatformClient, module_id: str) -> None:
    """
    Updates the Flow Module with a basic Set Data node.
    """
    # 1. Retrieve current definition (optimistic locking requires version)
    try:
        current_module = client.flow.get_flow_flowmodule(flow_module_id=module_id)
        current_version = current_module.version
        
        # 2. Define the Node
        # Generate a unique ID for the node
        node_id = str(uuid.uuid4())
        
        set_data_node = FlowModuleSetDataNode(
            id=node_id,
            name="Init Auth Status",
            type="setData",
            settings={
                "data": [
                    {
                        "key": "auth_status",
                        "value": "pending"
                    }
                ]
            }
        )
        
        # 3. Define the Edge (Start -> SetData)
        # The 'start' node is implicit in the definition, but we need to connect it.
        # In Flow Modules, the entry point is defined by the 'entry' node in the edges.
        
        # Note: The exact structure of 'edges' depends on the specific node types.
        # For a simple module, we often define the 'entry' edge pointing to our first node.
        
        new_definition = FlowModuleDefinition(
            nodes=[set_data_node],
            edges=[
                {
                    "id": str(uuid.uuid4()),
                    "source": "start", # Implicit start node
                    "target": node_id,
                    "type": "default"
                }
            ],
            # Define the entry point of the module
            entry={
                "nodeId": node_id
            }
        )
        
        # 4. Update the module
        updated_module = FlowModule(
            id=module_id,
            version=current_version, # Required for optimistic locking
            definition=new_definition
        )
        
        client.flow.update_flow_flowmodule(
            flow_module_id=module_id,
            body=updated_module
        )
        print(f"Updated Flow Module {module_id} with logic.")
        
    except Exception as e:
        print(f"Error updating Flow Module: {e}")
        raise

Step 3: Invoke the Shared Module from an Inbound Call Flow

Now that the module exists, you must link it to an inbound call flow. You do this by adding a Flow Module node to the inbound call flow.

API Endpoint: PUT /api/v2/flows/{flowId}
Scope: flow:flow:write

Python Implementation: Linking the Module

from genesyscloud.flows.model import (
    Flow,
    FlowDefinition,
    FlowNode,
    FlowEdge,
    FlowFlowModuleNode
)

def link_module_to_flow(client: PlatformClient, flow_id: str, module_id: str) -> None:
    """
    Adds a Flow Module node to an existing inbound call flow.
    """
    try:
        # 1. Get current flow definition
        current_flow = client.flow.get_flow_flow(flow_id=flow_id)
        current_version = current_flow.version
        
        # 2. Create the Flow Module Node
        node_id = str(uuid.uuid4())
        flow_module_node = FlowFlowModuleNode(
            id=node_id,
            name="Execute Shared Auth",
            type="flowModule",
            settings={
                "flowModuleId": module_id,
                # Optional: Map inputs/outputs if the module defines them
                # "inputMappings": [],
                # "outputMappings": []
            }
        )
        
        # 3. Update the Flow Definition
        # We append the node and add an edge from a hypothetical 'Start' or 'Queue' node.
        # For this example, assume we are connecting from the 'start' node.
        
        new_nodes = current_flow.definition.nodes + [flow_module_node]
        new_edges = current_flow.definition.edges + [
            {
                "id": str(uuid.uuid4()),
                "source": "start",
                "target": node_id,
                "type": "default"
            }
        ]
        
        new_definition = FlowDefinition(
            nodes=new_nodes,
            edges=new_edges,
            entry=current_flow.definition.entry
        )
        
        updated_flow = Flow(
            id=flow_id,
            version=current_version,
            definition=new_definition
        )
        
        # 4. Publish the update
        client.flow.update_flow_flow(flow_id=flow_id, body=updated_flow)
        print(f"Linked Module {module_id} to Flow {flow_id}")
        
    except Exception as e:
        print(f"Error linking module to flow: {e}")
        raise

Step 4: Reuse the Module in a Second Flow

To demonstrate reuse, repeat Step 3 with a different flow_id. The module_id remains the same. This confirms that the logic is centralized. If you update module_id in Step 2, both flows will automatically inherit the new logic upon their next execution (or immediately if the flow is already running and the module is re-fetched, depending on Genesys Cloud’s caching behavior for that specific session).

Complete Working Example

This script combines authentication, module creation, logic definition, and flow linking.

File: shared_flow_orchestrator.py

import os
import uuid
from dotenv import load_dotenv
from genesyscloud import PlatformClient
from genesyscloud.flows.model import (
    FlowModule, FlowModuleDefinition, FlowModuleSetDataNode,
    Flow, FlowDefinition, FlowFlowModuleNode
)

# Load environment variables
load_dotenv()

def main():
    # 1. Initialize Client
    client = PlatformClient(
        client_id=os.getenv('GENESYS_CLIENT_ID'),
        client_secret=os.getenv('GENESYS_CLIENT_SECRET'),
        org_id=os.getenv('GENESYS_ORG_ID')
    )

    # 2. Define Constants
    MODULE_NAME = "Shared_Auth_Library_v1"
    FLOW_ID_1 = os.getenv('INBOUND_FLOW_ID_1')
    FLOW_ID_2 = os.getenv('INBOUND_FLOW_ID_2')

    if not FLOW_ID_1 or not FLOW_ID_2:
        raise ValueError("Environment variables INBOUND_FLOW_ID_1 and INBOUND_FLOW_ID_2 must be set.")

    try:
        # 3. Create Flow Module
        print("Creating Flow Module...")
        module = FlowModule(
            name=MODULE_NAME,
            description="Centralized authentication logic for all inbound flows.",
            definition=FlowModuleDefinition(nodes=[], edges=[])
        )
        created_module = client.flow.create_flow_flowmodule(body=module)
        module_id = created_module.id
        print(f"Module Created: {module_id}")

        # 4. Add Logic to Module
        print("Adding logic to Module...")
        node_id = str(uuid.uuid4())
        set_data_node = FlowModuleSetDataNode(
            id=node_id,
            name="Set Auth Pending",
            type="setData",
            settings={"data": [{"key": "auth_status", "value": "pending"}]}
        )
        
        updated_def = FlowModuleDefinition(
            nodes=[set_data_node],
            edges=[{"id": str(uuid.uuid4()), "source": "start", "target": node_id, "type": "default"}],
            entry={"nodeId": node_id}
        )
        
        client.flow.update_flow_flowmodule(
            flow_module_id=module_id,
            body=FlowModule(id=module_id, version=created_module.version, definition=updated_def)
        )
        print("Module Logic Updated.")

        # 5. Link to Flow 1
        print(f"Linking Module to Flow 1 ({FLOW_ID_1})...")
        link_module(client, FLOW_ID_1, module_id)

        # 6. Link to Flow 2
        print(f"Linking Module to Flow 2 ({FLOW_ID_2})...")
        link_module(client, FLOW_ID_2, module_id)

        print("Success: Shared module linked to both flows.")

    except Exception as e:
        print(f"Critical Error: {e}")

def link_module(client, flow_id, module_id):
    """Helper to link a module to a specific flow."""
    current_flow = client.flow.get_flow_flow(flow_id=flow_id)
    current_version = current_flow.version
    
    node_id = str(uuid.uuid4())
    flow_module_node = FlowFlowModuleNode(
        id=node_id,
        name="Execute Shared Auth",
        type="flowModule",
        settings={"flowModuleId": module_id}
    )
    
    # Append node and edge from 'start'
    new_nodes = current_flow.definition.nodes + [flow_module_node]
    new_edges = current_flow.definition.edges + [
        {"id": str(uuid.uuid4()), "source": "start", "target": node_id, "type": "default"}
    ]
    
    new_def = FlowDefinition(nodes=new_nodes, edges=new_edges, entry=current_flow.definition.entry)
    
    client.flow.update_flow_flow(
        flow_id=flow_id,
        body=Flow(id=flow_id, version=current_version, definition=new_def)
    )

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 409 Conflict on Flow Update

• Cause: Optimistic locking failure. You attempted to update a flow or module using a version number that is older than the current server version. This happens if another process (or admin user) modified the flow between your GET and PUT requests.
• Fix: Implement a retry loop that re-fetches the resource, merges your changes into the latest version, and retries the update.

import time

def update_with_retry(client, flow_id, update_function, max_retries=3):
    for attempt in range(max_retries):
        try:
            current = client.flow.get_flow_flow(flow_id=flow_id)
            update_function(current) # Function should update the object in place or return new body
            return True
        except Exception as e:
            if "Conflict" in str(e) or e.response.status_code == 409:
                print(f"Version conflict. Retrying ({attempt+1}/{max_retries})...")
                time.sleep(1)
            else:
                raise
    raise Exception("Max retries exceeded for flow update.")

Error: 400 Bad Request - Invalid Node Type

• Cause: The type field in the node definition does not match a known Genesys Cloud node type. For Flow Modules, ensure you use flowModule as the node type when invoking it from a parent flow.
• Fix: Verify the type string in your FlowFlowModuleNode or FlowModuleSetDataNode definitions. Refer to the official Flow Node Types documentation.

Error: 403 Forbidden

• Cause: Missing OAuth scopes.
• Fix: Ensure your OAuth client has flow:flowmodule:write and flow:flow:write scopes granted in the Genesys Cloud Admin Console under Administration > Security > OAuth Clients.

Official References

• Genesys Cloud Flow Module API Documentation
• Genesys Cloud Flow API Documentation
• Python SDK Reference
• JavaScript SDK Reference