You’re hitting a hard limit enforced by Prefect Cloud: flow run parameters must be ≤ 512 KB after JSON serialization. This is why large dicts/DataFrames/etc. sent as parameters fail. It’s documented here: Flow parameters (Prefect 2) How to fix it (GCP-friendly patterns): - Store the big payload outside Prefect and pass a small reference - GCS: upload the payload to a bucket and pass a

gs://...

path as a parameter - BigQuery: write data to a table and pass the table ID - For moderately sized configuration, use Variables (limit ~5,000 chars) and fetch inside the flow - Docs: Variables Example: write JSON to GCS and pass a reference

from uuid import uuid4
import json
from prefect import flow, task
from prefect_gcp.credentials import GcpCredentials
from prefect_gcp.cloud_storage import (
    cloud_storage_upload_blob_from_string,
    cloud_storage_download_blob_as_bytes,
)

# Create these blocks once (UI or code), then load by name
CREDS_BLOCK = "my-gcp-creds"   # prefect_gcp.credentials.GcpCredentials block name
BUCKET = "my-bucket"

@task
def upload_payload_to_gcs(payload: dict) -> str:
    gcp_credentials = GcpCredentials.load(CREDS_BLOCK)
    blob = f"payloads/{uuid4()}.json"
    cloud_storage_upload_blob_from_string(
        data=json.dumps(payload),
        bucket=BUCKET,
        blob=blob,
        gcp_credentials=gcp_credentials,
        content_type="application/json",
    )
    return f"gs://{BUCKET}/{blob}"

@task
def download_payload_from_gcs(gs_uri: str) -> dict:
    _, bucket, *path = gs_uri.split("/", 3)
    blob = path[-1]
    gcp_credentials = GcpCredentials.load(CREDS_BLOCK)
    b = cloud_storage_download_blob_as_bytes(
        bucket=bucket,
        blob=blob,
        gcp_credentials=gcp_credentials,
    )
    return json.loads(b.decode())

@flow
def my_flow(gs_uri: str):
    payload = download_payload_from_gcs(gs_uri)
    # ...process payload...

# In your trigger path (outside Prefect), upload first and pass only the URI:
# gs_uri = upload_payload_to_gcs.submit(huge_payload).result()  # if done in Prefect
# Or upload from your app/ETL code via GCP SDK, then call the deployment with {"gs_uri": "..."}

Variables for config (small-to-medium)

import json
from prefect import flow
from prefect.variables import Variable

@flow
def use_config():
    var = Variable.get("app_config")  # value is a string
    cfg = json.loads(var.value)       # keep under ~5,000 chars
    # use cfg...

Notes: - Compressing the parameter won’t help if you still send it as a parameter; the 512 KB limit applies after serialization sent to the API. - Prefect Server (self-hosted) is more permissive, but Prefect Cloud enforces this limit. If you can share what you’re passing (type and approximate size) and how you trigger the run (UI, API, Deployment call), I can suggest the most direct refactor for your setup. Helpful docs: - Flow parameters (Prefect 2) - prefect-gcp collection docs

Short answer: use a JSON block for anything structured or > ~5 KB; use a Variable only for small string-y config. For secrets, use a Secret block. Why: - Variable - String only, ~5,000 character limit - Simple for small flags or short JSON strings - Not for secrets - Docs: Variables - JSON block - Stores any JSON-compatible structure (dicts/lists), no tiny 5 KB limit - Great for nested configs and larger payloads - Not for secrets - Docs: Blocks GCP-friendly note: - For truly large data (MBs+), prefer GCS. Put the file in GCS and pass a

gs://...

path as a parameter or store that path in a JSON block. Examples Using a JSON block for config

from prefect.blocks.system import JSON

# Create/update once
cfg = JSON(value={"db": {"host": "x", "port": 5432}, "features": {"a": True}})
cfg.save("app-config", overwrite=True)

# Load in your flow (avoids the 512KB parameter limit)
from prefect import flow
from prefect.blocks.system import JSON

@flow
def my_flow():
    cfg = JSON.load("app-config").value
    # use cfg...

Using a Variable for small config

import json
from prefect.variables import Variable
from prefect import flow

# Set once
Variable.set(name="app_cfg", value=json.dumps({"env":"prod"}), overwrite=True)

@flow
def my_flow():
    var = Variable.get("app_cfg")
    cfg = json.loads(var.value)   # keep under ~5,000 chars
    # use cfg...

