Task authoring and execution

Declaring a task with @task

When a Python function is the unit of work you want Flyte to schedule, start with an annotated function and the public task decorator. The annotations become the task interface, so the smallest declaration is:

from flytekit import task

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

For a plugin-backed task, pass the plugin's configuration object and task options through the same decorator:

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

These examples are the declarations shown in task.py. The decorator is flytekit's normal construction path: it creates task metadata, derives the function interface, selects a Python task implementation, and forwards options such as container_image, environment, resources, secrets, pod settings, decks, and a resolver. Coroutine functions are routed to AsyncPythonFunctionTask; ordinary functions use PythonFunctionTask. The resulting task instance is also registered in FlyteEntities.entities by the Task constructor.

Metadata and configuration

Use task metadata to describe execution behavior rather than putting that information in the function body. TaskMetadata is a dataclass in base_task.py with fields for retries, timeout, interruptibility, caching, pod-template name, deprecation text, deck generation, and eager execution.

from flytekit.core.base_task import TaskMetadata

metadata = TaskMetadata(
    retries=2,
    timeout=300,
    interruptible=True,
)

An integer timeout is converted to datetime.timedelta(seconds=...). A truthy timeout of another type raises ValueError. Caching has explicit validation: cache=True requires a non-empty cache_version, while cache_serialize and cache_ignore_input_vars require caching to be enabled. The decorator documentation marks the older cache arguments as deprecated in favor of the newer Cache configuration and rejects mixing the old and new cache forms.

TaskMetadata.retry_strategy turns retries into Flyte's retry model. to_taskmetadata_model() serializes the metadata into Flyte's task model and records the installed flytekit version as the SDK runtime version.

Python tasks have decks disabled by default. Set enable_deck=True to enable them and use deck_fields to select the generated fields. Supplying both enable_deck and the deprecated disable_deck raises ValueError; an invalid value in deck_fields also raises ValueError. Resource-related decorator options are forwarded to PythonAutoContainerTask. Do not combine resources with requests or limits.

The task abstraction layers

Flyte's task classes separate the platform contract from Python invocation:

Task
  -> PythonTask
       -> PythonAutoContainerTask
            -> PythonFunctionTask
                 -> AsyncPythonFunctionTask
                 -> EagerAsyncPythonFunctionTask

Task is closest to the Flyte IDL. Its constructor stores the task type, name, typed interface, metadata, task-type version, security context, and documentation. It defines the platform-facing methods dispatch_execute, pre_execute, and execute as abstract methods, plus serialization extension points such as get_container, get_k8s_pod, get_sql, get_custom, get_config, and get_extended_resources.

PythonTask adds a Python-native Interface. It transforms that interface into a Flyte typed interface, retains the Python input and output types, and provides task_config and environment properties. Its compile() method calls create_and_link_node, which is how a Python task becomes a node while a workflow is compiled.

PythonInstanceTask is the extension point for a Python-container task whose implementation is a class's execute method rather than a user function. Its documented usage is an instance such as x = MyInstanceTask(name="x", ...), invoked as x(a=5) according to its interface. For function-backed tasks, use PythonFunctionTask—normally through @task.

What happens when you invoke a task

A task call does not immediately mean “call the Python function.” Task.__call__ delegates to flyte_entity_call_handler, which chooses the behavior appropriate to the current Flyte context: workflow compilation, local execution, or runtime execution.

For local execution, Task.local_execute() follows this path:

native values or Promises
    -> translate_inputs_to_literals(...)
    -> LiteralMap
    -> sandbox_execute()
    -> dispatch_execute()
    -> output LiteralMap
    -> Promise objects (or VoidPromise)

The conversion uses the task's Flyte interface and, for PythonTask, its Python input types. If caching is enabled and local caching is configured, local_execute() checks LocalTaskCache using the task name, cache version, input literal map, and ignored input variables. A cache miss runs sandbox_execute() and stores the resulting output literal map; a cache hit skips execution. sandbox_execute() adjusts the execution parameters for a task sandbox and calls dispatch_execute().

PythonTask.dispatch_execute() converts the input literal map back to native values with TypeEngine.literal_map_to_kwargs, then calls execute(**native_inputs). It converts the returned native outputs back to literals using TypeEngine.async_to_literal. During local execution, input and user-code exceptions are re-raised with the task name in the message; on hosted execution, the relevant failures may be wrapped in Flyte runtime exceptions. Outputs are returned as Promise objects when the task has outputs. A task with no declared outputs returns VoidPromise(self.name).

For PythonFunctionTask, the implementation of execute in default mode is simply the captured function:

if self.execution_mode == self.ExecutionBehavior.DEFAULT:
    return self._task_function(**kwargs)

The task constructor obtains the interface with transform_function_to_interface, removes inputs named in ignore_input_vars, and derives the task name and module from extract_task_module. ignore_input_vars therefore changes the task interface even though the underlying callable may still have that parameter; flytekit documents this for client-side injection and recommends execution parameters instead.

Serialization and task rehydration

A hosted Python task must be identifiable when pyflyte-execute starts in its container. PythonAutoContainerTask.get_default_command() constructs the command with input and output locations, the resolver location, and resolver arguments:

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),
]

TaskResolverMixin defines the resolver contract: location, name(), load_task(loader_args), loader_args(settings, task), and get_all_tasks(). The default module-based convention uses loader arguments that identify the task's module and name. Consequently, a function using the default resolver must be accessible at module level. PythonFunctionTask rejects nested or local functions, except in test modules whose names begin with test_. If a custom decorator wraps a task function, preserve its metadata with functools.wraps or functools.update_wrapper, or provide a custom resolver.

