This Python source code shows how to call any VCF Automation workflow from Python runtime environment. This approach is of interest, because it allows us to ensure the seamless use of existing workflows in this context.

Execute Workflow from Python

The code contains a few functions. One is callWorkflow, which bundles all necessary function calls. The sequence is then getWorkflowId and executeWorkflow. The execution of a workflow is asynchronous, so the status is then queried cyclically with getWorkflowState. When the status is complete, the output parameters are queried with getOutputParameters, the logs with getWorkflowLog and getWorkflowSyslog. Each of them makes an API call to the orchestrator using the request function.

"""
@module de.stschnell

@author Stefan Schnell <mail@stefan-schnell.de>
@license MIT
@version 0.2.0

@runtime python:3.11

@memoryLimit 128000000

@inputType in_userName {string}
@inputType in_password {SecureString}

@outputType Properties
"""


import json
import time
from util.http import Http

http = Http()


def getCategoryPath(
    vcoUrl: str,
    bearerToken: str,
    categoryId: str
) -> str:
    """ Gets the path of a category.

    @param {string} vcoUrl - URL of Aria orchestrator
    @param {string} bearerToken
    @param {string} categoryId - ID of the category
    @returns {string}
    """

    returnValue = ""

    try:

        returnValue = http.request(
            url = vcoUrl + "/api/categories/" + categoryId,
            bearerToken = bearerToken
        )["path"]

    except Exception as err:
        raise ValueError(f"An error occurred at get category path - {err}") \
            from err

    return returnValue


def getWorkflowId(
    vcoUrl: str,
    bearerToken: str,
    workflowFolder: str,
    workflowName: str
) -> str:
    """ Gets the ID of a workflow.

    @param {string} vcoUrl - URL of Aria orchestrator
    @param {string} bearerToken
    @param {string} workflowFolder - Folder of the workflow
    @param {string} workflowName - Name of the workflow
    @returns {string}
    """

    returnValue: str = ""

    try:

        actions = http.request(
            url = vcoUrl + "/api/workflows",
            bearerToken = bearerToken
        )

        found = False
        for action in actions["link"]:
            for attribute in action["attributes"]:
                if attribute["name"] == "name" and \
                attribute["value"] == workflowName:
                    for attribute in action["attributes"]:
                        if attribute["name"] == "categoryId":
                            categoryId = attribute["value"]
                            categoryPath = getCategoryPath(
                                vcoUrl,
                                bearerToken,
                                categoryId
                            )
                            if categoryPath == workflowFolder:
                                for attribute in action["attributes"]:
                                    if attribute["name"] == "id":
                                        returnValue = attribute["value"]
                                        found = True
            if found:
                break

    except Exception as err:
        raise ValueError(f"An error occurred at get workflow ID - {err}") \
            from err

    return returnValue


def executeWorkflow(
    vcoUrl: str,
    bearerToken: str,
    workflowId: str,
    parameters: dict = {}
) -> dict:
    """ Executes a workflow.

    @param {string} vcoUrl - URL of Aria orchestrator
    @param {string} bearerToken
    @param {string} workflowId - ID of the workflow
    @param {dictionary} parameters - Parameters of the workflow
    @returns {dictionary}
    """

    returnValue = {}

    try:

        returnValue = http.request(
            url = vcoUrl + "/api/workflows/" + workflowId + "/executions",
            bearerToken = bearerToken,
            method = "POST",
            body = parameters
        )

    except Exception as err:
        raise ValueError(f"An error occurred at workflow executing - {err}") \
            from err

    return returnValue


def getWorkflowState(
    vcoUrl: str,
    bearerToken: str,
    workflowId: str,
    executionId: str
) -> str:
    """ Delivers the running state of a workflow.

    @param {string} vcoUrl - URL of Aria orchestrator
    @param {string} bearerToken
    @param {string} workflowId - ID of the workflow
    @param {string} executionId - ID of the execution, from executeWorkflow
    @returns {string}
    """

    returnValue: str = ""

    try:

        returnValue = http.request(
            url = vcoUrl + "/api/workflows/" + workflowId + "/executions/" + \
            executionId + "/state",
            bearerToken = bearerToken
        )["value"]

    except Exception as err:
        raise ValueError(f"An error occurred at get workflow state - {err}") \
            from err

    return returnValue


def getWorkflowLog(
    vcoUrl: str,
    bearerToken: str,
    workflowId: str,
    executionId: str
) -> dict:
    """ Delivers the workflow run log.

    @param {string} vcoUrl - URL of Aria orchestrator
    @param {string} bearerToken
    @param {string} workflowId - ID of the workflow
    @param {string} executionId - ID of the execution, from executeWorkflow
    @returns {dictionary}
    """

    returnValue: dict = {}

    try:

        returnValue = http.request(
            url = vcoUrl + "/api/workflows/" + workflowId + "/executions/" + \
            executionId + "/logs?maxResult=2147483647",
            bearerToken = bearerToken
        )

    except Exception as err:
        raise ValueError(f"An error occurred at get workflow log - {err}") \
            from err

    return returnValue


