Short answer: Use strict JSON. In the Prefect 2 Cloud UI Run modal, parameter values are parsed as JSON. For a list of dicts: - If you’re editing the single parameter field for your list param (e.g.,

records

), paste just the value:

[{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]

- If you’re using the “Custom” parameters editor (full params object), include the parameter name:

{
  "records": [
    {
      "id": 1,
      "name": "Alice"
    },
    {
      "id": 2,
      "name": "Bob"
    }
  ]
}

Gotchas: - Use double quotes for strings:

"text"

(not single quotes) - Booleans/null must be JSON: `true`/`false`/`null` (not Python `True`/`False`/`None`) - No trailing commas References: - Prefect Cloud API docs (parameters are sent as JSON) - Flows: Parameters (general guidance) If you share your parameter name(s), I can show the exact JSON you’d paste for your flow.

That error means the JSON-serialized parameters for the flow run exceed Prefect’s 512 KB limit. In Prefect 2, parameters are intended for small configs, not large payloads. Recommended pattern: - Put the large data somewhere durable (GCS in your case) - Pass a small reference (URI/path) as the parameter - Load the data inside the flow at runtime Example using prefect-gcp tasks:

from prefect import flow, task
from prefect_gcp.cloud_storage import cloud_storage_download_blob_as_bytes
import json

def split_gs_uri(uri: str):
    assert uri.startswith("gs://"), "Expected <gs://bucket/path>"
    path = uri[5:]
    bucket, blob = path.split("/", 1)
    return bucket, blob

@flow
def process_records(gcs_uri: str):
    bucket, blob = split_gs_uri(gcs_uri)
    data_bytes = cloud_storage_download_blob_as_bytes(bucket=bucket, blob=blob)
    records = json.loads(data_bytes.decode("utf-8"))
    # use records here...
    return len(records)

In the UI Run modal, pass:

{
  "gcs_uri": "<gs://my-bucket/path/to/records.json>"
}

If you need to upload the data first (outside the flow), stage it to GCS however you prefer (CLI, app code, etc.). If you want to do it within Prefect, the collection also provides helpers like: -

cloud_storage_upload_blob_from_string

-

cloud_storage_upload_blob_from_file

Docs: - prefect-gcp: Cloud Storage docs - Flows: Parameters If you share how you’re generating the list and triggering the run, I can tailor an example (e.g., writing the JSON to GCS, which parameter shape to use, and the flow code to read it).

Great question. In Prefect 2, parameters are meant for small config — not bulk data — which is why you hit the 512 KB limit. The best approaches for “local data” depend on how you want to run: Option A — Run the flow locally with the data in-memory - Ideal for development or ad-hoc runs where the code and data are on your machine. - Call the flow like a normal function; no parameter size constraints because you’re not sending the payload to the API.

from prefect import flow
import json

@flow
def process_records(records: list[dict]):
    # do work
    return len(records)

if __name__ == "__main__":
    with open("records.json") as f:
        data = json.load(f)
    process_records(data)  # runs locally

Tip: If you’ve done

prefect cloud login

, this local run can still report to Cloud for visibility. Option B — Stage the data and pass a small reference - Best for scheduled/remote runs (workers on GCP, Cloud UI, etc.). - Upload your local file to GCS and pass a

gs://...

URI as a parameter. The flow loads it at runtime. Upload from your machine (any one of these): - gsutil:

gsutil cp ./records.json <gs://my-bucket/path/records.json>

- Prefect task:

from prefect_gcp.cloud_storage import cloud_storage_upload_blob_from_file

cloud_storage_upload_blob_from_file(
    file="records.json",
    bucket="my-bucket",
    blob="path/records.json",
)

Flow reads from GCS:

from prefect import flow
from prefect_gcp.cloud_storage import cloud_storage_download_blob_as_bytes
import json

def split_gs_uri(uri: str):
    assert uri.startswith("gs://")
    bucket_blob = uri[5:]
    bucket, blob = bucket_blob.split("/", 1)
    return bucket, blob

@flow
def process_records(gcs_uri: str):
    bucket, blob = split_gs_uri(gcs_uri)
    data_bytes = cloud_storage_download_blob_as_bytes(bucket=bucket, blob=blob)
    records = json.loads(data_bytes.decode())
    # do work...
    return len(records)

In the Cloud UI Run modal use JSON:

{
  "gcs_uri": "<gs://my-bucket/path/records.json>"
}

Docs: prefect-gcp: Cloud Storage Option C — Pass a local file path that the runtime can see - If your worker/container has a mounted volume or shared path, pass a path like

/data/records.json

as the parameter and read it in the flow. - Useful if you’re running on Docker/Kubernetes with volume mounts. Be sure your job template mounts the same path at runtime. Option D — Use a data system reference - On GCP, consider putting the data in BigQuery or referencing a GCS prefix, then pass a small identifier like

project.dataset.table

or

<gs://bucket/prefix/>

. If you tell me: - Where your “local data” lives (your laptop, a VM, CI, etc.) - How you trigger runs (UI, CLI, API) - Where the flow executes (local machine, Docker, K8s, Cloud Run) I’ll recommend the cleanest setup and share exact code/commands for your case.

Great — since the JSON file is on your laptop and you want to run the flow on your machine, you have two good options: Option 1: Run the flow locally and pass the data directly (simplest) - This avoids the 512KB parameter limit because nothing is sent as parameters; you’re just calling a Python function. - If you want visibility in Prefect Cloud, log in first (

prefect cloud login

) then run your script — the run will show up in Cloud. Example:

from prefect import flow
import json
from pathlib import Path

@flow
def process_records(records: list[dict]):
    # do work here...
    return len(records)

if __name__ == "__main__":
    records = json.loads(Path("records.json").read_text())
    process_records(records)

Option 2: Trigger from the UI but still run on your laptop (local worker) - Create a Process work pool in the Cloud UI (Work Pools -> Create -> Type: Process). - Start a local worker that polls that pool:

prefect worker start --pool <your-pool-name>

- Create/point a deployment to that pool. - Pass a small parameter (a file path) in the UI, and read the file inside the flow. Use an absolute path so the worker can find it. Flow example:

from prefect import flow
import json
from pathlib import Path

@flow
def process_records(file_path: str):
    data = json.loads(Path(file_path).read_text())
    # do work...
    return len(data)

UI Run modal parameters:

{
  "file_path": "/absolute/path/to/records.json"
}

Notes: - Avoid passing the JSON content as a parameter (512KB limit). - Absolute paths are safest; the worker’s working directory may differ from your shell. - If you only need local development/testing, Option 1 is fastest. If you want to schedule/trigger from Cloud, go with Option 2. If you share your current flow code (or whether you already have a deployment), I can tailor the exact steps and commands.

Got it — let’s get Option 1 working. A few quick questions to pinpoint the issue: - What’s the exact error/stack trace you see? - How are you running it (python script, notebook, IDE)? - Do you expect to see the run in Prefect Cloud, and have you done

prefect cloud login

? Here’s a minimal, copy/paste example that should work on your laptop and (optionally) report to Cloud if you’re logged in:

from prefect import flow, get_run_logger
import json
from pathlib import Path

@flow(log_prints=True)
def process_records_from_file(file_path: str):
    logger = get_run_logger()
    p = Path(file_path).expanduser().resolve()
    <http://logger.info|logger.info>(f"Loading {p}")
    records = json.loads(p.read_text())
    <http://logger.info|logger.info>(f"Loaded {len(records)} records")
    # do work...
    return len(records)

if __name__ == "__main__":
    # Use an absolute path if you get FileNotFoundError
    process_records_from_file("records.json")

Run it:

# (Optional) to see runs in Cloud
prefect cloud login --key <YOUR_API_KEY> --workspace <account>/<workspace>

python your_script.py

If this still “doesn’t work”, please share: - The exact error/traceback - Output of

prefect version

- Whether

records.json

exists at the path you pass (try an absolute path) - Whether you expect it to appear in Cloud (and have run

prefect cloud login

) Common fixes: - Use an absolute file path (

/Users/you/path/records.json

) to avoid working-directory issues - Ensure the file is valid JSON (no trailing commas, proper quotes) - If you’re in a notebook, still call the flow like a function:

process_records_from_file("/abs/path.json")

- For extra visibility, set

PREFECT_LOGGING_LEVEL=DEBUG

before running If you paste your current code snippet and error, I’ll tailor the exact fix.

Thanks — that error tells me your flow is still receiving the entire JSON payload as a parameter. Even when you “run locally,” if you’re logged into Prefect Cloud, Prefect 2 sends the flow parameters to Cloud — and anything >512KB is rejected. Two easy fixes: A) Keep Cloud visibility, but don’t pass large data as a parameter - Change your flow to accept a small reference (e.g., a file path), and load the JSON inside the flow.

