Task authoring and execution

Declare a task with `@task`

Start with an annotated module-level function when you want Flyte to infer the task interface and execute your Python code:

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

The task decorator in task.py is flytekit's primary authoring path. It creates a PythonFunctionTask (or an asynchronous variant for a coroutine function), retains the wrapped function metadata with update_wrapper, and passes task configuration and execution options into the task object. PythonFunctionTask calls transform_function_to_interface to derive input and output types from the annotations, creates documentation from the function docstring, and derives the task name from the function's module and name.

For a plugin-backed task, pass its configuration and task options to the decorator:

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

Annotations are part of the task contract. A missing or incompatible annotation normally fails while transform_function_to_interface builds the interface. pickle_untyped=True is available on PythonFunctionTask as a convenience for untyped outputs, but its constructor documentation explicitly says that it is not recommended for production. ignore_input_vars removes named inputs from the Flyte interface for injected or client-side values; it does not alter the original Python callable's signature.

The task abstraction layers

Flytekit separates the Flyte-facing task model from the Python execution model:

Task
├── PythonTask             native Interface and Python execution pipeline
│   └── PythonAutoContainerTask
│       └── PythonFunctionTask  wraps a user Callable
└── other Task/PythonTask extensions, such as SQL and container tasks

Task in base_task.py is the IDL-oriented base. Its constructor stores the task type, name, typed interface, metadata, task-type version, security context, and documentation, and registers the instance in FlyteEntities.entities. It exposes interface, metadata, name, task_type, and hosted-task payload hooks such as get_container, get_k8s_pod, get_sql, get_custom, get_config, and get_extended_resources; these hooks return None by default.

A task call is routed through the context-sensitive call handler:

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

PythonTask adds a native Interface, an optional plugin task_config, environment variables, and deck configuration. It transforms that interface into Flyte's typed interface, exposes Python input and output type maps, and implements compile by calling create_and_link_node. Concrete non-function extensions use this layer too: SQL and container tasks provide their own execution representation, while mapping tasks orchestrate an inner Python task rather than wrapping a user function.

Use PythonInstanceTask when the task is represented by an object or class and has no user-defined function body. Its class documentation shows the intended shape:

x = MyInstanceTask(name="x", .....)
x(a=5) # depending on the interface of the defined task

A subclass supplies execute; PythonInstanceTask preserves the task name, configuration, task type, and optional resolver while PythonAutoContainerTask handles container execution. Use PythonFunctionTask when the task should invoke a particular Python callable. The distinction matters to integrations such as ArrayNodeMapTask: it accepts a default-mode PythonFunctionTask or a PythonInstanceTask, but rejects dynamic and eager function tasks.

Configure execution metadata

Pass operational metadata through @task options or construct TaskMetadata directly. The metadata object carries caching, retries, timeout, interruptibility, deprecation, pod-template, deck, and eager settings. For example, a mapped task source example supplies retry metadata explicitly:

map_task(
    my_mappable_task,
    metadata=TaskMetadata(retries=1),
    concurrency=10,
    min_success_ratio=0.75,
)(a=x).with_overrides(requests=Resources(cpu="10M"))

TaskMetadata.__post_init__ validates related cache settings:

cache=True requires a non-empty cache_version.
cache_serialize=True requires cache=True.
cache_ignore_input_vars requires cache=True.
An integer timeout is converted to datetime.timedelta(seconds=timeout); another non-timedelta timeout value raises ValueError.

The defaults are cache=False, retries=0, interruptible=None, timeout=None, generates_deck=False, and is_eager=False. retry_strategy converts the retry count to Flyte's retry model. to_taskmetadata_model maps these values to the Flyte task model and records the flytekit SDK version and Python runtime type.

The decorator also carries configuration for container images, resources, environment variables, secrets, pod templates, accelerators, shared memory, task resolvers, documentation, and deck behavior. Decks are disabled by default in PythonTask, even though a default tuple of deck fields exists. Enable them explicitly:

