Thanks for sharing the snippet. A few things that commonly make

client.read_flow_runs

appear to “hang” in Prefect 3.x: - If your API URL isn’t set/reachable, the client will retry on timeouts and certain HTTP errors. With the default 60s request timeout and up to 5 retries with exponential backoff, a call can take several minutes before failing. - If

get_client()

has to discover the server (no

PREFECT_API_URL

set), it can add additional delay. - Large/slow queries can exceed the read timeout. Even with

limit=1

, connectivity issues will still trigger the retry logic. - If you’re not actually awaiting the coroutine (e.g., not using

await get_runs(...)

or

asyncio.run(...)

), nothing executes. Quick steps to debug and make it fail fast 1) Verify connectivity and the resolved API URL

import asyncio, os, httpx
from prefect.client.orchestration import get_client

async def check():
    # Fail fast while debugging
    os.environ["PREFECT_CLIENT_MAX_RETRIES"] = "0"

    async with get_client(
        httpx_settings={
            "timeout": httpx.Timeout(connect=5.0, read=20.0, write=10.0, pool=5.0)
        }
    ) as client:
        print("API URL:", client.api_url)
        # Health check returns an error string if unhealthy or None if healthy
        err = await client.api_healthcheck()
        print("Health:", "OK" if not err else f"Error: {err}")
        runs = await client.read_flow_runs(limit=1)
        print("Query OK, got", len(runs), "run(s)")

asyncio.run(check())

2) Ensure you’re actually awaiting your function - From sync code:

asyncio.run(get_runs(deployment_id))

- From async code:

await get_runs(deployment_id)

3) Pass explicit timeouts to your original code

import httpx

async def get_runs(deployment_id):
    async with get_client(
        httpx_settings={
            "timeout": httpx.Timeout(connect=5.0, read=30.0, write=10.0, pool=5.0)
        }
    ) as client:
        runs = await client.read_flow_runs(
            flow_run_filter=FlowRunFilter(
                deployment_id=FlowRunFilterDeploymentId(any_=[deployment_id]),
            ),
            sort=FlowRunSort.START_TIME_DESC,
            limit=1,
        )
        return runs

4) Set the API URL explicitly (recommended) - For Prefect Cloud, use your workspace URL (e.g., `https://api.prefect.cloud/api/accounts/<account_id>/workspaces/<workspace_id>`) - For self-hosted, your server URL (e.g.,

http://<host>:<port>/api

)

export PREFECT_API_URL="<https://api.prefect.cloud/api/accounts/.../workspaces/...>"
# optionally fail faster while debugging
export PREFECT_CLIENT_MAX_RETRIES=0
export PREFECT_API_REQUEST_TIMEOUT=30.0

Notes on your code - Your filter usage is correct for Prefect 3.x:

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

. - Importing

DeploymentFilter

isn’t needed here (that’s for listing deployments, not flow runs). -

FlowRunSort.START_TIME_DESC

is valid and not known to cause hangs. If this still “never stops,” please share: - Your Prefect version (

python -c "import prefect; print(prefect.__version__)"

) - Whether you’re targeting Prefect Cloud or self-hosted, and your

PREFECT_API_URL

- Where you’re running this (local, container, network with proxy) - Whether the quick health check above prints an error or times out I’ll help you pinpoint the exact cause from there.

Got it — and glad the URL issue is sorted. Short answer: there isn’t a server-side filter for parameters in Prefect 3.x. You’ll need to fetch runs and filter in Python, or encode the parameter into tags/idempotency_key when creating the run so you can filter server-side. Examples - Client-side filter by parameter value:

from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterDeploymentId
from prefect.client.schemas.sorting import FlowRunSort

async def get_runs_by_param(deployment_id, param_name, param_value, page_size=200):
    results = []
    offset = 0
    async with get_client() as client:
        while True:
            runs = await client.read_flow_runs(
                flow_run_filter=FlowRunFilter(
                    deployment_id=FlowRunFilterDeploymentId(any_=[deployment_id])
                ),
                sort=FlowRunSort.START_TIME_DESC,
                limit=page_size,
                offset=offset,
            )
            if not runs:
                break
            for r in runs:
                if (r.parameters or {}).get(param_name) == param_value:
                    results.append(r)
            offset += page_size
    return results

- Filter where a parameter is present (regardless of value):

def has_param(run, name): 
    return run.parameters is not None and name in run.parameters

- Reduce scan size by adding time window (optional):

from datetime import datetime, timedelta, timezone
from prefect.client.schemas.filters import FlowRunFilterStartTime

time_filter = FlowRunFilterStartTime(
    after_=(datetime.now(timezone.utc) - timedelta(days=7))
)

runs = await client.read_flow_runs(
    flow_run_filter=FlowRunFilter(
        deployment_id=FlowRunFilterDeploymentId(any_=[deployment_id]),
        start_time=time_filter,
    ),
    limit=200,
)

If you can influence how runs are created: - Add a tag derived from the parameter (e.g.,

param:customer=acme

) and then filter with

FlowRunFilterTags

. - Or set an

idempotency_key

based on the parameter set and filter by

FlowRunFilterIdempotencyKey

. If you share your parameter shape and the selection rule (equals, contains, etc.), I can tailor a minimal snippet.