Here’s a Prefect 3.6.29 flow that checks all deployments across all work pools and alerts if a deployment has no active schedule. This uses the Prefect client directly and optionally sends a Slack notification via a

SlackWebhook

block.

python
from prefect import flow, task, get_run_logger
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import WorkPoolFilter
from prefect.blocks.notifications import SlackWebhook


@task
async def send_slack_alert(message: str, block_name: str) -> None:
    slack_webhook = await SlackWebhook.load(block_name)
    await slack_webhook.notify(message)


@flow
async def alert_on_unscheduled_deployments(
    slack_webhook_block_name: str | None = None,
) -> list[dict]:
    """
    Alert if any deployment in any work pool does not have an active schedule.

    A deployment is considered non-compliant if:
    - it has no schedules, or
    - all of its schedules are inactive
    """
    logger = get_run_logger()
    missing_active_schedule = []

    async with get_client() as client:
        work_pools = await client.read_work_pools()

        for pool in work_pools:
            deployments = await client.read_deployments(
                work_pool_filter=WorkPoolFilter(
                    name={"any_": [pool.name]}
                )
            )

            for deployment in deployments:
                schedules = deployment.schedules or []
                has_active_schedule = any(schedule.active for schedule in schedules)

                if not has_active_schedule:
                    missing_active_schedule.append(
                        {
                            "work_pool": pool.name,
                            "deployment_name": deployment.name,
                            "deployment_id": str(deployment.id),
                            "schedule_count": len(schedules),
                        }
                    )

    if missing_active_schedule:
        lines = [
            "*Deployments without an active schedule:*",
            "",
        ]
        for item in missing_active_schedule:
            lines.append(
                f"- work pool `{item['work_pool']}` | deployment `{item['deployment_name']}` "
                f"| id `{item['deployment_id']}` | schedules: {item['schedule_count']}"
            )

        message = "\n".join(lines)

        logger.warning(message)

        if slack_webhook_block_name:
            await send_slack_alert(message, slack_webhook_block_name)

    else:
        <http://logger.info|logger.info>("All deployments in all work pools have at least one active schedule.")

    return missing_active_schedule


if __name__ == "__main__":
    import asyncio

    asyncio.run(
        alert_on_unscheduled_deployments(
            slack_webhook_block_name=None  # e.g. "my-slack-webhook"
        )
    )

A few notes: - In Prefect 3.x, deployment schedules are available directly on

deployment.schedules

. - A deployment fails this check if: -

deployment.schedules

is empty, or - every schedule has

active=False

. -

SlackWebhook

is optional. If you don’t want Slack, you can remove that part and just rely on flow logs / state. If you want, I can also give you: - a version that fails the flow when any deployment is missing an active schedule - a version that uses email / Teams / PagerDuty instead of Slack - a

prefect deploy

config example to run this audit on a schedule

Yes — you can do this, but there’s an important caveat: Prefect does not have an official “compare server deployments to

prefect.yaml

in Git” API. So the practical approach is: 1. read all deployments from Prefect server 2. fetch

prefect.yaml

from your Git repo 3. parse the YAML yourself 4. compare the deployment definitions in Git vs the deployments registered in Prefect 5. alert on anything in Prefect that is not in the YAML For matching, the most reliable key is usually: -

deployment.name

-

deployment.entrypoint

