Writing tierkreis workers

One of the core concepts in tierkreis are workers. In this example, we will cover the conceptual ideas of workers and how to write one. For this we will write a simple worker from scratch. The tkr init cli provides a convenient set up for this which will automate most of the tasks described after.

Concepts

Workers encapsulate a set of functionalities which are described in their API (stubs). The API contains a list of typed function interfaces which are implemented by the worker. Tierkreis expects the functions to be stateless and will only checkpoint inputs and outputs of the function call. Although, it is still possible to introduce state, but this then has to be managed by the programmer.

Each worker is an independent program, can have multiple implementations (e.g. based on architecture) of these interfaces and has its own dependencies. There are two main ways to write workers:

  • Using the tierkreis Worker class for python based workers

  • Using a spec file to define the interface for non-python workers

In this example, we will write a simple worker, that greets a subject. greet: str -> str -> str

Worker Setup

A worker consist of four conceptual parts which we will set up in this example:=

  1. Dependency information. We recommend using uv and a pyproject.toml.

  2. A main entry point into the worker called main.py

  3. The api definition of the worker typically called stubs.py or api.py. For python workers this can be generated from the main file.

  4. (Optional) Library code containing the core logic.

1.Dependencies

Dependency information should be provide, such that the worker can run in its installed location. For python, this is canonically done through the pyproject.toml which tierkreis also uses.

[project]
name = "hello-world-worker"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = ["tierkreis"]

[dependency-groups]
# Optional dev dependency group
dev = ["ruff"]

Ruff is an optional dependency to format the code later.

2. Worker entry point

The entrypoint contains the implementations of the interface. We are going to use the build in Worker class to define the worker. Each worker needs a unique name, here we chose hello_world_worker. In tierkreis the worker is defined by its name and the directory it lives in.

%pip install tierkreis

/home/runner/work/tierkreis/tierkreis/.venv/bin/python3: No module named pip

Note: you may need to restart the kernel to use updated packages.

import logging
from pathlib import Path

from tierkreis import Worker

logger = logging.getLogger(__name__)
worker = Worker("hello_world_worker")

From here we can add functions to the worker by decorating python functions with @worker.task(). To make use of the type safety in tierkreis, you should provide narrow types to the function.

@worker.task()
def greet(greeting: str, subject: str) -> str:
    logger.info("%s %s", greeting, subject)
    return greeting + subject

Now we have defined a greet function that concatenates a two strings. It now lives in the namespace hello_world_worker.greet. To make sure the worker runs correctly we need to add the following lines:

%%script false --no-raise-error
from sys import argv
if __name__ == "__main__":
    worker.app(argv)

They make sure that tierkreis can call the worker correctly.

3. Api definition

For python workers, tierkreis can automatically generate the api definition. You can either run the app as uv run main.py --stubs-path ./api/stubs.py or call the function directly.

worker.namespace.write_stubs(Path("./api/stubs.py"))

1 file reformatted
Found 10 errors (10 fixed, 0 remaining).

This will generate a file stubs.py that contains the interface definition that can now be used as task in a tierkreis workflow. The code will look like this:

"""Code generated from hello_world_worker namespace. Please do not edit."""

from typing import NamedTuple
from tierkreis.controller.data.models import TKR


class greet(NamedTuple):
    greeting: TKR[str]  # noqa: F821 # fmt: skip
    subject: TKR[str]  # noqa: F821 # fmt: skip

    @staticmethod
    def out() -> type[TKR[str]]:  # noqa: F821 # fmt: skip
        return TKR[str]  # noqa: F821 # fmt: skip

    @property
    def namespace(self) -> str:
        return "hello_world_worker"

Which defines a class describing the function.