Task authoring and execution

Declaring a task with @task

Start with a module-level, annotated Python function. The task decorator in flytekit/core/task.py turns it into a PythonFunctionTask (or a compatible plugin task):

from flytekit import task

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

PythonFunctionTask in flytekit/core/python_function_task.py calls transform_function_to_interface() to derive the Flyte interface from the function annotations and uses extract_task_module() to derive the task name and module information. The resulting object retains the Python-native interface as well as the Flyte typed interface. Constructing its Task base also appends the entity to FlyteEntities.entities, which is how Flyte's serialization machinery discovers it.

You can configure execution properties at the decorator boundary:

@task(
    container_image="repo/image:0.0.0",
    retries=3,
    environment={"MODE": "production"},
)
def process(value: str) -> str:
    return value.upper()

The decorator builds a TaskMetadata instance, selects a registered Python-task plugin from task_config, and passes the function, metadata, resources, resolver, deck settings, and other options to that plugin. An asynchronous function is selected as an AsyncPythonFunctionTask when the selected plugin is the standard PythonFunctionTask; a plugin used with async def must instead subclass AsyncPythonFunctionTask.

Core task abstractions

Task in flytekit/core/base_task.py is the lowest-level abstraction. It stores a task type, name, TypedInterface, metadata, optional security context, and documentation. Its abstract execution contract consists of:

• pre_execute(user_params), called before input conversion;
• execute(**kwargs), which performs the task operation; and
• dispatch_execute(ctx, input_literal_map), which connects Flyte literals to that operation.

The base class also exposes serialization extension points such as get_container(), get_k8s_pod(), get_sql(), get_custom(), get_config(), and get_extended_resources(). Each returns None by default, so task implementations override only the representation they provide.

PythonTask adds a Python Interface, an optional task_config, environment variables, and Deck configuration. PythonFunctionTask is the function-backed specialization; use PythonInstanceTask when the task has no user-defined function body:

class PythonInstanceTask(PythonAutoContainerTask[T], ABC):
    ...

x = MyInstanceTask(name="x", ...)
x(a=5)

That pattern, documented in flytekit/core/python_function_task.py, captures the module-level variable so the task can be loaded again. It is the base pattern for class-based tasks such as shell and plugin tasks, where the class's execute() method supplies the behavior.

Configuring task metadata

Every Task has a TaskMetadata object. The task decorator populates it from decorator arguments, while lower-level task constructors can receive one directly.

Field	Behavior in TaskMetadata
cache	Enables caching.
cache_serialize	Requests serialized execution for identical cached inputs.
cache_version	Version used for the cached value.
cache_ignore_input_vars	Input names excluded from the cache key.
interruptible	Marks whether execution may be interrupted.
deprecated	Warning text for a deprecated task; an empty string means active.
retries	Retry count used to construct the task's retry strategy.
timeout	Maximum execution duration; an integer is interpreted as seconds and converted to datetime.timedelta.
pod_template_name	Name of an existing cluster PodTemplate.
generates_deck	Indicates that the task generates a Deck URI.
is_eager	Marks the task as eager.

For example, the decorator's cache options can be used in the compatibility form:

@task(cache=True, cache_serialize=True, cache_version="1.0")
def foo(i: str):
    print(f"{i}")

The newer form passes a Cache object through cache; task() obtains its version, serialization setting, and ignored inputs before constructing TaskMetadata. Do not combine a Cache object with the deprecated cache_serialize, cache_version, or cache_ignore_input_vars arguments: task() raises ValueError.

TaskMetadata.__post_init__() validates the final values. In particular, cache=True without a cache version, cache_serialize=True without caching, or ignored inputs without caching all raise ValueError. An integer timeout is converted to a duration; a timeout of any other type that is not a datetime.timedelta also raises ValueError. to_taskmetadata_model() maps these values to Flyte's task metadata model, including retries, timeout, cache discovery version, interruptibility, deprecation text, Deck generation, pod template name, and eager state.

How execution moves through a task

A task call is routed through Task.__call__() and Flyte's entity call handler. During local execution, Task.local_execute() first converts native arguments or Promises into a LiteralMap. It then calls sandbox_execute(), which switches the execution parameters into task-sandbox mode before invoking dispatch_execute().

For a PythonTask, dispatch_execute() implements the common lifecycle:

LiteralMap
  -> PythonTask._literal_map_to_python_input()
  -> pre_execute()
  -> execute(**native_inputs)
  -> post_execute()
  -> PythonTask._output_to_literal_map()
  -> LiteralMap

_literal_map_to_python_input() uses TypeEngine.literal_map_to_kwargs() with the task's Python input types. After execute() returns, _output_to_literal_map() maps the return value to declared output names and converts outputs with TypeEngine.async_to_literal(); conversions run through asyncio.gather(). post_execute() is a no-op by default and is available for cleanup or output adjustment.

The base local path wraps the resulting literals back into Promise objects. A task with no declared outputs returns a VoidPromise; a task with outputs returns the corresponding Promise or named output structure. Local cache handling surrounds sandbox_execute(): when metadata caching is enabled and local caching is enabled, LocalTaskCache.get() is checked first, and a miss executes the task and stores the resulting literal map with LocalTaskCache.set().

Choosing execution behavior

PythonFunctionTask.ExecutionBehavior has three values: DEFAULT, DYNAMIC, and EAGER. Normal @task uses DEFAULT.

Standard execution

In default mode, PythonFunctionTask.execute() simply invokes the retained function:

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

