Yes. In Prefect 3.x you define parameters as your flow’s function arguments, then pass values when you trigger a run. Quick examples: - Define parameters on a flow

from prefect import flow

@flow
def greet(name: str, excited: bool = False):
    msg = f"Hello, {name}"
    if excited:
        msg += "!!!"
    print(msg)

- Set default parameters at deploy/serve time (can be overridden at run time)

# Using work pools
greet.deploy(
    name="greet-deploy",
    work_pool_name="my-pool",
    parameters={"name": "Marvin", "excited": True},
)

# Using serve (static infra)
greet.serve(
    name="greet-serve",
    parameters={"name": "Marvin", "excited": True},
)

- Trigger with parameters via CLI

prefect deployment run greet/greet-deploy --param name=Zaphod --param excited=true
# You can pass structured JSON
prefect deployment run greet/greet-deploy --param meta='{"team":"hitchhikers"}'
# Or pass a full JSON object
prefect deployment run greet/greet-deploy --params '{"name":"Ford","excited":false}'

Notes: -

--param key=value

parses values as JSON when possible (e.g.,

true

,

42

,

{"a":1}

); otherwise it’s treated as a string. Quote values with spaces or structured JSON. - Trigger with parameters via Python API

from prefect.deployments import run_deployment

run = run_deployment(
    name="greet/greet-deploy",
    parameters={"name": "Trillian", "excited": True},
)

- Trigger from the UI - Go to your Deployment → Run or Custom Run → fill in the parameter form generated from your flow’s signature and any deployment defaults. Docs: - Flows: parameters https://docs.prefect.io/v3/concepts/flows#specify-flow-parameters - Deploy via Python (set defaults) https://docs.prefect.io/v3/how-to-guides/deployments/deploy-via-python - Run deployments (CLI/Python/UI) https://docs.prefect.io/v3/how-to-guides/deployments/run-deployments - CLI reference for running deployments https://docs.prefect.io/v3/api-ref/cli/deployments#prefect-deployments-run Would you like to pass params from the CLI, Python, or the UI—and are you using deployments or running flows ad hoc?

Yes—enter a JSON array for that parameter in the run modal. Best experience: give the parameter a list type so the UI knows it’s an array. Example flow

from prefect import flow
from typing import List  # or use built-in list[str] on 3.9+

@flow
def process_users(user_ids: list[int], tags: list[str] = []):
    ...

In the UI: - Open your deployment → Run (or Custom Run) - For

user_ids

or

tags

, either: - Use the form’s array control (you’ll see “Add item” if the param is typed as a list), or - Switch to the JSON editor and paste valid JSON, e.g.:

{
  "user_ids": [
    101,
    102,
    103
  ],
  "tags": [
    "vip",
    "beta"
  ]
}

Notes: - If you didn’t type-hint the param, the UI may treat it as a string; you can still pass a list by using the JSON editor with a JSON array. - Lists of objects are supported—match your type hints:

{
  "items": [
    {
      "id": 1,
      "qty": 2
    },
    {
      "id": 2,
      "qty": 1
    }
  ]
}

with

from pydantic import BaseModel

class Item(BaseModel):
    id: int
    qty: int

@flow
def process(items: list[Item]): ...

Docs: - Flow parameters https://docs.prefect.io/v3/concepts/flows#specify-flow-parameters - Advanced form building (UI schemas) https://docs.prefect.io/v3/advanced/form-building

Great question. In Prefect 3.x, log level is controlled by settings (env vars), so you configure it for a single run by setting environment variables in the run’s environment. Easiest options: One-off run via CLI (per-run) - Override job variables for just this run (works with any worker type):

prefect deployment run <FLOW>/<DEPLOYMENT> \
  -jv env.PREFECT_LOGGING_LEVEL=DEBUG

- Granular control for flow vs task logs:

prefect deployment run <FLOW>/<DEPLOYMENT> \
  -jv env.PREFECT_LOGGING_LOGGERS_PREFECT_FLOW_RUNS_LEVEL=DEBUG \
  -jv env.PREFECT_LOGGING_LOGGERS_PREFECT_TASK_RUNS_LEVEL=INFO

One-off run via Python

from prefect.deployments import run_deployment

run_deployment(
    name="my-flow/my-deployment",
    parameters={...},
    job_variables={"env": {"PREFECT_LOGGING_LEVEL": "DEBUG"}},
)

Set defaults on a deployment (persisted)

my_flow.deploy(
    name="my-deploy",
    work_pool_name="my-pool",
    job_variables={"env": {"PREFECT_LOGGING_LEVEL": "DEBUG"}},
)