That’s better than matching only on deployment name, since names can repeat across flows. Here’s a complete flow for Prefect 3.6.29. ```python from future import annotations import base64 from typing import Any import httpx import yaml from prefect import flow, task, get_run_logger from prefect.client.orchestration import get_client from prefect.blocks.notifications import SlackWebhook @task async def fetch_prefect_yaml_from_github_raw(raw_url: str) -> dict[str, Any]: """ Fetch prefect.yaml from a raw Git URL, for example: https://raw.githubusercontent.com/<org>/<repo>/<branch>/prefect.yaml """ async with httpx.AsyncClient(timeout=30.0) as client: response = await client.get(raw_url) response.raise_for_status() return yaml.safe_load(response.text) @task async def fetch_prefect_yaml_from_github_api( owner: str, repo: str, path: str = "prefect.yaml", ref: str = "main", token: str | None = None, ) -> dict[str, Any]: """ Fetch prefect.yaml using the GitHub contents API. Useful for private repos if you provide a GitHub token. """ url = f"https://api.github.com/repos/{owner}/{repo}/contents/{path}?ref={ref}" headers = {} if token: headers["Authorization"] = f"Bearer {token}" async with httpx.AsyncClient(timeout=30.0) as client: response = await client.get(url, headers=headers) response.raise_for_status() payload = response.json() content = base64.b64decode(payload["content"]).decode() return yaml.safe_load(content) def normalize_entrypoint(entrypoint: str | None) -> str: return (entrypoint or "").strip() def extract_yaml_deployment_keys(prefect_yaml: dict[str, Any]) -> set[tuple[str, str]]: """ Build a set of keys from prefect.yaml: (deployment_name, entrypoint) Expected prefect.yaml structure includes: deployments: - name: ... entrypoint: path/to/file.py:flow_func """ keys: set[tuple[str, str]] = set() for deployment in prefect_yaml.get("deployments", []) or []: name = (deployment.get("name") or "").strip() entrypoint = normalize_entrypoint(deployment.get("entrypoint")) if name and entrypoint: keys.add((name, entrypoint)) return keys @task async def find_orphaned_deployments( yaml_keys: set[tuple[str, str]], ) -> list[dict[str, str]]: """ Return deployments that exist in Prefect server but are not present in prefect.yaml. """ orphaned: list[dict[str, str]] = [] async with get_client() as client: deployments = await client.read_deployments() for deployment in deployments: deployment_name = (deployment.name or "").strip() entrypoint = normalize_entrypoint(getattr(deployment, "entrypoint", None)) key = (deployment_name, entrypoint) if key not in yaml_keys: orphaned.append( { "deployment_name": deployment_name, "entrypoint": entrypoint, "flow_id": str(deployment.flow_id), "deployment_id": str(deployment.id), "work_pool_name": getattr(deployment, "work_pool_name", "") or "", } ) return orphaned @task async def send_slack_alert(block_name: str, message: str) -> None:

slack = await SlackWebhook.load(block_name) await slack.notify(message) @flow async def detect_orphaned_deployments( github_raw_prefect_yaml_url: str | None = None, github_owner: str | None = None, github_repo: str | None = None, github_ref: str = "main", github_path: str = "prefect.yaml", github_token: str | None = None, slack_webhook_block_name: str | None = None, ) -> list[dict[str, str]]: """ Detect deployments that exist in Prefect server but are not defined in prefect.yaml in Git. Use one of: - github_raw_prefect_yaml_url - github_owner + github_repo (+ optional token for private repos) """ logger = get_run_logger() if github_raw_prefect_yaml_url: prefect_yaml = await fetch_prefect_yaml_from_github_raw( github_raw_prefect_yaml_url ) elif github_owner and github_repo: prefect_yaml = await fetch_prefect_yaml_from_github_api( owner=github_owner, repo=github_repo, path=github_path, ref=github_ref, token=github_token, ) else: raise ValueError( "Provide either github_raw_prefect_yaml_url or github_owner/github_repo." ) yaml_keys = extract_yaml_deployment_keys(prefect_yaml) logger.info(f"Found {len(yaml_keys)} deployment definitions in prefect.yaml") orphaned = await find_orphaned_deployments(yaml_keys) if orphaned: lines = [ "*Orphaned deployments found in Prefect server*", "These deployments exist in Prefect but are not defined in `prefect.yaml`:", "", ] for dep in orphaned: lines.append( f"- deployment

{dep['deployment_name']}

| " f"entrypoint

{dep['entrypoint']}

| " f"work_pool

{dep['work_pool_name']}

| " f"id `{dep['deployment_id']}`" ) message = "\n".join(lines) logger.warning(message) if slack_webhook_block_name: await send_slack_alert(slack_webhook_block_name, message) else: logger.info("No orphaned deployments found.") return orphaned

Example usage with a raw GitHub URL:

```python
if __name__ == "__main__":
    import asyncio

    asyncio.run(
        detect_orphaned_deployments(
            github_raw_prefect_yaml_url="<https://raw.githubusercontent.com/my-org/my-repo/main/prefect.yaml>",
            slack_webhook_block_name="my-slack-webhook",
        )
    )

Example usage with GitHub API for a private repo:

python
if __name__ == "__main__":
    import asyncio

    asyncio.run(
        detect_orphaned_deployments(
            github_owner="my-org",
            github_repo="my-repo",
            github_ref="main",
            github_path="prefect.yaml",
            github_token="ghp_xxx",
            slack_webhook_block_name="my-slack-webhook",
        )
    )

Important caveats - This assumes your

prefect.yaml

has deployment definitions with

name

and

entrypoint

. - Matching on

(name, entrypoint)

is recommended. - If a deployment was created manually or by another automation path, it will be reported as orphaned even if that was intentional. - If you want stricter matching, you could also compare: -

work_pool_name

-

tags

-

parameters

-

path

-

read_deployments()

may be paginated depending on your environment; if you expect many deployments, I can give you a paginated version. If you want, I can also give you: - a version that clones the repo instead of fetching raw YAML - a version that fails the flow when orphaned deployments are found - a version that deletes orphaned deployments after reporting them