POST /api/v2/flows/executions — Launching an Architect Flow Programmatically

What You Will Build

• This tutorial demonstrates how to trigger a Genesys Cloud CX Architect flow from an external application using the REST API.
• You will use the POST /api/v2/flows/executions endpoint to inject data into a flow and retrieve the execution ID.
• The implementation uses Python with the official genesyscloud SDK and the requests library for raw HTTP comparison.

Prerequisites

• OAuth Client Type: Service Account or Confidential Client.
• Required Scopes: flow:execution:start is mandatory. If your flow requires accessing user data or updating entities, you may need additional scopes such as user:read or contact:write.
• SDK Version: genesyscloud-python-sdk >= 2.0.0.
• Language/Runtime: Python 3.8+.
• External Dependencies:
  • genesyscloud (official SDK)
  • requests (for raw HTTP examples)
  • pyyaml (optional, for config management)

Install the dependencies using pip:

pip install genesyscloud requests

Authentication Setup

Genesys Cloud uses OAuth 2.0 for authentication. For server-to-server integrations, the Client Credentials Grant flow is the standard approach. You must store your Client ID, Client Secret, and Environment URL securely.

The following code initializes the Genesys Cloud Platform client. This client handles token acquisition, caching, and automatic refresh behind the scenes.

import os
import sys
from purecloudplatformclientv2 import (
    Configuration,
    ApiClient,
    PlatformClient,
    FlowApi
)

def get_platform_client() -> PlatformClient:
    """
    Initializes and returns the Genesys Cloud Platform Client.
    Reads credentials from environment variables.
    """
    # Environment variables must be set:
    # GENESYS_CLOUD_CLIENT_ID, GENESYS_CLOUD_CLIENT_SECRET, GENESYS_CLOUD_ENVIRONMENT
    client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
    environment = os.getenv("GENESYS_CLOUD_ENVIRONMENT", "mypurecloud.com")

    if not client_id or not client_secret:
        raise ValueError("Missing GENESYS_CLOUD_CLIENT_ID or GENESYS_CLOUD_CLIENT_SECRET")

    # Construct the base URL
    base_url = f"https://api.{environment}"
    
    # Create configuration
    configuration = Configuration(base_url=base_url)
    
    # Initialize API Client with OAuth credentials
    api_client = ApiClient(configuration=configuration)
    
    # Set up OAuth
    api_client.configuration.set_default_header(
        "Authorization", 
        f"Bearer {api_client.get_access_token(client_id, client_secret)}"
    )
    
    # Create the Platform Client which wraps the API clients
    platform_client = PlatformClient(api_client=api_client)
    
    return platform_client

Note on Token Refresh: The ApiClient automatically manages the access token lifecycle. If the token expires during a long-running batch operation, the SDK will attempt to refresh it. You do not need to implement manual refresh logic unless you are using raw HTTP requests without the SDK.

Implementation

Step 1: Identify the Flow ID

Before launching a flow, you must know its unique identifier. The Flow ID is a UUID string found in the URL when viewing the flow in the Architect UI, or via the GET /api/v2/flows endpoint.

If you do not know the Flow ID, you can search for it by name using the SDK.

def find_flow_by_name(platform_client: PlatformClient, flow_name: str) -> str:
    """
    Searches for a flow by name and returns its ID.
    Raises ValueError if the flow is not found or if multiple flows match.
    """
    flow_api = platform_client.flow_api
    try:
        # Search flows with a specific name
        # expand=false keeps the payload light; we only need the ID
        result = flow_api.post_flows_search(
            body={
                "query": f"name:{flow_name}",
                "size": 10
            }
        )
        
        if not result.entities:
            raise ValueError(f"No flow found with name: {flow_name}")
        
        if len(result.entities) > 1:
            raise ValueError(f"Multiple flows found with name: {flow_name}. Please specify a unique name.")
        
        return result.entities[0].id
    
    except Exception as e:
        print(f"Error searching for flow: {e}")
        raise

Step 2: Construct the Execution Request

The POST /api/v2/flows/executions endpoint accepts a JSON body that defines the initial context for the flow execution. This context is passed into the flow as variables.

Key parameters in the request body:

• flowId: The UUID of the flow to execute.
• context: An object containing key-value pairs. These keys become variables in your Architect flow.
• type: Optional. Usually omitted for standard external triggers.

OAuth Scope Required: flow:execution:start

Using the SDK

The SDK provides the FlowApi client. The method post_flows_executions corresponds to the REST endpoint.

