Got it—assuming you’re on Prefect 3.x and you want “scheduled” flow runs whose expected start time is before now, and that belong to the same flow as the code that’s running. Here are two common ways to scope “from the current flow”: Option A: Same flow (by flow name) - Good if you want all runs of this flow, regardless of deployment

import asyncio
from datetime import datetime, timezone

from prefect.client.orchestration import PrefectClient
from prefect.client.schemas.filters import (
    FilterSet,
    FlowFilter,
    FlowFilterName,
    FlowRunFilter,
    FlowRunFilterExpectedStartTime,
    FlowRunFilterStateType,
)
from prefect.client.schemas.objects import StateType
from prefect.runtime import flow_run as rt_flow_run

async def scheduled_runs_for_this_flow_before_now(limit=200):
    now = datetime.now(timezone.utc)

    async with PrefectClient() as client:
        filters = FilterSet(
            flows=FlowFilter(
                name=FlowFilterName(any_=[rt_flow_run.flow_name.get()])
            ),
            flow_runs=FlowRunFilter(
                state=FlowRunFilterStateType(any_=[StateType.SCHEDULED]),
                expected_start_time=FlowRunFilterExpectedStartTime(before_=now),
            ),
        )
        return await client.read_flow_runs(limit=limit, filters=filters)

runs = asyncio.run(scheduled_runs_for_this_flow_before_now())
for r in runs:
    print(r.id, r.expected_start_time, r.state.type)

Option B: Same deployment (most common when you mean “this scheduled series”) - Good if you only want runs created by the current deployment’s schedule

import asyncio
from datetime import datetime, timezone

from prefect.client.orchestration import PrefectClient
from prefect.client.schemas.filters import (
    FilterSet,
    FlowRunFilter,
    FlowRunFilterDeploymentId,
    FlowRunFilterExpectedStartTime,
    FlowRunFilterStateType,
)
from prefect.client.schemas.objects import StateType
from prefect.runtime import deployment as rt_deployment

async def scheduled_runs_for_this_deployment_before_now(limit=200):
    now = datetime.now(timezone.utc)

    async with PrefectClient() as client:
        filters = FilterSet(
            flow_runs=FlowRunFilter(
                deployment_id=FlowRunFilterDeploymentId(any_=[rt_deployment.id.get()]),
                state=FlowRunFilterStateType(any_=[StateType.SCHEDULED]),
                expected_start_time=FlowRunFilterExpectedStartTime(before_=now),
            ),
        )
        return await client.read_flow_runs(limit=limit, filters=filters)

runs = asyncio.run(scheduled_runs_for_this_deployment_before_now())
for r in runs:
    print(r.id, r.expected_start_time, r.state.type)

Notes: - Use timezone-aware datetimes (UTC) for time filters. - If you prefer sync code, swap in

from prefect.client.orchestration import SyncPrefectClient

and use

with SyncPrefectClient() as client:

then

client.read_flow_runs(...)

. - If you meant “project” as in Prefect 1.x, in Prefect 3.x that’s a workspace; the filters above scope by flow or deployment instead. Quick check: Do you want to scope by the same flow or by the same deployment? And are you running this from within a flow run (so

prefect.runtime.*

is available), or from a separate script?