Short answer: not natively with a single automation. In Prefect 3.x, Automations/Events don’t aggregate “all runs with tag X are finished” across a dynamic tag value. Triggers evaluate per resource/event; there isn’t a built-in “wait until no more runs with this tag exist” primitive. That said, you can get the behavior you want with one of these patterns: - Best if you can coordinate: a parent “batch” flow launches the subflows (all tagged with the batch UUID) and waits for them to finish. This gives you a single, event-driven place to run the follow-up. - If flows are created independently and only share the UUID tag: have an automation fire a small “checker” deployment on each flow-run completion for that tag, then make that checker decide whether it’s the last one. Use a short debounce (schedule_after) to coalesce multiple completions. The checker queries Prefect for any non-terminal runs with that tag; if none remain, it performs the one-time action. Example checker flow (works in Prefect 3.4.9, Python 3.12):

from prefect import flow, get_run_logger
from prefect.client.orchestration import get_client
from prefect.schemas.filters import FlowRunFilter, FlowRunFilterTags

TERMINAL = {"COMPLETED", "FAILED", "CANCELLED", "CRASHED"}

@flow
async def finalize_if_batch_done(batch_id: str):
    logger = get_run_logger()

    # Look up all runs with this tag
    async with get_client() as client:
        runs = await client.read_flow_runs(
            flow_run_filter=FlowRunFilter(
                tags=FlowRunFilterTags(all_=[batch_id])
            )
        )

    # Identify any non-terminal runs
    non_terminal = [
        r for r in runs
        if not getattr(r, "state", None) or r.state.type.value not in TERMINAL
    ]

    if non_terminal:
        <http://logger.info|logger.info>(f"Batch {batch_id}: {len(non_terminal)} runs still active; skipping.")
        return

    # All done — perform your one-time follow-up
    <http://logger.info|logger.info>(f"Batch {batch_id} is complete; running finalization.")
    # TODO: your single action here (e.g., trigger another deployment)

How to wire it up with Automations: - Create an event-based automation that listens for flow-run completed events filtered to runs that include the batch tag (in the UI: Match on Related > Tag equals your batch UUID; if you can, emit tags as “batch:<uuid>” to make matching/for_each grouping easier). - Add a small debounce: set within to e.g. 30–60s and schedule_after to e.g. PT30S. This reduces redundant triggers during bursts. - Action: RunDeployment for the checker, passing

batch_id

as a parameter (the batch UUID tag). The checker will no-op unless it’s the last one. Notes and caveats: - Because your batch size is unknown, you need a definition of “the batch is closed.” Without that, the checker can only use an inactivity window (debounce) to infer “no more runs are coming.” If you can produce an explicit “batch closed” signal (e.g., a tiny flow or event when you’re done launching runs), that’s more reliable. - If you’re worried about double-triggering the final action, add an idempotent guard (e.g., write a “finalized” flag to a durable store like S3/DynamoDB, or a Prefect Variable) and early-exit if it’s already set. - If you can switch to a parent flow pattern, you can do this cleanly with

run_deployment(..., tags=[batch_id])

and

wait_for_flow_run(...)

for each launched subflow. If you can share how you know a batch is “closed” (e.g., all expected runs have been scheduled vs. you rely on an inactivity timeout), I can sketch the exact automation config and the checker wiring for your setup.

Yes. In Prefect 3.x you can create deployments without installing your project’s dependencies (or even importing your flow code) in CI. Two common ways to do it: 1) Python API (source-first) - Point Prefect at your repo/storage and entrypoint; Prefect will extract flow metadata without executing your flow code if imports fail (it falls back to safe AST parsing). - Only Prefect (and any optional extras you use, e.g., prefect-aws) need to be installed in CI. Example (deploy from Git without your project deps installed):

# deploy.py - run this in CI with only `pip install prefect` (plus any extras you need)
from prefect import flow

