Launch plans, schedules, and fixed inputs

When you need to execute a workflow on a recurring schedule, lock certain arguments to prevent accidental overrides, or establish environment-specific runtime configurations (such as custom IAM roles or concurrency limits), defining the workflow alone is not sufficient. In flytekit, a Launch Plan (LaunchPlan in flytekit.core.launch_plan) wraps a workflow definition with bound inputs, schedules, notifications, and execution settings.

Every workflow registered in Flyte receives a default launch plan. You can also construct named launch plans to represent customized entry points for automated pipelines, scheduled jobs, or nested invocations inside other workflows.

---

Defining Launch Plans

Default Launch Plans

A default launch plan mirrors the workflow interface without modifying default values, adding fixed inputs, or configuring schedules.

from flytekit import workflow, task, LaunchPlan

@task
def calculate_metrics(threshold: float, dataset_name: str) -> float:
    return threshold * len(dataset_name)

@workflow
def metrics_workflow(threshold: float = 0.5, dataset_name: str = "production") -> float:
    return calculate_metrics(threshold=threshold, dataset_name=dataset_name)

# Retrieve or create the default launch plan
default_lp = LaunchPlan.get_or_create(workflow=metrics_workflow)

Internally, calling LaunchPlan.get_or_create(workflow=wf) without a name delegates to LaunchPlan.get_default_launch_plan(ctx, workflow). This extracts inputs and default values from workflow.python_interface using transform_inputs_to_parameters and stores the resulting plan in LaunchPlan.CACHE under workflow.name.

Constraint: You cannot specify custom parameters (such as default_inputs, fixed_inputs, or schedule) when retrieving or creating a default launch plan without specifying name. Supplying any extra parameters with name=None raises a ValueError: Only named launchplans can be created that have other properties. Drop the name if you want to create a default launchplan.

Named Launch Plans with Custom Parameters

To parameterize a workflow with custom defaults or schedule it, you must provide a unique name:

from flytekit import LaunchPlan

nightly_metrics_lp = LaunchPlan.get_or_create(
    name="nightly_metrics_lp",
    workflow=metrics_workflow,
    default_inputs={"threshold": 0.95},
    fixed_inputs={"dataset_name": "nightly_audit"},
    max_parallelism=10,
    overwrite_cache=True,
)

The combination of Flyte project, domain, version, and the launch plan name forms its unique identifier. During registration, flytekit registers all created instances recorded in FlyteEntities.entities.

---

Parameterizing Inputs: Default vs. Fixed

Flytekit distinguishes between inputs that callers can override and inputs that are permanently locked for that launch plan.

Workflow Interface: [threshold: float, dataset_name: str]
       │
       ├──> default_inputs: {"threshold": 0.95} ──> ParameterMap (Caller can override at execution)
       │
       └──> fixed_inputs:   {"dataset_name": "audit"} ──> LiteralMap (Removed from ParameterMap; immutable)

default_inputs

default_inputs supply fallback values for workflow arguments. If the underlying @workflow function signature already defines defaults, values in default_inputs take higher precedence. Callers launching the workflow (via the UI, CLI, or API) can override these defaults.

lp_custom_defaults = LaunchPlan.get_or_create(
    name="lp_custom_defaults",
    workflow=metrics_workflow,
    default_inputs={"threshold": 0.75},
)

fixed_inputs

fixed_inputs freeze arguments to specific values. Once set:

1. translate_inputs_to_literals converts the Python values into a flytekit.models.literals.LiteralMap.
2. The LaunchPlan.__init__ constructor filters the launch plan's ParameterMap, removing keys present in fixed_inputs.
3. The Flyte engine prohibits callers from overriding these values at execution time.

lp_fixed_dataset = LaunchPlan.get_or_create(
    name="lp_fixed_dataset",
    workflow=metrics_workflow,
    fixed_inputs={"dataset_name": "immutable_prod_v1"},
)

Executing Launch Plans Locally and in Workflows

You can invoke a launch plan directly inside another workflow or during local execution. When doing so, you must pass all arguments as keyword arguments.

@workflow
def parent_workflow() -> float:
    # Invoking a launch plan inside a workflow constructs a workflow node
    return lp_custom_defaults(threshold=0.85)

