Task authoring and execution

Tasks are the atomic unit of work in Flyte — independently versioned, typed, retryable pieces of code. Understanding how flytekit models a task, from the abstract base class down to the Python function you actually write, makes it much easier to reason about what happens when your workflow runs.

The Task Class Hierarchy

Flytekit structures tasks as a layered inheritance tree, each layer adding a distinct concern:

Task                        (flytekit/core/base_task.py)
└── PythonTask              (flytekit/core/base_task.py)
    └── PythonAutoContainerTask  (flytekit/core/python_auto_container.py)
        ├── PythonFunctionTask   (flytekit/core/python_function_task.py)
        │   └── AsyncPythonFunctionTask
        │       └── EagerAsyncPythonFunctionTask
        └── PythonInstanceTask   (flytekit/core/python_function_task.py)

Task is the root. It carries the raw IDL-level information: task_type (a string key that identifies the backend plugin), a fully-qualified name, a TypedInterface describing inputs and outputs in Flyte's type system, TaskMetadata, and an optional SecurityContext. Every Task instance registers itself into FlyteEntities.entities on construction, so the serialization toolchain can discover it.

PythonTask adds a Python-native Interface — types expressed as actual Python annotations rather than IDL protobuf descriptors — along with an environment dict and a generic task_config. It also implements dispatch_execute, the method that converts Flyte literals into Python values, calls execute, converts the outputs back to literals, and writes Deck files.

PythonAutoContainerTask (in flytekit/core/python_auto_container.py) layers on container-level concerns: container_image, Resources (requests and limits), pod_template, accelerator, shared_memory, and secret_requests. It owns the logic that constructs the pyflyte-execute command line stored in the serialized task's container spec.

PythonFunctionTask is what you use every day. It introspects a Python callable at decoration time to extract the task's name, module path, and type-annotated interface automatically.

Declaring a Task with @task

The @task decorator, defined in flytekit/core/task.py, is the standard entry point. At its simplest:

from flytekit import task

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

When this module is imported, @task calls transform_function_to_interface(double, ...) to extract {n: int} -> int as the Interface, then calls extract_task_module(double) to derive the fully-qualified task name (e.g., my_package.my_module.double). TaskPlugins.find_pythontask_plugin(type(None)) returns PythonFunctionTask (the default), and a PythonFunctionTask instance is constructed and stored in FlyteEntities.

For tasks that need specialized backends — Spark, Ray, distributed training — pass a task_config:

from flytekitplugins.spark import Spark

@task(task_config=Spark(spark_conf={"spark.executor.instances": "4"}), retries=2)
def compute_stats(df: pyspark.sql.DataFrame) -> dict:
    ...

Here TaskPlugins.find_pythontask_plugin(Spark) resolves to PysparkFunctionTask (registered at import time by the plugin package), and that class is instantiated instead of PythonFunctionTask.

If your function is a coroutine, @task detects inspect.iscoroutinefunction(fn) and substitutes AsyncPythonFunctionTask for PythonFunctionTask automatically.

Configuring Task Behavior with TaskMetadata

All per-execution knobs — retries, timeouts, caching, interruptibility — are packed into TaskMetadata, a dataclass defined in flytekit/core/base_task.py. You normally set these as keyword arguments to @task:

import datetime
from flytekit import task

@task(
    retries=3,
    timeout=datetime.timedelta(minutes=30),
    interruptible=True,
    deprecated="Use process_v2 instead",
)
def process(x: int) -> int:
    ...

TaskMetadata.__post_init__ normalizes the timeout: if you pass an int, it's converted to datetime.timedelta(seconds=n). A string or other type raises ValueError immediately.

interruptible has three-valued logic: None means "inherit from the workflow", True opts the task in regardless of workflow setting, and False opts it out even if the workflow is interruptible.

Caching

Caching in flytekit has two styles. The old-style explicit-version approach:

@task(cache=True, cache_version="v2", cache_serialize=True)
def expensive_query(date: str) -> dict:
    ...

TaskMetadata validates that if cache=True, then cache_version must be non-empty; if cache_serialize=True, then cache must also be True. These constraints are enforced in __post_init__, so violations raise ValueError at decoration time, not at runtime.

The new-style approach uses Cache objects from flytekit/core/cache.py:

from flytekit import task
from flytekit.core.cache import Cache

@task(cache=Cache(version="v2", serialize=True, ignored_inputs=("timestamp",)))
def expensive_query(date: str, timestamp: str) -> dict:
    ...

