Task authoring and execution

Writing a Flyte task looks like a simple decorator, but the @task wrapper in flytekit/core/task.py does more than register metadata: it auto-detects your function’s interface, validates that the function is importable at module level, and returns a PythonFunctionTask instance that behaves differently depending on whether you are compiling a workflow, running locally, or executing in a container. Misconfigure caching or pass a nested function, and you’ll hit hard validation errors before the task ever reaches the Flyte platform.

Task Abstraction Hierarchy

Flyte tasks are built as a stack of increasingly specialized base classes defined in flytekit/core/base_task.py and flytekit/core/python_auto_container.py.

At the bottom is Task. It stores the raw FlyteIDL-aligned contract—task_type, name, a TypedInterface, TaskMetadata, security_ctx, and docs—and defines the abstract hooks dispatch_execute, pre_execute, and execute. It also implements local_execute, which wraps sandbox_execute and handles local caching via LocalTaskCache.

PythonTask extends Task (and TrackedInstance) and is the first layer that understands Python-native types. It keeps a python_interface: Interface, a task_config, and environment variables. This is where dispatch_execute is actually implemented: it translates Flyte literals to Python inputs, calls execute, then translates the outputs back to literals. PythonTask also implements compile() by calling create_and_link_node(), which is why a task invocation inside a workflow turns into a workflow node instead of an immediate function call.

PythonAutoContainerTask extends PythonTask and adds container-specific concerns: container_image, resource requests/limits, secret_requests, pod_template, accelerator, and the task_resolver that knows how to reload the task at runtime. It builds the default pyflyte-execute command via get_default_command.

Finally, PythonFunctionTask extends PythonAutoContainerTask. It defines ExecutionBehavior(Enum) with DEFAULT, DYNAMIC, and EAGER. On initialization it inspects the decorated function with transform_function_to_interface, derives the task name with extract_task_module, and validates that the function is not a nested local function.

Declaring Tasks with @task

The task decorator in flytekit/core/task.py (lines 174–455) is the entry point most users see. For a simple Python task, the usage is minimal:

@task
def my_task(x: int, y: typing.Dict[str, str]) -> str:
    ...

When you need plugin-specific configuration or retries, pass them to the decorator:

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

Behind the scenes, the decorator inspects the function. If the function is a coroutine (inspect.iscoroutinefunction), it instantiates AsyncPythonFunctionTask; otherwise it instantiates PythonFunctionTask. It then calls functools.update_wrapper(task_instance, decorated_fn), so the returned object masquerades as the original function.

Parameters such as retries, timeout, interruptible, deprecated, and cache are funneled into TaskMetadata. Parameters such as container_image, requests, limits, environment, secret_requests, pod_template, pod_template_name, and accelerator are passed to PythonAutoContainerTask.

The decorator also maintains backward compatibility for deprecated cache arguments (cache_serialize, cache_version, cache_ignore_input_vars). If you pass cache=True without a version, and you do not use the newer Cache object, the wrapper constructs a default Cache object for you. If you mix the old parameters with a Cache object, it raises an error:

if isinstance(cache, Cache):
    if cache_serialize is not None or cache_version is not None or cache_ignore_input_vars is not None:
        raise ValueError(
            "cache_serialize, cache_version, and cache_ignore_input_vars are deprecated. Please use Cache object"
        )

How Tasks Execute: From Compilation to Runtime

When you call a task object, its __call__ method delegates to flyte_entity_call_handler in flytekit/core/promise.py. That handler branches based on the current FlyteContext:

• Compilation mode (ctx.compilation_state is active) — calls create_and_link_node to produce a workflow node and returns Promise objects.
• Local workflow execution — unwraps promises and runs the task natively.
• Standalone local execution — calls local_execute directly.

The local path

Task.local_execute (in flytekit/core/base_task.py) translates your Python kwargs into Flyte literals via translate_inputs_to_literals, wraps them in a LiteralMap, and checks LocalTaskCache if metadata.cache is enabled and LocalConfig.auto().cache_enabled is true. On a cache miss it calls sandbox_execute, which in turn calls dispatch_execute.

The runtime path

PythonTask.dispatch_execute is the full runtime pipeline. It lives in flytekit/core/base_task.py and performs the following steps in order:

1. Pre-execution setup — calls pre_execute(user_params) to mutate execution parameters (for example, to set up a SparkSession).
2. Input translation — converts the incoming LiteralMap to Python kwargs with _literal_map_to_python_input.
3. User code — invokes self.execute(**native_inputs).
4. Post-execution — calls post_execute(new_user_params, native_outputs) to allow cleanup or output mutation.
5. Output translation — converts Python outputs back to a LiteralMap via _output_to_literal_map.
6. Deck generation — if enable_deck is true, _write_decks renders source code, dependencies, timeline, input, and output decks.

The _output_to_literal_map method handles an important edge case: if the task declares exactly one output, it detects whether the user returned a single value or a length-one NamedTuple, and normalizes the result before passing it to the type engine.

if len(expected_output_names) == 1:
    if self.python_interface.output_tuple_name and isinstance(native_outputs, tuple):
        native_outputs_as_map = {expected_output_names[0]: native_outputs[0]}
    else:
        native_outputs_as_map = {expected_output_names[0]: native_outputs}

Dynamic Workflows

A dynamic workflow is not a special decorator—it is literally a functools.partial of @task with a different execution mode. In flytekit/core/dynamic_workflow_task.py, line 21:

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

This means @dynamic behaves like @task at registration time, but at execution time PythonFunctionTask.execute routes to dynamic_execute instead of calling the function directly.

