Task authoring and execution

When you write @task\ndef my_func(a: int) -> str: ... in flytekit, you are not just decorating a function — you are instantiating a PythonFunctionTask object that captures the function's type-annotated interface, serializes to a FlyteIDL TaskTemplate, and can be rehydrated at runtime inside a container via pyflyte-execute. This section walks through how that declaration, configuration, and execution machinery is actually wired together in flytekit, from the @task decorator down to the dispatch_execute pipeline.

The Class Hierarchy

Flytekit's task types form a layered inheritance chain, where each layer adds a specific capability:

Task                              (base_task.py)        — FlyteIDL-level: task_type, TypedInterface, TaskMetadata
 └─ PythonTask                    (base_task.py)        — Python-native Interface, deck config, compile()
     └─ PythonAutoContainerTask   (python_auto_container.py) — container image, resources, pyflyte-execute command, task_resolver
         ├─ PythonFunctionTask    (python_function_task.py)  — wraps a user Python function; the @task default
         │   └─ AsyncPythonFunctionTask                        — async function tasks and base for eager
         │       └─ EagerAsyncPythonFunctionTask               — eager workflows
         └─ PythonInstanceTask    (python_function_task.py)  — instance-based tasks with no function body

• Task (flytekit/core/base_task.py) is the root. Its constructor takes task_type, name, interface (a TypedInterface), and an optional TaskMetadata, then appends itself to FlyteEntities.entities so the serialization pass can discover it. python_interface returns None here because the base class has no Python-native type information.
• PythonTask (flytekit/core/base_task.py) adds a Python-native Interface (inputs/outputs as actual Python types), deck configuration, and a compile() method that calls create_and_link_node to produce a workflow node.
• PythonAutoContainerTask (flytekit/core/python_auto_container.py) adds the container concerns: container_image, Resources, PodTemplate, an accelerator, a task_resolver, and the get_default_command method that emits the pyflyte-execute command line.
• PythonFunctionTask (flytekit/core/python_function_task.py) is the concrete class that wraps a user-supplied callable. Its execute() method invokes that function. This is what @task produces by default.

PythonInstanceTask is a sibling branch for tasks that have a platform-defined execute method rather than a user function body — you construct it directly (x = MyInstanceTask(name="x", ...) and call x(a=5)), and it captures its module location so the loader can rehydrate it.

Declaring a Task with @task

The @task decorator (flytekit/core/task.py) is the primary entry point. A minimal task needs only a type-annotated function:

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

Its full signature accepts the parameters most users need, including cache, retries, timeout, container_image, requests/limits, secret_requests, enable_deck, deck_fields, pod_template, accelerator, resources, and more. The task_config parameter is how plugins receive their configuration object, and execution_mode defaults to PythonFunctionTask.ExecutionBehavior.DEFAULT.

Inside the wrapper closure, the decorator does three things:

1. Builds TaskMetadata from the cache/retries/timeout/etc. parameters.
2. Finds the plugin class via TaskPlugins.find_pythontask_plugin(type(task_config)). With no task_config this returns PythonFunctionTask itself. If the function is a coroutine (inspect.iscoroutinefunction(fn)), the decorator swaps in AsyncPythonFunctionTask (or asserts the chosen plugin already subclasses it).
3. Instantiates the task, passing task_config, the decorated function, the metadata, and all container/resource parameters through to the plugin constructor.

The relevant code from flytekit/core/task.py:

task_plugin = TaskPlugins.find_pythontask_plugin(type(task_config))
if inspect.iscoroutinefunction(fn):
    if task_plugin is PythonFunctionTask:
        task_plugin = AsyncPythonFunctionTask
    else:
        if not issubclass(task_plugin, AsyncPythonFunctionTask):
            raise AssertionError(f"Task plugin {task_plugin} is not compatible with async functions")

