Task authoring and execution

Declaring a task with @task

Every Flyte task you author starts with the @task decorator, published from flytekit.core.task (and re-exported from the top-level flytekit package). In its simplest form it wraps an annotated function:

from flytekit import task

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

When you call add_one(x=3), flytekit does not immediately run the body. It first decides what kind of task object to build. The task() function inspects your function and constructs one of two classes from flytekit.core.python_function_task:

if inspect.iscoroutinefunction(fn):
    task_plugin = AsyncPythonFunctionTask
...
task_instance = task_plugin(
    task_config,
    decorated_fn,
    metadata=_metadata,
    container_image=container_image,
    environment=environment,
    requests=requests,
    limits=limits,
    secret_requests=secret_requests,
    execution_mode=execution_mode,
    node_dependency_hints=node_dependency_hints,
    task_resolver=task_resolver,
    disable_deck=disable_deck,
    enable_deck=enable_deck,
    deck_fields=deck_fields,
    docs=docs,
    pod_template=pod_template,
    pod_template_name=pod_template_name,
    accelerator=accelerator,
    pickle_untyped=pickle_untyped,
    shared_memory=shared_memory,
    resources=resources,
)
update_wrapper(task_instance, decorated_fn)
return task_instance

A plain def becomes a PythonFunctionTask; a coroutine (async def) becomes an AsyncPythonFunctionTask, whose execute awaits the underlying function instead of calling it directly. The decorator returns the task instance itself — so add_one is no longer a bare function but a task object with a python_interface, metadata, and the rest of the lifecycle described below.

@task accepts a large set of keyword arguments that map onto task behavior, including retries, interruptible, deprecated, timeout, container_image, environment, requests/limits, secret_requests, pod_template, accelerator, shared_memory, and resources. Their docs appear in the task() docstring in flytekit/core/task.py:

@task(task_config=Spark(), retries=3)
def my_task(x: int, y: typing.Dict[str, str]) -> str:
    ...

The task_config parameter is the extension hook: it selects which task plugin handles the function (see Task plugins).

How PythonFunctionTask infers the interface

The core container-based task class is PythonFunctionTask, defined in flytekit/core/python_function_task.py. It does not need you to declare inputs and outputs separately — it derives the task's Interface from the function's Python type hints. That happens in the constructor:

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

transform_function_to_interface (in flytekit/core/interface.py) reads the signature and type hints:

type_hints = get_type_hints(fn, include_extras=True)
signature = inspect.signature(fn)
return_annotation = type_hints.get("return", None)
...
outputs = extract_return_annotation(return_annotation)
inputs: Dict[str, Tuple[Type, Any]] = OrderedDict()
for k, v in signature.parameters.items():
    annotation = type_hints.get(k, None)
    if annotation is None and not pickle_untyped:
        raise FlyteMissingTypeException(fn=fn, param_name=k)
    default = v.default if v.default is not inspect.Parameter.empty else None
    inputs[k] = (annotation, default)

A few things to notice:

• Missing type annotations are errors unless you pass pickle_untyped=True. Without it, an unannotated input raises FlyteMissingTypeException; with it, both the input and output are treated as Any.
• NamedTuple outputs keep their field names. extract_return_annotation detects typing.NamedTuple return annotations and records the user-chosen tuple name so output names are preserved:

nt1 = typing.NamedTuple("NT1", x_str=str, y_int=int)

@task
def my_task(a: int, b: str) -> nt1:
    return nt1(str(b), a * 2)

• Interface mutation happens before the parent constructor. The inputs you list in ignore_input_vars are removed from the interface, while the function itself still receives them at execution time. This is useful for injecting client-side-only values that should never appear in the task signature.

Configuring execution: TaskMetadata and caching

The decorator collects its runtime knobs — retries, timeout, interruptibility, deprecated message, and cache settings — into a TaskMetadata dataclass (flytekit.core.base_task). Every Task holds one in its _metadata attribute; Task.__init__ defaults it to an empty TaskMetadata() if none is provided.

TaskMetadata carries much of the per-task behavior:

• retries (int): number of retries on failure
• timeout: a datetime.timedelta or an int (interpreted as seconds)
• interruptible: whether the node can be placed on lower-QoS, pre-emptible resources
• deprecated: a warning string marking the task deprecated
• cache, cache_serialize, cache_version, cache_ignore_input_vars: caching behavior
• pod_template_name, generates_deck, is_eager