Cache accepts an optional version string. If version is omitted, Cache computes a SHA-256 hash by running a list of CachePolicy objects (implementations of the CachePolicy protocol, which requires a get_version(salt, params) method). When using a Cache object, the deprecated cache_version, cache_serialize, and cache_ignore_input_vars kwargs must not be set simultaneously — doing so raises ValueError.

The @task wrapper in task.py expands the Cache object into the underlying TaskMetadata fields:

# From flytekit/core/task.py
if isinstance(cache, Cache):
    cache_version = cache.get_version(VersionParameters(func=fn, ...))
    cache_serialize = cache.serialize
    cache_ignore_input_vars = cache.get_ignored_inputs()
    cache = True

During local execution, the Task.local_execute method checks LocalConfig.cache_enabled and uses LocalTaskCache to store and retrieve results keyed by task name, version, input literal map, and ignored variables. Set the environment variable FLYTE_LOCAL_CACHE_ENABLED=false to disable this during local testing.

Container and Resource Configuration

Container-level settings live in PythonAutoContainerTask:

from flytekit import task, Resources, Secret
from flytekit.image_spec import ImageSpec

gpu_image = ImageSpec(packages=["torch"], cuda="12.1")

@task(
    container_image=gpu_image,
    requests=Resources(cpu="2", mem="4Gi"),
    limits=Resources(cpu="4", mem="8Gi", gpu="1"),
    environment={"TORCH_LOGS": "+dynamo"},
    secret_requests=[Secret(group="my-secrets", key="api-key")],
    shared_memory=True,
)
def train(epochs: int) -> float:
    ...

The resources parameter provides a shorthand when you want to set both request and limit together:

@task(resources=Resources(cpu=("1", "4"), mem="8Gi"))
def analyze(data: list) -> dict:
    ...

Resources(cpu=("1", "4")) sets a request of 1 and a limit of 4. A single value sets both to the same amount. resources and the separate requests/limits parameters are mutually exclusive — using both raises ValueError in PythonAutoContainerTask.__init__.

container_image can be a string (a fully-qualified image reference) or an ImageSpec object that flytekit can build. When not set, the default image configured via FLYTE_INTERNAL_IMAGE is used.

secret_requests accepts a list of Secret objects. Internally, PythonAutoContainerTask wraps them into a SecurityContext and attaches it to the task.

For Kubernetes PodTemplate customization, either supply an inline template or reference a pre-existing cluster resource by name:

from flytekit import task
from flytekit.core.pod_template import PodTemplate

@task(pod_template_name="my-cluster-template")
def run(x: int) -> int:
    ...

Deck Configuration

Tasks can produce HTML visualization decks. By default decks are disabled (enable_deck=False). Enable them and choose which decks to emit:

from flytekit import task
from flytekit.deck import DeckField

@task(
    enable_deck=True,
    deck_fields=(DeckField.SOURCE_CODE, DeckField.INPUT, DeckField.OUTPUT),
)
def process(x: int) -> int:
    ...

DeckField values recognized by PythonTask and PythonFunctionTask are SOURCE_CODE, DEPENDENCIES, TIMELINE, INPUT, and OUTPUT. PythonFunctionTask._write_decks overrides the base implementation to additionally render the function's source code using SourceCodeRenderer and its Python dependencies using PythonDependencyRenderer.

The disable_deck parameter still works but has been deprecated since flytekit 1.10.0 and emits a FutureWarning. Setting both disable_deck and enable_deck at once raises ValueError.

How Execution Works

Calling a task — double(n=5) — goes through Task.__call__, which delegates to flyte_entity_call_handler. That function inspects the current FlyteContext to decide what to do:

• Compile time (inside @workflow): returns a Promise. PythonTask.compile() calls create_and_link_node, wiring the task into the workflow DAG.
• Local execution (called with plain Python values): goes through Task.local_execute → Task.sandbox_execute → PythonTask.dispatch_execute.

dispatch_execute is the core execution pipeline:

1. pre_execute — called before any type conversion. Plugins override this to set up external sessions (Spark, Ray).
2. Translate LiteralMap inputs to Python values via TypeEngine.literal_map_to_kwargs.
3. execute(**native_inputs) — the actual user function call.
4. post_execute — called with the raw return value; plugins may clean up or transform outputs.
5. _output_to_literal_map — converts Python return values back to a LiteralMap using async type conversion.
6. _write_decks — emits deck HTML if enabled.

