Launch plans, schedules, and fixed inputs

A workflow function is just a definition — it has no execution configuration. The moment you want to run wf(a=3, c="hello") on a schedule, or with some inputs baked in so they can't be changed, you need a launch plan. In flytekit, that's the LaunchPlan class in flytekit/core/launch_plan.py, which wraps a workflow together with default inputs, fixed inputs, a schedule or trigger, notifications, labels, and execution options.

The default launch plan

Every workflow registered with Flyte gets a default launch plan — a plan with no defaults, fixed values, or schedules. You create it explicitly with:

from flytekit import workflow, LaunchPlan

@workflow
def wf(a: int, c: str) -> str:
    ...

lp = LaunchPlan.get_or_create(workflow=wf)

The default plan inherits whatever default values are defined in the workflow function signature. Internally, LaunchPlan.get_default_launch_plan calls transform_inputs_to_parameters(ctx, workflow.python_interface) to build the parameter map, leaves fixed_inputs as an empty LiteralMap, and stores the signature defaults in the plan's _saved_inputs so they're available when the plan is called.

One important rule: the default launch plan cannot carry any extra configuration. If you call get_or_create with no name but pass default_inputs, schedule, or any other attribute, flytekit raises:

ValueError("Only named launchplans can be created that have other properties. Drop the name if you want to create a default launchplan. Default launchplans cannot have any other associations")

Give the plan a unique name to use any of those features.

Default inputs vs fixed inputs

Named launch plans can parameterize a workflow in two distinct ways:

• default_inputs — values the caller can still override at launch time. These become entries in the launch plan's ParameterMap.
• fixed_inputs — values baked in permanently. They are translated to Flyte Literal objects (via translate_inputs_to_literals) into a LiteralMap, and the LaunchPlan.__init__ constructor actively strips them from the parameter map. In __init__:

# Ensure fixed inputs are not in parameter map
parameter_map = {k: v for k, v in parameters.parameters.items() if k not in fixed_inputs.literals}

So a caller simply cannot supply a new value for a fixed input — it isn't part of the launch plan's interface anymore. (Internally, flytekit/core/array_node.py relies on this too: when a launch plan is a map-over target, its fixed inputs are excluded from the resulting interface.)

A named launch plan looks like this:

from flytekit import LaunchPlan
from flytekit.models.common import Annotations, Labels, RawOutputDataConfig

my_lp = LaunchPlan.get_or_create(
    name="my_scheduled_lp",
    workflow=wf,
    default_inputs={"c": "hello"},
    fixed_inputs={"a": 42},
)

Both are plain Python native values — flytekit does the type translation for you. create() also merges fixed inputs into _saved_inputs (default_inputs.update(fixed_inputs)), so when you call the launch plan locally, the fixed values are included in the invocation.

If the same input appears in both default_inputs and fixed_inputs, the fixed value wins: create() removes it from the defaults as part of building the parameter map.

The caching contract — and its sharp edges

All launch plans live in LaunchPlan.CACHE, a class-level Dict[str, LaunchPlan]. Caching exists, per the source comment, "simply because users may get the default launch plan twice for a single Workflow. We don't want to create two defaults, could be confusing."

Two rules follow from this:

• create() raises AssertionError immediately if the name is already in the cache.
• get_or_create() with an existing name returns the cached plan — but compares every attribute (schedule, notifications, default/fixed inputs, labels, annotations, raw_output_data_config, max_parallelism, security_context, overwrite_cache, auto_activate). Any mismatch raises an AssertionError telling you to use a different launch plan name. It also checks the workflow identity: two launch plans with the same name for different workflows are rejected.

Names must be globally unique per project/domain/version — they form part of the primary key on the Admin side.

Schedules and triggers

There are two syntaxes for attaching a schedule to a launch plan.

The classic schedule= argument takes a schedule model directly:

from datetime import timedelta

from flytekit import CronSchedule, FixedRate, LaunchPlan

lp = LaunchPlan.get_or_create(
    name="every_min",
    workflow=wf,
    fixed_inputs={"a": 1},
    schedule=CronSchedule(schedule="*/1 * * * *"),
)

lp2 = LaunchPlan.get_or_create(
    name="every_10_min",
    workflow=wf,
    schedule=FixedRate(duration=timedelta(minutes=10)),
)

The newer (alpha) trigger= syntax wraps the schedule in an OnSchedule object (flytekit/core/schedule.py), which implements LaunchPlanTriggerBase:

from flytekit import OnSchedule

lp3 = LaunchPlan.get_or_create(
    name="triggered",
    workflow=wf,
    trigger=OnSchedule(schedule=CronSchedule(schedule="@daily")),
)