When to choose what - Small, simple config (string < 5 KB): Variable - Structured or larger config: JSON block - Sensitive values: Secret block (

prefect.blocks.system.Secret

) - Very large payloads or files: store in GCS and pass/load a

gs://...

reference Either approach avoids the “Flow run parameters must be less than 512KB” error as long as you load the value inside the flow instead of passing it as a parameter.

Great question. If you want a JSON block to exist only for the lifetime of a flow run, create it at runtime and delete it in a finally block (or in a cleanup task) so it’s removed even if the run fails. Pattern 1: create-use-delete within a single flow

from prefect import flow
from prefect.blocks.system import JSON

@flow
def ephemeral_config_flow():
    name = f"run-config-{__import__('uuid').uuid4()}"
    try:
        # create
        JSON(value={"big": "config", "v": 1}).save(name)

        # use
        cfg = JSON.load(name).value
        # ... do work with cfg ...

    finally:
        # cleanup (won't error if not found)
        try:
            JSON.delete(name)
        except Exception:
            pass

Pattern 2: split into tasks so cleanup is observable and retried

from prefect import flow, task
from prefect.blocks.system import JSON

@task
def create_json_block(name: str, value: dict):
    JSON(value=value).save(name)

@task
def use_json_block(name: str):
    return JSON.load(name).value

@task
def delete_json_block(name: str):
    JSON.delete(name)

@flow
def ephemeral_config_flow(value: dict):
    name = f"run-config-{__import__('uuid').uuid4()}"
    try:
        create_json_block(name, value)
        cfg = use_json_block(name)
        # ... use cfg ...
    finally:
        # ensure cleanup runs
        delete_json_block.submit(name)

Notes and tips: - Name uniquely: include the flow run ID if available for easier traceability: -

name = f"cfg-{prefect.runtime.flow_run.id}"

(available inside a flow) - Guard deletes: wrap in try/except so a missing block doesn’t fail the run. - If you need to share across subflows/tasks in the same run, pass just the block name (small string), not the config itself—avoids the 512 KB parameter limit. - If the payload is very large or many runs happen in parallel, consider GCS with a TTL instead of blocks to reduce control-plane churn. Reference APIs: - Create/save:

JSON(value=...).save(name, overwrite=True|False)

- Load:

JSON.load(name).value

- Delete:

JSON.delete(name)

or

Block.delete("json/<name>")

In Prefect 2.x there isn’t a public

load_from_id

helper. To load a block by its block document ID (UUID), use the client to read the document, then convert it to a block: Sync-friendly helper

from uuid import UUID
from prefect.blocks.core import Block
from prefect.client.orchestration import PrefectClient
from prefect.utilities.asyncutils import sync_compatible
from prefect.exceptions import ObjectNotFound

@sync_compatible
async def load_block_by_id(block_document_id: UUID, include_secrets: bool = True):
    async with PrefectClient() as client:
        try:
            doc = await client.read_block_document(
                block_document_id=block_document_id,
                include_secrets=include_secrets,
            )
        except ObjectNotFound:
            raise

        # Convert the document to the correct Block subclass instance
        return Block._from_block_document(doc)

# Usage
block = load_block_by_id(UUID("12345678-1234-5678-1234-567812345678"))
print(type(block), getattr(block, "_block_document_name", None))

Notes: -

Block._from_block_document(...)

is a “private” method by name but is the intended way to hydrate a stored block document into its concrete Block subclass. - If you know the block name, the simpler public API is

MyBlockSubclass.load("name")

or

Block.load("type-slug/name")

. - The

include_secrets

flag controls whether secret fields (e.g., in Secret blocks or credential blocks) are populated.

I’m happy to help, but I’m missing the specifics. What problem are you trying to solve in Prefect 3? Could you share: - The goal (what you’re trying to do) - Any error messages or stack traces - How you deploy today (agents/workers, storage, infra) - Your GCP target (Cloud Run, GKE, Compute Engine) and how code is stored (Git, GCS, image) Since you’re currently on Prefect 2 (Cloud) and run on GCP, here are the most common “best solution” patterns in Prefect 3 you might be looking for, with links: - Migrate to Prefect 3: Upgrade to Prefect 3 guide - Agents → Workers: Upgrade agents to workers - GCP Cloud Run (hybrid worker): GCP Cloud Run worker guide - GCP Cloud Run (serverless push work pool): Serverless (push) work pools - Deployments in 3.x: Deploy via Python and prefect.yaml If your question is about “what’s the best way to deploy on GCP in Prefect 3,” these are the two recommended options: - Cloud Run push work pool (no long-running worker)