When running remotely, the container receives a pyflyte-execute command that looks like:

pyflyte-execute --inputs s3://path/inputs.pb --output-prefix s3://outputs/ \
  --resolver flytekit.core.python_auto_container.default_task_resolver \
  -- task-module my_package.my_module task-name double

The resolver location and module/name args are baked in at serialization time by TaskResolverMixin. At container startup, default_task_resolver.load_task(["task-module", "my_package.my_module", "task-name", "double"]) imports the module and looks up the task by name, rehydrating the PythonFunctionTask object.

The IgnoreOutputs Escape Hatch

In distributed training scenarios, worker ranks other than rank 0 should not upload results. Raising IgnoreOutputs from inside execute() (or post_execute()) signals pyflyte-execute to skip writing outputs.pb:

from flytekit import task
from flytekit.core.base_task import IgnoreOutputs

@task
def worker(rank: int, data: list) -> list:
    result = run_training(rank, data)
    if rank != 0:
        raise IgnoreOutputs(f"Rank {rank} outputs ignored")
    return result

In flytekit/bin/entrypoint.py, the entrypoint catches FlyteUserRuntimeException wrapping IgnoreOutputs and returns without writing any output file. IgnoreOutputs is also re-exported from flytekit.extend.

Execution Behavior Modes

PythonFunctionTask supports three execution modes, controlled by the ExecutionBehavior enum (DEFAULT, DYNAMIC, EAGER). The @task decorator defaults to DEFAULT.

Default Mode

The execute method simply calls self._task_function(**kwargs). This is the path for the overwhelming majority of tasks.

Dynamic Mode (@dynamic)

@dynamic in flytekit/core/dynamic_workflow_task.py is implemented as:

dynamic = functools.partial(task, execution_mode=PythonFunctionTask.ExecutionBehavior.DYNAMIC)

The function body runs at execution time (not compile time), and its task-call expressions are compiled into a DynamicJobSpec — a workflow definition produced at runtime — rather than executed immediately. This allows using Python control flow to build task graphs dynamically:

from flytekit import task, dynamic
import typing

@task
def t1(a: int) -> str:
    return "fast-" + str(a + 2)

@dynamic
def ranged_int_to_str(a: int) -> typing.List[str]:
    s = []
    for i in range(a):
        s.append(t1(a=i))
    return s

Locally, dynamic_execute runs the function body directly and returns a LiteralMap. On the backend, compile_into_workflow is called instead, producing a DynamicJobSpec that Flyte propeller then executes as a sub-workflow.

Two caveats: first, node_dependency_hints — needed when a dynamic task calls a LaunchPlan that Flyte can't discover statically — is only valid on DYNAMIC mode tasks; passing it to a DEFAULT task raises ValueError. Second, the module-level docstring in flytekit/core/dynamic_workflow_task.py recommends keeping dynamic workflows under 50 nodes; large loops can produce thousands of nodes.

Eager Mode (@eager)

Eager workflows go further: the Python process itself acts as the execution engine, with each flyte entity call creating a real remote execution on the Flyte cluster. The function must be async:

from flytekit import task
from flytekit.core.task import eager

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

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

@eager
async def eager_workflow(x: int) -> int:
    out = await add_one(x=x)
    return await double(x=out)

@eager instantiates EagerAsyncPythonFunctionTask, always sets execution_mode=EAGER and metadata.is_eager=True. The execution_mode kwarg is silently deleted from kwargs if passed. Locally, the function runs with asyncio; you can test it with asyncio.run(eager_workflow(x=1)).

On the backend, execute() constructs a Controller (from flytekit.core.worker_queue) that dispatches sub-executions to Flyte Admin. The Controller also installs SIGINT/SIGTERM handlers — signal handlers must be installed on the main thread, so this happens inside execute() which runs on the pyflyte-execute main thread.

Eager workflows automatically get a failure cleanup node via get_as_workflow():

# From tests/flytekit/unit/core/test_eager_cleanup.py
imperative_wf = simple_eager_workflow.get_as_workflow()
assert imperative_wf.failure_node.flyte_entity.container_image == simple_eager_workflow.container_image

get_as_workflow() wraps the eager task in an ImperativeWorkflow with an EagerFailureHandlerTask as the on_failure handler. If the eager workflow fails, this handler queries Flyte Admin for all child executions tagged with the parent execution ID and terminates any that are still running, preventing resource leaks.