from prefect import flow, get_run_logger
from pathlib import Path
import json

@flow(log_prints=True)
def process_records(file_path: str):
    logger = get_run_logger()
    p = Path(file_path).expanduser().resolve()
    data = json.loads(p.read_text())
    <http://logger.info|logger.info>(f"Loaded {len(data)} records")
    # do work...
    return len(data)

if __name__ == "__main__":
    process_records("/absolute/path/to/records.json")

B) Run fully offline (no Cloud) for this local test - This avoids sending parameters anywhere, so large in-memory args are fine. - Temporarily disconnect from Cloud for this run: - Easiest: use a local profile -

prefect profile create local

-

prefect profile use local

- Run your script - When done:

prefect profile use <your-cloud-profile>

- Or temporarily unset the API URL just for the command: - On macOS/Linux:

PREFECT_API_URL= python your_script.py

- On Windows (PowerShell):

$env:PREFECT_API_URL=''; python your_script.py

If you prefer to trigger from the Cloud UI, keep your worker running locally but pass only a file path parameter in the UI:

{
  "file_path": "/absolute/path/to/records.json"
}

If you share your current flow function signature and how you’re calling it, I can point out the minimal change needed.

Thanks — that explains it. When you’re logged into Prefect Cloud, even “local” flow runs are orchestrated by Cloud and the parameters are sent to the API. So the 512 KB limit still applies, which is why Option 1 didn’t work if you passed the whole JSON payload as a parameter. You’ve got two clean options: Option A — Keep using Cloud, but pass a small reference (file path) instead of the data - Change your flow to accept a file path and load the JSON inside the flow. The parameter stays tiny and your run still shows up in Cloud.

from prefect import flow, get_run_logger
from pathlib import Path
import json

@flow(log_prints=True)
def process_records_from_file(file_path: str):
    logger = get_run_logger()
    p = Path(file_path).expanduser().resolve()
    <http://logger.info|logger.info>(f"Loading {p}")
    records = json.loads(p.read_text())
    <http://logger.info|logger.info>(f"Loaded {len(records)} records")
    # do work...
    return len(records)

if __name__ == "__main__":
    process_records_from_file("/absolute/path/to/records.json")

Option B — Run fully offline to bypass the limit (won’t appear in Cloud) - If you absolutely must pass the large object directly, disable orchestration for the run so nothing is sent to Cloud:

PREFECT_API_ENABLE_ORCHESTRATION=false python your_script.py

or set it in code before the flow is imported/created:

import os
os.environ["PREFECT_API_ENABLE_ORCHESTRATION"] = "false"
from your_module import process_records  # then call your flow

Recommendation: Use Option A. It keeps Cloud visibility and avoids size limits. If you later want to run remotely, you can swap the local file path for a

gs://...

URI and load from GCS the same way. If you share your current flow signature, I can show the minimal change to switch it to

file_path

and keep everything else the same.