Task authoring and execution
You declare a task by decorating a Python function with @task. Flyte automatically inspects the function signature, registers the task with a global FlyteEntities registry, and wires it into the execution engine. The decorator produces a PythonFunctionTask instance — the central class in flytekit's task authoring model.
from flytekit import task, workflow
@task
def process_data(x: int) -> str:
return f"processed: {x}"
@workflow
def my_workflow(x: int) -> str:
return process_data(x=x)
The Task Class Hierarchy
The task system follows a clear inheritance chain. Task (in flytekit/core/base_task.py) is the abstract root — it captures configuration but lacks Python-native interfaces. PythonTask extends Task and adds a typed Interface, environment variable support, and deck generation. PythonAutoContainerTask (in flytekit/core/python_auto_container.py) extends PythonTask and auto-configures the container image and resource settings used when the task runs on a Flyte cluster. Finally, PythonFunctionTask extends PythonAutoContainerTask and wraps an actual Python callable.
The @task decorator creates a PythonFunctionTask. When the decorated function is async, flytekit creates an AsyncPythonFunctionTask instead (detected via inspect.iscoroutinefunction(fn) in flytekit/core/task.py, line 411). Eager workflows use EagerAsyncPythonFunctionTask, which sets metadata.is_eager = True automatically in its constructor.
TaskMetadata: Configuring Execution Behavior
Every task carries a TaskMetadata object that controls retries, caching, timeouts, and more. Pass these as keyword arguments to @task:
import datetime
from flytekit import task, Cache
# Enable caching with explicit version (required when cache=True)
@task(cache=True, cache_version="1.0.0")
def cached_task(x: float) -> float:
return x * 2
# Cache with ignored inputs and serialization
@task(cache=Cache(ignored_inputs=["debug_flag"], serialize=True), cache_version="v2")
def robust_cache(data: str, debug_flag: bool = False) -> int:
return len(data)
# Retry on failure and set a timeout
@task(retries=3, timeout=datetime.timedelta(minutes=5))
def reliable_task(data: str) -> int:
return len(data)
# Mark as interruptible for lower-cost scheduling
@task(interruptible=True)
def preemptible_task(n: int) -> int:
return n ** 2
TaskMetadata (defined in flytekit/core/base_task.py) validates its fields in __post_init__. If cache=True but cache_version is empty, it raises ValueError. The same validation applies to cache_serialize (requires cache=True) and cache_ignore_input_vars (also requires cache=True).
The timeout parameter accepts either a datetime.timedelta or an integer (interpreted as seconds):
# Both are equivalent
@task(timeout=300)
def with_int_timeout(x: int) -> int:
return x + 1
@task(timeout=datetime.timedelta(seconds=300))
def with_timedelta_timeout(x: int) -> int:
return x + 1
How Tasks Are Invoked
When you call a task — either directly or within a workflow — the flyte_entity_call_handler function in flytekit/core/promise.py (line 1442) decides what happens based on the current ExecutionState.Mode:
- Compilation mode (
ctx.compilation_state.mode == 1): The task call producesPromiseobjects and creates a node in the workflow graph viacreate_and_link_node. - Local execution mode (one of
LOCAL_TASK_EXECUTION,LOCAL_WORKFLOW_EXECUTION,EAGER_LOCAL_EXECUTION,LOCAL_DYNAMIC_TASK_EXECUTION): The task'slocal_executemethod runs the actual function. - Eager execution mode (
EAGER_EXECUTION): The call is forwarded toasync_flyte_entity_call_handler, which queues the task for remote execution via aController.
Inside local_execute (in Task, line 677 of base_task.py), the inputs are first translated from LiteralMap to native Python values via _literal_map_to_python_input. If caching is enabled and a hit is found in LocalTaskCache, the cached LiteralMap is returned immediately. Otherwise, sandbox_execute creates a task-sandbox context and calls dispatch_execute, which invokes the concrete execute method.
Tasks that return no outputs still return a VoidPromise (from flytekit/core/promise.py).
Dynamic Workflows
A dynamic workflow compiles a sub-workflow at runtime. Use the @dynamic decorator, which is a partial of the @task decorator with execution_mode=PythonFunctionTask.ExecutionBehavior.DYNAMIC (see flytekit/core/dynamic_workflow_task.py, line 21):
from flytekit import dynamic, task
@task
def t1(a: int) -> str:
return str(a)
@task
def t2(b: str) -> int:
return len(b)
@dynamic
def dynamic_subwf(a: int, b: int) -> int:
s = t1(a=a)
return t2(b=s)
Inside PythonFunctionTask.dynamic_execute (line 295 of python_function_task.py), the function body is executed to produce a workflow. The workflow is serialized into a DynamicJobSpec containing nodes and task templates, which FlytePropeller executes as a sub-workflow. During local execution, dynamic_execute runs the function directly and returns a LiteralMap.
The node_dependency_hints parameter on @task (or @dynamic) lists tasks, launch plans, or workflows that the dynamic function depends on at runtime. This is useful when calling launch plans, which must be registered on FlyteAdmin before they can run:
from flytekit import dynamic, LaunchPlan
@dynamic(node_dependency_hints=[some_launch_plan])
def dynamic_with_launchplan():
return [some_launch_plan()] * 10
Eager Workflows
Eager workflows let you write Flyte workflows using native Python async/await syntax. Every Flyte entity called inside an @eager-decorated function runs as a separate execution on the Flyte cluster, with results fetched asynchronously. Use the @eager decorator from flytekit:
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
if __name__ == "__main__":
import asyncio
result = asyncio.run(eager_workflow(x=1))
print(f"Result: {result}") # Result: 4
@eager produces an EagerAsyncPythonFunctionTask (line 425 of python_function_task.py). The constructor sets metadata.is_eager = True and execution_mode = ExecutionBehavior.EAGER. When execute is called, it checks whether the execution is local. If local, it runs the async function directly. If remote, it constructs a Controller that uses FlyteRemote to launch sub-executions and fetches their results.
During eager local execution, calling a regular @task from within an eager workflow invokes that task's local_execute and returns native Python values (not Promise objects). This is handled in flyte_entity_call_handler (line 1514 of promise.py) by checking for EAGER_LOCAL_EXECUTION mode and calling create_native_named_tuple to unwrap the results.
Decks: Task Visualization
Flyte generates HTML decks for tasks, showing source code, dependencies, timeline, inputs, and outputs. Control this with enable_deck, disable_deck (deprecated), and deck_fields:
@task(enable_deck=True)
def with_deck(x: int) -> int:
return x + 1
@task(enable_deck=True, deck_fields=(DeckField.SOURCE_CODE, DeckField.INPUT))
def minimal_deck(x: int) -> int:
return x * 2
The deck_fields parameter accepts a tuple of DeckField values from flytekit.deck. By default, decks are disabled (enable_deck=False), but the @task decorator sets a default of (SOURCE_CODE, DEPENDENCIES, TIMELINE, INPUT, OUTPUT) when deck_fields is not explicitly set. The disable_deck parameter is deprecated as of version 1.10.0.
Ignoring Task Outputs
The IgnoreOutputs exception (in flytekit/core/base_task.py, line 191) signals that a task's outputs can be safely discarded. This is useful for distributed training or peer-to-peer parallel algorithms where outputs are written to shared storage rather than returned:
from flytekit import task
from flytekit.core.base_task import IgnoreOutputs
@task
def distributed_trainer(rank: int) -> None:
if rank == 0:
# Master writes model to shared storage
save_model(...)
return None
else:
# Workers don't produce meaningful outputs
raise IgnoreOutputs()
Defining Interfaces Without Type Annotations
The kwtypes helper (in flytekit/core/base_task.py, line 100) creates an OrderedDict of types from keyword arguments. Use this when defining task interfaces that don't come from decorated Python functions — for example, with PythonTask subclasses that override execute:
from flytekit import PythonTask
from flytekit.core.base_task import kwtypes
class MyCustomTask(PythonTask):
_TASK_TYPE = "my-custom-task"
def __init__(self, name: str, **kwargs):
super().__init__(
task_type=self._TASK_TYPE,
name=name,
interface=Interface(
inputs=kwtypes(data=str, count=int),
outputs=kwtypes(result=str)
),
**kwargs
)
def execute(self, data: str, count: int) -> str:
return data * count
Common Gotchas
Nested functions are forbidden. The default task resolver cannot serialize nested functions. Flytekit raises ValueError if you try to use an inner function as a task unless it is wrapped with functools.wraps or the module name starts with test_:
@task
def outer():
@task # ValueError: TaskFunction cannot be a nested/inner or local function
def inner():
return 1
return inner()
Custom decorators on task functions must also use functools.wraps or functools.update_wrapper, otherwise flytekit loses track of the underlying function.
Cache requires a version. Setting cache=True without cache_version raises ValueError in TaskMetadata.__post_init__. Use the Cache object to bundle caching options cleanly.
Eager and dynamic are incompatible. AsyncPythonFunctionTask does not handle dynamic execution — eager and dynamic execution modes cannot be combined in the same task.
Reference tasks cannot be used inside dynamic tasks. If you try to include a ReferenceTask in a dynamic workflow, compile_into_workflow raises ValueError because it would require a network call to FlyteAdmin.
Single-length NamedTuples need special handling. When a dynamic workflow output is a NamedTuple with one field, the code in _output_to_literal_map uses output_tuple_name as a proxy to detect and handle this case correctly.