if __name__ == "__main__":
    flow.from_source(
        source="<https://github.com/org/repo.git>",
        entrypoint="path/to/flow_file.py:my_flow",  # file.py:function
    ).deploy(
        name="my-deployment",
        work_pool_name="my-work-pool",
        # Prefer a pre-built runtime image with your deps, or install at runtime (see below)
        image="<http://123456.dkr.ecr.us-east-1.amazonaws.com/my-image:latest|123456.dkr.ecr.us-east-1.amazonaws.com/my-image:latest>",
        build=False,
        push=False,
        # For simple cases, you can install deps at runtime on the worker:
        # job_variables={"env": {"EXTRA_PIP_PACKAGES": "pandas requests"}}
    )

2) CLI in CI - If your CI has the repo checked out, you can run:

prefect deploy --entrypoint path/to/flow_file.py:my_flow --name my-deployment --pool my-work-pool

- Prefect will attempt to import, but if your project deps aren’t installed it will safely fall back and still create the deployment. You only need Prefect installed in CI. Where do the dependencies get installed? - At runtime, on the worker/infrastructure that executes your flow: - Best practice: use a pre-built image (Docker/K8s/ECS) that already contains your project deps and point the deployment at that image (

image=...

,

build=False

,

push=False

). - Alternatively: have the worker install deps at startup using

EXTRA_PIP_PACKAGES

via

job_variables

(supported by Docker/K8s workers). Example:

flow.from_source(
    source="<https://github.com/org/repo.git>",
    entrypoint="path/to/flow_file.py:my_flow",
).deploy(
    name="my-deployment",
    work_pool_name="docker-pool",
    job_variables={"env": {"EXTRA_PIP_PACKAGES": "pandas==2.2.2 boto3==1.34.*"}},
)

Notes for AWS CodeBuild/CI and Python 3.12 (Prefect 3.4.9): - In CI, install only what you need to define the deployment:

pip install prefect

(plus any extras for blocks you reference during deploy, e.g.,

pip install 'prefect[aws]'

if you use S3/Git blocks at deploy time). - Your execution environment (e.g., Docker image run by your worker, ECS task, or K8s pod) should include your project deps. If you can share which work pool/infrastructure you’re using (Docker, Kubernetes, ECS, etc.) and where your code lives (GitHub/GitLab/S3), I can give you an exact deploy snippet wired to that setup.

You’re right: when deploying via the Python API, there isn’t a parameter to directly set “pull steps.” That knob exists in prefect.yaml, but Flow.deploy()/flow.from_source(...).deploy(...) do not expose a pull_steps field. How to handle this in Python: - Use a storage object as the source. Prefect will generate the correct pull steps for you behind the scenes. - Git: pass a GitRepository (with optional credentials/branch/directories) to flow.from_source - Remote storage (S3/GCS/Azure): pass a RemoteStorage URL (e.g., s3://bucket/path) - Local path: pass a local directory string (mainly for Process workers) Examples: - Git repo (public or private with credentials block)

from prefect import flow
from prefect.runner.storage import GitRepository
from prefect_github import GitHubCredentials  # optional

github = GitRepository(
    url="<https://github.com/org/repo.git>",
    branch="main",
    # credentials=GitHubCredentials.load("my-gh-creds")  # if private
    # directories=["flows"]  # for sparse checkout
)

flow.from_source(
    source=github,
    entrypoint="flows/my_flow.py:my_flow",
).deploy(
    name="my-deployment",
    work_pool_name="my-pool",
)

- S3 (or other fsspec storage)

from prefect import flow
from prefect.runner.storage import RemoteStorage

flow.from_source(
    source=RemoteStorage(url="<s3://my-bucket/my-folder>"),
    entrypoint="flows/my_flow.py:my_flow",
).deploy(
    name="s3-deployment",
    work_pool_name="my-pool",
)

What if you truly need custom pull steps? - Use prefect.yaml. The YAML “pull” section lets you compose steps like git_clone, set_working_directory, pull_from_remote_storage, etc., then run

prefect deploy

. For complex sequences, YAML is the supported path today. - There is no supported public Python API to inject arbitrary pull steps directly; using internal/private classes to do so isn’t recommended (they can change between versions). Given your CI/CD constraints (no project venv), the storage-first approach is the intended Python path: Prefect will generate the appropriate pull logic from the storage object, and your workers will fetch code at runtime. If you share your storage and worker type (Docker/K8s/ECS/Process), I can tailor a snippet to your exact setup.