Short answer: in Prefect 2.x, “active” typically means task runs in the RUNNING state. You can get that count either by calling the REST count endpoint (most efficient) or by listing and counting. Recommended (fast) — use the REST count endpoint via the Python client

import asyncio
from prefect.client.orchestration import get_client

async def count_active_task_runs(flow_run_ids=None, deployment_ids=None):
    # Build filters; include only what you need
    payload = {
        "flows": None,
        "flow_runs": {"id": {"any_": [str(i) for i in flow_run_ids]}} if flow_run_ids else None,
        "task_runs": {"state": {"type": {"any_": ["RUNNING"]}}},
        "deployments": {"id": {"any_": [str(i) for i in deployment_ids]}} if deployment_ids else None,
    }
    async with get_client() as client:
        resp = await <http://client._client.post|client._client.post>("/task_runs/count", json=payload)
        return resp.json()  # integer

print(asyncio.run(count_active_task_runs()))  # count across your workspace

Within a running flow (scope to this flow run)

from prefect import flow
from prefect.context import get_run_context
from prefect.client.orchestration import get_client
import asyncio

@flow
def my_flow():
    ctx = get_run_context()
    flow_run_id = str(ctx.flow_run.id)

    async def count_active_here():
        async with get_client() as client:
            payload = {
                "flow_runs": {"id": {"any_": [flow_run_id]}},
                "task_runs": {"state": {"type": {"any_": ["RUNNING"]}}},
            }
            r = await <http://client._client.post|client._client.post>("/task_runs/count", json=payload)
            return r.json()

    print("Active tasks in this flow run:", asyncio.run(count_active_here()))

Alternative (simpler but less efficient at scale) — list and count

import asyncio
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import TaskRunFilter, TaskRunFilterState, TaskRunFilterStateType
from prefect.client.schemas.objects import StateType

async def count_by_listing():
    async with get_client() as client:
        runs = await client.read_task_runs(
            task_run_filter=TaskRunFilter(
                state=TaskRunFilterState(
                    type=TaskRunFilterStateType(any_=[StateType.RUNNING])
                )
            ),
            limit=10000  # set high enough or paginate
        )
        return len(runs)

print(asyncio.run(count_by_listing()))

Optional: curl against the API directly (Cloud or Server)

# Cloud: set PREFECT_API_KEY and use your account/workspace route
curl -X POST \
  "<https://api.prefect.cloud/api/accounts/<ACCOUNT_ID>/workspaces/<WORKSPACE_ID>/task_runs/count>" \
  -H "Authorization: Bearer $PREFECT_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"task_runs": {"state": {"type": {"any_": ["RUNNING"]}}}}'

# Local server:
curl -X POST "<http://127.0.0.1:4200/api/task_runs/count>" \
  -H "Content-Type: application/json" \
  -d '{"task_runs": {"state": {"type": {"any_": ["RUNNING"]}}}}'

Notes - If your definition of “active” includes other states (e.g., CANCELLING, PENDING, SCHEDULED), add them to

any_

, e.g.

["RUNNING", "CANCELLING"]

. - The Python client in Prefect 2.x does not have a dedicated

count_task_runs()

helper; the REST

/task_runs/count

endpoint is the most accurate/performant way to get counts. Would you like this scoped to a specific deployment, flow, or just your whole workspace?