# Local invocation delegates to the underlying workflow function
result = lp_custom_defaults(threshold=0.85)

Inside LaunchPlan.__call__, flytekit checks for positional arguments:

if len(args) > 0:
    raise AssertionError("Only Keyword Arguments are supported for launch plan executions")

When compiling inside a workflow context (ctx.compilation_state is not None), __call__ invokes create_and_link_node(ctx, entity=self, **inputs) to attach the launch plan node to the execution graph. In local execution, it passes the merged dictionary (saved_inputs updated with kwargs) directly to self.workflow(*args, **inputs).

---

Scheduling Launch Plans

Flytekit supports recurring workflow execution through schedules defined in flytekit.core.schedule. You can configure time-based execution using CronSchedule or fixed interval execution using FixedRate.

Cron Schedules (CronSchedule)

Use CronSchedule to run workflows according to a standard 5-token cron expression or a predefined cron alias.

from datetime import datetime
from flytekit import workflow, task, LaunchPlan, CronSchedule

@task
def sync_data(run_time: datetime):
    print(f"Syncing data for run timestamp: {run_time}")

@workflow
def scheduled_sync_wf(run_time: datetime):
    sync_data(run_time=run_time)

hourly_sync_lp = LaunchPlan.get_or_create(
    name="hourly_sync_lp",
    workflow=scheduled_sync_wf,
    schedule=CronSchedule(
        schedule="0 * * * *",                   # Runs at minute 0 past every hour
        kickoff_time_input_arg="run_time",       # Injects execution timestamp into workflow input
    ),
    auto_activate=True,
)

CronSchedule features and constraints:

• schedule: Accepts a 5-field cron expression (parsed with croniter) or a standard alias from CronSchedule._VALID_CRON_ALIASES: "hourly", "@hourly", "daily", "@daily", "weekly", "@weekly", "monthly", "@monthly", "yearly", "annually", etc.
• kickoff_time_input_arg: When specified, the Flyte scheduler injects the scheduled kickoff timestamp into the matching workflow input parameter (as a Python datetime).
• offset: An optional ISO 8601 duration string (such as "PT1H" or "P1D") validated against the _OFFSET_PATTERN regex.
• Deprecated cron_expression: Passing cron_expression raises an AssertionError. Use the schedule parameter instead.

Interval Schedules (FixedRate)

Use FixedRate to run workflows at regular periodic intervals specified as a datetime.timedelta.

from datetime import timedelta
from flytekit import LaunchPlan, FixedRate

ten_minute_lp = LaunchPlan.get_or_create(
    name="ten_minute_lp",
    workflow=metrics_workflow,
    schedule=FixedRate(
        duration=timedelta(minutes=10),
    ),
    auto_activate=True,
)

FixedRate._translate_duration enforces the following rules:

1. Minimum Granularity: The duration cannot have microsecond precision and must be divisible by whole minutes (duration.seconds % 60 == 0). Finer granularities raise an AssertionError.
2. Unit Conversion: The duration is automatically mapped to FixedRateUnit.DAY, FixedRateUnit.HOUR, or FixedRateUnit.MINUTE.

Triggers (OnSchedule and LaunchPlanTriggerBase)

Flytekit provides a trigger abstraction via the LaunchPlanTriggerBase protocol. You can wrap a schedule inside OnSchedule and pass it to the trigger parameter of LaunchPlan.get_or_create:

from flytekit import LaunchPlan
from flytekit.core.schedule import OnSchedule, FixedRate
from datetime import timedelta

interval_trigger_lp = LaunchPlan.get_or_create(
    name="interval_trigger_lp",
    workflow=metrics_workflow,
    trigger=OnSchedule(
        schedule=FixedRate(duration=timedelta(hours=2)),
    ),
)

Both schedule and trigger populate schedule metadata during protobuf serialization through to_flyte_idl().

---

Execution Controls and Metadata

Launch plans accept infrastructure and runtime configuration parameters:

