Deployments are server-side representations of flows that can be executed:

This guide covers how to trigger deployments on demand.

Prerequisites

In order to run a deployment, you need to have:

Run a deployment from the CLI

The simplest way to trigger a deployment run is using the Prefect CLI:

prefect deployment run my-flow/my-deployment

CLI options

Add parameters and customize the run:

# Pass parameters
prefect deployment run my-flow/my-deployment \
  --param my_param=42 \
  --param another_param="hello"

# Schedule for later
prefect deployment run my-flow/my-deployment --start-in "2 hours"

# Watch the run until completion
prefect deployment run my-flow/my-deployment --watch

# Set custom run name
prefect deployment run my-flow/my-deployment --name "custom-run-name"

# Add tags to the run
prefect deployment run my-flow/my-deployment --tag production --tag critical

Run a deployment from Python

Use the run_deployment function for programmatic control:

from prefect.deployments import run_deployment

# Basic usage
flow_run = run_deployment(
    name="my-flow/my-deployment"
)

# With parameters
flow_run = run_deployment(
    name="my-flow/my-deployment",
    parameters={
        "my_param": 42,
        "another_param": "hello"
    }
)

# With job variables (environment variables, etc.)
flow_run = run_deployment(
    name="my-flow/my-deployment",
    parameters={"my_param": 42},
    job_variables={"env": {"MY_ENV_VAR": "production"}}
)

# Don't wait for completion
flow_run = run_deployment(
    name="my-flow/my-deployment",
    timeout=0  # returns immediately
)

# Wait with custom timeout (seconds)
flow_run = run_deployment(
    name="my-flow/my-deployment",
    timeout=300  # wait up to 5 minutes
)

# Schedule for later
from datetime import datetime, timedelta

flow_run = run_deployment(
    name="my-flow/my-deployment",
    scheduled_time=datetime.now() + timedelta(hours=2)
)

# With custom tags
flow_run = run_deployment(
    name="my-flow/my-deployment",
    tags=["production", "critical"]
)

By default, deployments triggered via run_deployment from within another flow will be treated as a subflow of the parent flow in the UI. To disable this, set as_subflow=False.

Async usage

In an async context, you can use the run_deployment function as a coroutine:

import asyncio
from prefect.deployments import run_deployment

async def trigger_deployment():
    flow_run = await run_deployment(
        name="my-flow/my-deployment",
        parameters={"my_param": 42}
    )
    return flow_run

# Run it
flow_run = asyncio.run(trigger_deployment())

Further reading