Inside dynamic_execute, flytekit creates a cached PythonFunctionWorkflow from the decorated function body. If the context is local, it runs the workflow and returns a LiteralMap. If the context is a real task execution, it calls compile_into_workflow, which serializes the generated workflow into a DynamicJobSpec containing the sub-task TaskTemplates and workflow nodes. The Flyte backend then executes that generated workflow as a subworkflow.

@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

Because Flyte cannot always statically infer every dependency inside a dynamic block, you can pass node_dependency_hints when you need to guarantee that upstream tasks, workflows, or launch plans are registered before the dynamic task runs. If you pass node_dependency_hints on a non-dynamic task, PythonFunctionTask.__init__ raises a ValueError:

if self._node_dependency_hints is not None and self._execution_mode != self.ExecutionBehavior.DYNAMIC:
    raise ValueError(
        "node_dependency_hints should only be used on dynamic tasks. On static tasks and "
        "workflows its redundant because flyte can find the node dependencies automatically"
    )

Async and Eager Execution

If the decorated function is an async def, the @task decorator instantiates AsyncPythonFunctionTask. Its __call__ awaits async_flyte_entity_call_handler, and its execute attribute is set via loop_manager.synced(async_execute) so that the normal dispatch_execute path still works:

async def async_execute(self, *args, **kwargs) -> Any:
    assert not args
    if self.execution_mode == self.ExecutionBehavior.DEFAULT:
        return await self._task_function(**kwargs)
    elif self.execution_mode == self.ExecutionBehavior.DYNAMIC:
        raise NotImplementedError

execute = loop_manager.synced(async_execute)

Notice that async tasks do not support ExecutionBehavior.DYNAMIC; mixing @dynamic with async def will raise NotImplementedError.

Eager workflows

EagerAsyncPythonFunctionTask (used by the @eager decorator in flytekit/core/task.py) forces ExecutionBehavior.EAGER and sets metadata.is_eager = True. In local execution, it simply runs the user function. In backend execution, it constructs a Controller backed by FlyteRemote and sets ExecutionState.Mode.EAGER_EXECUTION. Every nested Flyte entity call inside an eager workflow is then submitted as a remote execution and awaited:

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)

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

Eager workflows only support @task, @workflow, and @eager entities. Conditionals are not supported—use a plain Python if statement instead. When you point an eager workflow at a remote cluster that requires authentication, pass client_secret_group and client_secret_key so the internal FlyteRemote can authenticate.

Task Resolvers and Container Runtime

When a task runs on a hosted Flyte cluster, the container needs to know which Python object to execute. TaskResolverMixin in flytekit/core/base_task.py defines the contract: location, name, load_task(loader_args), and loader_args(settings, t).

The default implementation, DefaultTaskResolver in flytekit/core/python_auto_container.py, builds a command line that looks like this:

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

DefaultTaskResolver.loader_args returns ["task-module", m, "task-name", t], and load_task rehydrates the task by importing the module and calling getattr:

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

That is why PythonFunctionTask rejects nested or local functions: the default resolver cannot look them up by module and name. It allows an exception for test modules and functions wrapped with functools.wraps.

Testing Tasks Locally

For unit testing, flytekit.core.testing provides task_mock, a context manager that temporarily replaces PythonTask.execute with a MagicMock:

from flytekit.core.testing import task_mock

@task
def t1(i: int) -> int:
    pass

with task_mock(t1) as m:
    m.side_effect = lambda x: x
    t1(10)

There is also patch, a decorator that performs the same substitution for the duration of a test function:

from flytekit.core.testing import patch

@patch(t1)
def test_t1(mock, i):
    mock.side_effect = lambda x: x
    return t1(i)

Both utilities validate that the target is a PythonTask, WorkflowBase, or ReferenceEntity; otherwise they raise ValueError.

Validation Rules and Common Pitfalls

The codebase enforces several hard constraints that surface as immediate errors if you violate them.

Function definition constraints

PythonFunctionTask.__init__ rejects nested or inner functions because the default resolver cannot reload them:

if (
    not istestfunction(func=task_function)
    and isnested(func=task_function)
    and not is_functools_wrapped_module_level(task_function)
):
    raise ValueError(
        "TaskFunction cannot be a nested/inner or local function. "
        "It should be accessible at a module level for Flyte to execute it."
    )

Cache and metadata constraints

TaskMetadata.__post_init__ validates that caching parameters are consistent:

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(
        f"Cache ignore input vars are specified ... but ``cache`` is not enabled."
    )

Deck parameter conflicts

You cannot set both disable_deck and enable_deck at the same time. PythonTask.__init__ raises:

configured_deck_params = [disable_deck is not None, enable_deck is not None]
if sum(configured_deck_params) > 1:
    raise ValueError("only one of [disable_deck, enable_deck] can be set")

disable_deck is deprecated; prefer enable_deck.

Map task restrictions

ArrayNodeMapTask in flytekit/core/array_node_map_task.py only supports tasks with zero or one outputs, and only PythonFunctionTask with ExecutionBehavior.DEFAULT or PythonInstanceTask:

n_outputs = len(actual_task.python_interface.outputs)
if n_outputs > 1:
    raise ValueError("Only tasks with a single output are supported in map tasks.")

Dynamic workflow restrictions

• node_dependency_hints is only allowed for dynamic tasks.
• Reference tasks are unsupported inside dynamic tasks; compile_into_workflow raises ValueError("Reference tasks are currently unsupported within dynamic tasks").
• pickle_untyped=True is exposed as a convenience flag but is explicitly discouraged for production use.

These constraints are not cosmetic—each one guards a serialization or runtime assumption made by the layers above, from TaskMetadata up to the container resolver.