async def METHOD_NAME(project_id):
    flow = prefect.Flow(
        name="Test Flow",
        schedule=prefect.schedules.IntervalSchedule(
            start_date=pendulum.datetime(2018, 1, 1),
            interval=datetime.timedelta(days=1),
        ),
    )
    flow.add_edge(
        prefect.Task("t1", tags={"red", "blue"}),
        prefect.Task("t2", tags={"red", "green"}),
    )
    flow.add_task(prefect.Parameter("x", default=1))
    METHOD_NAME = await api.flows.create_flow(
        project_id=project_id,
        serialized_flow=flow.serialize(),
        set_schedule_active=False,
    )
    return METHOD_NAME