For class-managed tasks, ClassStorageTaskResolver in class_based_resolver.py keeps a list of task instances. add() appends a task; loader_args() returns its integer index as a string; and load_task() requires exactly one argument and uses it as that index. The resolver therefore depends on the same task registration sequence when the task is rehydrated.

Synchronous and asynchronous function tasks

A regular decorated function executes directly in PythonFunctionTask.ExecutionBehavior.DEFAULT:

from flytekit import task

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

For a coroutine function, the decorator selects AsyncPythonFunctionTask. Its asynchronous __call__ uses async_flyte_entity_call_handler, and async_execute() awaits the captured function in default mode. Async and dynamic execution are deliberately not combined: AsyncPythonFunctionTask.async_execute() raises NotImplementedError when its execution mode is dynamic.

Dynamic tasks

Use dynamic when the function must use native Python inputs to construct a workflow at execution time. In dynamic_workflow_task.py, dynamic is a partial application of task.task with execution_mode=PythonFunctionTask.ExecutionBehavior.DYNAMIC.

from flytekit import dynamic

@dynamic
def my_dynamic_subwf(a: int, b: int) -> int:
    x = t1(a=a)
    return t2(b=b, x=x)

A dynamic body can also use ordinary Python control flow over a native input:

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

At runtime, PythonFunctionTask.dynamic_execute() distinguishes execution states. In local execution it builds and executes a PythonFunctionWorkflow; in task execution it calls compile_into_workflow(), which serializes the generated workflow and returns a DynamicJobSpec; in local task execution it directly invokes the function. A generated workflow with no nodes can instead return a literal map of its strict outputs. Dynamic compilation rejects ReferenceTask entities and expects serialized task entities to be TaskSpec objects. Because a loop can generate very large workflows, the module documentation recommends keeping dynamic workflows under roughly fifty tasks.

node_dependency_hints is another dynamic-only option. Passing it to a static task or workflow raises ValueError; dynamic tasks may use it to identify task, launch-plan, or workflow dependencies that cannot be inferred before runtime.

Eager workflows

Eager tasks are async function tasks with a different execution model: EagerAsyncPythonFunctionTask forces ExecutionBehavior.EAGER and sets TaskMetadata.is_eager=True, ignoring an execution_mode supplied in kwargs. The repository contains this runnable local 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)

if __name__ == "__main__":
    import asyncio

    result = asyncio.run(eager_workflow(x=1))
    print(f"Result: {result}")  # "Result: 4"

Locally, eager execution changes the context to EAGER_LOCAL_EXECUTION and awaits the function. During a real execution, execute() constructs or reuses a Controller and worker queue; run_with_backend() changes the mode to EAGER_EXECUTION, awaits the function, renders the controller as an Eager Executions Deck, and converts an EagerException into a non-recoverable system failure. A user-facing execution context with an execution ID is required for remote eager execution. The project also exposes run(remote, ss, **kwargs) for local testing against a remote backend.

Eager execution and dynamic execution are different: dynamic tasks compile a runtime-generated workflow into a DynamicJobSpec, while eager functions run async Python and use a worker queue to submit task invocations. get_as_workflow() wraps an eager task in an ImperativeWorkflow and attaches an EagerFailureHandlerTask as its failure handler. That handler uses EagerFailureTaskResolver with fixed loader arguments ['eager', 'failure', 'handler']; remotely it polls unfinished executions tagged with execution_tag.key=eager-exec and terminates them.

Integration and testing patterns

Python function tasks are consumed by other task abstractions. map_task accepts a function task and can apply per-map metadata, concurrency, partial-success settings, and resource overrides:

from flytekit import task, workflow, map_task
from flytekit.core.base_task import TaskMetadata
from flytekit import Resources

@task
def my_mappable_task(a: int) -> str | None:
    return str(a)

@workflow
def my_wf(x: list[int]) -> list[str | None]:
    return map_task(
        my_mappable_task,
        metadata=TaskMetadata(retries=1),
        concurrency=10,
        min_success_ratio=0.75,
    )(a=x).with_overrides(requests=Resources(cpu="10M"))

SQLTask extends PythonTask with an explicit interface and SQL-specific execution. Array and legacy map implementations consume PythonFunctionTask or PythonInstanceTask; array mapping rejects dynamic and eager function tasks.

For unit tests, task_mock() temporarily replaces a Python-interface task's execute method with a MagicMock:

from flytekit import task
from flytekit.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)

The mock is valid only inside the context manager.

Troubleshooting task authoring

Symptom	flytekit behavior and correction
A decorated function is nested or local	The default resolver cannot load it. Move it to module scope, preserve wrapper metadata with functools.wraps/update_wrapper, or supply a TaskResolverMixin; test modules beginning with test_ are explicitly allowed.
TaskMetadata(cache=True) fails	Set a non-empty cache_version. Do not use cache_serialize or cache_ignore_input_vars unless cache=True.
Both deck switches are supplied	PythonTask raises ValueError; use enable_deck, because disable_deck is deprecated.
An async dynamic task raises NotImplementedError	AsyncPythonFunctionTask does not support combining async execution with dynamic mode.
Remote eager execution asserts about execution context	Eager remote setup requires user-facing parameters containing an execution ID; project and domain are derived from that ID.
A dynamic task produces too many nodes	The dynamic-workflow module documentation warns that loops can create thousands of nodes and recommends keeping generated dynamic workflows under about fifty tasks.
You cannot find matching task tests or an examples/ directory	The researched repository slice contains no matching test*python*function*.py, test*base*task*.py, or examples/ files. The runnable examples above come from embedded source documentation and helper documentation instead.