Core execution concepts and context

Two context layers: execution machinery and task-visible values

A task needs two different kinds of context. Flytekit keeps compilation and dispatch machinery in FlyteContext, while task code reads runtime values from ExecutionParameters through flytekit.current_context().

import flytekit

ctx = flytekit.current_context()
ctx.logging.info("running task")

flytekit.current_context() is defined in flytekit/__init__.py and returns:

return FlyteContextManager.current_context().execution_state.user_space_params

That return value is an ExecutionParameters object, not a FlyteContext. It contains the values intended for user task code: a working directory, logger, stats handle, execution identifiers, secrets, checkpoints, decks, and task-specific attributes.

FlyteContext is the internal-facing object used to convert values, compile workflows, serialize entities, and execute tasks. Its fields include:

• file_access, used for local sandboxes and data persistence;
• optional compilation_state and execution_state;
• an optional Flyte client and SerializationSettings;
• output_metadata_tracker and worker_queue;
• in_a_condition, a nesting level, and the originating stack frame.

Its user_space_params property delegates through the execution state:

@property
def user_space_params(self) -> Optional[ExecutionParameters]:
    if self.execution_state:
        return self.execution_state.user_space_params
    return None

Use FlyteContext when you are extending Flytekit internals or need to create a scoped compilation or execution context. Use flytekit.current_context() inside task code when you need runtime parameters.

Representing compilation and execution state

FlyteContext.compilation_state and FlyteContext.execution_state carry different phases of Flytekit operation. FlyteContext.new_compilation_state() creates a CompilationState with the default Python task resolver and an optional prefix. FlyteContext.new_execution_state() creates an ExecutionState using the file-access provider's local sandbox when no working directory is supplied, and carries forward the current user parameters.

ExecutionState records the values needed while a task or local workflow runs:

class ExecutionState(object):
    def __init__(
        self,
        working_dir,
        mode=None,
        engine_dir=None,
        branch_eval_mode=None,
        user_space_params=None,
    ):
        if not working_dir:
            raise ValueError("Working directory is needed")
        self.working_dir = working_dir
        self.mode = mode
        self.engine_dir = engine_dir if engine_dir else os.path.join(self.working_dir, "engine_dir")
        pathlib.Path(self.engine_dir).mkdir(parents=True, exist_ok=True)
        self.branch_eval_mode = branch_eval_mode
        self.user_space_params = user_space_params

The execution state's working_dir is the directory associated with execution machinery. Its engine_dir defaults to working_dir/engine_dir and is created by the constructor. The separate ExecutionParameters.working_directory is the directory exposed to user code.

To derive an execution state, use with_params() rather than rebuilding all fields yourself:

es = ctx.execution_state.with_params(
    mode=ExecutionState.Mode.LOCAL_TASK_EXECUTION,
    user_space_params=execution_parameters,
)

with_params() constructs a new ExecutionState while retaining values that are not replaced. The same pattern is used by FlyteContext builders to derive a new context without directly changing the context currently on the stack.

Execution modes

ExecutionState.Mode distinguishes the runtime being represented. The enum contains:

Mode	Meaning in Flytekit
TASK_EXECUTION	A task execution that mimics the hosted runtime environment.
LOCAL_WORKFLOW_EXECUTION	A workflow is being run locally; task results are wrapped as workflow outputs such as NodeOutput.
LOCAL_TASK_EXECUTION	A task is run locally without a container or Propeller.
DYNAMIC_TASK_EXECUTION	A dynamic task execution.
EAGER_EXECUTION	Eager execution.
EAGER_LOCAL_EXECUTION	Local eager execution.
LOCAL_DYNAMIC_TASK_EXECUTION	Local dynamic task execution.

The distinction between TASK_EXECUTION and LOCAL_TASK_EXECUTION matters for task behavior. The comments on ExecutionState.Mode specifically contrast dynamic tasks: in local task execution a dynamic task runs as an ordinary function, while task execution extracts a runtime specification.

ExecutionState.is_local_execution() returns True for LOCAL_TASK_EXECUTION, LOCAL_WORKFLOW_EXECUTION, EAGER_LOCAL_EXECUTION, and LOCAL_DYNAMIC_TASK_EXECUTION. TASK_EXECUTION, DYNAMIC_TASK_EXECUTION, and EAGER_EXECUTION are not included in that check.

Derive contexts with builders

FlyteContext.Builder is a mutable builder for creating a derived FlyteContext. It copies the current fields, lets you replace selected values, and increments the nesting level when build() creates the context:

def build(self) -> FlyteContext:
    return FlyteContext(
        level=self.level + 1,
        file_access=self.file_access,
        compilation_state=self.compilation_state,
        execution_state=self.execution_state,
        flyte_client=self.flyte_client,
        serialization_settings=self.serialization_settings,
        in_a_condition=self.in_a_condition,
        output_metadata_tracker=self.output_metadata_tracker,
        worker_queue=self.worker_queue,
    )

The builder exposes with_execution_state, with_compilation_state, with_file_access, with_serialization_settings, with_client, with_output_metadata_tracker, and with_worker_queue. For example, the context-manager test in tests/flytekit/unit/core/test_context_manager.py overrides an inherited client in a nested context:

ctx = FlyteContextManager.current_context()
b = ctx.new_builder()
b.flyte_client = SampleTestClass(value=1)
with FlyteContextManager.with_context(b) as outer:
    assert outer.flyte_client.value == 1
    b = outer.new_builder()
    b.flyte_client = SampleTestClass(value=2)
    with FlyteContextManager.with_context(b) as ctx:
        assert ctx.flyte_client.value == 2
    with FlyteContextManager.with_context(outer.with_new_compilation_state()) as ctx:
        assert ctx.flyte_client.value == 1

Do not confuse this builder with ExecutionParameters.Builder. The latter clones user-visible runtime parameters. Its add_attr() method adds task-specific values, and build() creates the configured working directory before constructing ExecutionParameters.

Nested contexts and cleanup

FlyteContextManager owns a singleton stack backed by a ContextVar. The normal scoped API is with_context():

with FlyteContextManager.with_context(ctx.new_builder()) as ctx:
    # Work with the derived FlyteContext here.
    ...

Internally, with_context() calls builder.build(), pushes the result, records the stack depth, and yields the new context. Its finally block pops contexts until the recorded depth is restored. This cleanup is significant for conditionals: Flytekit's conditional syntax does not itself use a Python with block, so a failed conditional compilation or evaluation can leave a branch context on the stack. with_context() removes such extra contexts when its scope exits.

push_context() and pop_context() are also available, but the FlyteContextManager docstring marks manual push/pop as not recommended. If you use them, pair every push with a pop. pop_context() raises an assertion if removing the context would leave the stack empty.

current_context() initializes the manager if its ContextVar has no value. initialize() resets the entire stack, creates <Config.local_sandbox_path>/user_space, installs the SIGINT dispatcher on the main thread, and creates default local ExecutionParameters with a local execution identifier, UTC execution date, mock stats, the user-space logger, and the local user-space directory.

Context warnings

• FlyteContextManager is explicitly documented as not thread-safe. A new thread can have an empty ContextVar and cause current_context() to initialize a default context; shared context mutation should not be treated as safe.
• Prefer with_context() over manual push_context() and pop_context() so the stack is restored even when conditional syntax has leaked a context.
• FlyteContext.current_context() exists for backward compatibility. User task code should use flytekit.current_context(), which returns ExecutionParameters; FlyteContextManager.current_context() returns the internal FlyteContext.

Hosted execution and local task execution

The hosted and local paths populate the same context model differently. The hosted task entrypoint in flytekit/bin/entrypoint.py creates runtime parameters from backend identifiers and execution services, then installs a TASK_EXECUTION state:

execution_parameters = ExecutionParameters(
    execution_id=_identifier.WorkflowExecutionIdentifier(
        project=exe_project, domain=exe_domain, name=exe_name,
    ),
    execution_date=datetime.datetime.now(datetime.timezone.utc),
    stats=_get_stats(
        cfg=StatsConfig.auto(),
        prefix=f"{tk_project}.{tk_domain}.{tk_name}.user_stats",
        tags={
            "exec_project": exe_project,
            "exec_domain": exe_domain,
            "exec_workflow": exe_wf,
            "exec_launchplan": exe_lp,
            "api_version": _api_version,
        },
    ),
    logging=user_space_logger,
    tmp_dir=user_workspace_dir,
    raw_output_prefix=raw_output_data_prefix,
    output_metadata_prefix=output_metadata_prefix,
    checkpoint=checkpointer,
    task_id=_identifier.Identifier(
        _identifier.ResourceType.TASK, tk_project, tk_domain, tk_name, tk_version
    ),
)

