Python Function Tasks and Dynamic Tasks

Every @task-decorated function in flytekit becomes an instance of PythonFunctionTask. Understanding how that class works — and how its three execution modes differ — is the key to understanding what happens when your task runs on a Flyte backend, when you call it locally, and when you want it to generate sub-workflows at runtime.

PythonFunctionTask: The Foundation

PythonFunctionTask (in flytekit/core/python_function_task.py) wraps a Python callable and extends PythonAutoContainerTask. When @task is applied to a function, two things happen immediately:

1. Interface detection. transform_function_to_interface inspects the function's type annotations to build a typed Interface object (inputs and outputs).
2. Name extraction. extract_task_module derives the task name from the function's module path and qualified name.

from flytekit import task

@task
def process(x: int, label: str) -> float:
    return float(x) * 2.0

The resulting PythonFunctionTask has process.name set to something like my_module.process, auto-detected from the function. Its .task_function property holds the original callable, and its .execution_mode is ExecutionBehavior.DEFAULT.

The class hierarchy is:

PythonAutoContainerTask
    └── PythonFunctionTask
            ├── AsyncPythonFunctionTask
            │       └── EagerAsyncPythonFunctionTask
            └── (plugin subclasses via TaskPlugins)

The ExecutionBehavior Enum

PythonFunctionTask has an inner enum that controls how the task behaves at execution time:

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

The execute() method branches on this value:

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)

EAGER is handled by EagerAsyncPythonFunctionTask, which overrides execute() entirely.

Nested Functions Are Forbidden

The default task resolver needs to import the task function at execution time by module path. If the function is defined inside another function (a closure), it can't be imported. PythonFunctionTask.__init__ enforces this:

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. ..."
    )

The exemption for test modules (names starting with test_) allows test files to define tasks inline. If you're wrapping task functions with your own decorators, use functools.wraps to preserve the module-level identity, or implement a custom TaskResolverMixin.

---

DEFAULT Mode: Plain Task Execution

A plain @task creates a PythonFunctionTask with ExecutionBehavior.DEFAULT. When the task runs on the cluster, execute() calls the wrapped function directly with the deserialized input values.

@task
def add(a: int, b: int) -> int:
    return a + b

No special compilation, no sub-workflow generation. The function runs, returns a value, and flytekit serializes it.

---

DYNAMIC Mode: Runtime Workflow Compilation

A dynamic task starts life as a regular task on the Flyte backend. At execution time, instead of returning a result, its function body runs under a compilation context and produces a new workflow specification, which Flyte's engine then executes as a subworkflow.

The @dynamic decorator in flytekit/core/dynamic_workflow_task.py is exactly this:

# flytekit/core/dynamic_workflow_task.py
dynamic = functools.partial(task.task, execution_mode=PythonFunctionTask.ExecutionBehavior.DYNAMIC)

It's just @task with execution_mode preset. Your function sees Python-native values for its inputs, so you can use range(), if, append, and any other Python control flow:

from flytekit import task, dynamic
import typing

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

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

How Dynamic Compilation Actually Works

When execute() detects DYNAMIC mode, it delegates to dynamic_execute(), which checks the current execution context:

• Local execution (is_local_execution() is true): The function runs in LOCAL_DYNAMIC_TASK_EXECUTION mode, behaving like a workflow that executes tasks locally. You get full Python results back.

  # This runs locally and returns actual values:
  result = my_subwf(a=3)
  assert result == ["fast-2", "fast-3", "fast-4"]

• Backend task execution (ExecutionState.Mode.TASK_EXECUTION): dynamic_execute() calls compile_into_workflow(). This method creates a PythonFunctionWorkflow from the task function via _create_and_cache_dynamic_workflow(), enters a compilation context (ExecutionState.Mode.DYNAMIC_TASK_EXECUTION), compiles the workflow, then serializes it via get_serializable to produce a WorkflowSpec. The result is packaged into a DynamicJobSpec:

  dj_spec = _dynamic_job.DynamicJobSpec(
      min_successes=len(workflow_spec.template.nodes),
      tasks=tts,
      nodes=workflow_spec.template.nodes,
      outputs=workflow_spec.template.outputs,
      subworkflows=workflow_spec.sub_workflows,
  )
  return dj_spec

  If no nodes are produced (the dynamic function returned constant values without calling any tasks), compile_into_workflow() returns a plain LiteralMap instead.

The _wf cache on the task object (self._wf) is None at construction time and only populated on demand by _create_and_cache_dynamic_workflow().

node_dependency_hints: Pre-Registering Dynamic Dependencies

