<@ULVA73B9P> how can I query flow run parameters u...
# ask-marvin
d
@Marvin how can I query flow run parameters using the Prefect API?
m
thought for 457.4 seconds
Great question! In Prefect 3.x, flow run parameters are stored on the flow run record and are returned by the API as a
parameters
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
Copy code
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
Copy code
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):
Copy code
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:
Copy code
{
  "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:
Copy code
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.
d
@Marvin how can I query all flow runs for certain parameters? Can you give an example?
m
thought for 25.0 seconds
Short answer: In Prefect 3.x there isn’t a server-side filter for parameter values. The pattern is: 1) query flow runs using normal filters (e.g., by deployment, tags, state), and 2) filter by parameter values in your client code. If you need true server-side filtering, consider also recording key parameters as tags or emitting an artifact keyed by those values so you can filter by tags/artifacts in the API. Below are working examples you can copy/paste. Python (sync) — filter client-side by parameters
Copy code
from 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)
Copy code
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)
Copy code
# 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_run
If you share which parameters you care about and how large your result set is, I can tailor the server-side narrowing (filters, pagination strategy) and suggest a tagging/artifact approach if you want true server-side lookups by parameter values.