Its __post_init__ enforces the invariants between these fields:

if self.timeout:
    if isinstance(self.timeout, int):
        self.timeout = datetime.timedelta(seconds=self.timeout)
    elif not isinstance(self.timeout, datetime.timedelta):
        raise ValueError("timeout should be duration represented as either a datetime.timedelta or int seconds")
if self.cache and not self.cache_version:
    raise ValueError("Caching is enabled ``cache=True`` but ``cache_version`` is not set.")
if self.cache_serialize and not self.cache:
    raise ValueError("Cache serialize is enabled ``cache_serialize=True`` but ``cache`` is not enabled.")
if self.cache_ignore_input_vars and not self.cache:
    raise ValueError(...)

So caching is only valid with a cache version, and cache_serialize/cache_ignore_input_vars are only valid alongside cache=True. An int timeout is normalized to a timedelta here; any other non-timedelta type raises ValueError.

The decorator assembles this metadata and forwards it into the task constructor:

_metadata = TaskMetadata(
    cache=cache,
    cache_serialize=cache_serialize,
    cache_version=cache_version,
    cache_ignore_input_vars=cache_ignore_input_vars,
    retries=retries,
    interruptible=interruptible,
    deprecated=deprecated,
    timeout=timeout,
)

Newer @task calls pass a Cache object instead of the three deprecated cache_* kwargs. When cache=True and no cache_version is given, the decorator builds a Cache and computes a version:

if isinstance(cache, bool) and cache is True and cache_version is None:
    cache = Cache(
        serialize=cache_serialize if cache_serialize is not None else False,
        ignored_inputs=cache_ignore_input_vars if cache_ignore_input_vars is not None else tuple(),
    )
...
if isinstance(cache, Cache):
    cache_version = cache.get_version(
        VersionParameters(
            func=fn,
            container_image=container_image,
            pod_template=pod_template,
            pod_template_name=pod_template_name,
        )
    )

Task plugins: the TaskPlugins registry

task_config is how flytekit dispatches to plugin-specific task implementations. The TaskPlugins class (also in flytekit/core/task.py) maps a config type to a PythonFunctionTask subclass:

class TaskPlugins(object):
    _PYTHONFUNCTION_TASK_PLUGINS: Dict[type, Type[PythonFunctionTask]] = {}

    @classmethod
    def register_pythontask_plugin(cls, plugin_config_type: type, plugin: Type[PythonFunctionTask]):
        ...
        cls._PYTHONFUNCTION_TASK_PLUGINS[plugin_config_type] = plugin

    @classmethod
    def find_pythontask_plugin(cls, plugin_config_type: type) -> Type[PythonFunctionTask]:
        if plugin_config_type in cls._PYTHONFUNCTION_TASK_PLUGINS:
            return cls._PYTHONFUNCTION_TASK_PLUGINS[plugin_config_type]
        return PythonFunctionTask

The task() decorator looks up the plugin by the runtime type of the config you passed:

task_plugin = TaskPlugins.find_pythontask_plugin(type(task_config))

If nothing is registered for that config type, you get the base PythonFunctionTask. Plugin packages such as Spark, Athena, Hive, PyTorch, TensorFlow, and Pod register their own subclasses so that e.g. @task(task_config=Spark(...)) returns a Spark-specific task. These plugins live in the plugins/**/task.py modules and subclass PythonFunctionTask or PythonInstanceTask — the latter being the base for tasks that have no user-defined function body but define a platform execute() method instead.

The execution lifecycle

A task can be invoked in three situations, and Task.__call__ routes all of them through one entry point:

def __call__(self, *args, **kwargs):
    return flyte_entity_call_handler(self, *args, **kwargs)

flyte_entity_call_handler (in flytekit/core/promise.py) validates your arguments against entity.python_interface.inputs, rejects tuples passed as inputs, and checks the current ExecutionState to decide what "calling" means:

1. Compilation mode — when a task is called from inside a @workflow body. The handler sees a compilation state and calls create_and_link_node, which builds a Node bound to this task and returns Promise objects for each output. This is how a workflow becomes a DAG without running any user code.

2. Within a local execution — when one task calls another while a workflow/task is already running locally, it goes through local_execute().