prefect work-pool create --type cloud-run:push --provision-infra my-cloud-run-pool
# then deploy your flow (Python or prefect.yaml based)
prefect deploy

- Hybrid worker on Cloud Run

# Create a Cloud Run work pool
prefect work-pool create --type cloud-run my-cloud-run-pool

# Deploy a worker to Cloud Run
gcloud run deploy prefect-worker \
  --image=prefecthq/prefect:3-latest \
  --set-env-vars PREFECT_API_URL=$PREFECT_API_URL,PREFECT_API_KEY=$PREFECT_API_KEY \
  --args "prefect","worker","start","--with-healthcheck","-p","my-cloud-run-pool","-t","cloud-run"

And a minimal 3.x deployment pattern using Git/GCS storage:

from prefect import flow

@flow
def my_flow():
    ...

# Example: Git or GCS
flow.from_source(
    source="<https://github.com/org/repo.git>",  # or "<gs://my-bucket/path>"
    entrypoint="flows/my_flow.py:my_flow",
).deploy(
    name="my-deployment",
    work_pool_name="my-cloud-run-pool",
)

If you can share the exact problem, I’ll tailor the best 3.x solution and give you step-by-step commands/code.

That error means the JSON-serialized parameters for your flow run exceed Prefect’s hard 512 KB limit. This limit exists in both Prefect 2 and 3 and can’t be raised in Cloud. Best practice: don’t pass large data as parameters. Pass a reference and load the data inside the flow. Recommended fixes (GCP-friendly): 1) Store the payload in GCS and pass a gs:// path - Create a

GcpCredentials

block and (optionally) a

GcsBucket

block. - Pass a small string (bucket/key or gs:// URL) as the parameter. - Load the bytes in your flow. Example using the GCP integration functions:

from prefect import flow
from prefect_gcp.credentials import GcpCredentials
from prefect_gcp.cloud_storage import cloud_storage_download_blob_as_bytes

@flow
def process_from_gcs(bucket: str, blob: str):
    creds = GcpCredentials.load("my-gcp-creds")  # saved once in UI/CLI
    data_bytes = cloud_storage_download_blob_as_bytes(
        bucket=bucket,
        blob=blob,
        gcp_credentials=creds,
    )
    # deserialize/process data_bytes here
    ...

# Trigger with small params, e.g. {"bucket": "my-bucket", "blob": "inputs/run-123.json"}

2) Persist large task/flow results to GCS instead of passing them around - Configure result persistence to a GCS block so large outputs don’t travel as parameters. - For subflows, pass a reference (e.g., a GCS key) instead of the large object. 3) If you’re only slightly over the limit, compress before passing - Compress + base64 a small JSON payload and ensure the resulting string is still <512 KB; then decode inside the flow. This is a last resort—prefer GCS references. Key notes: - You cannot increase the 512 KB limit. - The same pattern applies to subflows—if you pass a large object to a subflow as a parameter, you’ll hit the same limit. Store it externally and pass a pointer. Docs: - Flows concept (notes the 512 KB limit): Flows concepts - GCP integration (GCS, credentials, examples): prefect-gcp integration If you can share what you’re passing today (e.g., a big JSON, DataFrame, model, etc.) and how you trigger the run, I can tailor an exact pattern and code snippet for your setup on Prefect 2 (Cloud).

Short answer: Yes—with an important nuance. - Don’t store the large payload itself in a “block document.” Block documents live in the API and are not meant for big blobs. - Do use a storage block (e.g., GCS) to hold the payload, and load it inside the flow. The block stores credentials/config; the actual data lives in GCS. You then pass only a small reference (bucket/key) as a parameter. GCP-friendly pattern (works in Prefect 2 and 3): 1) One-time: create blocks

from prefect_gcp.credentials import GcpCredentials
from prefect_gcp.cloud_storage import GcsBucket

# Save credentials
GcpCredentials(service_account_info=... ).save("my-gcp-creds", overwrite=True)

# Save a GCS bucket block (stores bucket + creds config)
GcsBucket(bucket="my-bucket", gcp_credentials=GcpCredentials.load("my-gcp-creds")).save(
    "my-gcs-block", overwrite=True
)

