Great question. In Prefect 3.x, from an async flow A you can start async flow B in a few ways depending on whether you want to wait for it or not. Quick clarification: Do you want to await B’s result, fire-and-forget, or trigger B’s deployment on a worker? Below are examples for each. Option 1 — Call B as a subflow and await (A waits for B)

from prefect import flow

@flow
async def flow_b(x: int) -> int:
    return x + 1

@flow
async def flow_a():
    result = await flow_b(10)  # runs as a subflow, A waits
    print(f"B result: {result}")

- If you need the state instead of the value:

state = await flow_b(10, return_state=True)
result = await state.result()

Option 2 — Start B without waiting (fire-and-forget) from A

from prefect import flow
from prefect.utilities.asyncutils import create_task

@flow
async def flow_b(x: int) -> int:
    return x + 1

@flow
async def flow_a():
    task = create_task(flow_b(10))  # schedule B immediately; A keeps going
    print("B started in the background")
    # optionally later: result = await task

Notes: - Simply calling

flow_b(10)

creates a coroutine but does not start it. Use

create_task(...)

(or

asyncio.create_task(...)

) to actually start it. Option 3 — Trigger B’s deployment via the API (run on a worker) If you have B deployed and want to start a run on its infrastructure: A) Async client (recommended in async flows)

from prefect import flow, get_client

@flow
async def flow_a():
    async with get_client() as client:
        flow_run = await client.create_flow_run_from_deployment(
            deployment_id=<UUID>,        # or look up by name first
            parameters={"x": 10},
            tags=["triggered-from-A"],
        )
        print(f"Started B run: {flow_run.id}")
        # fire-and-forget: just return; or wait for completion:
        # final = await client.wait_for_flow_run(flow_run.id)
        return flow_run.id

B) Look up by name, then trigger

from prefect import flow, get_client

@flow
async def flow_a():
    async with get_client() as client:
        deployments = await client.read_deployments(
            flow_filter={"name": {"eq_": "flow_b"}},
            deployment_filter={"name": {"eq_": "prod"}},
        )
        if not deployments:
            raise ValueError("Deployment not found")
        dep = deployments[0]

        fr = await client.create_flow_run_from_deployment(
            deployment_id=dep.id,
            parameters={"x": 10},
        )
        return fr.id

C) Helper function (synchronous) — fine for quick triggers

from prefect import flow
from prefect.deployments import run_deployment

@flow
async def flow_a():
    fr = run_deployment(
        "flow_b/prod",          # "<flow-name>/<deployment-name>"
        parameters={"x": 10},
        as_subflow=True,        # keep lineage under A
    )
    return fr.id

Which approach should we tailor for you? - Await B’s result within A - Fire-and-forget inside A - Trigger B’s deployment on a worker (with or without waiting)