task_instance = task_plugin(
    task_config,
    decorated_fn,
    metadata=_metadata,
    container_image=container_image,
    environment=environment,
    requests=requests,
    limits=limits,
    secret_requests=secret_requests,
    execution_mode=execution_mode,
    node_dependency_hints=node_dependency_hints,
    task_resolver=task_resolver,
    disable_deck=disable_deck,
    enable_deck=enable_deck,
    deck_fields=deck_fields,
    docs=docs,
    pod_template=pod_template,
    pod_template_name=pod_template_name,
    accelerator=accelerator,
    pickle_untyped=pickle_untyped,
    shared_memory=shared_memory,
    resources=resources,
)
update_wrapper(task_instance, decorated_fn)
return task_instance

TaskMetadata and Configuration

TaskMetadata (flytekit/core/base_task.py) is a dataclass that holds execution-policy fields that map directly to the FlyteIDL TaskMetadata proto. Its fields are:

Field	Type	Default	Purpose
cache	bool	False	Enable output caching
cache_serialize	bool	False	Serialize identical cache-keyed executions
cache_version	str	""	Version string for cache entries
cache_ignore_input_vars	Tuple[str, ...]	()	Inputs excluded from cache key
interruptible	Optional[bool]	None	Allow preemption / lower QoS
deprecated	str	""	Deprecation warning message
retries	int	0	Retry count on failure
timeout	Optional[Union[datetime.timedelta, int]]	None	Per-execution wall-clock limit
pod_template_name	Optional[str]	None	Name of a cluster PodTemplate
generates_deck	bool	False	Whether a Deck URI is produced
is_eager	bool	False	Whether this is an eager task

Validation in __post_init__

Several ValueErrors fire at construction time, so misconfiguration fails fast rather than at execution:

def __post_init__(self):
    if self.timeout:
        if isinstance(self.timeout, int):
            self.timeout = datetime.timedelta(seconds=self.timeout)
        elif not isinstance(self.timeout, datetime.timedelta):
            raise ValueError("timeout should be duration represented as either a datetime.timedelta or int seconds")
    if self.cache and not self.cache_version:
        raise ValueError("Caching is enabled ``cache=True`` but ``cache_version`` is not set.")
    if self.cache_serialize and not self.cache:
        raise ValueError("Cache serialize is enabled ``cache_serialize=True`` but ``cache`` is not enabled.")
    if self.cache_ignore_input_vars and not self.cache:
        raise ValueError(
            f"Cache ignore input vars are specified ``cache_ignore_input_vars={self.cache_ignore_input_vars}`` but ``cache`` is not enabled."
        )

So timeout accepts either datetime.timedelta or an int (seconds, auto-converted). Enabling cache requires a non-empty cache_version. cache_serialize and cache_ignore_input_vars are no-ops unless cache is also True.

The @task decorator handles the cache-version requirement for you: when you pass cache=True without an explicit version, it constructs a Cache object and calls cache.get_version(VersionParameters(...)) to derive one automatically. Passing the deprecated cache_serialize/cache_version/cache_ignore_input_vars keyword arguments alongside a Cache object raises a ValueError.

Caching at Local Execution Time

During local execution, Task.local_execute checks the in-process LocalTaskCache before running anything. The cache is only consulted when both self.metadata.cache is set and LocalConfig.auto().cache_enabled is true:

if self.metadata.cache and local_config.cache_enabled:
    if local_config.cache_overwrite:
        outputs_literal_map = None
        logger.info("Cache overwrite, task will be executed now")
    else:
        logger.info(
            f"Checking cache for task named {self.name}, cache version {self.metadata.cache_version} "
            f", inputs: {kwargs}, and ignore input vars: {self.metadata.cache_ignore_input_vars}"
        )
        outputs_literal_map = LocalTaskCache.get(
            self.name, self.metadata.cache_version, input_literal_map, self.metadata.cache_ignore_input_vars
        )
        if outputs_literal_map is None:
            logger.info("Cache miss, task will be executed now")
        else:
            logger.info("Cache hit")
    if outputs_literal_map is None:
        outputs_literal_map = self.sandbox_execute(ctx, input_literal_map)
        LocalTaskCache.set(
            self.name,
            self.metadata.cache_version,
            input_literal_map,
            self.metadata.cache_ignore_input_vars,
            outputs_literal_map,
        )