2) Before triggering the flow: upload the big payload to GCS (so you only pass a tiny pointer)

from prefect_gcp.credentials import GcpCredentials
from prefect_gcp.cloud_storage import cloud_storage_upload_blob_from_string

creds = GcpCredentials.load("my-gcp-creds")
cloud_storage_upload_blob_from_string(
    bucket="my-bucket",
    blob="inputs/run-123.json",
    data=big_json_string,        # or use cloud_storage_upload_blob_from_file for files
    gcp_credentials=creds,
)

3) Inside your flow: load from GCS using the block

from prefect import flow
from prefect_gcp.credentials import GcpCredentials
from prefect_gcp.cloud_storage import cloud_storage_download_blob_as_bytes

@flow
def process_from_gcs(bucket: str, blob: str):
    creds = GcpCredentials.load("my-gcp-creds")
    data_bytes = cloud_storage_download_blob_as_bytes(
        bucket=bucket,
        blob=blob,
        gcp_credentials=creds,
    )
    # Deserialize/process data_bytes here
    ...

Trigger the run with tiny params, e.g.:

{
  "bucket": "my-bucket",
  "blob": "inputs/run-123.json"
}

This avoids the 512 KB parameter limit and is the recommended approach. Refs: - Flows concept (512 KB param limit): Flows concepts - GCP integration (GCS, credentials): prefect-gcp integration If you prefer, you can also store a default blob path in a custom block or in the GCS block and load that inside the flow—but for per-run data, passing the blob path as a parameter is simplest.

Yes—there are several ways to avoid the 512 KB parameter limit without using GCS. The idea is the same: store the large payload somewhere accessible, then pass a small reference (path/ID/URL) as the flow parameter and load it inside the flow. Good options without GCS: 1) Shared filesystem (LocalFileSystem block) - Works great if your workers run on VMs/Kubernetes with a shared mount (NFS/Filestore/etc.). - Save a LocalFileSystem block pointing at the mount; pass a relative path as the parameter. Example:

from prefect.filesystems import LocalFileSystem
from prefect import flow

# One-time setup (can be done in code or UI)
LocalFileSystem(basepath="/mnt/shared").save("shared-fs", overwrite=True)

@flow
def process_from_local(relative_path: str):
    fs = LocalFileSystem.load("shared-fs")
    data = fs.read_path(relative_path)   # bytes
    # deserialize/process data and continue

2) RemoteFileSystem / SMB (no cloud object store required) - Use SFTP, SMB, HTTP(S), or other fsspec-supported backends. - Create a RemoteFileSystem or SMB block and pass a small path reference. Docs: - Filesystems API: prefect.filesystems API - Blocks concepts: Blocks concepts 3) Database staging (BigQuery, Cloud SQL Postgres/MySQL, etc.) - Store the payload in a table (or as a file path in your infra) and pass a key/ID as the parameter. - Inside the flow, query by ID and load the data. - On GCP, BigQuery is a solid choice for tabular/JSON payloads; Cloud SQL works for arbitrary blobs. Docs: - GCP integration (BigQuery examples): prefect-gcp integration 4) HTTP(S) URL with auth/signed URL - Host the payload behind a URL (any web server, artifact server, etc.), pass the URL as the param, and fetch in the flow. - Keep the auth token small (header or query string). Example:

import requests
from prefect import flow

@flow
def process_from_url(url: str, token: str | None = None):
    headers = {"Authorization": f"Bearer {token}"} if token else {}
    r = requests.get(url, headers=headers, timeout=60)
    r.raise_for_status()
    data = r.content
    # process data

5) If it’s just slightly over 512 KB: compress and pass - Compress + base64 a small JSON payload and pass the string (only if it ends up <512 KB). - Prefer external storage for anything moderately large. Notes and constraints: - Serverless (Cloud Run push pools) cannot access local paths; use a network-accessible option (HTTP/DB/SFTP/SMB). - Don’t store big blobs in block documents (blocks are for config/credentials; they live in the API). - The 512 KB parameter limit applies in both Prefect 2 and 3 and isn’t configurable in Cloud. Related docs: - 512 KB limit noted here: Flows concepts - Persisting results (for large outputs): Persist workflow results - Filesystems API: prefect.filesystems API If you share your runtime (Cloud Run serverless vs. VM/K8s workers) and what storage/database options you do have, I’ll recommend the best-fit pattern and provide a tailored snippet.