def getWorkflowSyslog(
    vcoUrl: str,
    bearerToken: str,
    workflowId: str,
    executionId: str
) -> dict:
    """ Delivers the workflow scripting and system log.

    @param {string} vcoUrl - URL of Aria orchestrator
    @param {string} bearerToken
    @param {string} workflowId - ID of the workflow
    @param {string} executionId - ID of the execution, from executeWorkflow
    @returns {dictionary}
    """

    returnValue = {}

    try:

        returnValue = http.request(
            url = vcoUrl + "/api/workflows/" + workflowId + "/executions/" + \
            executionId + "/syslogs?maxResult=2147483647",
            bearerToken = bearerToken
        )

    except Exception as err:
        raise ValueError(f"An error occurred at get workflow syslog - {err}") \
            from err

    return returnValue


def getOutputParameters(
    vcoUrl: str,
    bearerToken: str,
    workflowId: str,
    executionId: str
) -> dict:
    """ Delivers the output parameters of the workflow run.

    @param {string} vcoUrl - URL of Aria orchestrator
    @param {string} bearerToken
    @param {string} workflowId - ID of the workflow
    @param {string} executionId - ID of the execution, from executeWorkflow
    @returns {dictionary}
    """

    returnValue: dict = {}

    try:

        returnValue = http.request(
            url = vcoUrl + "/api/workflows/" + workflowId + "/executions/" + \
            executionId,
            bearerToken = bearerToken
        )["output-parameters"]

    except Exception as err:
        raise ValueError(f"An error occurred at get workflow syslog - {err}") \
            from err

    return returnValue


def callWorkflow(
    vcoUrl: str,
    bearerToken: str,
    workflowFolder: str,
    workflowName: str,
    parameters: dict
) -> dict:
    """ Calls a workflow

    @param {string} vcoUrl - URL of Aria orchestrator
    @param {string} bearerToken
    @param {string} workflowFolder - Folder of the action
    @param {string} workflowName - Name of the workflow
    @param {dictionary} parameters - Parameters of the workflow
    @returns {dictionary}
    """

    returnValue: dict = {}

    try:

        workflowId = getWorkflowId(
            vcoUrl,
            bearerToken,
            workflowFolder,
            workflowName
        )

        executionResult = executeWorkflow(
            vcoUrl,
            bearerToken,
            workflowId,
            parameters
        )

        executionId = executionResult["id"]

        while True:

            workflowState = getWorkflowState(
                vcoUrl,
                bearerToken,
                workflowId,
                executionId
            )

            if workflowState == "completed":
                break
            else:
                time.sleep(2)

        outputParameters = getOutputParameters(
            vcoUrl,
            bearerToken,
            workflowId,
            executionId
        )

        workflowLog = getWorkflowLog(
            vcoUrl,
            bearerToken,
            workflowId,
            executionId
        )["logs"]

        workflowSyslog = getWorkflowSyslog(
            vcoUrl,
            bearerToken,
            workflowId,
            executionId
        )["logs"]

        returnValue["outputParameters"] = outputParameters
        returnValue["runLogs"] = workflowLog
        returnValue["sysLogs"] = workflowSyslog

    except Exception as err:
        raise ValueError(f"An error occurred at call workflow - {err}") \
            from err

    return returnValue


def handler(context: dict, inputs: dict) -> dict:
    """ Aria Automation standard handler, the main function.
    """

    vcoUrl: str = context["vcoUrl"]
    bearerToken: str = context["getToken"]()

    output: dict = {}

    try:

        # Begin of workflow call ---------------------------------------

        # @example
        workflowName: str = "Get virtual machines by name"
        workflowFolder: str = \
            "Library/vCenter/Virtual Machine management/Basic"
        parameters: dict = { "parameters":
            [
                {
                    "name": "criteria",
                    "type": "string",
                    "value": {
                        "string": {
                            "value": "HP0STB"
                        }
                    }
                }
            ]
        }

        output = callWorkflow(
            vcoUrl,
            bearerToken,
            workflowFolder,
            workflowName,
            parameters
        )

        # End of workflow call -----------------------------------------

        outputs = {
            "status": "done",
            "error": None,
            "result": output
        }

    except Exception as err:

        outputs = {
            "status": "incomplete",
            "error": repr(err),
            "result": output
        }

    return outputs

vcf automation action to execute workflows from python

Python Code from Terminal Executes Workflow

We can also execute the same code from the console with a few modifications. It is necessary to add the function getBearerToken and to implement a main function to call the handler.

python code from terminal executes workflow

References

Hint: The setting of the parameters is described in the REST API Calling Conventions. The only difference compared to the use in an action is that the parameters must be bundled in an array with the same name.
HTTP Class

VCF Automation Blog

from Stefan Schnell

Execute Workflow from Python

Python Code from Terminal Executes Workflow

References