Skip to main content

Python Function Tasks and Execution Behaviors

PythonFunctionTask is the class that underlies almost every task you write in flytekit. When you apply @task to a function, you get back a PythonFunctionTask instance — not the original function. Understanding what that instance does, and how its three execution modes work, is the key to understanding dynamic workflows, eager workflows, and the plugin system.

From Function to Task: PythonFunctionTask​

The @task decorator lives in flytekit/core/task.py and wraps a Python callable in a PythonFunctionTask (or a subclass). The constructor immediately inspects the function:

self._native_interface = transform_function_to_interface(
    task_function, Docstring(callable_=task_function), pickle_untyped=pickle_untyped
)

Type annotations become the task's Flyte interface. A function like def add(a: int, b: int) -> int yields an interface with two integer inputs and one integer output — no extra registration step required. The task name comes from extract_task_module(task_function), which produces a fully qualified name like mypackage.mymodule.add.

A basic task looks like this:

from flytekit import task, Resources

@task(
    retries=2,
    cache=True,
    cache_version="v1",
    requests=Resources(cpu="1", mem="1Gi"),
    container_image="ghcr.io/myorg/myimage:latest",
    environment={"LOG_LEVEL": "DEBUG"},
)
def add(a: int, b: int) -> int:
    return a + b

The decorator accepts a wide range of configuration: retries, cache/cache_version, requests/limits/resources, container_image, environment, secret_requests, interruptible, timeout, pod_template, and accelerator, among others.

The Nested Function Restriction​

One sharp edge: if your function is defined inside another function (a closure or inner function), PythonFunctionTask.__init__ raises a ValueError at definition time:

TaskFunction cannot be a nested/inner or local function.
It should be accessible at a module level for Flyte to execute it.

The check uses extract_task_module and the isnested / is_functools_wrapped_module_level helpers. The exception is code in test modules (test_*.py) where nesting is allowed for convenience. If you use a custom decorator, wrapping the inner function with functools.wraps at module level satisfies the check.

Async Functions Become AsyncPythonFunctionTask​

When @task is applied to a coroutine function, task.py substitutes AsyncPythonFunctionTask for PythonFunctionTask:

if inspect.iscoroutinefunction(fn):
    if task_plugin is PythonFunctionTask:
        task_plugin = AsyncPythonFunctionTask
    else:
        if not issubclass(task_plugin, AsyncPythonFunctionTask):
            raise AssertionError(f"Task plugin {task_plugin} is not compatible with async functions")

AsyncPythonFunctionTask overrides __call__ to be async, delegates to async_flyte_entity_call_handler, and provides async_execute which awaits the user's coroutine. The synchronous execute is derived via loop_manager.synced(async_execute) so the task can still be called from synchronous dispatch paths:

execute = loop_manager.synced(async_execute)
@task
async def fetch_data(url: str) -> str:
    # runs as an AsyncPythonFunctionTask; the decorated function is awaited
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.text()

AsyncPythonFunctionTask supports ExecutionBehavior.DEFAULT only. Calling it with DYNAMIC raises NotImplementedError — async and dynamic execution modes do not mix.

The Three Execution Behaviors​

PythonFunctionTask contains a nested enum that controls how execute() behaves:

class ExecutionBehavior(Enum):
    DEFAULT = 1
    DYNAMIC = 2
    EAGER = 3

DEFAULT is what @task gives you. DYNAMIC is what @dynamic gives you. EAGER is what @eager gives you. All three share the same PythonFunctionTask base but diverge significantly at execution time.

DEFAULT: Run the Function Directly​

In DEFAULT mode, execute() is straightforward:

def execute(self, **kwargs) -> Any:
    if self.execution_mode == self.ExecutionBehavior.DEFAULT:
        return self._task_function(**kwargs)

The platform deserializes inputs, calls your function with native Python values, and serializes the outputs. This is the path for every @task-decorated function.

DYNAMIC: Compile the Function Body Into a Workflow​

@dynamic is implemented as:

# flytekit/core/dynamic_workflow_task.py
dynamic = functools.partial(task.task, execution_mode=PythonFunctionTask.ExecutionBehavior.DYNAMIC)

It is literally @task with execution_mode=DYNAMIC. The function body does not run to produce a result — it runs to produce a workflow spec. When the backend executes a dynamic task, it calls dynamic_execute, which calls compile_into_workflow:

def execute(self, **kwargs) -> Any:
    elif self.execution_mode == self.ExecutionBehavior.DYNAMIC:
        return self.dynamic_execute(self._task_function, **kwargs)

Inside compile_into_workflow, flytekit wraps the user function in a PythonFunctionWorkflow, compiles it (which runs the function body under a compilation context so task calls produce node edges rather than outputs), serializes it to a WorkflowSpec, and returns a DynamicJobSpec. The backend then schedules that spec as a subworkflow.

@task
def process(item: int) -> str:
    return f"item-{item}"