Parameter	Type	Behavior
auto_activate	bool	If True, the launch plan schedule is automatically activated upon registration in Flyte Admin without requiring manual UI/CLI activation. Default is False.
overwrite_cache	Optional[bool]	If True, all task executions initiated by this launch plan bypass and overwrite cached outputs.
max_parallelism	Optional[int]	Caps the maximum number of task nodes that can run concurrently within a single execution of this workflow.
security_context	Optional[SecurityContext]	Configures execution identities, such as Identity(iam_role="...", k8s_service_account="..."). (Replaces deprecated auth_role)
labels	Optional[Labels]	Key-value pairs attached to Kubernetes resources spawned by the execution.
annotations	Optional[Annotations]	Metadata annotations applied to the execution.
notifications	Optional[List[Notification]]	List of Notification objects (email, Slack, PagerDuty) triggered on phase transitions (SUCCEEDED, FAILED, ABORTED).
raw_output_data_config	Optional[RawOutputDataConfig]	Specifies offloaded storage paths (such as custom S3/GCS buckets) for raw task outputs.

Example configuring execution controls:

from flytekit import LaunchPlan
from flytekit.models.common import Annotations, Labels
from flytekit.models.security import Identity, SecurityContext

secure_metrics_lp = LaunchPlan.get_or_create(
    name="secure_metrics_lp",
    workflow=metrics_workflow,
    default_inputs={"threshold": 0.8},
    fixed_inputs={"dataset_name": "restricted_data"},
    max_parallelism=5,
    security_context=SecurityContext(
        run_as=Identity(
            iam_role="arn:aws:iam::123456789012:role/AnalyticsExecutionRole",
            k8s_service_account="analytics-sa",
        )
    ),
    labels=Labels({"team": "data-eng", "env": "production"}),
    annotations=Annotations({"owner": "analytics"}),
    auto_activate=False,
)

Deriving Launch Plans with clone_with

To create variations of an existing launch plan without redefining all parameters, use clone_with:

staging_metrics_lp = secure_metrics_lp.clone_with(
    name="staging_metrics_lp",
    fixed_inputs=None, # Reverts or overrides fixed inputs
    max_parallelism=2,
)

---

Referencing Remote Launch Plans

When constructing workflows that invoke launch plans registered in another Flyte project, domain, or version, use the @reference_launch_plan decorator from flytekit.core.launch_plan.

from flytekit import workflow
from flytekit.core.launch_plan import reference_launch_plan

@reference_launch_plan(
    project="shared_services",
    domain="production",
    name="aggregate_daily_metrics_lp",
    version="v1.2.0",
)
def remote_metrics_plan(threshold: float, dataset_name: str) -> float:
    ...

@workflow
def pipeline_wf() -> float:
    # remote_metrics_plan is compiled as an external ReferenceLaunchPlan node
    return remote_metrics_plan(threshold=0.9, dataset_name="daily_logs")

The decorator inspects the stub function signature using transform_function_to_interface(fn, is_reference_entity=True) and creates a ReferenceLaunchPlan without initiating network calls to Flyte Admin.

---

Common Gotchas and Failure Modes

• Name Collision and Inconsistent Definitions: LaunchPlan.CACHE indexes plans by name. Calling LaunchPlan.get_or_create multiple times with the same name but different arguments (such as differing default_inputs, schedule, or workflow) raises an AssertionError warning that conflicting configurations were detected for that name.
• Overriding Fixed Inputs: Passing an argument that is declared in fixed_inputs when launching the workflow or invoking the plan inside another workflow results in an error during execution or node compilation.
• Positional Arguments Disallowed: Launch plans must be invoked using keyword arguments only (lp(threshold=0.5)). Invoking lp(0.5) raises AssertionError: Only Keyword Arguments are supported for launch plan executions.
• Sub-minute FixedRate Durations: FixedRate(duration=timedelta(seconds=45)) raises AssertionError because Flyte scheduler intervals require minute-level granularity or coarser.
• Deprecated Cron Parameter: Passing CronSchedule(cron_expression="...") raises AssertionError. You must supply 5-field cron strings or aliases to schedule="0 * * * *".
• AuthRole vs SecurityContext: Supplying both auth_role and security_context to LaunchPlan.create or LaunchPlan.get_or_create raises ValueError: Use of AuthRole is deprecated. You cannot specify both AuthRole and SecurityContext. Always use security_context.
• Default Launch Plan Customizations: Calling LaunchPlan.get_or_create(workflow=wf, default_inputs={...}) without providing a name raises ValueError because unnamed plans default strictly to the bare workflow signature.