from purecloudplatformclientv2 import FlowExecutionRequest

def launch_flow_sdk(
    platform_client: PlatformClient, 
    flow_id: str, 
    context_data: dict
) -> str:
    """
    Launches a Genesys Cloud flow using the SDK.
    
    Args:
        platform_client: The initialized PlatformClient.
        flow_id: The UUID of the flow to execute.
        context_data: A dictionary of variables to pass into the flow.
        
    Returns:
        The execution ID of the newly created flow execution.
    """
    flow_api = platform_client.flow_api
    
    # Construct the request body
    # The SDK maps this to the JSON payload expected by the API
    request_body = FlowExecutionRequest(
        flow_id=flow_id,
        context=context_data
    )
    
    try:
        # Execute the API call
        response = flow_api.post_flows_executions(body=request_body)
        
        # The response contains the execution ID and other metadata
        execution_id = response.id
        print(f"Flow executed successfully. Execution ID: {execution_id}")
        return execution_id
        
    except Exception as e:
        # Handle API errors (400, 401, 403, 429, 500)
        print(f"Failed to launch flow: {e}")
        raise

Using Raw HTTP (Requests Library)

Sometimes you need more control over the HTTP request or are integrating into a system where the SDK is not available. Here is the equivalent raw HTTP implementation.

import requests

def launch_flow_raw(
    base_url: str, 
    access_token: str, 
    flow_id: str, 
    context_data: dict
) -> str:
    """
    Launches a Genesys Cloud flow using raw HTTP requests.
    """
    url = f"{base_url}/api/v2/flows/executions"
    
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "flowId": flow_id,
        "context": context_data
    }
    
    try:
        response = requests.post(url, headers=headers, json=payload)
        
        # Check for success status codes
        if response.status_code == 200:
            result = response.json()
            execution_id = result.get("id")
            print(f"Flow executed successfully. Execution ID: {execution_id}")
            return execution_id
        else:
            # Log the error for debugging
            print(f"API Error {response.status_code}: {response.text}")
            raise Exception(f"API Request Failed: {response.status_code}")
            
    except requests.exceptions.RequestException as e:
        print(f"Network error: {e}")
        raise

Step 3: Handling Context Variables and Data Types

The context object is critical. It allows you to pass data from your external application into the Genesys Cloud flow.

Rules for Context Data:

1. Keys must be strings.
2. Values can be strings, numbers, booleans, or null. Complex objects (lists, nested dicts) are serialized as JSON strings by the SDK or must be manually serialized if using raw HTTP.
3. Variable Names in Architect: If you pass {"orderId": "12345"} in the context, you must create a variable named orderId in your Architect flow to receive it. The case sensitivity matters.

Example: Passing Complex Data

If you need to pass a list or a nested object, you should serialize it to a JSON string before placing it in the context.

import json

def prepare_context(order_id: str, customer_email: str, items: list) -> dict:
    """
    Prepares the context dictionary for the flow execution.
    Serializes complex objects to JSON strings.
    """
    # Serialize the items list to a JSON string
    items_json = json.dumps(items)
    
    context = {
        "orderId": order_id,
        "customerEmail": customer_email,
        "orderItems": items_json, # Architect will see this as a string
        "timestamp": json.dumps({"year": 2023, "month": 10}) # Another example
    }
    
    return context

In your Architect flow, you would use a Set Variable step or a JSON Parse step to convert the orderItems string back into an array if you need to iterate over it.

Step 4: Error Handling and Retries

Genesys Cloud APIs enforce rate limits. A 429 Too Many Requests response indicates you have exceeded the quota. Your code should implement exponential backoff.

import time

def launch_flow_with_retry(
    platform_client: PlatformClient, 
    flow_id: str, 
    context_data: dict,
    max_retries: int = 3
) -> str:
    """
    Launches a flow with exponential backoff retry logic for 429 errors.
    """
    flow_api = platform_client.flow_api
    
    for attempt in range(max_retries + 1):
        try:
            request_body = FlowExecutionRequest(
                flow_id=flow_id,
                context=context_data
            )
            
            response = flow_api.post_flows_executions(body=request_body)
            return response.id
            
        except Exception as e:
            # Check if the error is a 429 Too Many Requests
            # The SDK raises exceptions, but the underlying response status code is available
            if hasattr(e, 'status') and e.status == 429:
                if attempt < max_retries:
                    wait_time = 2 ** attempt
                    print(f"Rate limited. Retrying in {wait_time} seconds...")
                    time.sleep(wait_time)
                    continue
                else:
                    raise Exception("Max retries exceeded due to rate limiting.")
            else:
                # Non-retryable error
                print(f"Non-retryable error: {e}")
                raise