@dynamic
def fan_out(n: int) -> typing.List[str]:
    # This body runs at execution time to build a workflow spec.
    # `n` is available as a real Python int.
    results = []
    for i in range(n):
        results.append(process(item=i))
    return results

Locally, dynamic_execute detects it is in a local execution and runs the resulting PythonFunctionWorkflow directly, giving you the same result as production without a real Flyte cluster.

Be careful with size. The dynamic_workflow_task.py module-level docstring states: "Please keep dynamic workflows to under fifty tasks. For large-scale identical runs, we recommend the upcoming map task." A loop that produces thousands of nodes creates an enormous workflow spec that the backend must process.

node_dependency_hints for Dynamic LaunchPlan Dependencies​

When a dynamic task calls a LaunchPlan, the platform cannot discover that dependency during static registration (it's only visible at runtime). Pass node_dependency_hints to pre-register those entities:

@workflow
def my_workflow() -> int:
    ...

my_lp = LaunchPlan.get_or_create(my_workflow, "my_lp")

@dynamic(node_dependency_hints=[my_lp])
def launch_dynamically() -> typing.List[int]:
    # my_lp is already registered, so this works
    return [my_lp() for _ in range(5)]

Using node_dependency_hints on a non-dynamic task raises a ValueError immediately at definition time: "node_dependency_hints should only be used on dynamic tasks."

EAGER: Python Becomes the Orchestrator​

Eager workflows are a different model entirely. Instead of compiling a workflow spec, the Python process itself drives execution, using a FlyteRemote connection to launch each task call as a real Flyte execution on the backend. The docstring puts it well:

"Python becomes propeller, and every task invocation, creates a stack frame on the Flyte cluster in the form of an execution rather than on the actual memory stack."

The @eager decorator creates an EagerAsyncPythonFunctionTask. The decorated function must be async:

from flytekit import task, eager
from flytekit.exceptions.eager import EagerException

@task
def add_one(x: int) -> int:
    return x + 1

@task
def double(x: int) -> int:
    return x * 2

@eager
async def my_eager_workflow(x: int) -> int:
    out = add_one(x=x)       # call — returns a value locally or launches remotely
    return double(x=out)

Locally, await my_eager_workflow(x=3) runs both tasks inline and returns 8. On the backend, each task call becomes a full execution tagged with eager-exec.

EagerAsyncPythonFunctionTask.__init__ forcibly sets execution_mode=EAGER and TaskMetadata.is_eager=True regardless of what the caller passes. Any execution_mode kwarg in **kwargs is silently deleted before the super-call:

if "execution_mode" in kwargs:
    del kwargs["execution_mode"]

if "metadata" in kwargs:
    kwargs["metadata"].is_eager = True
else:
    kwargs["metadata"] = TaskMetadata(is_eager=True)

Parallelism with asyncio​

Because eager tasks are async, asyncio.create_task and asyncio.gather give you parallelism. Two eager sub-workflows launch concurrently and their results are awaited together:

@eager
async def parent_wf(a: int, b: int) -> typing.Tuple[int, int]:
    t1 = asyncio.create_task(base_wf(x=a))
    t2 = asyncio.create_task(base_wf(x=b))
    i1, i2 = await asyncio.gather(t1, t2)
    return i1, i2

Both base_wf(x=a) and base_wf(x=b) are in-flight simultaneously.

Error Handling with EagerException​

When a task inside an eager workflow fails, flytekit wraps the underlying error in an EagerException from flytekit.exceptions.eager. Your workflow code catches it using a plain try/except:

@eager
async def safe_workflow(x: int) -> int:
    try:
        out = await add_one(x=x)
    except EagerException:
        # The original ValueError is caught and re-raised as EagerException
        raise
    return await double(x=out)

EagerException is the signal that a sub-execution failed — inspect it or propagate it as needed.

Backend Execution and the Controller​

On the backend, EagerAsyncPythonFunctionTask.execute() constructs a Controller (the worker queue) on the main thread. Signal handlers for SIGINT and SIGTERM are registered at this point — the source comment explains this is required because signal handlers can only be installed on the main thread. The Controller gets a FlyteRemote reference derived from the current execution's project and domain, and uses an execution prefix derived from the task name:

prefix = self.name.split(".")[-1][:8]
prefix = f"e-{prefix}-{tag[:5]}"
prefix = _dnsify(prefix)
c = Controller(remote=remote, ss=ss, tag=tag, root_tag=root_tag, exec_prefix=prefix)

The root_tag is read from the _F_EE_ROOT environment variable (EAGER_ROOT_ENV_NAME) if present, otherwise it defaults to the current execution name. This tracks the root eager execution when eager workflows are nested.

The EagerFailureHandlerTask Cleanup Mechanism​

What happens to still-running child executions when an eager workflow fails mid-flight? EagerAsyncPythonFunctionTask.get_as_workflow() produces an ImperativeWorkflow that wraps the eager task and attaches an EagerFailureHandlerTask as its on_failure handler:

def get_as_workflow(self):
    cleanup = EagerFailureHandlerTask(
        name=f"{self.name}-cleanup",
        container_image=self.container_image,
        inputs=self.python_interface.inputs,
    )
    wb = ImperativeWorkflow(name=self.name)
    ...
    wb.add_on_failure_handler(cleanup)
    return wb

During serialization (serialize_helpers.py), every EagerAsyncPythonFunctionTask is registered via this workflow wrapper, not as a standalone task. The EagerFailureHandlerTask uses the EagerFailureTaskResolver (a singleton eager_failure_task_resolver at module level in python_function_task.py) to reconstruct itself at remote execution time without needing the original Python interface.

When the failure handler runs, its dispatch_execute method queries Flyte admin for all child executions tagged eager-exec=<parent_execution_name> that are still in QUEUED or RUNNING phase and terminates them in a loop:

key_filter = ValueIn("execution_tag.key", ["eager-exec"])
value_filter = ValueIn("execution_tag.value", [name])
phase_filter = ValueIn("phase", ["UNDEFINED", "QUEUED", "RUNNING"])

EagerFailureHandlerTask.execute() raises AssertionError — it is only valid at remote dispatch time, never called locally.

Eager + Dynamic Don't Mix​

Eager and dynamic are mutually exclusive. AsyncPythonFunctionTask.async_execute() explicitly raises NotImplementedError for DYNAMIC mode. You can call a @dynamic task from an @eager workflow (as a regular call), but you cannot apply both decorators to the same function.

Similarly, eager tasks cannot be used in map tasks. ArrayNodeMapTask only accepts PythonFunctionTask instances with DEFAULT mode or PythonInstanceTask instances.

PythonInstanceTask: Tasks Without a Function Body​

Not every task wraps a user-written function. PythonInstanceTask is the abstract base for tasks that are instantiated as objects and provide their own execute() implementation. Where PythonFunctionTask is built around a Callable, PythonInstanceTask extends PythonAutoContainerTask directly:

class PythonInstanceTask(PythonAutoContainerTask[T], ABC):
    def __init__(
        self,
        name: str,
        task_config: T,
        task_type: str = "python-task",
        task_resolver: Optional[TaskResolverMixin] = None,
        **kwargs,
    ):
        super().__init__(name=name, task_config=task_config, task_type=task_type, task_resolver=task_resolver, **kwargs)

The class docstring shows the usage pattern:

x = MyInstanceTask(name="x", task_config=..., ...)
x(a=5)

ShellTask in flytekit/extras/tasks/shell.py is the canonical example — the task is an object you configure, not a function you decorate. The module loader captures the variable name (x) and module path to reconstruct the task at execution time, which is why the base class is designed the way it is.

Extending PythonFunctionTask: The Plugin Pattern​

Flytekit's plugin ecosystem is built on two facts: PythonFunctionTask is generic over T (the task config type), and TaskPlugins maps config types to task classes.

To create a custom task type, subclass PythonFunctionTask, define a config dataclass, and register the mapping:

from flytekit.core.python_function_task import PythonFunctionTask
from flytekit.extend import TaskPlugins

@dataclass
class MyConfig:
    option: str = "default"

class MyTask(PythonFunctionTask[MyConfig]):
    _TASK_TYPE = "my_task"

    def __init__(self, task_config, task_function, **kwargs):
        super().__init__(
            task_config=task_config or MyConfig(),
            task_type=self._TASK_TYPE,
            task_function=task_function,
            **kwargs,
        )

    def get_custom(self, settings) -> dict:
        return {"option": self.task_config.option}

TaskPlugins.register_pythontask_plugin(MyConfig, MyTask)

After registration, @task(task_config=MyConfig(option="fast")) automatically creates a MyTask instance. The task.py decorator calls TaskPlugins.find_pythontask_plugin(type(task_config)) to resolve the class. The flytekit-mmcloud, flytekit-ray, and flytekit-kf-pytorch plugins all follow this pattern (MMCloudTask, RayFunctionTask, PyTorchFunctionTask).

If the task config type is async-compatible, the plugin class must subclass AsyncPythonFunctionTask instead — the @task decorator enforces this with an AssertionError if you pass an async function to a plugin that doesn't inherit from AsyncPythonFunctionTask.

Summary of the Class Hierarchy​

PythonAutoContainerTask
├── PythonInstanceTask (ABC)          # for object-based tasks like ShellTask
└── PythonFunctionTask                # @task (DEFAULT and DYNAMIC modes)
    └── AsyncPythonFunctionTask       # @task on async functions (DEFAULT only)
        └── EagerAsyncPythonFunctionTask  # @eager (EAGER mode exclusively)

EagerFailureHandlerTask sits outside this hierarchy — it extends PythonAutoContainerTask directly and has a fixed task type of "eager_failure_handler_task". It is never instantiated by user code; EagerAsyncPythonFunctionTask.get_as_workflow() creates it automatically when an eager workflow is serialized.