Recommendation: start agent programmatically instead of subprocess

[mike-lev-c]

https://github.com/anyscale/prefect-anyscale/blob/df77b6a0e7198c937d5779dd495bbe1d668a670d/start_anyscale_service.py#L39C47-L39C88

We ran into runtime_env issues when running this inside a serve deployment part of a multi app serve cluster. Solution that works for us is to start via the underlying cli start callback:

async def agent_start(
    work_queues: List[str],
    work_pool_name: str,
    run_once: bool,
    prefetch_seconds: int,
    limit: Optional[int] = None,
):
    """
    Start an agent process to poll one or more work queues for flow runs.
    """
    work_queues = work_queues or []
    async with PrefectAgent(
        work_queues=work_queues,
        work_pool_name=work_pool_name,
        prefetch_seconds=prefetch_seconds,
        limit=limit,
    ) as agent:
        print("AGENT STARTED")
        
        print(
            "Agent started! Looking for work from "
            f"work pool '{work_pool_name}'..."
        )
        
        async with anyio.create_task_group() as tg:
            tg.start_soon(
                partial(
                    critical_service_loop,
                    agent.get_and_submit_flow_runs,
                    PREFECT_AGENT_QUERY_INTERVAL.value(),
                    printer=print,
                    run_once=run_once,
                    jitter_range=0.3,
                    backoff=4,  # Up to ~1 minute interval during backoff
                )
            )

            tg.start_soon(
                partial(
                    critical_service_loop,
                    agent.check_for_cancelled_flow_runs,
                    PREFECT_AGENT_QUERY_INTERVAL.value() * 2,
                    printer=print,
                    run_once=run_once,
                    jitter_range=0.3,
                    backoff=4,
                )
            )

    print("Agent stopped!")