Writing workflows in Python

YAML works well when each task is a command. Use Python when a task needs real logic, such as reading and transforming files, calling a library, or prompting the user. Prompting is something YAML cannot do.

You build the same workflow object, but you define tasks as Python functions with the @FunctionTask.task decorator and run the workflow with render_workflow, the same live dashboard horus run uses.

A minimal Python workflow

Two tasks wired by an edge: one writes a greeting, the next reads it and uppercases it.

Download hello_workflow.py

hello_workflow.py

from pathlib import Path

from horus_builtin.artifact.file import FileArtifact
from horus_builtin.event.tui_subscriber import render_workflow
from horus_builtin.target.local import LocalTarget
from horus_builtin.task.function import FunctionTask
from horus_builtin.workflow.horus_workflow import HorusWorkflow
from horus_runtime.context import HorusContext
from horus_runtime.core.artifact.base import BaseArtifact
from horus_runtime.core.workflow.edge import WorkflowEdge

# Boot the runtime once before building or running anything.
HorusContext.boot()

wf = HorusWorkflow(name="hello_python")

work = LocalTarget(working_directory="./horus-work")
greeting = FileArtifact(id="greeting", path=Path("horus-out/greeting.txt"))
shout = FileArtifact(id="shout", path=Path("horus-out/shout.txt"))


# The task id and name default to the function name. The parameters after
# `task` are the task's artifacts, matched by their id.
@FunctionTask.task(wf, outputs=[greeting], target=work)
async def make_greeting(
    task: FunctionTask, greeting: FileArtifact
) -> list[BaseArtifact]:
    """Write a greeting to the output artifact."""
    greeting.write("Hello from a Python workflow!")
    return []


@FunctionTask.task(wf, inputs=[greeting], outputs=[shout], target=work)
async def shout_it(
    task: FunctionTask, greeting: FileArtifact, shout: FileArtifact
) -> list[BaseArtifact]:
    """Read the greeting and write an uppercased version."""
    shout.write(greeting.read().upper())
    return []


wf.edges.append(
    WorkflowEdge(
        source="make_greeting",
        source_output="greeting",
        target="shout_it",
        target_input="greeting",
    )
)

if __name__ == "__main__":
    render_workflow(wf, trigger_id="make_greeting")

python hello_workflow.py

How it fits together

HorusContext.boot() starts the runtime. Call it once, before building or running the workflow.
@FunctionTask.task(wf, inputs=, outputs=, target=) registers the decorated function as a task on wf. Its id and name default to the function name. The function receives task plus one argument per artifact, named by the artifact's id.
Inside a task, read and write artifacts directly with artifact.read() and artifact.write(...).
WorkflowEdge(...) wires producer to consumer, exactly like YAML edges.
render_workflow(wf, trigger_id=...) runs the workflow with the live TUI. To run without it, use asyncio.run(wf.run(trigger_id=...)).

Prompting the user

A task can ask the user questions through task.interaction. Under the live dashboard, the display pauses while the prompt is shown and resumes once you answer.

Download interactive_workflow.py

interactive_workflow.py

from pathlib import Path

from horus_builtin.artifact.file import FileArtifact
from horus_builtin.event.tui_subscriber import render_workflow
from horus_builtin.interaction.common.confirm import ConfirmInteraction
from horus_builtin.interaction.common.string import StringInteraction
from horus_builtin.target.local import LocalTarget
from horus_builtin.task.function import FunctionTask
from horus_builtin.workflow.horus_workflow import HorusWorkflow
from horus_runtime.context import HorusContext
from horus_runtime.core.artifact.base import BaseArtifact

HorusContext.boot()

wf = HorusWorkflow(name="interactive")

work = LocalTarget(working_directory="./horus-work")
greeting = FileArtifact(id="greeting", path=Path("horus-out/greeting.txt"))