The implementation branch is return self._task_function(**kwargs). The surrounding PythonTask.dispatch_execute() still performs conversion, hooks, and output serialization.

Dynamic tasks

Use @dynamic when the function constructs a workflow at execution time. A real example from the dynamic-task tests is:

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

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

The decorator creates a PythonFunctionTask with execution_mode=DYNAMIC. During local execution, dynamic_execute() creates or reuses a PythonFunctionWorkflow and executes it using the local dynamic execution state. During a real TASK_EXECUTION, compile_into_workflow() compiles the generated workflow and returns a DynamicJobSpec containing its nodes, task templates, outputs, and subworkflows. Reference tasks are explicitly rejected inside that dynamic compilation path.

node_dependency_hints is available for dynamic tasks when Flyte cannot infer dependencies before runtime. PythonFunctionTask rejects it for DEFAULT and other non-dynamic modes.

Eager execution

Use @eager with an async def when Python code should orchestrate Flyte entities as executions rather than compile into one workflow specification:

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)

The eager() decorator constructs an EagerAsyncPythonFunctionTask, always enables Decks, and sets TaskMetadata.is_eager=True. Its execute() creates a Controller and worker queue for a remote execution when needed; run_with_backend() runs the async function and renders the controller's execution information into an Eager Executions Deck. The implementation uses the current execution ID for the eager execution tags and honors _F_EE_ROOT for nested eager execution roots.

EagerAsyncPythonFunctionTask.get_as_workflow() provides a workflow representation with an EagerFailureHandlerTask as its failure handler. That handler uses Flyte Admin to find queued or running child executions tagged with the parent eager execution and terminates them.

Asynchronous Python functions

For an ordinary asynchronous task, write async def and let @task select AsyncPythonFunctionTask:

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

AsyncPythonFunctionTask.__call__() awaits async_flyte_entity_call_handler(). Its async_execute() awaits the retained function, while execute is a synchronized wrapper supplied by loop_manager. Dynamic execution is not supported for this class: selecting DYNAMIC causes async_execute() to raise NotImplementedError. Eager tasks use the separate EagerAsyncPythonFunctionTask path instead.

Extending task authoring

Explicit interfaces with kwtypes()

Function tasks derive interfaces from annotations. Class-based tasks can create ordered name-to-type mappings with kwtypes():

kwtypes(a=int, b=str)

The helper in flytekit/core/base_task.py returns an OrderedDict containing the keyword names and their Python types. Plugins use this when constructing an Interface for a PythonInstanceTask subclass.

Custom task resolution

Hosted execution must reconstruct a task object in the worker. TaskResolverMixin defines the resolver contract: location identifies the resolver, loader_args() emits task identifiers, load_task() rehydrates one task from those identifiers, and get_all_tasks() exposes resolver-managed tasks. PythonAutoContainerTask uses the default resolver unless you pass task_resolver= to @task or a task constructor.

The default serialized command has the form described in TaskResolverMixin's source documentation:

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 imports the task module and looks up the task name. Consequently, a standard function task must be accessible at module level. PythonFunctionTask rejects nested or local functions when using the default resolver, except for test functions, and allows wrapped module-level functions when functools.wraps or functools.update_wrapper preserves the module-level identity. For other loading schemes, implement TaskResolverMixin and pass the resolver explicitly.

Plugin task types

A plugin supplies a configuration type and a PythonFunctionTask subclass, then registers the pair. The MMCloud plugin is a concrete example:

@dataclass
class MMCloudConfig(object):
    submit_extra: str = ""

class MMCloudTask(PythonFunctionTask):
    _TASK_TYPE = "mmcloud_task"

    def __init__(self, task_config, task_function, container_image=None, requests=None, limits=None, **kwargs):
        super().__init__(
            task_config=task_config or MMCloudConfig(),
            task_type=self._TASK_TYPE,
            task_function=task_function,
            container_image=container_image,
            **kwargs,
        )
        self._mmcloud_resources = flyte_to_mmcloud_resources(requests=requests, limits=limits)

    def execute(self, **kwargs) -> Any:
        return PythonFunctionTask.execute(self, **kwargs)

TaskPlugins.register_pythontask_plugin(MMCloudConfig, MMCloudTask)

When task_config is an MMCloudConfig, the decorator's plugin lookup selects MMCloudTask. The plugin can override serialization hooks such as get_custom(); MMCloud serializes submit_extra and its translated resources into the task's custom data.

Runtime output control and common constraints

A task can raise IgnoreOutputs, the exception defined in flytekit/core/base_task.py, when generated outputs should be ignored. At the container entry point, a wrapped FlyteUserRuntimeException containing IgnoreOutputs causes execution to return without uploading outputs.pb. This is used by distributed-task implementations where a worker should not publish outputs.

Keep these implementation constraints in mind:

• Define standard task functions at module level when using the default resolver; nested and local functions are rejected at task construction time.
• Set a cache version whenever caching is enabled. cache_serialize and ignored cache inputs also require caching.
• Pass only one of disable_deck and enable_deck; disable_deck is deprecated, and PythonTask raises if both are supplied. Use enable_deck and, when enabled, select valid DeckField values.
• Use node_dependency_hints only with dynamic tasks.
• pickle_untyped=True permits pickling untyped outputs, but PythonFunctionTask documents that it is not recommended for production use.
• When configuring resources through the task decorator, the resources form is distinct from separate requests and limits configuration; the decorator and resource handling reject mutually exclusive combinations.