From the UI - Edit your deployment → set Environment Variables (or Job variables) to include: -

PREFECT_LOGGING_LEVEL=DEBUG

- or targeted: -

PREFECT_LOGGING_LOGGERS_PREFECT_FLOW_RUNS_LEVEL=DEBUG

-

PREFECT_LOGGING_LOGGERS_PREFECT_TASK_RUNS_LEVEL=INFO

Local/ad hoc runs (no deployment) - Set the env var where the flow executes:

PREFECT_LOGGING_LEVEL=DEBUG python flow.py

Common settings - `PREFECT_LOGGING_LEVEL`: default level for Prefect logs (INFO by default) - `PREFECT_LOGGING_LOGGERS_PREFECT_FLOW_RUNS_LEVEL`: flow-run logger - `PREFECT_LOGGING_LOGGERS_PREFECT_TASK_RUNS_LEVEL`: task-run logger - Valid levels: DEBUG, INFO, WARNING, ERROR, CRITICAL Docs - Logging customization https://docs.prefect.io/v3/advanced/logging-customization - Settings reference https://docs.prefect.io/v3/api-ref/settings-ref - Run deployments (CLI/Python) https://docs.prefect.io/v3/how-to-guides/deployments/run-deployments - CLI: deployments run https://docs.prefect.io/v3/api-ref/cli/deployments#prefect-deployments-run If you tell me how you’re running (process/Docker/K8s worker, UI vs CLI), I can give you the exact command or UI steps.

Yes—automation templates can access asset metadata, but you do it via the event that triggered the automation (not via a special

asset

object). In templates (Slack, email, etc.), use

event.resource

and

event.payload

. Common fields for asset events: -

{{ event.resource["prefect.resource.id"] }}

— the asset key/ID -

{{ event.resource["prefect.resource.name"] }}

— the asset display name (if set) -

{{ event.resource["prefect.asset.description"] }}

— description (if set) -

{{ event.resource["prefect.asset.url"] }}

— URL (if set) -

{{ event.payload }}

— runtime metadata you attached during materialization (e.g., counts, hashes, etc.) Example Slack subject/body

Subject:
✅ Asset materialized: {{ event.resource["prefect.resource.name"] or event.resource["prefect.resource.id"] }}

Body:
Key: {{ event.resource["prefect.resource.id"] }}
Name: {{ event.resource.get("prefect.resource.name", "N/A") }}
Description: {{ event.resource.get("prefect.asset.description", "N/A") }}
URL: {{ event.resource.get("prefect.asset.url", "N/A") }}

Metadata:
{{ event.payload | tojson(indent=2) }}

Filtering assets in the trigger - Set your Event Trigger to expect asset events like: -

prefect.asset.materialization.succeeded

-

prefect.asset.materialization.failed

-

prefect.asset.referenced

- Optionally match the asset key: -

match: { "prefect.resource.id": "s3://*" }

Notes and limitations: - There is no dedicated

asset

object in templates; use

event.resource

and

event.payload

. - Any runtime metadata you add in code (e.g., via

add_asset_metadata

) will be available under

event.payload

. Helpful docs: - Automations overview and templating https://docs.prefect.io/v3/concepts/automations - Access values in templates https://docs.prefect.io/v3/how-to-guides/automations/access-parameters-in-templates - Assets concepts and metadata https://docs.prefect.io/v3/concepts/assets - Event triggers https://docs.prefect.io/v3/concepts/event-triggers If you share which fields you want to include, I can draft the exact Slack template and trigger config.

Short answer: Yes. That message is coming from Cloud Run and indicates the container hit its configured memory limit and the job failed. How to confirm - Open the flow run logs in Prefect; you’ll see the Cloud Run message “The configured memory limit was reached.” - Check the Cloud Run Job execution logs in GCP; you’ll typically see OOM/“memory limit was reached” there as well. How to fix (increase resources) - One-off run override (CLI):

prefect deployment run <FLOW>/<DEPLOYMENT> \
  -jv memory=2Gi \
  -jv cpu=2000m

- Persist on the deployment (Python):

my_flow.deploy(
  name="my-deployment",
  work_pool_name="my-cloud-run-pool",
  job_variables={"memory": "2Gi", "cpu": "2000m"},
)

- Or edit the deployment in the UI and set Job Variables: - memory: 2Gi - cpu: 2000m Notes - Valid memory formats: e.g., 512Mi, 1024M, 2Gi, 2G (units required). - Default Cloud Run worker memory is often too low for data-heavy workloads; start with 2–4 GiB and adjust. - If memory keeps spiking, consider reducing in-process data size, batching/streaming, or lowering parallelism in your code.