cache_overwrite forces a re-execution and overwrites the stored entry. The cache key is computed from the task name, cache version, and input literal map (minus any cache_ignore_input_vars).

Execution Modes

PythonFunctionTask declares an ExecutionBehavior enum that determines how execute() dispatches:

class ExecutionBehavior(Enum):
    DEFAULT = 1
    DYNAMIC = 2
    EAGER = 3

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)

Default tasks

A plain @task produces ExecutionBehavior.DEFAULT. Calling execute() invokes the wrapped function directly with native Python kwargs.

Dynamic tasks

The @dynamic decorator (flytekit/core/dynamic_workflow_task.py) is a thin wrapper that fixes the execution mode:

dynamic = functools.partial(task.task, execution_mode=PythonFunctionTask.ExecutionBehavior.DYNAMIC)

When a dynamic task runs, dynamic_execute compiles the function body into a PythonFunctionWorkflow at execution time and, in production, returns a DynamicJobSpec rather than native outputs. The workflow that gets compiled can contain an arbitrary number of nodes determined by the inputs:

@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

Two constraints apply to dynamic tasks:

• node_dependency_hints is only valid on dynamic tasks. Passing it on a static task raises ValueError("node_dependency_hints should only be used on dynamic tasks..."). This parameter lets you declare dependencies (e.g., launch plans that must be pre-registered) that flyte cannot infer statically.
• Reference tasks are unsupported inside dynamic tasks — compile_into_workflow raises ValueError("Reference tasks are currently unsupported within dynamic tasks").

Eager tasks

The @eager decorator (flytekit/core/task.py) bypasses the normal TaskPlugins lookup and constructs an EagerAsyncPythonFunctionTask directly, setting is_eager=True in the metadata:

def eager(_fn, *args, **kwargs):
    if _fn is None:
        return partial(eager, **kwargs)
    if "enable_deck" in kwargs:
        del kwargs["enable_deck"]
    et = EagerAsyncPythonFunctionTask(task_config=None, task_function=_fn, enable_deck=True, **kwargs)
    return et

Eager tasks are async — "Python becomes propeller," as the class docstring puts it: each task invocation inside an eager function creates a stack frame on the Flyte cluster as an execution rather than on the local memory stack. EagerAsyncPythonFunctionTask.execute sets up a Controller and worker queue, installs signal handlers, and runs async_execute, which either runs the task function locally or delegates to run_with_backend for a live cluster run.

Eager and dynamic modes are mutually exclusive — AsyncPythonFunctionTask.async_execute raises NotImplementedError if the execution mode is DYNAMIC.

The Execution Pipeline

Whether a task runs locally or on the platform, the core execution path goes through dispatch_execute (defined on PythonTask in base_task.py). The pipeline is:

pre_execute(user_params)
    ↓
translate input LiteralMap → native Python kwargs   (_literal_map_to_python_input)
    ↓
execute(**native_inputs)                              (calls _task_function or dynamic_execute)
    ↓
post_execute(user_params, native_outputs)             (no-op by default; can alter outputs)
    ↓
if result is LiteralMap or DynamicJobSpec → return as-is (short-circuit)
    ↓
translate native outputs → LiteralMap                 (_output_to_literal_map)
    ↓
_write_decks(...)
    ↓
return LiteralMap

Key details from dispatch_execute:

• pre_execute runs before input translation. The default implementation returns the params unchanged, but subclasses override it to set up things like a SparkSession before type transformers fire.
• Input translation uses TypeEngine.literal_map_to_kwargs to turn the incoming LiteralMap into native Python values matching self.python_interface.inputs. On failure during local execution the original exception is re-raised with a task-name prefix; during remote execution it's wrapped in FlyteNonRecoverableSystemException.
• execute is where the user function actually runs. Exceptions during local execution are re-raised with context; remotely they become FlyteUserRuntimeException.
• post_execute is a no-op by default but can be overridden to clean up or transform outputs. If it raises IgnoreOutputs, that exception bubbles up — this is the mechanism for distributed-training-style tasks where only one worker should emit outputs.
• Output translation maps return values back to literals. Single-output tasks handle the single-length-NamedTuple edge case specially. If an output receives a tuple where a scalar type was expected, TypeError is raised: f"Output({k}) in task '{self.name}' received a tuple {v}, instead of {py_type}".
• Short-circuit: if execute (or post_execute) returns a LiteralMap or DynamicJobSpec, the output-translation step is skipped entirely. This is how dynamic tasks return their compiled workflow spec and how no-op dynamic tasks return a pre-built LiteralMap.

Local execution vs. remote execution

local_execute is the local-only entry point. It does extra work that dispatch_execute alone does not:

1. Unwraps incoming Promise/native-constant kwargs into a LiteralMap via translate_inputs_to_literals.
2. Consults LocalTaskCache (as described above).
3. Calls sandbox_execute, which builds a fresh ExecutionParameters via .with_task_sandbox() and then calls dispatch_execute.
4. Wraps the resulting literals back into Promise objects (or a VoidPromise if there are no outputs).

Task.__call__ delegates to flyte_entity_call_handler, which decides at call time whether to compile a node (inside a workflow) or to run local_execute (at the top level).

Task Resolution at Runtime

When flytekit serializes a task for the platform, it needs to embed enough information for a fresh container to find and rehydrate that same task object. This is the job of TaskResolverMixin (base_task.py) and its default implementation DefaultTaskResolver (python_auto_container.py).

PythonAutoContainerTask.get_default_command produces the command line that runs inside the container:

def get_default_command(self, settings: SerializationSettings) -> List[str]:
    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),
    ]
    return container_args

DefaultTaskResolver.loader_args returns the module and attribute name of the task:

def loader_args(self, settings, task):
    _, m, t, _ = extract_task_module(task)
    return ["task-module", m, "task-name", t]

At runtime, DefaultTaskResolver.load_task reverses this:

def load_task(self, loader_args):
    _, task_module, _, task_name, *_ = loader_args
    task_module = importlib.import_module(name=task_module)
    task_def = getattr(task_module, task_name)
    return task_def

So a task defined in repo_root.workflows.example as t1 serializes to a container command ending in:

--resolver flytekit.core.python_auto_container.default_task_resolver \
-- task-module repo_root.workflows.example task-name t1

You can supply a custom resolver via the task_resolver parameter of @task or PythonAutoContainerTask. The EagerFailureTaskResolver is an example of a custom resolver — it always loads an EagerFailureHandlerTask regardless of loader args, because the failure handler doesn't need to rehydrate a specific user function.

Nested-Function Constraint

Because the default resolver rehydrates tasks by importing their module and looking up an attribute, the task function must be reachable at module level. PythonFunctionTask.__init__ enforces this when the default resolver is in use:

if self._task_resolver is default_task_resolver:
    if (
        not istestfunction(func=task_function)
        and isnested(func=task_function)
        and not is_functools_wrapped_module_level(task_function)
    ):
        raise ValueError(
            "TaskFunction cannot be a nested/inner or local function. "
            "It should be accessible at a module level for Flyte to execute it. Test modules with "
            "names beginning with `test_` are allowed to have nested tasks. "
            "If you're decorating your task function with custom decorators, use functools.wraps "
            "or functools.update_wrapper on the function wrapper. "
            "Alternatively if you want to create your own tasks with custom behavior use the TaskResolverMixin"
        )

Three exemptions exist: test modules (filenames starting with test_), functions wrapped with functools.wraps/functools.update_wrapper that remain module-level, and tasks using a non-default resolver.

Deck Generation

PythonTask accepts enable_deck (and the deprecated disable_deck) along with a deck_fields tuple. The default deck_fields are:

deck_fields: Optional[Tuple[DeckField, ...]] = (
    DeckField.SOURCE_CODE,
    DeckField.DEPENDENCIES,
    DeckField.TIMELINE,
    DeckField.INPUT,
    DeckField.OUTPUT,
)

Setting both disable_deck and enable_deck raises ValueError("only one of [disable_deck, enable_deck] can be set"). Decks are disabled by default (_disable_deck = True unless enable_deck=True).

PythonFunctionTask._write_decks extends the base behavior by rendering the task function's source code (via SourceCodeRenderer) and Python dependencies (via PythonDependencyRenderer), wrapped in suppress(OSError, TypeError) so missing source doesn't break execution. The INPUT and OUTPUT decks use TypeEngine.to_html to render native values.

The Plugin System

TaskPlugins (flytekit/core/task.py) is the registry that lets specialized task types (Spark, Athena, PyTorch, etc.) plug into the @task decorator:

class TaskPlugins(object):
    _PYTHONFUNCTION_TASK_PLUGINS: Dict[type, Type[PythonFunctionTask]] = {}

    @classmethod
    def register_pythontask_plugin(cls, plugin_config_type: type, plugin: Type[PythonFunctionTask]):
        if plugin_config_type in cls._PYTHONFUNCTION_TASK_PLUGINS:
            found = cls._PYTHONFUNCTION_TASK_PLUGINS[plugin_config_type]
            if found == plugin:
                return
            raise TypeError(
                f"Requesting to register plugin {plugin} - collides with existing plugin {found}"
                f" for type {plugin_config_type}"
            )
        cls._PYTHONFUNCTION_TASK_PLUGINS[plugin_config_type] = plugin

    @classmethod
    def find_pythontask_plugin(cls, plugin_config_type: type) -> Type[PythonFunctionTask]:
        if plugin_config_type in cls._PYTHONFUNCTION_TASK_PLUGINS:
            return cls._PYTHONFUNCTION_TASK_PLUGINS[plugin_config_type]
        return PythonFunctionTask

A plugin registers a mapping from a config-object type to a PythonFunctionTask subclass. When you call @task(task_config=Spark()), the decorator inspects type(Spark()), finds the registered plugin class, and instantiates it with the config and function. With no task_config (or an unrecognized config type), find_pythontask_plugin falls back to PythonFunctionTask.

Async and Eager Task Variants

When the decorated function is a coroutine, the @task decorator switches the plugin to AsyncPythonFunctionTask. This subclass overrides __call__ to use async_flyte_entity_call_handler and defines async_execute (which awaits _task_function). The synchronous execute is set to loop_manager.synced(async_execute), bridging the async function into the standard dispatch_execute pipeline.

EagerAsyncPythonFunctionTask further extends this. Its execute method sets up a Controller (with signal handlers for SIGINT/SIGTERM), attaches a worker queue to the context, and runs async_execute synchronously via loop_manager.run_sync. get_as_workflow wraps the eager task in an ImperativeWorkflow with an EagerFailureHandlerTask as the on-failure handler — that handler terminates any still-running child executions tagged with the parent's execution name.

Common Pitfalls

A few constraints surface as runtime errors worth keeping in mind:

• Cache without a version: TaskMetadata.__post_init__ raises ValueError if cache=True but cache_version is empty. Use the Cache object in @task to let flytekit derive a version automatically.
• cache_serialize / cache_ignore_input_vars without cache: both raise ValueError in __post_init__.
• Nested functions: raise ValueError unless in a test_ module, wrapped with functools.wraps, or using a custom resolver.
• node_dependency_hints on non-dynamic tasks: raises ValueError.
• Reference tasks in dynamic tasks: compile_into_workflow raises ValueError.
• Eager + dynamic: AsyncPythonFunctionTask.async_execute raises NotImplementedError for DYNAMIC mode.
• Async function with incompatible plugin: raises AssertionError(f"Task plugin {task_plugin} is not compatible with async functions").
• Deck parameter conflict: setting both disable_deck and enable_deck raises ValueError.
• disable_deck is deprecated since flytekit 1.10.0 — use enable_deck instead.