@FunctionTask.task(wf, outputs=[greeting], target=work)
async def greet_user(
    task: FunctionTask, greeting: FileArtifact
) -> list[BaseArtifact]:
    """Ask the user for their name, confirm, then write a greeting."""
    name = await task.interaction.ask(
        StringInteraction(
            value_key="name",
            prompt="What is your name?",
            default="World",
        )
    )

    shout = await task.interaction.ask(
        ConfirmInteraction(
            value_key="shout",
            prompt="Shout the greeting?",
            default=False,
        )
    )

    message = f"Hello, {name}!"
    greeting.write(message.upper() if shout else message)
    return []


if __name__ == "__main__":
    render_workflow(wf, trigger_id="greet_user")

Built-in interactions

from horus_builtin.interaction.common.string import StringInteraction

answer: str = await task.interaction.ask(
    StringInteraction(
        value_key="name",
        prompt="What is your name?",
        default="World",     # used when the user enters nothing
        placeholder="e.g. Ada",
    )
)

from horus_builtin.interaction.common.confirm import ConfirmInteraction

ok: bool = await task.interaction.ask(
    ConfirmInteraction(
        value_key="proceed",
        prompt="Continue?",
        default=True,
    )
)

The answer is validated and coerced to the right type: str for StringInteraction, bool for ConfirmInteraction. An answer that cannot be parsed prompts again. To add your own interaction types, see the Interaction reference.

Subworkflows

wf.subworkflow(id=, body=, port_overrides=None, max_depth=None, name=None, target=None) inlines another workflow into wf as a single task. body is a child HorusWorkflow, built the same way as any top-level workflow:

from pathlib import Path

from horus_builtin.artifact.file import FileArtifact
from horus_builtin.event.tui_subscriber import render_workflow
from horus_builtin.executor.shell import ShellExecutor
from horus_builtin.runtime.command import CommandRuntime
from horus_builtin.target.local import LocalTarget
from horus_builtin.task.horus_task import HorusTask
from horus_builtin.workflow.horus_workflow import HorusWorkflow
from horus_runtime.context import HorusContext
from horus_runtime.core.workflow.edge import WorkflowEdge

HorusContext.boot()

work = LocalTarget(working_directory="./horus-work")


def _shell_task(task_id, command, *, inputs=None, outputs=None):
    return HorusTask(
        id=task_id,
        name=task_id,
        runtime=CommandRuntime(command=command),
        executor=ShellExecutor(),
        target=work,
        inputs=inputs or [],
        outputs=outputs or [],
    )


# The child is an ordinary workflow: a root artifact feeding two tasks.
child = HorusWorkflow(
    name="child",
    artifacts=[FileArtifact(id="seed", path=Path("seed_in.txt"))],
    tasks=[
        _shell_task(
            "upper",
            "tr a-z A-Z < $seed > $upped",
            inputs=[FileArtifact(id="seed", path=Path("seed_in.txt"))],
            outputs=[FileArtifact(id="upped", path=Path("upped.txt"))],
        ),
        _shell_task(
            "report",
            "cat $upped > $report_out",
            inputs=[FileArtifact(id="upped", path=Path("upped.txt"))],
            outputs=[FileArtifact(id="report_out", path=Path("report.txt"))],
        ),
    ],
    edges=[
        WorkflowEdge(
            source="artifact-seed", source_output="seed",
            target="upper", target_input="seed",
        ),
        WorkflowEdge(
            source="upper", source_output="upped",
            target="report", target_input="upped",
        ),
    ],
)

wf = HorusWorkflow(name="parent")
wf.subworkflow(id="sub", body=child)

if __name__ == "__main__":
    render_workflow(wf, trigger_id="sub")

Ports work exactly as in YAML: in-ports are the child's root artifacts, out-ports are its unconsumed leaf outputs (here, seed in and report_out out), and port_overrides={"derived_name": "parent_facing_name"} renames one. Wire a parent task to a port with a normal WorkflowEdge, but set transfer=False since the edge crosses the subworkflow boundary — the real data edge gets rewired directly to the inlined inner task at validation time. Inlined tasks are reachable afterwards by their prefixed id, e.g. sub/report. See Subworkflows for the full rules on ports, nesting, and max_depth.

Next, Running workflows covers the CLI and the live dashboard.

Writing workflows (Python)