Task authoring and execution

Declare a task with @task

Use an annotated Python function when you want Flyte to infer the task interface from the function itself. The source-level example in task.py uses the public decorator this way:

from flytekit import task

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

You can also pass plugin configuration and task options through the decorator. For example, task.py shows a Spark configuration with three retries:

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

The decorator constructs a PythonFunctionTask for ordinary functions. Coroutine functions are instead represented by AsyncPythonFunctionTask. The function's annotations are converted by transform_function_to_interface; the resulting Python-native interface is exposed through PythonTask.python_interface, while Task.interface contains the transformed Flyte typed interface used for serialization and literal conversion. The function's module-level name and module are also captured for task rehydration, and its docstring is used to populate task documentation.

Configure execution metadata

TaskMetadata carries settings that are independent of the Python function body. Its constructor accepts retries, timeout, caching, interruptibility, deprecation text, pod-template name, deck generation, and eager status. Integer timeouts are converted to datetime.timedelta values. Caching is deliberately validated when metadata is created: cache=True requires a non-empty cache_version, and cache_serialize or cache_ignore_input_vars require caching to be enabled.

from flytekit import TaskMetadata

metadata = TaskMetadata(
    cache=True,
    cache_version="v1",
    retries=2,
    timeout=60,
    cache_serialize=True,
)

TaskMetadata.to_taskmetadata_model() maps these values to Flyte's task model, including the SDK runtime metadata and retry strategy. During local execution, Task.local_execute consults LocalTaskCache only when both the task metadata enables caching and LocalConfig enables the local cache. A cache hit returns the cached literal outputs; a miss executes the task and stores the resulting literal map.

Python task configuration also includes the options supplied by PythonAutoContainerTask, such as container_image, resources, secrets, pod templates, and environment. PythonTask stores the environment dictionary and plugin-specific task_config. Decks are disabled by default. Set enable_deck=True to turn them on and optionally select deck_fields; disable_deck is deprecated. Supplying both flags raises ValueError.

The task abstraction and a task call

Task is Flytekit's IDL-oriented base abstraction. It stores the task type, name, typed interface, metadata, task-type version, security context, and documentation, and registers each instance in FlyteEntities.entities. It defines the lifecycle methods that concrete tasks must provide: pre_execute, dispatch_execute, and execute.

Calling a task does not have one fixed meaning. Task.__call__ delegates to flyte_entity_call_handler, which examines the current FlyteContext:

my_task(x=1)
  -> flyte_entity_call_handler
     -> workflow compilation: create and link a node
     -> local execution: Task.local_execute and native outputs/promises
     -> eager execution: submit through the eager worker queue

For local execution, Task.local_execute translates native arguments and promises into a Flyte LiteralMap, invokes sandbox_execute, and wraps the returned literals in Promise objects. A task with no declared outputs returns a VoidPromise. PythonTask supplies the Python-specific conversion: _literal_map_to_python_input uses TypeEngine.literal_map_to_kwargs, dispatch_execute calls execute(**native_inputs), and _output_to_literal_map converts the native return value back through TypeEngine.

In a compiled workflow, PythonTask.compile calls create_and_link_node, so the same expression creates a node rather than immediately calling the Python function. This is why task code should use task calls and promises instead of assuming that a task invocation always returns an ordinary Python value.

Ordinary and custom Python tasks

PythonFunctionTask is the function-backed implementation normally produced by @task. In its default ExecutionBehavior, its execute method simply calls the captured function:

return self._task_function(**kwargs)

PythonInstanceTask is the alternative base for a task object with no user-defined function body. A subclass provides execute, while the instance is assigned at module level so the resolver can capture and reconstruct it:

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

x(a=5)

The actual interface and required configuration depend on the subclass. PythonTask is also the base for other task types that implement their own dispatch behavior rather than wrapping a Python function.

Dynamic tasks

Use @dynamic when the function body constructs a workflow at execution time. The decorator creates a PythonFunctionTask with ExecutionBehavior.DYNAMIC:

@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

