@task
def task_a(x: int) -> int:
    get_run_logger().info(f"Running task A with {x}")
    sleep(5)
    return x


@task
def task_b(x: int) -> int:
    get_run_logger().info(f"Running task B with {x}")
    sleep(10)
    return x


@task
def task_c(x: int) -> int:
    get_run_logger().info(f"Running task C with {x}")
    sleep(5)
    return x


@flow
def run_process(x: int) -> int:
    r_a = task_a(x)
    r_b = task_b(r_a)
    r_c = task_c(r_b)
    return r_c

@flow
async def master_process(x_list: List[int]):
    await asyncio.gather(
        *[
            run_deployment(
                name=f"run-process/deployment-{x}",
                flow_run_name=f"{x}"
            )
            for x in x_list
        ]
    )


if __name__ == "__main__":
    asyncio.run(master_process([1, 2]))

Deployment.build_from_flow(run_process, f"deployment-1", parameters={"x": 1}).apply()

Deployment.build_from_flow(run_process, f"deployment-2", parameters={"x": 2}).apply()