Flyte can't statically determine what tasks a dynamic workflow will call — the call graph is only known at runtime. If your dynamic task conditionally calls different tasks based on its input, those tasks won't be discovered during static compilation and may not be registered. node_dependency_hints solves this:

@task
def t1() -> int:
    return 0

@task
def t2() -> int:
    return 0

@dynamic(node_dependency_hints=[t1, t2])
def dt(mode: int) -> int:
    if mode == 1:
        return t1()
    if mode == 2:
        return t2()
    raise ValueError("Invalid mode")

The hints tell flytekit's serializer to pre-register t1 and t2 before the dynamic task itself. Under the hood, get_serializable_task() in the translator serializes the hinted entities first, ensuring they appear in the entity mapping before dt.

node_dependency_hints is silently restricted to dynamic tasks — setting it on a plain @task immediately raises:

ValueError: node_dependency_hints should only be used on dynamic tasks. On static
tasks and workflows its redundant because flyte can find the node dependencies automatically

Dynamic Task Constraints

• Reference tasks are not supported inside dynamic tasks. compile_into_workflow() raises ValueError if a ReferenceTask appears in the serialized entity mapping.
• Keep node counts low. The dynamic_workflow_task.py module docstring explicitly warns: "It's rare to see a manually written workflow that has 5000 nodes for instance, but you can easily get there with a loop. Please keep dynamic workflows to under fifty tasks."
• Cannot mix with @eager. AsyncPythonFunctionTask.async_execute() raises NotImplementedError for DYNAMIC mode.

---

EAGER Mode: Python as the Orchestrator

Eager workflows take a different approach to dynamic execution. Instead of compiling a subworkflow at runtime, the Python function body itself acts as the orchestrator: every Flyte entity called inside the function becomes an actual execution on the backend.

Defining an Eager Workflow

The @eager decorator (in flytekit/core/task.py) creates an EagerAsyncPythonFunctionTask:

def eager(_fn=None, *args, **kwargs) -> Union[EagerAsyncPythonFunctionTask, partial]:
    ...
    et = EagerAsyncPythonFunctionTask(task_config=None, task_function=_fn, enable_deck=True, **kwargs)
    return et

The decorated function must be an async def:

from flytekit import task, eager
import asyncio

@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)
    doubled = double(x=out)
    return await doubled

EagerAsyncPythonFunctionTask forces execution_mode=ExecutionBehavior.EAGER and sets is_eager=True in the task metadata. Deck is always enabled.

Local vs. Backend Execution

EagerAsyncPythonFunctionTask.async_execute() handles two paths:

• Local execution: The function runs in EAGER_LOCAL_EXECUTION mode within the current event loop. Task calls produce Python-native values as you'd expect. This lets you test eager workflows with plain asyncio.run().

• Backend execution: execute() sets up a Controller — the worker queue (from flytekit.core.worker_queue) — which dispatches each sub-entity call as a real Flyte execution via FlyteRemote. The run_with_backend() method then awaits the async function with this infrastructure in place.

  # Inside run_with_backend():
  mode = ExecutionState.Mode.EAGER_EXECUTION
  with FlyteContextManager.with_context(builder) as ctx:
      result = await self._task_function(**kwargs)
      html = cast(Controller, ctx.worker_queue).render_html()
      Deck("Eager Executions", html)

Concurrency is built-in: since the function is async, you can use asyncio.gather or asyncio.create_task to fan out executions:

@eager
async def parent_wf(a: int, b: int) -> typing.Tuple[int, int]:
    t1 = asyncio.create_task(base_wf(x=a))
    t2 = asyncio.create_task(base_wf(x=b))
    i1, i2 = await asyncio.gather(t1, t2)
    return i1, i2

Error Handling in Eager Workflows

Sub-task failures surface as EagerException (from flytekit.exceptions.eager). Catch it to implement fallback logic:

from flytekit.exceptions.eager import EagerException

@eager
async def eager_workflow(x: int) -> int:
    try:
        out = await add_one(x=x)
    except EagerException:
        out = 0
    return await double(x=out)

Failure Cleanup: EagerFailureHandlerTask

When an eager workflow fails mid-flight, sub-executions it launched may still be running. To address this, each eager task has a paired EagerFailureHandlerTask and the two are bundled together via get_as_workflow():

def get_as_workflow(self):
    from flytekit.core.workflow import ImperativeWorkflow
    cleanup = EagerFailureHandlerTask(
        name=f"{self.name}-cleanup",
        container_image=self.container_image,
        inputs=self.python_interface.inputs,
    )
    wb = ImperativeWorkflow(name=self.name)
    ...
    wb.add_on_failure_handler(cleanup)
    return wb