3. Starting a fresh local execution — when you call a task directly from your script with native Python values.

Local execution

Task.local_execute (in flytekit/core/base_task.py) converts your Python-level inputs into Flyte literals, then runs the task inside a sandboxed execution context:

literals = translate_inputs_to_literals(
    ctx, incoming_values=kwargs,
    flyte_interface_types=self.interface.inputs,
    native_types=self.get_input_types(),
)
input_literal_map = _literal_models.LiteralMap(literals=literals)
...
outputs_literal_map = self.sandbox_execute(ctx, input_literal_map)

local_execute also implements a local cache layer: when self.metadata.cache is true and LocalConfig.auto().cache_enabled is set, it looks up LocalConfig for the task name + cache version + input hash, and stores the result if there was a miss.

sandbox_execute builds a fresh ExecutionParameters with a task-sandbox, wraps it in a new FlyteContext, and hands control to the abstract dispatch_execute:

def sandbox_execute(self, ctx, input_literal_map):
    es = cast(ExecutionState, ctx.execution_state)
    b = cast(ExecutionParameters, es.user_space_params).with_task_sandbox()
    ctx = ctx.current_context().with_execution_state(es.with_params(user_space_params=b.build())).build()
    return self.dispatch_execute(ctx, input_literal_map)

dispatch_execute is the core of every task. Its contract, spelled out in the base Task, is to "translate Flyte's Type system based input values and invoke the actual call to the executor", and it is the method invoked both locally and at runtime.

PythonTask supplies the real dispatch_execute implementation. It calls pre_execute (a hook for plugins to set up user-space state such as Spark sessions), converts the input LiteralMap back to native Python values via TypeEngine.literal_map_to_kwargs, and then invokes your code:

native_inputs = self._literal_map_to_python_input(input_literal_map, exec_ctx)
...
native_outputs = self.execute(**native_inputs)

For PythonFunctionTask, execute() is a thin dispatch on execution mode:

def execute(self, **kwargs):
    if self.execution_mode == self.ExecutionBehavior.DEFAULT:
        return self._task_function(**kwargs)
    elif self.execution_mode == self.ExecutionBehavior.DYNAMIC:
        return self.dynamic_execute(self._task_function, **kwargs)

After your function returns, dispatch_execute runs post_execute, then converts the native Python outputs back into a LiteralMap via _output_to_literal_map. Finally, local_execute wraps those literals back into Promise objects so downstream workflow nodes can consume them. Tasks with no declared outputs short-circuit and return a VoidPromise.

Rehydrating tasks at runtime: resolvers and the executor command

On a hosted Flyte platform your function task actually runs in a container. The bridge between the declared task and that container is the task resolver. PythonAutoContainerTask (in flytekit/core/python_auto_container.py) holds a task_resolver and generates the container command that tells the container which task to run:

def get_default_command(self, settings: SerializationSettings) -> List[str]:
    container_args = [
        "pyflyte-execute",
        "--inputs", "{{.input}}",
        "--output-prefix", "{{.outputPrefix}}",
        "--raw-output-data-prefix", "{{.rawOutputDataPrefix}}",
        "--checkpoint-path", "{{.checkpointOutputPrefix}}",
        "--prev-checkpoint", "{{.prevCheckpointPrefix}}",
        "--resolver", self.task_resolver.location,
        "--",
        *self.task_resolver.loader_args(settings, self),
    ]
    return container_args

For a simple @task, this resolves to a command like the one documented in TaskResolverMixin:

pyflyte-execute --inputs s3://path/inputs.pb --output-prefix s3://outputs/location \
--raw-output-data-prefix /tmp/data \
--resolver flytekit.core.python_auto_container.default_task_resolver \
-- \
task-module repo_root.workflows.example task-name t1

The resolver is the piece that re-creates the task at execution time. TaskResolverMixin (in Keepflytasks/base_task.py) is the abstract contract; the out-of-box implementation is DefaultTaskResolver:

class DefaultTaskResolver(TrackedInstance, TaskResolverMixin):
    def load_task(self, loader_args: List[str]) -> PythonAutoContainerTask:
        _, task_module, _, task_name, *_ = loader_args
        task_module = importlib.import_module(name=task_module)
        task_def = getattr(task_module, task_name)
        return task_def

    def loader_args(self, settings, task):
        _, m, t, _ = extract_task_module(task)
        return ["task-module", m, "task-name", t]

