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.

Important 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.
"""
@module de.stschnell

@version 0.1.0

@runtime python:3.10

@memoryLimit 128000000

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

@outputType Properties
"""


import base64
import json
import ssl
import time
import urllib.request


def request(
    url,
    user = None,
    password = None,
    bearerToken = None,
    method = "GET",
    body = {},
    contentType = "application/json;charset=utf-8",
    accept = "application/json"
):
    """ Executes a REST request

    @param {string} url - URL to execute the request
    @param {string} user
    @param {string} password
    @param {string} bearerToken
    @param {string} method - Method of request, e.g. GET, POST, etc
    @param {dictionary} body - Body of request
    @param {string} contentType - MIME type of the body for the request
    @param {string} accept - MIME type of content expect/prefer as response
    @returns {dictionary or bytes}
    """

    returnValue = {}

    try:

        request = urllib.request.Request(
            url = url,
            method = method,
            data = bytes(json.dumps(body).encode("utf-8"))
        )

        if user and password:
            authorization = base64.b64encode(
                bytes(user + ":" + password, "UTF-8")
            ).decode("UTF-8")
            request.add_header(
                "Authorization", "Basic " + authorization
            )

        if bearerToken:
            request.add_header(
                "Authorization", "Bearer " + bearerToken
            )

        request.add_header(
            "Content-Type", contentType
        )

        request.add_header(
            "Accept", accept
        )

        response = urllib.request.urlopen(
            request,
            context = ssl._create_unverified_context()
        )

        if response.getcode() == 200 or response.getcode() == 202:
            if "json" in accept:
                returnValue = json.loads(response.read())
            else:
                returnValue = response.read()

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

    return returnValue


def getCategoryPath(
    vcoUrl,
    bearerToken,
    categoryId
):
    """ 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 = 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,
    bearerToken,
    workflowFolder,
    workflowName
):
    """ 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 = ""

    try:

        actions = 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,
    bearerToken,
    workflowId,
    parameters = {}
):
    """ 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 = 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,
    bearerToken,
    workflowId,
    executionId
):
    """ 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 = ""

    try:

        returnValue = 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,
    bearerToken,
    workflowId,
    executionId
):
    """ 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 = {}

    try:

        returnValue = 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,
    bearerToken,
    workflowId,
    executionId
):
    """ 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 = 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,
    bearerToken,
    workflowId,
    executionId
):
    """ 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 = {}

    try:

        returnValue = 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,
    bearerToken,
    workflowFolder,
    workflowName,
    parameters
):
    """ 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 = {}

    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,
    inputs
):
    """ Aria Automation standard handler, the main function.
    """

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

    output = {}

    try:

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

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

        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 Blog

from Stefan Schnell

Execute Workflow from Python
