Got it! Here’s a very simple script that uses the Prefect 3.x Python client to fetch all flow runs for a given deployment ID. It paginates until it retrieves them all. Notes: - Set

PREFECT_API_URL

and (for Cloud)

PREFECT_API_KEY

in your environment, or ensure you’re logged in via

prefect cloud login

. - Provide the deployment ID via the

DEPLOYMENT_ID

environment variable.

import os
from uuid import UUID
from prefect.client.orchestration import SyncPrefectClient
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterDeploymentId

def main():
    deployment_id = os.environ.get("DEPLOYMENT_ID")
    if not deployment_id:
        raise SystemExit("Please set the DEPLOYMENT_ID environment variable (UUID of the deployment).")

    # Build a filter to select flow runs for the given deployment
    fr_filter = FlowRunFilter(
        deployment_id=FlowRunFilterDeploymentId(any_=[UUID(deployment_id)])
    )

    all_runs = []
    limit = 200
    offset = 0

    with SyncPrefectClient() as client:
        while True:
            runs = client.read_flow_runs(
                flow_run_filter=fr_filter,
                limit=limit,
                offset=offset,
            )
            if not runs:
                break
            all_runs.extend(runs)
            offset += len(runs)

    print(f"Fetched {len(all_runs)} flow runs for deployment {deployment_id}")
    for r in all_runs:
        # Keep output simple and robust across schema types
        name = getattr(r, "name", None)
        print(f"{r.id}\t{name}")

if __name__ == "__main__":
    main()

If you’d prefer to pass the deployment ID as a CLI arg instead of an env var, I can tweak this to use

argparse

. References: - Prefect 3.x API client guide - PrefectClient API reference

Thanks for the report — that error happens because

SyncPrefectClient

in Prefect 3.x requires an

api

argument. The simplest fix is to use

get_client(sync_client=True)

, which reads your API URL and credentials from the environment. Here’s a corrected minimal script:

import os
from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterDeploymentId

def main():
    deployment_id = os.environ.get("DEPLOYMENT_ID")
    if not deployment_id:
        raise SystemExit("Please set the DEPLOYMENT_ID environment variable (UUID of the deployment).")

    fr_filter = FlowRunFilter(
        deployment_id=FlowRunFilterDeploymentId(any_=[UUID(deployment_id)])
    )

    all_runs = []
    limit = 200
    offset = 0

    # Automatically uses PREFECT_API_URL and PREFECT_API_KEY if set
    with get_client(sync_client=True) as client:
        while True:
            runs = client.read_flow_runs(
                flow_run_filter=fr_filter,
                limit=limit,
                offset=offset,
            )
            if not runs:
                break
            all_runs.extend(runs)
            offset += len(runs)

    print(f"Fetched {len(all_runs)} flow runs for deployment {deployment_id}")
    for r in all_runs:
        print(f"{r.id}\t{getattr(r, 'name', '')}")

if __name__ == "__main__":
    main()

If you prefer to use

SyncPrefectClient

directly, pass the API URL (and API key for Cloud):

import os
from uuid import UUID
from prefect.client.orchestration import SyncPrefectClient
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterDeploymentId

api_url = os.environ["PREFECT_API_URL"]  # e.g. <https://api.prefect.cloud/api/accounts/.../workspaces/>...
api_key = os.environ.get("PREFECT_API_KEY")

with SyncPrefectClient(api=api_url, api_key=api_key) as client:
    fr_filter = FlowRunFilter(
        deployment_id=FlowRunFilterDeploymentId(any_=[UUID(os.environ["DEPLOYMENT_ID"])])
    )
    runs = client.read_flow_runs(flow_run_filter=fr_filter, limit=200, offset=0)
    for r in runs:
        print(r.id, getattr(r, "name", ""))

Docs: - Prefect 3.x API client guide - Client API reference Want me to switch this to accept a deployment name (flow/deployment) instead of a UUID?