loader_args serializes the module and the name the task is bound to in that module; load_task reverses it with importlib.import_module + getattr on the other side. Because the container has to reconstruct the task this way, the task function must be importable at module level — which is why nested/local functions are rejected (see Gotchas). The module is imported by the resolver, then dispatch_execute runs the task just as in local execution. Custom resolvers are supported by implementing TaskResolverMixin and passing the instance via task_resolver=.

Dynamic tasks (@dynamic)

@dynamic is not a separate decorator implementation. It is a partially-applied task() that pins execution_mode to DYNAMIC. In flytekit/core/dynamic_workflow_task.py:

import functools
from flytekit.core import task
from flytekit.core.python_function_task import PythonFunctionTask

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

A dynamic task is therefore still a PythonFunctionTask whose execute() takes the DYNAMIC branch shown earlier and calls dynamic_execute. Inside a dynamic function you can use ordinary Python control flow — for loops, range, conditionals — to call other Flyte entities:

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

dynamic_execute compiles the function body into a workflow at execution time. In production (a TASK_EXECUTION state), dynamic_execute calls compile_into_workflow, which serializes the function's nodes and returns a DynamicJobSpec for downstream orchestration instead of running them inline. When run locally it instead builds a PythonFunctionWorkflow from the function and executes that workflow directly. Dynamic tasks support optional node_dependency_hints — a list of tasks, launch plans, or workflows the body depends on — but the constructor rejects them unless the mode is DYNAMIC:

if self._node_dependency_empty is not None and self._execution_mode != self.ExecutionBehavior.DYNAMIC:
    raise ValueError(
        "node_dependency_hints should only be used on dynamic tasks."
    )

Eager tasks (@eager)

@eager is a distinct decorator (also in flytekit/core/task.py) that produces an EagerAsyncPythonFunctionTask. Eager execution inverts the model: instead of compiling a graph, every call to a Flyte entity inside the eager function kicks off a real execution, and Python waits for each result. The decorator's own docstring gives a complete runnable example:

from flytekit import task, 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 = add_one(x=x)
    return double(x=out)

# run locally with asyncio
if __name__ == "__main__":
    import asyncio
    result = asyncio.run(eager_workflow(x=1))
    print(f"Result: {result}")  # "Result: 4"

Note the eager function is declared async, and the underlying task invocations inside it are regular (synchronous) calls — EagerAsyncPythonFunctionTask.async_ex awaits them. The constructor forces execution_mode=EAGER and tags the metadata:

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

For real (non-local) runs, eager execution installs a Controller (the worker queue) and installs SIGINT/SIGTERM handlers before running; calling the entity inside an eager workflow routes through flyte_entity_call_handler's eager branch, which submits the entity to ctx.worker_queue rather than running it inline.

Gotchas and common mistakes

• Nested functions are rejected. PythonFunctionTask.__init__ raises a ValueError if the function is nested or local under the default resolver, unless it is a test function (module name begins with test_) or a functools-wrapped module-level function. The error suggests functools.wraps/update_wrapper for custom decorators, or a custom TaskResolverMixin.
• Caching requires a version. TaskMetadata.resolved_post_init raises at decoration time if you set cache=True without a cache_version. Pass a Cache object and the decorator computes one; the old cache_version/cache_serialize/cache_ignore_input_vars kwargs conflict with Cache and raise a ValueError if combined.
• ignore_input_vars hides inputs, not arguments. The inputs are removed from the task's interface and workflow, but the function still receives those kwargs at execution time — a client-only injection point.
• Only some tasks can be mapped. ArrayNodeMapTask and the legacy MapPythonTask accept only PythonFunctionTask in DEFAULT mode or PythonInstanceTask, and require at most one output.
• Timeout types are strict. Pass a timedelta or int seconds; any other type raises ValueError during TaskMetadata initialization.
• Nesting tasks is unsupported. Calling a task from within another running task logs a "You are not supposed to nest @task/@workflow" warning.
• enable_deck vs disable_deck. disable_deck is deprecated in favor of enable_deck; setting both raises ValueError.
• @async functions need async-able plugins. task() switches to AsyncPythonFunctionTask for coroutines, and raises an AssertionError if the registered plugin is not a subclass of AsyncPythonFunctionTask.