<@ULVA73B9P> can I use flows to schedule others wi...
# marvin-ai
o
@Marvin can I use flows to schedule others with custom params (but not run as subflows)
m
thought for 166.7 seconds
Yes. In Prefect 3.x you can have one flow trigger/schedule runs of other flows (via their deployments) with custom parameters, and keep them independent (not subflows). Two good options: 1) High-level helper: run_deployment (recommended) - Use
as_subflow=False
to avoid linking as a subflow in the UI - Pass
parameters={...}
to override defaults - Use
scheduled_time=<datetime>
to schedule for the future - Use
timeout=0
for fire-and-forget (don’t wait) Example:
Copy code
from datetime import datetime, timedelta
from prefect import flow
from prefect.deployments import run_deployment

@flow
def orchestrator():
    # Schedule another deployment for 2 hours from now with custom params
    run_deployment(
        name="target-flow/prod-deploy",
        parameters={"date": "2025-11-21", "priority": "high"},
        scheduled_time=datetime.utcnow() + timedelta(hours=2),
        as_subflow=False,   # keep it independent
        timeout=0           # return immediately
    )
2) Low-level client: create_flow_run_from_deployment - Gives full control; async only - If you don’t set
parent_task_run_id
, it will not be a subflow - Set initial state to
Scheduled(scheduled_time=...)
for future runs Example:
Copy code
import asyncio
from datetime import datetime, timedelta
from prefect.client.orchestration import get_client
from prefect.states import Scheduled

async def trigger():
    async with get_client() as client:
        dep = await client.read_deployment_by_name("target-flow/prod-deploy")
        fr = await client.create_flow_run_from_deployment(
            deployment_id=dep.id,
            parameters={"date": "2025-11-21", "priority": "high"},
            state=Scheduled(scheduled_time=datetime.utcnow() + timedelta(hours=2)),
            name="orchestrated-run",
            tags=["orchestrated"],
        )
        return fr

asyncio.run(trigger())
Notes and tips: - The target flow must have a deployment defined (created via
flow.deploy(...).apply()
or
flow.serve(...)
). -
run_deployment
can also accept
flow_run_name
,
tags
,
job_variables
,
idempotency_key
, and
work_queue_name
. - If you do want to wait for completion, omit
timeout
or set it to a positive number;
timeout=0
is fire-and-forget. Docs: - Run deployments programmatically - Create and manage schedules - Chaining deployments with events (alternative pattern) Would you like an example tailored to your specific deployment names/parameters or scheduling pattern (e.g., a batch of future times)?