Got it — if you don’t want to use file/object storage, you can still avoid the 512 KB parameter limit by passing a small reference and loading the large payload at runtime from a non-filesystem backend. Good non-file options on GCP: - BigQuery (store the payload as a row; pass the row ID) - Cloud SQL (Postgres/MySQL) or Firestore/Datastore (store JSON/doc; pass the key) - Pub/Sub (let the flow pull messages; don’t pass the message content as a param) - Redis/Memorystore (cache large payloads by UUID; pass the key) - Your own HTTP API (flow fetches by ID/token) Examples 1) BigQuery (recommended for JSON/tabular) - Write payload ahead of time:

from prefect_gcp.credentials import GcpCredentials
from prefect_gcp.bigquery import bigquery_insert_stream

creds = GcpCredentials.load("my-gcp-creds")
row_id = "run-123"  # your unique ID
bigquery_insert_stream(
    dataset="my_dataset",
    table="payloads",
    rows=[{"id": row_id, "payload": big_json_dict}],  # JSON column
    gcp_credentials=creds,
)

- Flow loads by ID:

from prefect import flow
from prefect_gcp.credentials import GcpCredentials
from prefect_gcp.bigquery import bigquery_query

@flow
def process_from_bq(row_id: str):
    creds = GcpCredentials.load("my-gcp-creds")
    rows = bigquery_query(
        query="""
        SELECT payload
        FROM `my_project.my_dataset.payloads`
        WHERE id = @id
        """,
        gcp_credentials=creds,
        query_params=[("id", row_id)],
        to_dataframe=False,
    )
    payload = rows[0]["payload"]
    # process payload...

2) Redis / Memorystore (in-memory cache, no files)

# producer side
import redis, json, uuid
r = redis.Redis(host="my-memorystore-host", port=6379)
key = f"payload:{uuid.uuid4()}"
r.setex(key, 3600, json.dumps(big_payload))  # TTL 1h

# flow side
from prefect import flow
import redis, json

@flow
def process_from_redis(key: str):
    r = redis.Redis(host="my-memorystore-host", port=6379)
    raw = r.get(key)
    payload = json.loads(raw)
    # process payload...

3) Firestore (document DB)

# producer side
from google.cloud import firestore
db = firestore.Client()
doc_ref = db.collection("payloads").document("run-123")
doc_ref.set({"payload": big_json_dict})

# flow side
from prefect import flow
from google.cloud import firestore

@flow
def process_from_firestore(doc_id: str):
    db = firestore.Client()
    doc = db.collection("payloads").document(doc_id).get()
    payload = doc.to_dict()["payload"]
    # process payload...

4) Pub/Sub (event-driven, no param payload) - Have the flow poll/consume from a subscription at runtime (don’t pass the message content as a parameter; the flow just knows which subscription to read from). This avoids the param limit entirely and keeps payloads in Pub/Sub. Notes and caveats - The 512 KB parameter limit still applies in both Prefect 2 and 3 — you can’t raise it in Cloud. Keep parameters to small IDs/paths. - Avoid storing big blobs in block documents; blocks live in the API and aren’t for large payloads. - If you’re only slightly over the limit, you can compress + base64 a small JSON payload and pass the string (only if final size <512 KB), but prefer one of the backends above. Useful links - 512 KB param limit (Flows concepts): Flows concepts - GCP integration (BigQuery, credentials): prefect-gcp integration If you share which backend you prefer (BigQuery, Cloud SQL, Firestore, Pub/Sub, Redis), I’ll tailor a minimal pattern and permissions setup for your environment.

Short answer: No. With only Prefect and Prefect Cloud, there is no supported way to pass or store large (>512 KB) inputs for a flow run. You must keep parameters under 512 KB. Prefect Cloud is not a general-purpose blob store. Why: - Flow run parameters: hard limit of 512 KB (both 2.x and 3.x). Flows concepts - Variables: max ~5,000 characters; not for large data and not encrypted. Variables - Blocks: store configuration/credentials; block documents live in the API and are not for big blobs. Blocks - Artifacts: meant for UI-friendly outputs (links/markdown/tables/images/progress), not for shipping large inputs; they’re created during runs, not before. Artifacts What you can do instead without “file storage”: - Fetch the payload inside the flow from a service you control (HTTP API, Pub/Sub, DB, Redis/Memorystore, etc.) using a tiny ID/token as the parameter.

