Task authoring and execution

Declaring a task

When you have a typed Python function that should be independently executable by Flyte, use @task. The annotations are not merely documentation: task() passes the function to PythonFunctionTask, which derives the task's Python interface from them.

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

For a task type that needs plugin-specific configuration, pass task_config; task-level retry configuration can be supplied at the same declaration site:

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

These are the examples in the task() docstring (flytekit/core/task.py). The decorator returns a task object, so a call such as my_task(x=...) goes through Task.__call__, which delegates to Flyte's entity call handler. In a workflow context that call is compiled into a node; in a local execution context it follows the local task execution path.

The task abstraction layers

Flytekit separates the serialized task description from Python-native task behavior:

• Task in flytekit/core/base_task.py is the IDL-oriented base. It stores the task type, name, typed interface, TaskMetadata, task type version, security context, and documentation. Constructing a Task also appends it to FlyteEntities.entities, so task construction has registration side effects.
• PythonTask extends Task with a Python-native Interface. It exposes Python input and output types, stores plugin task_config and environment variables, constructs node metadata, and implements compile() by calling create_and_link_node().
• PythonFunctionTask is the callable-backed implementation used for ordinary annotated functions. PythonInstanceTask is the alternative for a class/platform implementation whose subclass supplies execute() rather than a user function.

For programmatic interfaces, kwtypes() returns an insertion-ordered mapping of names to Python types. The base-task documentation uses this form:

kwtypes(a=int, b=str)

Task itself is intentionally not a user-function interface. Its python_interface is None, and subclasses must implement dispatch_execute(), pre_execute(), and execute(). PythonTask supplies the Python-native conversion and compilation behavior needed by most Python-backed tasks.

Configuring execution metadata

Pass retry, timeout, interruptibility, deprecation, and caching settings to @task; the decorator constructs a TaskMetadata instance before constructing the task. Its defaults are no cache, zero retries, no timeout, and no interruptibility.

@task(retries=3, timeout=60, interruptible=True)
def work(x: int) -> int:
    return x + 1

TaskMetadata.__post_init__ interprets an integer timeout as seconds by replacing it with datetime.timedelta(seconds=timeout). A non-integer value must be a datetime.timedelta; otherwise construction raises ValueError. Caching has explicit consistency checks:

• cache=True requires a non-empty cache_version when TaskMetadata is constructed directly.
• cache_serialize=True requires cache=True.
• cache_ignore_input_vars requires cache=True.

The decorator also supports a Cache object. When cache=True is used without an explicit version, task() creates a Cache; when a Cache object is supplied, it derives the version from the function and settings such as the container image and pod template, then rejects the deprecated cache keyword arguments. Unknown decorator keywords are rejected immediately by task() with ValueError.

The metadata's retry_strategy property creates the model retry object. to_taskmetadata_model() converts the Python metadata to Flyte's task metadata model and includes the Flyte SDK runtime type, SDK version, and python runtime language. The same model carries timeout, retries, interruptibility, cache discovery/version, deprecation text, deck generation, pod-template name, and eager state.

The decorator forwards container and runtime configuration to the task plugin, including container_image, environment, requests, limits, secret_requests, resources, accelerator, shared_memory, pod_template, pod_template_name, and task_resolver. PythonAutoContainerTask, the immediate parent of function and instance tasks, resolves these settings for container execution. Its serialized container command has this shape:

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

That code is from flytekit/core/python_auto_container.py. If an ImageSpec has runtime packages, the auto-container implementation exports them through _F_RUNTIME_PACKAGES. A pod template changes the serialized representation: get_container() returns None and get_k8s_pod() supplies the pod definition.

Deck controls

PythonTask disables decks by default. Set enable_deck=True to enable them and optionally choose fields with deck_fields. disable_deck is the deprecated spelling; setting both disable_deck and enable_deck raises ValueError, and disable_deck emits a FutureWarning. Every value in deck_fields must be a DeckField member.

When enabled, PythonTask.dispatch_execute() writes input and output decks. PythonFunctionTask additionally writes source-code and dependency decks when those fields are selected and source/dependency rendering succeeds. Timeline decks are added to the execution parameters when the timeline field is selected.

What happens when a task runs

A task call and execution move through these concrete stages:

Task.__call__
  -> flyte_entity_call_handler
  -> workflow node (compile) or Task.local_execute
  -> sandbox_execute
  -> PythonTask.dispatch_execute
  -> pre_execute -> TypeEngine input conversion -> execute
  -> post_execute -> TypeEngine output conversion
  -> LiteralMap -> Promise / VoidPromise

PythonTask.compile() calls create_and_link_node(), and construct_node_metadata() copies the task name, timeout, retry strategy, and interruptibility into the node metadata.

For local execution, Task.local_execute() first calls translate_inputs_to_literals() on native values, Promises, lists, or dictionaries of values. If metadata caching is enabled and LocalConfig enables the cache, it looks up LocalTaskCache using the task name, cache version, input literal map, and ignored inputs. A cache miss calls sandbox_execute() and stores the resulting literal map; a hit skips task execution. sandbox_execute() creates task-sandbox execution parameters before calling dispatch_execute().

PythonTask.dispatch_execute() then:

1. calls pre_execute();
2. converts the input LiteralMap to native values with TypeEngine.literal_map_to_kwargs();
3. invokes execute(**native_inputs);
4. calls post_execute();
5. converts returned values back through TypeEngine.async_to_literal(); and
6. returns a Flyte LiteralMap, after writing enabled decks.

For a PythonFunctionTask in default mode, execute() simply calls the captured function. Task.local_execute() turns the returned literals into Promise objects named from the declared output interface, or returns VoidPromise(self.name) when there are no outputs. It also asserts that the number of returned literals equals the number of declared outputs.

