@Marvin: Can I use a flow run to trigger another flow run that’s “disconnected” from the parent one? So the parent one can complete, and the other continues on. I’d like to chain them together, but keep them disconnected.

✅ thought for 244.0 seconds

tldr 1.

prefect.flow-run.Completed

DeploymentTrigger on downstream one 2.

run_deployment

in

on_completion

hook of upstream one

Haha appreciated. I think I’m missing something, but how come you need both? If you’ve got the trigger on completion, then that could trigger the downstream one? It’s actually the same deployment, I just want to basically keep at least one always running!

Yes—there are a couple of good ways to do this in Prefect 3.x so the parent flow finishes immediately while the downstream flow keeps running, without a parent/child link. Option 1: Fire-and-forget a deployment from your flow - Use

run_deployment(...)

with

timeout=0

so the call returns immediately, and

as_subflow=False

so there’s no parent/child relationship in the UI.

from prefect import flow
from prefect.deployments import run_deployment

@flow
def parent():
    # Triggers a new flow run and returns immediately
    run = run_deployment(
        name="downstream-flow/downstream-deployment",
        timeout=0,          # don't wait
        as_subflow=False,   # no parent/child linkage
        parameters={"x": 1},
        idempotency_key="unique-key-123",  # optional, for deduplication
    )
    return run.id

- Docs: Run deployments • run_deployment API Option 2: Chain deployments with Automations (fully decoupled, event-driven) - Use

DeploymentEventTrigger

so the downstream deployment runs when the upstream completes—no parent/child linkage at all.

from prefect import flow, serve
from prefect.events import DeploymentEventTrigger

@flow
def upstream():
    ...

@flow
def downstream():
    ...

if __name__ == "__main__":
    up = upstream.to_deployment(name="upstream")
    down = downstream.to_deployment(
        name="downstream",
        triggers=[
            DeploymentEventTrigger(
                expect={"prefect.flow-run.Completed"},
                # Match the upstream run; see docs for matching options
                match_related={"prefect.resource.name": "upstream"}
            )
        ],
    )
    serve(up, down)

- Docs: Chaining deployments with events • Creating deployment triggers • DeploymentEventTrigger API Option 3: Lower-level control via the client - Create a run directly from a deployment. If you don’t provide a

parent_task_run_id

, it’s not linked as a subflow.

from uuid import UUID
from prefect.client.orchestration import get_client

deployment_id = UUID("...")

async def trigger():
    async with get_client() as client:
        fr = await client.create_flow_run_from_deployment(
            deployment_id=deployment_id,
            parameters={"x": 1},
            name="custom-decoupled-run",
            idempotency_key="unique-key-123",
            work_queue_name="default"
            # parent_task_run_id=None by default -> no parent/child link
        )
        return fr.id

- Docs: PrefectClient API Which route fits your needs best? If you want explicit chaining without any coupling, Automations are great. If you just want to “kick off and forget” from within a running flow,

run_deployment(timeout=0, as_subflow=False)

is the simplest. Happy to tailor an example if you share how you’re deploying and whether you need to pass data or delays between runs.

I forgot about

as_subflow

! And Marvin copied your trigger answer, haha

✅ thought for 383.6 seconds

trigger is the first class way to solve it,

run_deployment

is a more manual but more transparent way to do the same thing

That makes sense, thank you!

yea, it will often show you stuff you don't need to know, like manually setting the idempotency key, bc it thinks its being helpful, but messing with the defaults in those cases is actually very not helpful

It is quite verbose by default, you’re right

Great question. There isn’t a single “keep one always running” switch, but you can achieve it reliably with two pieces: 1) Enforce at most one at a time (singleton) - Set a deployment-level concurrency limit of 1. This prevents overlap and handles any accidental double-starts.

from prefect import flow
from prefect.client.schemas.objects import ConcurrencyLimitConfig, ConcurrencyLimitStrategy

@flow
def always_on():
    ...

# Example: deploy with a strict singleton policy
always_on.deploy(
    name="always-on",
    work_pool_name="my-pool",
    concurrency_limit=ConcurrencyLimitConfig(
        limit=1,
        collision_strategy=ConcurrencyLimitStrategy.ENQUEUE,  # or CANCEL_NEW
        grace_period_seconds=300
    )
)

- Docs: Deployments: concurrency limiting 2) Auto-restart when it exits (ensure at least one keeps running) Option A — Self-restarting deployment via Automations/DeploymentEventTrigger: - Add a trigger to the same deployment so a new run is created whenever the last one finishes (Completed/Failed/Cancelled/Timeout). This keeps the deployment continuously running, with no parent/child linkage.

from datetime import timedelta
from prefect import flow, serve
from prefect.events import DeploymentEventTrigger
from prefect.client.schemas.objects import ConcurrencyLimitConfig

@flow
def always_on():
    ...

if __name__ == "__main__":
    dep = always_on.to_deployment(
        name="always-on",
        concurrency_limit=ConcurrencyLimitConfig(limit=1),
        triggers=[
            DeploymentEventTrigger(
                expect={
                    "prefect.flow-run.Completed",
                    "prefect.flow-run.Failed",
                    "prefect.flow-run.Cancelled",
                    "prefect.flow-run.Timeout",
                },
                # Match this deployment’s runs; name is "<flow>/<deployment>"
                match_related={"prefect.resource.name": "always_on/always-on"},
                # Optional small cooldown to prevent hot loops
                schedule_after=timedelta(seconds=5),
            )
        ],
    )
    serve(dep)

- Docs: Chaining deployments with events • Creating deployment triggers • DeploymentEventTrigger API Option B — “Watchdog” that checks and restarts if nothing is running: - Periodically run this (via another small deployment or a cron) to ensure one is running. It counts RUNNING flow runs for the deployment and fires a new run if zero. Combine with the concurrency limit (limit=1) to prevent overlap. ``` from uuid import UUID from prefect import flow from prefect.deployments import run_deployment from prefect.client.orchestration import get_client from prefect.client.schemas.filters import ( FlowRunFilter, FlowRunFilterState, FlowRunFilterStateType, FlowRunFilterDeploymentId ) from prefect.client.schemas.objects import StateType DEPLOYMENT_NAME = "always_on/always-on" # "<flow>/<deployment>" async def _count_running_for_deployment(deployment_id: UUID) -> int: async with get_client() as client: fr_filter = FlowRunFilter( deployment_id=FlowRunFilterDeploymentId(any_=[deployment_id]), state=FlowRunFilterState(type=FlowRunFilterStateType(any_=[StateType.RUNNING])), ) return await client.count_flow_runs(flow_run_filter=fr_filter) @flow async def watchdog(): # Resolve deployment ID from name async with get_client() as client: dep = await client.read_deployment_by_name(DEPLOYMENT_NAME) deployment_id = dep.id count = await _count_running_for_deployment(deployment_id) if count == 0: # Fire-and-forget, no parent/child linkage

Verbose, but helpful tbf

sorry i missed this

Haha appreciated. I think I’m missing something, but how come you need both? If you’ve got the trigger on completion, then that could trigger the downstream one?

It’s actually the same deployment, I just want to basically keep at least one always running!

i wasn't suggesting you need both, just that you can do either if you want one always running, i'd just do the

run_deployment

in

on_completion

thing. that'd just be an infinite chain of the deployment finishing then invoking itself again

Oh nice, I thought it was possibly that too, of being one or the other!