from prefect import flow
import requests

@flow
def process_from_api(payload_id: str, api_base: str, token: str):
    r = requests.get(f"{api_base}/payloads/{payload_id}",
                     headers={"Authorization": f"Bearer {token}"}, timeout=60)
    r.raise_for_status()
    data = r.content  # or r.json()
    # process data...

- Redesign so the flow produces the data itself (fetch/generate in an upstream task) instead of pushing it in as a parameter. - If you’re only slightly over the limit, compress + base64 and pass it only if the final string is still <512 KB (a last resort; prefer fetching at runtime). If you can share the payload size, source, and why external storage/services aren’t an option, I can suggest the cleanest “fetch-at-runtime” pattern for your environment.

In Prefect 3.x, flow run parameters are serialized to JSON and sent to the API when you create a run. Key points: - Parameters must be JSON-serializable. Prefect uses Pydantic v2 to convert your inputs to JSON-friendly objects via

model_dump(mode="json")

. - The HTTP client (httpx) then encodes that dict with Python’s standard

json.dumps

and sends it to the Prefect API. - The API stores parameters as JSON. Prefect Cloud enforces a hard 512 KB limit on the serialized parameters payload. Important details: - Datetimes are serialized to ISO strings. - Pydantic BaseModels in your params are converted to JSON data first. - There’s no setting to change the parameter serializer in 3.x (unlike some 2.x internals that used orjson). The limit applies after JSON serialization. What this means for you: - Keep parameters small and JSON-friendly. If you need to pass big data, pass a compact reference (ID/path/URL) and load data at runtime. - If you hit the limit unexpectedly, check the size of

json.dumps(params)

locally to see the actual serialized size. Docs: - Flows concepts (mentions the 512 KB limit): Flows concepts

Yes. You can compress the dict into a base64 string (so it’s JSON-safe), pass that string to the subflow, then decode/decompress inside the subflow. This works in both Prefect 2 and 3. Just make sure the final base64 string is still under 512 KB after JSON serialization. Example utilities and flow/subflow:

import json
import zlib
import base64

# Utilities
def compress_dict_to_b64(d: dict, compression_level: int = 9) -> str:
    """
    JSON-serialize, zlib-compress, then base64-encode a dict.
    Returns a UTF-8 string safe for JSON parameters.
    """
    json_bytes = json.dumps(d, separators=(",", ":")).encode("utf-8")
    compressed = zlib.compress(json_bytes, level=compression_level)
    b64 = base64.b64encode(compressed).decode("utf-8")
    return b64

def decompress_b64_to_dict(b64: str) -> dict:
    """
    Base64-decode, zlib-decompress, then JSON-deserialize back to a dict.
    """
    compressed = base64.b64decode(b64.encode("utf-8"))
    json_bytes = zlib.decompress(compressed)
    return json.loads(json_bytes)

def approx_param_size_bytes(param_object) -> int:
    """
    Approximate the final request size impact by measuring the JSON-encoded bytes
    for the params you’ll send. Use compact separators to reduce overhead.
    """
    return len(json.dumps(param_object, separators=(",", ":")).encode("utf-8"))

from prefect import flow

@flow
def child_flow(packed_payload_b64: str):
    payload = decompress_b64_to_dict(packed_payload_b64)
    # ...process payload...
    return {"ok": True, "keys": list(payload.keys())}

@flow
def parent_flow(big_payload: dict):
    packed = compress_dict_to_b64(big_payload)

    # Optional: safety check against 512 KB limit
    approx_size = approx_param_size_bytes({"packed_payload_b64": packed})
    if approx_size > 512 * 1024:
        raise ValueError(f"Compressed payload still too large: ~{approx_size} bytes")

    return child_flow(packed)

Notes and tips: - Base64 adds ~33% overhead, so this helps mostly when your payload compresses well (e.g., large, repetitive JSON). Measure

approx_param_size_bytes

before calling the subflow. - If you’re just slightly over the limit, this approach is fine. If you’re still over, you’ll need to pass a reference and fetch at runtime (preferred for truly large inputs). - Keep payloads JSON-serializable. If you have numpy/pandas types, convert them to plain Python before compressing (e.g.,

.tolist()

for arrays). - Security: compression is not encryption. If the data is sensitive, encrypt (e.g., Fernet/KMS) before base64, then decrypt in the subflow. Reference on the 512 KB parameter limit: Flows concepts