EagerFailureHandlerTask.dispatch_execute() runs on the backend and queries the Flyte admin API, finding all executions tagged with eager-exec: <parent_execution_name> that are still in QUEUED or RUNNING state, and terminates them. Its execute() method raises AssertionError — it's only ever invoked through dispatch_execute, because at rehydration time it doesn't have access to the Python type interface of the original eager task.

During serialization, serialize_helpers.py specially handles EagerAsyncPythonFunctionTask by calling get_as_workflow() and registering the wrapper workflow and a default launch plan. This means an @eager function produces three registered entities: the task spec, the workflow spec, and the launch plan.

Eager Workflow Constraints

• Only @task, @workflow, and @eager entities are supported inside an eager workflow. Flyte-style Conditional objects are not supported — use plain Python if statements.
• Interactive mode registration (FlyteRemote) is not supported for eager tasks; the remote module raises FlyteAssertion in this path.
• @eager and @dynamic cannot be combined; AsyncPythonFunctionTask.async_execute() raises NotImplementedError for DYNAMIC mode.

---

Async Tasks (Without Eager)

If you decorate an async def function with plain @task (without @eager), flytekit detects the coroutine function via inspect.iscoroutinefunction and selects AsyncPythonFunctionTask instead of PythonFunctionTask:

# In task.py:
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")

AsyncPythonFunctionTask overrides __call__ to be async and implements async_execute() which awaits the task function. The class-level execute is a synchronous wrapper created by loop_manager.synced(async_execute), so the execution machinery that calls execute() remains synchronous.

If your task_config maps via TaskPlugins to a custom plugin, that plugin class must be a subclass of AsyncPythonFunctionTask for async task functions, or an AssertionError is raised at decoration time.

---

PythonInstanceTask: Class-Based Tasks Without a User Function

For tasks where you want to encapsulate behavior in a class rather than wrapping a user function, flytekit provides PythonInstanceTask (also in flytekit/core/python_function_task.py). The contract: subclass it, override execute(), and construct the task as a module-level instance:

x = MyInstanceTask(name="x", task_config=SomeConfig(...))

# The instance is callable like any other task
x(a=5)

PythonInstanceTask is the base for tasks like ShellTask and various plugin integrations where the business logic is defined in the class, not in a user-supplied function body.

---

PythonCustomizedContainerTask: External Container Plugins

Sometimes a task should run in a different container from the user's registered image — for example, a pre-built database query image or ML framework container. PythonCustomizedContainerTask (in flytekit/core/python_customized_container_task.py) provides the scaffolding for this pattern.

The class extends both ExecutableTemplateShimTask and PythonTask. Its key distinction: the container image is specified explicitly, and all execution logic is placed in a separate ShimTaskExecutor subclass, rather than in the task itself.

The Pattern: Three Moving Parts

Building a customized-container task requires:

1. A config dataclass — holding any parameters the executor needs.
2. A ShimTaskExecutor subclass — implementing execute_from_model(tt, **kwargs), where tt is the TaskTemplate and kwargs are the deserialized inputs.
3. A PythonCustomizedContainerTask subclass — connecting the two, specifying the container image, and overriding get_custom() to serialize config into the TaskTemplate.

get_custom() must be overridden — the base class raises NotImplementedError:

def get_custom(self, settings: SerializationSettings) -> Dict[str, Any]:
    # Overriding base implementation to raise an error, force task author to implement
    raise NotImplementedError

SQLite3Task: A Complete Example

SQLite3Task in flytekit/extras/sqlite3/task.py illustrates the full pattern:

@dataclass
class SQLite3Config:
    uri: str
    compressed: bool = False


class SQLite3Task(PythonCustomizedContainerTask[SQLite3Config], SQLTask[SQLite3Config]):
    _SQLITE_TASK_TYPE = "sqlite"

    def __init__(self, name, query_template, inputs=None, task_config=None,
                 output_schema_type=None, container_image=None, **kwargs):
        super().__init__(
            name=name,
            task_config=task_config,
            container_image=container_image or DefaultImages.default_image(),
            executor_type=SQLite3TaskExecutor,  # links to the executor
            task_type=self._SQLITE_TASK_TYPE,
            query_template=query_template,
            inputs=inputs,
            outputs=outputs,
            **kwargs,
        )

    def get_custom(self, settings: SerializationSettings) -> Dict[str, Any]:
        return {
            "query_template": self.query_template,
            "uri": self.task_config.uri,
            "compressed": self.task_config.compressed,
        }