Dynamic and eager modes cannot be combined — AsyncPythonFunctionTask.async_execute raises NotImplementedError for DYNAMIC mode.

Extending the Task System

Plugin Tasks via TaskPlugins

Plugin packages create specialized task behavior by subclassing PythonFunctionTask and registering against a config type:

# Pattern from plugins/flytekit-ray/flytekitplugins/ray/task.py
class RayFunctionTask(PythonFunctionTask):
    _RAY_TASK_TYPE = "ray"

    def __init__(self, task_config: RayJobConfig, task_function, **kwargs):
        super().__init__(
            task_config=task_config,
            task_type=self._RAY_TASK_TYPE,
            task_function=task_function,
            **kwargs,
        )

    def pre_execute(self, user_params):
        ray.init(address=self._task_config.address)
        return user_params

TaskPlugins.register_pythontask_plugin(RayJobConfig, RayFunctionTask)

Once registered, @task(task_config=RayJobConfig(...)) automatically instantiates RayFunctionTask. The task_type string is what the Flyte backend uses to route the task to the right plugin executor.

Override pre_execute to set up external sessions before any type conversion happens; override post_execute to clean up or transform outputs. If you override execute in a PythonFunctionTask subclass, you must also handle DYNAMIC mode yourself (check self.execution_mode) or dynamic dispatch will break.

Instance Tasks

For tasks that don't wrap a Python function — shell tasks, connector-based tasks, DBT tasks — subclass PythonInstanceTask:

from flytekit.core.python_function_task import PythonInstanceTask

class MyPluginTask(PythonInstanceTask[MyConfig]):
    def __init__(self, name: str, task_config: MyConfig, **kwargs):
        super().__init__(name=name, task_config=task_config, task_type="my-plugin", **kwargs)

    def execute(self, **kwargs):
        # platform-defined execution logic
        ...

x = MyPluginTask(name="my_task", task_config=MyConfig(...))

PythonInstanceTask inherits from PythonAutoContainerTask with ABC, so execute must be overridden. The DefaultTaskResolver handles rehydration by capturing the module and variable name at construction time via TrackedInstance.

Custom Task Resolvers

If DefaultTaskResolver doesn't work for your use case — nested functions, dynamically generated tasks, tasks from external packages — implement TaskResolverMixin:

from flytekit.core.base_task import TaskResolverMixin, Task
from flytekit.configuration import SerializationSettings

class MyResolver(TaskResolverMixin):
    @property
    def location(self) -> str:
        # Must be an importable dotted path to the singleton instance
        return "mypackage.resolvers.my_resolver"

    def name(self) -> str:
        return "my_resolver"

    def loader_args(self, settings: SerializationSettings, t: Task) -> list:
        return ["task-id", t.name]

    def load_task(self, loader_args: list) -> Task:
        task_id = loader_args[1]
        return TASK_REGISTRY[task_id]

    def get_all_tasks(self) -> list:
        return list(TASK_REGISTRY.values())

my_resolver = MyResolver()

The location property must be the dotted import path of the singleton instance (e.g., my_resolver), not the class. The loader_args you return get appended after -- in the pyflyte-execute command line; load_task receives those same args at container startup.

Common Gotchas

Nested functions are rejected. PythonFunctionTask raises ValueError if the decorated function is a closure or inner function when using DefaultTaskResolver. The task must be importable at module level. Workaround: use functools.wraps on wrapper functions, or supply a custom TaskResolverMixin:

# This raises ValueError:
def make_task():
    @task
    def inner(x: int) -> int:   # inner is a nested function
        return x
    return inner

# Fix with functools.wraps:
import functools

def my_decorator(fn):
    @functools.wraps(fn)   # preserves module-level identity
    def wrapper(*args, **kwargs):
        return fn(*args, **kwargs)
    return wrapper

Caching invariants fail at decoration time, not runtime. cache=True with no cache_version raises ValueError immediately. So does cache_serialize=True without cache=True. Check these before deploying.

Don't mix Cache() object with old-style params. Passing cache=Cache(...) alongside cache_version=... raises ValueError. Choose one style.

node_dependency_hints is only valid on @dynamic tasks. Passing it to a DEFAULT-mode task raises ValueError at construction.

disable_deck is deprecated. Use enable_deck=True instead. Setting both raises ValueError. The deprecation warning is FutureWarning, emitted at task construction.

Unrecognized @task kwargs raise ValueError. A typo like @task(retrie=3) will raise at decoration time, listing the unrecognized key.