You’ve got two separate concerns here: 1) Returning a generator from a task - Prefect will eagerly consume generators at task completion and convert them to a list. That means: - No streaming across task boundaries - The task won’t finish until the generator is fully consumed - For large/indefinite streams, this is not what you want 2) Ensuring the download completes before the task ends - Put the streaming/consumption loop inside the task and write to a file (or other external sink). The task will not complete until your loop finishes. Return the path (or other reference), not the data. Sync example (httpx.stream)

from prefect import task, flow
import httpx

@task(persist_result=False)
def download_file(url: str, output_path: str) -> str:
    with httpx.stream("GET", url, timeout=None) as resp:
        resp.raise_for_status()
        with open(output_path, "wb") as f:
            for chunk in resp.iter_bytes(chunk_size=65536):
                f.write(chunk)
    return output_path  # downstream tasks will only run after this finishes

@flow
def pipe():
    path = download_file("<https://example.com/big.zip>", "/tmp/big.zip")
    # safe to use the file here; download is finished

Async example

from prefect import task, flow
import httpx

@task(persist_result=False)
async def download_file_async(url: str, output_path: str) -> str:
    async with httpx.AsyncClient() as client:
        async with client.stream("GET", url, timeout=None) as resp:
            resp.raise_for_status()
            with open(output_path, "wb") as f:
                async for chunk in resp.aiter_bytes(chunk_size=65536):
                    f.write(chunk)
    return output_path

@flow
async def pipe():
    path = await download_file_async("<https://example.com/big.zip>", "/tmp/big.zip")
    # safe to use the file here

Why this fixes your issue: - If your task returns the httpx generator, Prefect will try to realize it (turn it into a list) to store the result. That’s memory-heavy and not truly streaming. - By iterating inside the task and writing to disk/object storage, the task only completes after the download is done, so the parent/next task waits as expected. Tips - Return a small reference (e.g., file path, S3 URL), not the content. - Set

persist_result=False

for tasks returning paths to avoid persisting them as results if you don’t need to. - Add progress logs with

get_run_logger()

if you want visibility during the stream. If you paste your current task code, I can show the minimal changes to make it wait and avoid returning a generator.

Short answer: yes—with a small tweak. In Prefect 3.x,

task_run_name

supports Python’s

str.format

templates with your task’s parameters. For dicts, use bracket indexing (not dot-notation). So instead of

some-task-name-{input_dict.input_key}

, use:

from prefect import task, flow

@task(task_run_name="some-task-name-{input_dict[input_key]}")
def my_task(input_dict: dict):
    return input_dict["input_key"]

@flow
def f():
    my_task({"input_key": "abc"})  # task run name: "some-task-name-abc"

Notes: - Dot notation works for objects with attributes (e.g.,

{obj.attr}

), but for Python dicts use

{dict[key]}

. You can nest:

{payload[user][id]}

. - If keys might be missing or you want custom logic, use a callback instead:

@task(task_run_name=lambda parameters: f"some-task-name-{parameters['input_dict'].get('input_key', 'missing')}")
def my_task(input_dict: dict):
    ...

- To include literal braces, escape them with double braces:

{{

and

}}

. Docs: - How-to guide with examples: Write and run a workflow - Task decorator API (shows

task_run_name

options): prefect.tasks API

That error means the thing you’re passing as

input_dict

isn’t actually a dict at runtime. Python’s formatter is trying to do

parameters["input_dict"]["input_key"]

, but your

input_dict

is a

FileStoreResource

object, which isn’t subscriptable. You’ve got a few ways to fix it: - If the object exposes an attribute, use dot notation:

@task(task_run_name="some-task-name-{input_dict.input_key}")
def my_task(input_dict):
    ...

- If you intended to pass a real dict, convert it before calling the task (or inside a callback) — e.g.

input_dict.model_dump()

for Pydantic models, or

asdict(dataclass_obj)

for dataclasses. - Make it robust with a parameter-aware callback:

@task(task_run_name=lambda parameters: (
    f"some-task-name-{(
        parameters['input_dict'].get('input_key')
        if isinstance(parameters['input_dict'], dict)
        else getattr(parameters['input_dict'], 'input_key', 'unknown')
    )}"
))
def my_task(input_dict):
    ...

If you can share what

FileStoreResource

is (class/type and how you construct it), I can suggest the exact template (attribute vs dict conversion). Refs: - How-to guide with task_run_name examples: Write and run a workflow - Task decorator API: prefect.tasks API