es = ctx.new_execution_state().with_params(
    mode=ExecutionState.Mode.TASK_EXECUTION,
    user_space_params=execution_parameters,
)
omt = OutputMetadataTracker()
cb = ctx.new_builder().with_execution_state(es).with_output_metadata_tracker(omt)

with FlyteContextManager.with_context(cb) as ctx:
    yield ctx

The complete entrypoint also replaces the file-access provider with one configured for the raw output prefix and attaches serialization settings to the builder before entering the context.

Local task dispatch uses a per-task sandbox instead. BaseTask.sandbox_execute() calls ExecutionParameters.with_task_sandbox(), which creates a temporary directory, creates a __cp subdirectory, and configures a SyncCheckpoint whose destination is that directory. It then replaces the execution state's user parameters before dispatching the task:

def sandbox_execute(self, ctx, input_literal_map):
    es = cast(ExecutionState, ctx.execution_state)
    b = cast(ExecutionParameters, es.user_space_params).with_task_sandbox()
    ctx = ctx.current_context().with_execution_state(
        es.with_params(user_space_params=b.build())
    ).build()
    return self.dispatch_execute(ctx, input_literal_map)

This local sandbox path is not invoked during hosted runtime execution, as stated in BaseTask.sandbox_execute().

Conditional compilation and local branch evaluation

BranchEvalMode has two values, BRANCH_ACTIVE and BRANCH_SKIPPED; None represents being outside a conditional context. It is primarily used for local conditional evaluation.

Constructing a ConditionalSection pushes a derived context immediately:

ctx = FlyteContextManager.current_context()
FlyteContextManager.push_context(ctx.enter_conditional_section().build())

FlyteContext.Builder.enter_conditional_section() clones the compilation state with the same prefix. For local execution, the first conditional section receives BRANCH_SKIPPED by default. A nested conditional preserves the skipped state by deriving another execution state. When a branch expression evaluates true, ExecutionState.take_branch() sets branch_eval_mode to BRANCH_ACTIVE; branch_complete() sets it back to BRANCH_SKIPPED.

The conditional dispatcher selects its implementation from the current context:

ctx = FlyteContextManager.current_context()

if ctx.compilation_state:
    return ConditionalSection(name)
elif ctx.execution_state:
    if ctx.execution_state.is_local_execution():
        if ctx.execution_state.branch_eval_mode == BranchEvalMode.BRANCH_SKIPPED:
            return SkippedConditionalSection(name)
        return LocalExecutedConditionalSection(name)
raise AssertionError("Branches can only be invoked within a workflow context!")

At the last branch, ConditionalSection.end_branch() pops the conditional context before constructing the branch node. This is why conditional context management must be understood together with the manager's cleanup behavior.

Runtime parameters available to tasks

ExecutionParameters is constructed with the execution date, temporary directory, stats handle, execution identifier, logger, output prefixes, optional checkpoint, decks, task identifier, and arbitrary keyword attributes. The most commonly used properties are:

• working_directory: a directory for temporary or user-created files;
• logging: the task's logger;
• stats: a tagged stats handle;
• execution_id, execution_date, and task_id: identifiers associated with the workflow or task;
• raw_output_prefix and output_metadata_prefix: output locations supplied by the runtime;
• secrets: a SecretsManager created for this parameter object;
• checkpoint: the configured checkpoint handle;
• decks, default_deck, timeline_deck, and enable_deck.

Flytekit's docstrings explicitly say that execution_date and execution_id should be used as tags or for debugging, not to drive production logic. task_id is populated at production runtime from backend-provided information.

Working directories and checkpoints

Use working_directory for files associated with user code:

ctx = flytekit.current_context()
ctx.logging.info(f"working in {ctx.working_directory}")

To create a task-specific directory with checkpoint support, derive parameters with with_task_sandbox() and build them. The builder creates __cp and attaches SyncCheckpoint. If no checkpoint was configured, reading ctx.checkpoint raises:

NotImplementedError(
    "Checkpointing is not available, please check the version of the platform."
)

Do not assume that every ExecutionParameters instance has a usable checkpoint. Hosted entrypoint setup only creates SyncCheckpoint when checkpoint_path is provided; otherwise it passes None.

Logging, stats, custom attributes, and decks

ExecutionParameters.Builder preserves the current values when initialized with an existing object. It also supports arbitrary task-specific attributes:

builder = ExecutionParameters.new_builder(current_parameters)
builder.add_attr("SPARK_SESSION", session)
parameters = builder.build()

Custom attributes are looked up case-insensitively by uppercasing the requested name. If the attribute is not available, __getattr__ raises an AssertionError, not an AttributeError:

session = flytekit.current_context().SPARK_SESSION

Deck state is stored in ExecutionParameters.decks. The default_deck property returns a new Deck("default"). timeline_deck searches the existing decks for a TimeLineDeck; if none exists, it lazily creates one with auto_add_to_deck=False and caches it. Accessing timeline_deck alone therefore does not append it to decks.

FlyteContext.get_deck() renders the deck produced by the last execution. The source docstring shows the intended scoped usage:

with flytekit.new_context() as ctx:
    my_task(...)
ctx.get_deck()

Task-level deck settings affect what is placed into user_space_params.decks. The deck tests exercise these real task options:

@task(**kwargs)
def t1(a: int) -> str:
    return str(a)

t1(a=3)
assert len(ctx.user_space_params.decks) == expected_decks

In those tests, disable_deck=True produces zero decks, while the default enabled behavior produces five decks for the tested task; selecting specific deck_fields changes the count.

Secrets: environment first, files second

Read a task secret through the runtime context:

sec = flytekit.current_context().secrets
value = sec.get("group", "key")

Flytekit also supports attribute syntax. The shell-task test uses both forms:

sec = flytekit.current_context().secrets
envvar = sec.get_secrets_env_var("group", "key")
os.environ[envvar] = "value"
assert sec.get("group", "key") == "value"

t = ShellTask(
    name="test",
    script="""
        export EXEC={ctx.execution_id}
        export SECRET={ctx.secrets.group.key}
        cat {inputs.f}
        """,
    inputs=kwtypes(f=CSVFile, y=FlyteDirectory, j=datetime.datetime),
    debug=True,
)

ctx.secrets.group.key calls SecretsManager.__getattr__(), which returns a group proxy; the proxy then calls SecretsManager.get(group, key).

SecretsManager uses SecretsConfig.auto() unless a SecretsConfig is passed explicitly. Its lookup order is:

1. Check the configured environment prefix plus uppercase GROUP, optional GROUP_VERSION, and KEY.
2. During local execution, also check the same uppercase name without the configured prefix.
3. Check the configured file path, using lowercase path components.
4. Raise ValueError if neither source contains the secret.

For example, the context-manager tests set ABC_XYZ during LOCAL_TASK_EXECUTION and _FSEC_ABC_XYZ during TASK_EXECUTION:

@pytest.mark.parametrize("is_local_execution, prefix", [(True, ""), (False, "_FSEC_")])
def test_secrets_manager_execution(monkeypatch, is_local_execution, prefix):
    if not is_local_execution:
        execution_state = context_manager.ExecutionState.Mode.TASK_EXECUTION
    else:
        execution_state = context_manager.ExecutionState.Mode.LOCAL_TASK_EXECUTION

    sec = SecretsManager()
    monkeypatch.setenv(f"{prefix}ABC_XYZ", "my-abc-secret")

    ctx = FlyteContext.current_context()
    with FlyteContextManager.with_context(
        ctx.with_execution_state(ctx.execution_state.with_params(mode=execution_state))
    ):
        assert sec.get(group="ABC", key="XYZ") == "my-abc-secret"

The configured names come from these settings:

• FLYTE_SECRETS_ENV_PREFIX controls the environment prefix;
• FLYTE_SECRETS_DEFAULT_DIR controls the base directory for file lookup;
• FLYTE_SECRETS_FILE_PREFIX is prepended to the lowercase final key in the filename.

For group="ABC", group_version="v1", and key="XYZ", the file lookup is assembled as <default_dir>/abc/v1/<file_prefix>xyz. encode_mode="r" reads text, while encode_mode="rb" reads a binary file. Values from either source are stripped before being returned. A missing value raises ValueError with a message asking for the corresponding Secret(group=..., key=...) declaration in the task.

Secrets and context warnings

• Hosted TASK_EXECUTION checks the configured prefixed environment variable; the unprefixed fallback is added only when the current execution is absent or local.
• SecretsManager is recreated for every ExecutionParameters instance. Pass a custom SecretsConfig directly to SecretsManager when you need non-default resolution settings.
• ExecutionParameters.with_params() and ExecutionState.with_params() use truthiness checks when selecting replacement values. An empty string or another falsey value does not act as an explicit reset; it falls back to the existing value.
• ExecutionParameters.Builder.build() creates its working directory unless it is an AutoDeletingTempDir. The source comments require AutoDeletingTempDir to be used with a with block so its lifecycle is managed correctly.