@task(enable_deck=True)
def my_task(x: int) -> int:
    return x + 1

Do not set enable_deck and the deprecated disable_deck together: PythonTask raises ValueError. Supplying disable_deck emits a FutureWarning, and an invalid member of deck_fields also raises ValueError. For a Python function, PythonFunctionTask._write_decks adds source-code and dependency decks when those fields are enabled; the base PythonTask adds input and output rendering and, where configured, timeline content.

What happens when a task is called

The same task call has different effects depending on the active Flyte context:

workflow compilation  -> Node + Promises
local execution       -> native values at the caller boundary
hosted execution      -> serialized literals + task runtime

During compilation, PythonTask.compile creates and links a node. construct_node_metadata puts the task name, timeout, retry strategy, and interruptible setting into the node metadata. Positional arguments are converted according to interface order, but flytekit's examples and internal calls generally use keyword arguments. Unexpected keywords, too many positional arguments, duplicate values, and output-cardinality mismatches are rejected by the call path.

During local execution, Task.local_execute first translates native values and Promise values into a LiteralMap. It then calls sandbox_execute, which derives a task sandbox execution context before invoking dispatch_execute. If caching is enabled in both the task metadata and LocalConfig.auto().cache_enabled, it looks up the literal inputs in LocalTaskCache; cache_overwrite skips the read and rewrites the entry after execution. Finally, local_execute converts the output literals into Promise objects, or returns VoidPromise(self.name) for a task with no outputs.

PythonTask.dispatch_execute is the common runtime pipeline:

Call pre_execute so a subclass can modify ExecutionParameters before input conversion.
Configure deck execution when enabled.
Convert the input LiteralMap to Python values with TypeEngine.literal_map_to_kwargs.
Invoke execute(**native_inputs).
Call post_execute.
Convert native outputs back to literals with TypeEngine.async_to_literal.
Attach tracked output metadata and write decks.

PythonFunctionTask.execute supplies the function-specific step:

def execute(self, **kwargs) -> Any:
    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)

Input conversion and user-code exceptions preserve their original form during local execution. For hosted execution, input failures are wrapped as FlyteNonRecoverableSystemException and user-code failures as FlyteUserRuntimeException. Output conversion is concurrent. A one-output NamedTuple receives special handling, multiple outputs are matched by declared output order, and a tuple supplied as the value of an individual output raises TypeError with the output name and task name.

IgnoreOutputs is a marker exception for implementations whose outputs may safely be discarded, including the distributed-training or peer-to-peer case described by its docstring. The Python task dispatch path allows it to bubble to the caller layer rather than converting outputs.

Serialization and task rehydration

A hosted Python task must be reconstructed inside the execution container. PythonAutoContainerTask builds a pyflyte-execute command containing the resolver location and resolver arguments. TaskResolverMixin defines the contract used to make that reconstruction possible:

location identifies the resolver.
name() supplies its name.
loader_args(settings, task) produces task-identifying arguments.
load_task(loader_args) reconstructs one Task.
get_all_tasks() exposes a resolver's task collection.
task_name(task) may override the task name and defaults to None.

The default convention is a module and an attribute, conceptually represented by the source docstring as task-module repo_root.workflows.example task-name t1. At runtime the default resolver imports the module and retrieves the task attribute. Consequently, a PythonFunctionTask using the default resolver must be discoverable at module level. The constructor rejects nested, inner, or local functions, except for allowed test functions and module-level functions wrapped with functools.wraps or functools.update_wrapper. Supply a custom TaskResolverMixin when a different storage or loading scheme is required.

Dynamic tasks

Use the dynamic execution mode when the task body constructs task calls at runtime. The source examples use the @dynamic alias:

@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
def my_dynamic_subwf(a: int, b: int) -> int:
    x = t1(a=a)
    return t2(b=b, x=x)

