Workflow Service Run Task

task api name: run

Runs a workflow defined as a YAML file reference or an inline dictionary. Workflow steps call other NorFab services and collect the per-step job results under the workflow name.

Inputs

Parameter	Required	Description
workflow	Yes	Workflow definition dictionary or URL to a YAML workflow file

Output

Returns workflow execution results keyed by workflow name and step name:

{
    "workflow_1": {
        "step1": {
            "nornir-worker-1": {
                "result": {"ceos-spine-1": {"show version": "..."}},
                "failed": False,
                "errors": [],
            },
        },
        "step2": {
            "nornir-worker-2": {
                "result": {"ceos-leaf-1": {"show hostname": "..."}},
                "failed": False,
                "errors": [],
            },
        },
    },
}

Workflow File Example

Workflow service run task uses YAML files to execute workflow steps:

workflow-1.yaml

name: workflow_1
description: Sample workflow with two steps.

step1:
  service: nornir
  task: cli
  kwargs:
    FC: spine
    commands:
      - show version
      - show ip int brief

step2:
  service: nornir
  task: cli
  kwargs:
    FC: leaf
    commands:
      - show hostname
      - show ntp status

Store the file on the broker, for example as nf://workflow/workflow-1.yaml, before running the workflow.

Examples

CLIPython

Run a workflow from a broker file:

nf#workflow run workflow nf://workflow/workflow-1.yaml

Run with a longer timeout:

nf#workflow run workflow nf://workflow/workflow-1.yaml timeout 900

Context manager - run from a broker file:

import pprint

from norfab.core.nfapi import NorFab

with NorFab(inventory="./inventory.yaml") as nf:
    client = nf.make_client()

    result = client.run_job(
        service="workflow",
        task="run",
        workers="any",
        kwargs={
            "workflow": "nf://workflow/workflow-1.yaml",
        },
    )
    pprint.pprint(result)

Direct lifecycle - run an inline workflow:

import pprint

from norfab.core.nfapi import NorFab

workflow = {
    "name": "inline_workflow",
    "description": "Collect basic command output from lab devices.",
    "show_version": {
        "service": "nornir",
        "task": "cli",
        "kwargs": {
            "FC": "spine",
            "commands": ["show version"],
        },
    },
    "show_hostname": {
        "service": "nornir",
        "task": "cli",
        "kwargs": {
            "FC": "leaf",
            "commands": ["show hostname"],
        },
    },
}

nf = NorFab(inventory="./inventory.yaml")
try:
    nf.start()
    client = nf.make_client()

    result = client.run_job(
        service="workflow",
        task="run",
        workers="any",
        kwargs={"workflow": workflow},
    )
    pprint.pprint(result)
finally:
    nf.destroy()

NORFAB Workflow Run Command Shell Reference

NorFab shell supports these command options for Workflow run task:

nf# man tree workflow
root
└── workflow:    Workflow service
    └── run:    Run workflows
        ├── timeout:    Job timeout
        ├── workers:    Filter worker to target, default 'all'
        ├── workflow:    Workflow to run
        └── progress:    Display progress events, default 'True'
nf#

Python API Reference

Executes a workflow defined by a dictionary.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata.	required
workflow	Union[str, Dict]	The workflow to execute. This can be a URL to a YAML file.	required

Returns:

Name	Type	Description
Dict	Result	A dictionary containing the results of the workflow execution.

Raises:

Type	Description
ValueError	If the workflow is not a valid URL or dictionary.

Source code in norfab\workers\workflow_worker\workflow_worker.py

@Task(
    input=RunInput,
    output=RunResult,
    fastapi={"methods": ["POST"]},
    mcp={
        "annotations": {
            "title": "Run Workflow",
            "readOnlyHint": False,
            "destructiveHint": True,
            "idempotentHint": False,
            "openWorldHint": True,
        }
    },
)
def run(self, job: Job, workflow: Union[str, Dict]) -> Result:
    """
    Executes a workflow defined by a dictionary.

    Args:
        job (Job): NorFab Job object containing relevant metadata.
        workflow (Union[str, Dict]): The workflow to execute. This can be a URL to a YAML file.

    Returns:
        Dict: A dictionary containing the results of the workflow execution.

    Raises:
        ValueError: If the workflow is not a valid URL or dictionary.
    """
    ret = Result(task=f"{self.name}:run", result={})

    # load workflow from URL
    if self.is_url(workflow):
        workflow_name = (
            os.path.split(workflow)[-1].replace(".yaml", "").replace(".yml", "")
        )
        workflow = self.fetch_file(workflow)
        workflow = yaml.safe_load(workflow)

    # extract workflow parameters
    workflow_name = workflow.pop("name", "workflow")
    workflow_description = workflow.pop("description", "")
    remove_no_match_results = workflow.pop("remove_no_match_results", True)

    job.event(f"starting workflow '{workflow_name}'")
    log.info(f"Starting workflow '{workflow_name}': {workflow_description}")

    ret.result[workflow_name] = {}

    # run each step in the workflow
    for step, data in workflow.items():
        # check if need to skip step based on run_if_x flags
        skip_status, message = self.skip_step_check(
            ret.result[workflow_name], step, data
        )
        if skip_status is True:
            ret.result[workflow_name][step] = {
                "all-workers": {
                    "failed": False,
                    "result": None,
                    "status": "skipped",
                    "task": data["task"],
                    "errors": [],
                    "messages": [message],
                    "juuid": None,
                }
            }
            job.event(
                f"skipping workflow step '{step}', one of run_if_x conditions not satisfied"
            )
            continue
        # stop workflow execution on error
        elif skip_status == "error":
            ret.result[workflow_name][step] = {
                "all-workers": {
                    "failed": True,
                    "result": None,
                    "status": "error",
                    "task": data["task"],
                    "errors": [message],
                    "messages": [],
                    "juuid": None,
                }
            }
            job.event(message)
            log.error(message)
            break

        job.event(f"doing workflow step '{step}'")

        ret.result[workflow_name][step] = self.client.run_job(
            service=data["service"],
            task=data["task"],
            workers=data.get("workers", "all"),
            kwargs=data.get("kwargs", {}),
            args=data.get("args", []),
            timeout=data.get("timeout", 600),
        )

        # check if need to stop workflow based on stop_if_fail flag
        if (
            self.stop_workflow_check(ret.result[workflow_name][step], step, data)
            is True
        ):
            job.event(
                f"stopping workflow, step '{step}' failed and has stop_if_fail flag"
            )
            break

    if remove_no_match_results:
        ret.result[workflow_name] = self.remove_no_match_results(
            ret.result[workflow_name]
        )

    log.info(
        f"Completed workflow '{workflow_name}' with {len(ret.result[workflow_name])} step result(s)"
    )

    return ret