Exception behavior differs by execution context. During local execution, input conversion and user exceptions are re-raised with task context in their messages, and output conversion errors remain local exceptions. During hosted execution, user-function failures are wrapped in FlyteUserRuntimeException, while conversion failures are wrapped in FlyteNonRecoverableSystemException.

Output shape matters. PythonTask._output_to_literal_map() maps zero outputs to {}, multiple outputs by position, and a single output to its declared name. It has special handling for a one-element NamedTuple; the per-output conversion path rejects an output value that is itself a tuple. Declare and return outputs in the shape represented by the task interface.

Function tasks, instance tasks, and rehydration

Use PythonFunctionTask when the task body is a Python callable. Its constructor requires task_function, derives the interface with transform_function_to_interface(), removes any names in ignore_input_vars, and derives the task name from the function's module and name. The default resolver requires the function to be accessible at module level. Nested or local functions raise ValueError, except in test modules; a custom decorator should preserve metadata with functools.wraps or functools.update_wrapper, or you should provide a custom TaskResolverMixin.

Use PythonInstanceTask when the behavior is supplied by a task class or platform implementation rather than a user-defined function. Its constructor accepts name and task_config, defaults task_type to "python-task", and forwards the optional resolver and other settings to PythonAutoContainerTask. Subclasses provide the interface and override execute().

TaskResolverMixin defines the rehydration contract used by containerized tasks: an implementation supplies location, name(), load_task(loader_args), loader_args(settings, task), and get_all_tasks(). The resolver location and loader arguments are appended to the pyflyte-execute command so the worker can import and reconstruct the task.

For class-backed tasks, ClassStorageTaskResolver (flytekit/core/class_based_resolver.py) keeps instances in mapping. Call add() before serialization; loader_args() returns the task's list index as one string, and load_task() requires exactly one argument and converts it to an integer index. It raises ValueError if the task is absent and RuntimeError for ambiguous loader arguments.

Execution modes

Ordinary synchronous tasks

The ordinary @task path uses PythonFunctionTask.ExecutionBehavior.DEFAULT, so execute() invokes the function directly. Async functions are detected by task() and are automatically assigned AsyncPythonFunctionTask; its async call handler awaits the user function. AsyncPythonFunctionTask explicitly raises NotImplementedError for dynamic execution.

Dynamic tasks

dynamic is implemented as a partial application of task.task with execution_mode=PythonFunctionTask.ExecutionBehavior.DYNAMIC. Dynamic code receives native input values and can use Python control flow to create task invocations:

@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

The same source module documents dependencies between generated tasks:

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

Locally, PythonFunctionTask.dynamic_execute() creates or reuses a PythonFunctionWorkflow and executes it with native inputs. In task execution mode, it compiles that generated workflow into a DynamicJobSpec; a no-node dynamic workflow can instead return a literal map of its strict outputs. Dynamic compilation gathers the generated task templates and currently rejects ReferenceTask entities inside the dynamic task. The dynamic-task module recommends keeping generated workflows under fifty tasks.

Supply node_dependency_hints only for dynamic tasks. PythonFunctionTask raises ValueError if hints are supplied for a static task, because static dependencies are discovered automatically.

Map wrappers are narrower than dynamic execution: ArrayNodeMapTask accepts only a default-mode PythonFunctionTask or a PythonInstanceTask, rejects @dynamic and @eager function tasks, and rejects wrapped tasks with more than one output.

Eager tasks

EagerAsyncPythonFunctionTask is the implementation selected by @eager. It removes any supplied execution_mode, sets TaskMetadata.is_eager=True, and enables decks by default. The following runnable example is embedded in flytekit/core/task.py:

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)

# run locally with asyncio
if __name__ == "__main__":
    import asyncio

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

For local execution, the eager task changes the execution mode to EAGER_LOCAL_EXECUTION and awaits the function. For a real execution, execute() creates a Controller and worker queue when one is not already present, installs SIGINT and SIGTERM handlers, and runs the function in eager execution mode. run_with_backend() renders the controller queue into an Eager Executions deck and converts an EagerException into FlyteNonRecoverableSystemException after rendering.

Remote eager execution requires a user-facing execution ID outside local execution. The controller uses that ID for tags and uses _F_EE_ROOT when present to propagate the root eager execution name. run(remote, ss, **kwargs) is the helper for local testing while pointing an eager parent at a remote. get_as_workflow() creates an ImperativeWorkflow, wires the eager task's inputs and outputs, and adds an EagerFailureHandlerTask as its failure handler. Eager workflows support ordinary Python if statements; the researched implementation does not support Flyte conditionals, and eager and dynamic modes are not combined.

Testing task calls

Use task_mock() from flytekit/core/testing.py when you want a task call to exercise the normal call path without running its original body:

@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 within this context

task_mock() temporarily replaces the task's execute method with a wrapper that calls the returned MagicMock, then restores the original method when the context exits. It accepts Python-native PythonTask objects (as well as workflow and reference entities handled by the same utility) and raises ValueError for unsupported objects.

Practical constraints

• Keep default-resolver function tasks at module scope. Preserve wrappers with functools.wraps/update_wrapper, or implement TaskResolverMixin for another loading strategy.
• Make caching metadata consistent: caching needs a version when TaskMetadata is built directly, and serialization/ignored-input options require caching.
• Treat integer timeouts as seconds; other timeout values must be datetime.timedelta instances.
• Do not combine node_dependency_hints with static tasks, or use eager/dynamic function tasks as array/map task inputs.
• Expect local and hosted failures to have different exception wrappers, as described above.
• Match declared output count and shape. Single-output NamedTuple handling is special, and tuple values are rejected by the individual output conversion path.
• Remember that constructing a task registers it globally in FlyteEntities.entities; import and initialization order can therefore affect registration and serialization.