Complete Working Example

This script combines all the steps into a single runnable module. It searches for a flow by name, prepares context data, and launches the execution.

import os
import sys
import json
from purecloudplatformclientv2 import (
    Configuration,
    ApiClient,
    PlatformClient,
    FlowApi,
    FlowExecutionRequest
)

def get_platform_client() -> PlatformClient:
    client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
    environment = os.getenv("GENESYS_CLOUD_ENVIRONMENT", "mypurecloud.com")

    if not client_id or not client_secret:
        raise ValueError("Missing GENESYS_CLOUD_CLIENT_ID or GENESYS_CLOUD_CLIENT_SECRET")

    base_url = f"https://api.{environment}"
    configuration = Configuration(base_url=base_url)
    api_client = ApiClient(configuration=configuration)
    
    # Authenticate
    api_client.configuration.set_default_header(
        "Authorization", 
        f"Bearer {api_client.get_access_token(client_id, client_secret)}"
    )
    
    return PlatformClient(api_client=api_client)

def find_flow_by_name(platform_client: PlatformClient, flow_name: str) -> str:
    flow_api = platform_client.flow_api
    try:
        result = flow_api.post_flows_search(
            body={
                "query": f"name:{flow_name}",
                "size": 10
            }
        )
        
        if not result.entities:
            raise ValueError(f"No flow found with name: {flow_name}")
        
        return result.entities[0].id
    
    except Exception as e:
        print(f"Error searching for flow: {e}")
        raise

def main():
    try:
        # 1. Initialize Client
        platform_client = get_platform_client()
        
        # 2. Configuration
        FLOW_NAME = "External Order Processing Flow" # Replace with your flow name
        ORDER_ID = "ORD-998877"
        CUSTOMER_EMAIL = "user@example.com"
        
        # 3. Find Flow ID
        print(f"Searching for flow: {FLOW_NAME}")
        flow_id = find_flow_by_name(platform_client, FLOW_NAME)
        print(f"Found Flow ID: {flow_id}")
        
        # 4. Prepare Context
        context_data = {
            "orderId": ORDER_ID,
            "customerEmail": CUSTOMER_EMAIL,
            "sourceSystem": "ExternalPythonApp",
            "priority": "High"
        }
        
        # 5. Launch Flow
        print(f"Launching flow with context: {json.dumps(context_data, indent=2)}")
        execution_id = launch_flow(platform_client, flow_id, context_data)
        print(f"Success! Execution ID: {execution_id}")
        
    except Exception as e:
        print(f"Fatal error: {e}")
        sys.exit(1)

def launch_flow(platform_client: PlatformClient, flow_id: str, context_data: dict) -> str:
    flow_api = platform_client.flow_api
    request_body = FlowExecutionRequest(
        flow_id=flow_id,
        context=context_data
    )
    
    try:
        response = flow_api.post_flows_executions(body=request_body)
        return response.id
    except Exception as e:
        print(f"Failed to launch flow: {e}")
        raise

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token is invalid, expired, or missing.
• Fix: Ensure your Client ID and Secret are correct. Verify that the get_access_token call succeeded. If using a long-running script, ensure the SDK is refreshing the token automatically.

Error: 403 Forbidden

• Cause: The OAuth client lacks the required scope.
• Fix: Add flow:execution:start to the scopes of your OAuth client in the Genesys Cloud Admin Console. Navigate to Integrations > OAuth > Select Client > Scopes.

Error: 400 Bad Request

• Cause: Invalid Flow ID or malformed context.
• Fix: Verify the Flow ID is a valid UUID. Ensure the context object does not contain circular references or unsupported data types. Check the error message in the response body for specific validation failures.

Error: Flow Not Found

• Cause: The flow name search returned no results.
• Fix: Ensure the flow name matches exactly, including case sensitivity. Flows must be Published to be executed. Draft flows cannot be triggered via API.

Error: 429 Too Many Requests

• Cause: You have exceeded the rate limit for flow executions.
• Fix: Implement exponential backoff as shown in Step 4. Reduce the frequency of flow triggers if possible.

Official References

• Genesys Cloud Flow Execution API Documentation
• Genesys Cloud Python SDK Reference
• Genesys Cloud OAuth Documentation