Devin Flake
10/21/2025, 9:46 PMMarvin
10/21/2025, 9:46 PMMarvin
10/21/2025, 9:53 PMparameters dictionary. You can read them via the Python API client or directly over HTTP. Note: there is no server-side filtering by parameter values; if you need to “filter by parameters,” fetch runs and filter client-side.
Key docs:
- API client guide: Prefect 3.x API client
- REST API: Read flow run and Filter/paginate flow runs
- Runtime access (inside a flow): prefect.runtime.flow_run
Python (sync) example
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterName
flow_run_id = "00000000-0000-0000-0000-000000000000"
with get_client(sync_client=True) as client:
# Read a single flow run and get its parameters
fr = client.read_flow_run(flow_run_id)
print("Parameters:", fr.parameters) # dict
# List flow runs (filter by other fields, e.g., name), then filter by parameter values in Python
runs = client.read_flow_runs(
flow_runs_filter=FlowRunFilter(name=FlowRunFilterName(any_=["etl-run"]))
)
us_runs = [r for r in runs if (r.parameters or {}).get("country") == "US"]
print("Count with country=US:", len(us_runs))
Python (async) example
import asyncio
from prefect.client.orchestration import get_client
async def main():
flow_run_id = "00000000-0000-0000-0000-000000000000"
async with get_client() as client:
fr = await client.read_flow_run(flow_run_id)
print("Parameters:", fr.parameters)
asyncio.run(main())
Raw HTTP examples
- Read a single flow run (Cloud URL shown; replace with your account/workspace and run ID; for OSS use your server’s base URL):
curl -H "Authorization: Bearer <PREFECT_API_KEY>" \
<https://api.prefect.cloud/api/accounts/<ACCOUNT_ID>/workspaces/<WORKSPACE_ID>/flow-runs/<FLOW_RUN_ID>>
Response will include a parameters object:
{
"id": "...",
"name": "...",
"state_type": "COMPLETED",
"parameters": {
"country": "US",
"date": "2025-10-01"
},
...
}
- List/filter flow runs by other fields (e.g., state), then filter parameters client-side:
curl -X POST -H "Content-Type: application/json" \
-H "Authorization: Bearer <PREFECT_API_KEY>" \
-d '{
"flow_runs": {
"state": { "type": { "any_": ["COMPLETED"] } }
},
"limit": 100
}' \
<https://api.prefect.cloud/api/accounts/<ACCOUNT_ID>/workspaces/<WORKSPACE_ID>/flow-runs/filter>
Then, in your app, iterate results and select those where parameters.country == "US".
Notes and caveats
- There is no server-side parameter filter in the FlowRun filters. Supported filters include id, name, tags, deployment_id, state fields, time ranges, etc. See: Filter classes.
- Parameters are JSON-serializable and stored with the flow run. Large or non-JSON-serializable objects should be avoided.
- Don’t confuse flow run parameters with Flow Run Input (real-time key/value inputs). Those are separate endpoints: Flow Run Input API.
- Inside a running flow, you can access parameters via prefect.runtime.flow_run.parameters.
If you can share whether you prefer Python or raw HTTP and what you want to filter by, I can tailor an example specifically to your use case.Devin Flake
10/21/2025, 10:00 PMMarvin
10/21/2025, 10:00 PMMarvin
10/21/2025, 10:01 PMfrom prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
FlowRunFilter, FlowRunFilterDeploymentId, FlowRunFilterStateType
)
deployment_id = "00000000-0000-0000-0000-000000000000"
with get_client(sync_client=True) as client:
# Narrow the search server-side to reduce payload size
base_filter = FlowRunFilter(
deployment_id=FlowRunFilterDeploymentId(any_=[deployment_id]),
state_type=FlowRunFilterStateType(any_=["COMPLETED"])
)
matches = []
limit = 200
offset = 0
while True:
page = client.read_flow_runs(
flow_runs_filter=base_filter,
limit=limit,
offset=offset,
)
if not page:
break
for run in page:
params = run.parameters or {}
# Example condition: country == "US" and batch == 42
if params.get("country") == "US" and params.get("batch") == 42:
matches.append(run)
offset += len(page)
print(f"Found {len(matches)} matching runs")
for r in matches[:5]:
print(r.id, r.name, r.parameters)
Python (async)
import asyncio
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
FlowRunFilter, FlowRunFilterDeploymentId, FlowRunFilterStateType
)
async def main():
deployment_id = "00000000-0000-0000-0000-000000000000"
async with get_client() as client:
base_filter = FlowRunFilter(
deployment_id=FlowRunFilterDeploymentId(any_=[deployment_id]),
state_type=FlowRunFilterStateType(any_=["COMPLETED"])
)
matches = []
limit = 200
offset = 0
while True:
page = await client.read_flow_runs(
flow_runs_filter=base_filter,
limit=limit,
offset=offset,
)
if not page:
break
for run in page:
params = run.parameters or {}
if params.get("country") == "US" and params.get("batch") == 42:
matches.append(run)
offset += len(page)
print(f"Found {len(matches)} matching runs")
asyncio.run(main())
Raw HTTP + client-side filtering (example with jq)
# Narrow by state or deployment server-side, then filter parameters client-side
curl -s -X POST \
-H "Content-Type: application/json" \
-H "Authorization: Bearer <PREFECT_API_KEY>" \
-d '{
"flow_runs": {
"state": { "type": { "any_": ["COMPLETED"] } }
},
"limit": 200
}' \
<https://api.prefect.cloud/api/accounts/<ACCOUNT_ID>/workspaces/<WORKSPACE_ID>/flow-runs/filter> \
| jq '.[] | select(.parameters.country=="US" and .parameters.batch==42)'
Useful docs:
- API client guide: Prefect 3.x API client
- REST API (read/paginate flow runs):
- Read flow run
- Filter/paginate flow runs
- Runtime access (inside a flow): prefect.runtime.flow_runMarvin
10/21/2025, 10:01 PM