PythonFunctionTask.dynamic_execute takes different paths based on execution state. In local execution it creates and executes the cached PythonFunctionWorkflow; in task execution it calls compile_into_workflow, which serializes the generated workflow and returns a DynamicJobSpec (or a literal map when there are no generated nodes).

Pass node_dependency_hints only to a dynamic task. The constructor raises ValueError if hints are supplied for a static task, because static workflow dependencies are discovered automatically. Dynamic tasks can also return the result of another task directly:

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

Async and eager execution

An async def function is represented by AsyncPythonFunctionTask; its asynchronous call handler awaits the function in default mode. Its implementation explicitly raises NotImplementedError for dynamic execution, so async function execution and dynamic workflow generation are separate modes.

For eager workflows, use @eager with an asynchronous function. The source includes this locally 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)

if __name__ == "__main__":
    import asyncio

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

EagerAsyncPythonFunctionTask forces ExecutionBehavior.EAGER and marks its TaskMetadata with is_eager=True, ignoring an execution_mode supplied through keyword arguments. Locally it changes the execution state to EAGER_LOCAL_EXECUTION and awaits the function. During a real execution, execute creates a Controller and worker queue; task calls inside the eager function become child executions submitted through that queue. run_with_backend is the main live-backend entry point, while run supports local testing against a FlyteRemote.

Remote eager execution requires a user-facing execution ID. The implementation derives child-execution tags from that ID and uses _F_EE_ROOT (EAGER_ROOT_ENV_NAME) when set, falling back to the current execution ID. It renders an Eager Executions deck and converts eager failures into FlyteNonRecoverableSystemException. get_as_workflow wraps an eager task in an ImperativeWorkflow and adds an EagerFailureHandlerTask cleanup handler. That handler is a remote-dispatch task: it finds active child executions tagged eager-exec for the parent and terminates them; its ordinary execute method intentionally raises an assertion.

To configure client-credential authentication for an eager remote, task.py shows passing a configured FlyteRemote and secret identifiers to @eager:

from flytekit.remote import FlyteRemote
from flytekit.configuration import Config

@eager(
    remote=FlyteRemote(config=Config.auto(config_file="config.yaml")),
    client_secret_group="my_client_secret_group",
    client_secret_key="my_client_secret_key",
)
async def eager_workflow(x: int) -> int:
    out = await add_one(x)
    return await double(out)

Serialize and rehydrate tasks

Hosted task execution must reconstruct the task inside the execution container. TaskResolverMixin defines that contract through location, name, loader_args(settings, task), load_task(loader_args), and get_all_tasks(). Auto-container serialization places the resolver and its loader arguments after the pyflyte-execute arguments. The base_task.py documentation illustrates the resulting shape:

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 default resolver imports the module and looks up the task name. Consequently, a default-resolver PythonFunctionTask must refer to a module-level function; nested, local, or inner functions are rejected, except in test functions. If a decorator obscures the function, preserve its metadata with functools.wraps or functools.update_wrapper, or implement a custom TaskResolverMixin. PythonInstanceTask is the corresponding module-level-instance pattern for custom task objects.

Constraints worth checking first

Situation	Flytekit behavior
cache=True without cache_version	TaskMetadata raises ValueError.
cache_serialize or ignored cache inputs without cache=True	TaskMetadata raises ValueError.
Nested function with the default resolver	PythonFunctionTask raises ValueError during construction.
node_dependency_hints on a static task	PythonFunctionTask raises ValueError; hints are dynamic-only.
Both disable_deck and enable_deck	PythonTask raises ValueError; disable_deck also emits a deprecation warning when used alone.
Dynamic async task	AsyncPythonFunctionTask.async_execute raises NotImplementedError.
Eager task with a supplied execution mode	EagerAsyncPythonFunctionTask replaces it with eager mode.
Map task wrapping a dynamic or eager function task	The map-task implementations reject it; they support default-mode Python tasks and a single output.

Use TaskMetadata for task-level execution properties, PythonFunctionTask behavior through the public decorators for function-backed tasks, and a resolver or PythonInstanceTask when the task cannot be represented by the default module-level function lookup.