Internally this selects PythonFunctionTask.ExecutionBehavior.DYNAMIC. dynamic_execute uses native input values. In a hosted task execution, compile_into_workflow creates or reuses a PythonFunctionWorkflow, compiles it with a dynamic compilation context, and returns a DynamicJobSpec containing generated nodes, task templates, outputs, and subworkflows. If no nodes are produced, it returns a LiteralMap instead. In local execution, flytekit runs the generated workflow in LOCAL_DYNAMIC_TASK_EXECUTION and converts its native results to a LiteralMap.

Set node_dependency_hints only for dynamic tasks:

@task(execution_mode=PythonFunctionTask.ExecutionBehavior.DYNAMIC)
def my_dynamic_task(x: int) -> int:
    ...

The PythonFunctionTask constructor raises ValueError when hints are supplied for a static task or workflow, because static dependencies are discovered automatically. Dynamic and eager modes are not interchangeable: the normal function task handles DEFAULT and DYNAMIC, while AsyncPythonFunctionTask explicitly raises NotImplementedError for dynamic execution. Array and legacy map wrappers likewise reject dynamic/eager function tasks and multi-output tasks.

Async and eager execution

For an async function, flytekit uses AsyncPythonFunctionTask. Its __call__ delegates to async_flyte_entity_call_handler, and its synchronized execute wrapper awaits the function in default mode. Eager tasks use EagerAsyncPythonFunctionTask, which forces ExecutionBehavior.EAGER, sets TaskMetadata.is_eager=True, and defaults decks to enabled.

The source example composes ordinary tasks inside an eager function and runs it locally with asyncio:

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, EagerAsyncPythonFunctionTask changes the execution mode to EAGER_LOCAL_EXECUTION and awaits the function. During a real execution it creates or uses a Controller worker queue, installs SIGINT and SIGTERM handlers, and runs the function in EAGER_EXECUTION. Nested entity calls are submitted to that queue and awaited, so each call can become a Flyte execution rather than only a Python in-memory call. The controller renders the eager call stack into the Eager Executions deck. The _F_EE_ROOT environment setting propagates the root eager execution tag; absent that setting, flytekit uses the current execution name.

Remote eager execution requires a current execution ID and user-space context. If no worker queue exists, the task constructs a remote using the current project and domain (or the documented defaults), then asserts that an execution ID is available. run(remote, ss, **kwargs) is the helper for local testing against a FlyteRemote. When client-credentials authentication is configured through the decorator, the source documentation requires a Flyte configuration file and client_secret_group and client_secret_key values.

get_as_workflow provides a workflow-shaped representation with an ImperativeWorkflow and an EagerFailureHandlerTask failure handler. That internal cleanup task uses a fixed EagerFailureTaskResolver; its dispatch_execute polls executions tagged for the eager run and terminates those in UNDEFINED, QUEUED, or RUNNING phases. Its execute method intentionally raises because dispatch, not native execution, is the supported path.

Extending task execution safely

When you add a task type, choose the narrowest existing abstraction:

Subclass Task when you need the IDL-facing task contract without a Python-native interface.
Subclass PythonTask when you have a native Interface, plugin configuration, and the standard literal conversion/runtime pipeline.
Use PythonFunctionTask for a user callable and PythonInstanceTask for an object-backed task whose subclass implements execute.
Override pre_execute to prepare execution parameters before type conversion, and post_execute to clean up or normalize the return value.
Implement task payload hooks such as get_container or get_custom when the hosted representation needs plugin-specific data.
Implement TaskResolverMixin when the task cannot be rehydrated by module-and-attribute lookup.

Keep the operational constraints visible while authoring: provide annotations, keep default-resolver functions module-discoverable, use a cache version with caching, do not combine the two deck flags, restrict dependency hints to dynamic tasks, and do not pass dynamic or eager tasks to map wrappers that require default-mode function tasks.

Declare a task with @task​

The task abstraction layers​

Configure execution metadata​

What happens when a task is called​

Serialization and task rehydration​

Dynamic tasks​

Async and eager execution​

Extending task execution safely​