class SQLite3TaskExecutor(ShimTaskExecutor[SQLite3Task]):
    def execute_from_model(self, tt: task_models.TaskTemplate, **kwargs) -> Any:
        # tt.custom holds exactly what get_custom() returned
        ctx = FlyteContext.current_context()
        ctx.file_access.get_data(tt.custom["uri"], local_path)
        interpolated_query = SQLite3Task.interpolate_query(tt.custom["query_template"], **kwargs)
        with contextlib.closing(sqlite3.connect(local_path)) as con:
            return pd.read_sql_query(interpolated_query, con)

TaskTemplateResolver: Loading at Runtime

At execution time, PythonCustomizedContainerTask uses TaskTemplateResolver (the default, default_task_template_resolver) instead of the normal resolver. Its loader_args() always returns the same pair of strings:

def loader_args(self, settings, t: PythonCustomizedContainerTask) -> List[str]:
    return ["{{.taskTemplatePath}}", f"{t.executor_type.__module__}.{t.executor_type.__name__}"]

At runtime, load_task() fetches the serialized TaskTemplate from {{.taskTemplatePath}}, deserializes it, then imports the executor class by its module path, and returns an ExecutableTemplateShimTask wrapping the two:

def load_task(self, loader_args: List[str]) -> ExecutableTemplateShimTask:
    ctx = FlyteContext.current_context()
    task_template_local_path = os.path.join(ctx.execution_state.working_dir, "task_template.pb")
    ctx.file_access.get_data(loader_args[0], task_template_local_path)
    task_template_model = _task_model.TaskTemplate.from_flyte_idl(...)
    executor_class = load_object_from_module(loader_args[1])
    return ExecutableTemplateShimTask(task_template_model, executor_class)

Note that TaskTemplateResolver.get_all_tasks() always returns an empty list — it does not track loaded tasks.

Lazy Template Serialization

PythonCustomizedContainerTask has a class-level SERIALIZE_SETTINGS constant with placeholder project/domain/version values:

SERIALIZE_SETTINGS = SerializationSettings(
    project="PLACEHOLDER_PROJECT",
    domain="LOCAL",
    version="PLACEHOLDER_VERSION",
    ...
)

The task_template property serializes on first access using these placeholders if no real settings have been provided yet. Proper identifiers are only stamped in when serialize_to_model() is called with real serialization settings during registration.

---

TaskPlugins: The Plugin Registry

TaskPlugins (defined in flytekit/core/task.py) is a class-level dictionary that maps config object types to PythonFunctionTask subclasses:

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

    @classmethod
    def register_pythontask_plugin(cls, plugin_config_type: type, plugin: Type[PythonFunctionTask]):
        ...
        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  # default

When you write @task(task_config=MyConfig()), the task() function calls TaskPlugins.find_pythontask_plugin(type(task_config)) to select the task class. If no plugin is registered for MyConfig, it falls back to PythonFunctionTask.

Plugin authors register their plugin at module import time:

# From flytekit-k8s-pod plugin:
class PodFunctionTask(PythonFunctionTask[Pod]):
    def __init__(self, task_config: Pod, task_function: Callable, **kwargs):
        super().__init__(
            task_config=task_config,
            task_type="sidecar",
            task_function=task_function,
            task_type_version=2,
            **kwargs,
        )

    def get_k8s_pod(self, settings) -> _task_models.K8sPod:
        ...

    def get_container(self, settings) -> _task_models.Container:
        return None  # signals to use k8s_pod instead


TaskPlugins.register_pythontask_plugin(Pod, PodFunctionTask)

Registering the same config type twice with different plugin classes raises TypeError. Registering with the same plugin class is idempotent.

---

Summary of Key Constraints and Gotchas

Situation	Behavior
@task wrapping a nested/inner function	ValueError at decoration time (unless in a test_ module)
node_dependency_hints on a non-dynamic task	ValueError at decoration time
ReferenceTask inside @dynamic	ValueError during compile_into_workflow()
@dynamic inside an @eager task	NotImplementedError in async_execute()
async def function with @task	Automatically uses AsyncPythonFunctionTask
async def function with a non-async-compatible plugin	AssertionError at decoration time
@eager task registered via FlyteRemote interactive mode	FlyteAssertion raised
Custom decorators on @task functions	Must use functools.wraps to preserve module-level identity
PythonCustomizedContainerTask without get_custom() override	NotImplementedError at serialization time