Short answer: yes. In Prefect 3.x, when you

submit()

a task you get a

PrefectFuture

you can poll or wait on, then trigger other flows. A few options: 1) Simple wait, then trigger a deployment

from prefect import task, flow
from prefect.deployments import run_deployment

@task
def do_work(x): 
    return x * 2

@flow
def parent():
    fut = do_work.submit(10)
    fut.wait()  # blocks until the task finishes

    if fut.state.is_completed():
        result = fut.result()
        run_deployment(
            name="downstream-flow/downstream-deployment",
            parameters={"upstream_result": result},
            as_subflow=True,   # shows as a child in the UI
        )
    elif fut.state.is_failed():
        # handle error or trigger a remediation flow
        pass

2) If you really want a while-not-done loop

from time import sleep

fut = do_work.submit(10)

while not fut.state.is_final():
    sleep(1)

if fut.state.is_completed():
    run_deployment(name="downstream-flow/downstream-deployment")

Tip:

.wait()

(or

.result()

) is cleaner than manual polling. 3) Fire actions as tasks finish (multiple tasks)

from prefect import task, flow
from prefect.futures import as_completed
from prefect.deployments import run_deployment

@task
def do_work(i): 
    return i

@flow
def parent():
    futures = [do_work.submit(i) for i in range(5)]
    for fut in as_completed(futures):
        if fut.state.is_completed():
            run_deployment(
                name="downstream-flow/downstream-deployment",
                parameters={"i": fut.result()},
            )

4) Lightweight callback (client-side)

def on_done(fut):
    if fut.state.is_completed():
        run_deployment(name="downstream-flow/downstream-deployment")

fut = do_work.submit(10)
fut.add_done_callback(on_done)

Note: callbacks are client-side; if the process dies, the callback won’t fire. For production-grade chaining, prefer Automations. Production-grade alternative: Automations / deployment triggers - You can trigger a downstream deployment automatically on events like

prefect.flow-run.Completed

or

prefect.task-run.Completed

. This is server-side and reliable even if your client code exits. If that’s what you’re after, I can share a minimal example using a

DeploymentEventTrigger

or an automation definition. Quick checks on future state: -

fut.wait()

or

fut.result()

-

fut.state.is_final()

,

fut.state.is_completed()

,

fut.state.is_failed()

-

prefect.futures.wait([...])

and

prefect.futures.as_completed([...])

for groups Would you like a server-side automation example that triggers a flow when a specific task (or flow) completes, or are you looking to do this entirely within a single parent flow run?

Great questions. Async tasks - Yes, this works with async tasks. The key is how you consume the result: - In a sync flow:

my_async_task.submit()

+

future.wait()

/

future.result()

is fine. - In an async flow: don’t call

future.wait()

(it blocks the event loop). Prefer: - Call the task directly and await it

from prefect import flow, task
      import asyncio

      @task
      async def do_work(x):
          await asyncio.sleep(1)
          return x * 2

      @flow
      async def parent():
          # Run one
          res = await do_work(10)

          # Run many concurrently
          results = await asyncio.gather(*(do_work(i) for i in range(5)))
          return results

- If you must use futures in an async flow, use the async result method instead of `wait()`:

fut = do_work.submit(10)
      # Prefer an async result call (method name depends on future type)
      # e.g., await fut.aresult() or await fut.result_async()
      # If you're not sure, the simplest/most robust is: await do_work(10)

- If you really want a while-not-done loop in an async flow, don’t block:

import asyncio
      fut = do_work.submit(10)
      while not fut.state.is_final():
          await asyncio.sleep(1)

- Triggering other flows still works the same. After completion you can call

run_deployment(...)

to kick off a downstream deployment. Signature (verified):

run_deployment(name, client=None, parameters=None, scheduled_time=None, flow_run_name=None, timeout=None, poll_interval=5, tags=None, idempotency_key=None, work_queue_name=None, as_subflow=True, job_variables=None)

. Prefect 2.x - Yes, the same basic pattern exists in Prefect 2: -

task.submit()

returns a future with

state

,

wait()

, and

result()

. - You can poll

future.state.is_final()

or just call

future.result()

. - You can trigger other flows with

run_deployment(...)

in 2.x as well. - Differences to note: - Some 2.x deployment/CLI APIs were removed/renamed in 3.x (e.g., don’t use

Deployment.build_from_flow()

or

prefect deployment build

in 3.x). If you’re still on 2.x, those older commands are fine there, but we recommend upgrading to 3.x for the current APIs and Automations. If you share whether your flow is sync or async (and if you’re on 2.x or 3.x), I can tailor a minimal snippet for your exact setup.