OnSchedule.to_flyte_idl() delegates to the wrapped schedule's protobuf serialization.

CronSchedule rules

CronSchedule(schedule=...) accepts either a cron alias — the class validates against _VALID_CRON_ALIASES, which includes "hourly", "@daily", "@weekly", "@monthly", "@yearly", and friends — or any 5-field expression parseable by croniter. Anything else raises ValueError("Schedule is invalid. It must be set to either a cron alias or valid cron expression.").

Do not use the cron_expression parameter — it raises an AssertionError immediately ("cron_expression is deprecated and should not be used. Use schedule instead."). The legacy AWS-style 6-field validation (_validate_expression) still exists in the code: it requires exactly 6 fields and a ? in either day-of-month or day-of-week, and validates the first 5 fields with croniter.

CronSchedule also supports an ISO 8601 offset (validated with the regex _OFFSET_PATTERN = re.compile("([-+]?)P([-+0-9YMWD]+)?(T([-+0-9HMS.,]+)?)?")), and kickoff_time_input_arg, which delivers the scheduled kickoff time as a workflow input:

from datetime import datetime

from flytekit import CronSchedule, workflow


@workflow
def my_wf(kickoff_time: datetime):
    ...


schedule = CronSchedule(
    schedule="*/1 * * * *",
    kickoff_time_input_arg="kickoff_time",
)

The docstring adds a caveat: "until Flyte has an atomic clock, there could be a few seconds here and there" — a run scheduled for 3pm UTC may actually start at 15:00:02.

FixedRate rules

FixedRate(duration=timedelta(minutes=10)) translates the duration into the largest fitting unit (day, hour, or minute) in _translate_duration. Sub-minute granularity is rejected outright:

AssertionError("Granularity of less than a minute is not supported for FixedRate schedules. Received: ...")

Any duration whose seconds aren't a whole number of minutes fails this check.

Calling a launch plan locally

LaunchPlan.__call__ lets you invoke a launch plan like a workflow. It accepts keyword arguments only — positional args raise AssertionError("Only Keyword Arguments are supported for launch plan executions"). The call merges the plan's saved inputs (defaults plus fixed inputs) with your kwargs:

inputs = self.saved_inputs
inputs.update(kwargs)
return self.workflow(*args, **inputs)

During compilation (when ctx.compilation_state is set) it instead calls create_and_link_node to wire the launch plan into the graph being built — this is how launch plans participate as nodes inside workflows, handled in flytekit/core/node_creation.py.

The saved_inputs property deliberately returns a copy, because call sites mutate the returned dict before updating it. Note the code's own TODO about custom classes as input types here.

Referencing launch plans that already exist

ReferenceLaunchPlan(ReferenceEntity, LaunchPlan) points at a launch plan already registered on your Flyte installation without making a network call — you supply the expected interface yourself, and registration-time compilation errors if it doesn't match:

from flytekit import reference_launch_plan


@reference_launch_plan(
    project="flytesnacks",
    domain="development",
    name="my_lp",
    version="v1",
)
def my_lp_ref(x: int) -> int:
    ...

Launch plans in dynamic workflows

To run a sub-launch plan from a dynamic workflow, the plan must already be registered on Admin. Pass it as a node dependency hint so flytekit registers it alongside the dynamic task:

launchplan0 = LaunchPlan.get_or_create(workflow=wf)


@dynamic(node_dependency_hints=[launchplan0])
def dynamic_wf():
    ...
    return [launchplan0] * 10

(This pattern is documented on the node_dependency_hints parameter in flytekit/core/task.py.)

Other constructor options

Beyond inputs and schedules, create()/get_or_create() accept notifications, labels, annotations, raw_output_data_config (offloaded data location for S3, etc.), max_parallelism (caps task nodes running in parallel across the workflow — MapTasks count as one unit and have their own concurrency), overwrite_cache, and auto_activate (activates the launch plan on registration; exposed as should_auto_activate).

Two deprecated paths to avoid:

• auth_role — passing both auth_role and security_context raises ValueError("Use of AuthRole is deprecated. You cannot specify both AuthRole and SecurityContext"); passing only auth_role silently converts it into a SecurityContext(run_as=Identity(...)).
• CronSchedule(cron_expression=...) — raises immediately, as described above.

Finally, clone_with() produces a copy of a launch plan with overridden attributes. Watch one asymmetry: every field uses new or self.field except trigger, which is passed through as given — a new trigger=None fully replaces the original, while a new schedule=None falls back to the original schedule.