https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I’m getting this error when I try to ...
# ask-community
g
@Marvin I’m getting this error when I try to use Prefect python package from python interpreter. I set PREFECT_API_URL env var from prefect import get_client File “<frozen importlib._bootstrap>“, line 1412, in _handle_fromlist File “/Users/giacomochiarella/github/dwh_dbt/venv313/lib/python3.13/site-packages/prefect/__init__.py”, line 171, in getattr module = importlib.import_module(mname, package=package) File “/usr/local/Cellar/python@3.13/3.13.7/Frameworks/Python.framework/Versions/3.13/lib/python3.13/importlib/__init__.py”, line 88, in import_module return _bootstrap._gcd_import*(name[level:], package, level)* ~~~~~~~~~~~~~~~~~~~~~~*^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^* File “/Users/giacomochiarella/github/dwh_dbt/venv313/lib/python3.13/site-packages/prefect/main.py”, line 6, in <module> from prefect.transactions import Transaction File “/Users/giacomochiarella/github/dwh_dbt/venv313/lib/python3.13/site-packages/prefect/transactions.py”, line 27, in <module> from prefect.context import ContextModel File “/Users/giacomochiarella/github/dwh_dbt/venv313/lib/python3.13/site-packages/prefect/context.py”, line 990, in <module> GLOBAL_SETTINGS_CONTEXT: SettingsContext = root_settings_context*()* ~~~~~~~~~~~~~~~~~~~~~*^^* File “/Users/giacomochiarella/github/dwh_dbt/venv313/lib/python3.13/site-packages/prefect/context.py”, line 949, in root_settings_context profiles = prefect.settings.load_profiles() File “/Users/giacomochiarella/github/dwh_dbt/venv313/lib/python3.13/site-packages/prefect/settings/profiles.py”, line 307, in load_profiles current_settings = get_current_settings() File “/Users/giacomochiarella/github/dwh_dbt/venv313/lib/python3.13/site-packages/prefect/settings/context.py”, line 21, in get_current_settings return Settings() File “/Users/giacomochiarella/github/dwh_dbt/venv313/lib/python3.13/site-packages/pydantic_settings/main.py”, line 195, in init **__pydantic_self__._settings_build_values*(* ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~*^* values, ^^^^^^^ ...<25 lines>... _secrets_dir=_secrets_dir, ^^^^^^^^^^^^^^^^^^^^^^^^^^ ) ^ File “/Users/giacomochiarella/github/dwh_dbt/venv313/lib/python3.13/site-packages/pydantic_settings/main.py”, line 438, in _settings_build_values source_state = source() File “/Users/giacomochiarella/github/dwh_dbt/venv313/lib/python3.13/site-packages/pydantic_settings/sources/base.py”, line 512, in call raise SettingsError( f’error parsing value for field “{field_name}” from source “{self.class.name}“' ) from e pydantic_settings.exceptions.SettingsError: error parsing value for field “api” from source “EnvFilterSettingsSource”
m
✅ thought for 327.8 seconds
Thanks for the stack trace — that error usually means Prefect is seeing a top-level env var for the nested settings model 
api
and trying (and failing) to parse it as JSON. Root cause - If 
PREFECT_API
is set (e.g., 
PREFECT_API=http://...
), Prefect treats it as the entire 
api
settings object and tries to JSON-parse it. A plain string isn’t valid for that, so you get: pydantic_settings.exceptions.SettingsError: error parsing value for field "api" from source "EnvFilterSettingsSource" Fix - Unset any 
PREFECT_API
env var and only use 
PREFECT_API_URL
(and optionally 
PREFECT_API_KEY
, etc.) Quick checks 1) Check what’s set in your shell:
Copy code
env | grep PREFECT_API
If you see 
PREFECT_API=...
, unset it: - bash/zsh:
Copy code
unset PREFECT_API
- fish:
Copy code
set -e PREFECT_API
- PowerShell:
Copy code
Remove-Item Env:PREFECT_API
2) Verify Prefect’s view of your config and sources:
Copy code
prefect config view --show-sources
If you want to validate explicitly:
Copy code
prefect config validate
Correct ways to set the API URL - Environment variable:
Copy code
export PREFECT_API_URL="<http://127.0.0.1:4200/api>"
# For Cloud:
# export PREFECT_API_URL="<https://api.prefect.cloud/api/accounts/<ACCOUNT_ID>/workspaces/<WORKSPACE_ID>>"
# export PREFECT_API_KEY="pcu_xxxxx"
- In a profile (recommended):
Copy code
prefect config set PREFECT_API_URL="<http://127.0.0.1:4200/api>"
- In ~/.prefect/profiles.toml (env-style keys):
Copy code
[profiles.myprofile]
PREFECT_API_URL = "<http://127.0.0.1:4200/api>"
If you’re still seeing the error after unsetting 
PREFECT_API
, check for: - An entry like 
api = "http://..."
in 
~/.prefect/profiles.toml
. Replace it with 
PREFECT_API_URL = "..."
or use a nested table:
Copy code
[profiles.myprofile.api]
url = "<http://127.0.0.1:4200/api>"
Once cleaned up, importing should work:
Copy code
from prefect import get_client
If it persists, please share: - Output of 
env | grep PREFECT_API
- Output of 
prefect config view --show-sources
(mask secrets) Docs for reference: Settings and Profiles
g
@Marvin I set PREFECT_API_URL to my address. The address is correct. I’m using python 3.13 and prefect 3.6.4
m
✅ thought for 76.9 seconds
Got it — thanks for confirming Python 3.13 and Prefect 3.6.4. Since the error is for the top-level field “api”, the most likely cause is that something else (in your env or a config file) is setting 
api
incorrectly, even though 
PREFECT_API_URL
is correct. Here’s how to pinpoint and fix it: 1) Check for a stray PREFECT_API env var - This is the most common culprit.
Copy code
env | grep '^PREFECT_API'
If you see 
PREFECT_API=...
, unset it: - bash/zsh:
Copy code
unset PREFECT_API
- fish:
Copy code
set -e PREFECT_API
- PowerShell:
Copy code
Remove-Item Env:PREFECT_API
2) Inspect Prefect’s view of your settings and their sources - This will show exactly where 
api
is being set from.
Copy code
prefect config view --show-sources
- Validate the active profile:
Copy code
prefect config validate
Look for any source that sets 
api
as a string (instead of 
api.url
) — that will cause the parsing error. 3) Check config files that may set api incorrectly - ~/.prefect/profiles.toml - A local .env in your working directory - A local prefect.toml or pyproject.toml Wrong (causes your error): - In profiles.toml:
Copy code
[profiles.myprofile]
api = "http://..."
- In prefect.toml:
Copy code
api = "http://..."
Right: - Env-style in profiles.toml:
Copy code
[profiles.myprofile]
PREFECT_API_URL = "<http://127.0.0.1:4200/api>"
- Or nested table in prefect.toml:
Copy code
[api]
url = "<http://127.0.0.1:4200/api>"
4) Minimal sanity check in a clean shell - Start a fresh shell, only set the URL, and import:
Copy code
export PREFECT_API_URL="<http://127.0.0.1:4200/api>"
python -c "from prefect.settings import get_current_settings as g; print(g().api.url)"
python -c "from prefect import get_client; print('ok')"
If you’re still getting the error, please share: - Output of:
Copy code
env | grep '^PREFECT_API'
prefect config view --show-sources
(mask any API key), and we’ll zero in on the conflicting setting. Reference: Settings and Profiles
g
@Marvin I’m getting Unable to connect to ‘wss://myprefectwebsite.net/api/events/in’. Please check your network settings to ensure websocket connections to the API are allowed. Otherwise event data (including task run data) may be lost. Reason: server rejected WebSocket connection: HTTP 404. Set PREFECT_DEBUG_MODE=1 to see the full error. My Prefect instance is at <https//w|https//>myprefectwebsite.net and the api url https://myprefectwebsite.net/api works. I’m getting this warning when I try to read a block with the following
Copy code
class JsonConfig(Block):
    value: Dict[str, Any]
for block_name in blocks:
    print(JsonConfig.load(block_name).value)
It is strange because the block is read, it is printed but I get printed
Copy code
<block-value>
<warning>
<block-value>
<warning>
<block-value>
<warning>
…
why?
m
✅ thought for 212.3 seconds
You’re seeing that warning because every time you load a block, Prefect 3 emits an “I loaded a block” event and tries to send it to the API over a WebSocket at wss://<your-api>/api/events/in. Since your server/proxy is returning HTTP 404 for that WebSocket handshake, Prefect logs a warning. Because you load multiple blocks, you get one warning per load — hence the interleaving of - your printed block value - the warning line repeated. Why it’s 404 - In Prefect 3.x, events are sent via WebSocket to /api/events/in. A 404 during the WebSocket handshake typically means your reverse proxy (nginx, ALB, Cloudflare, etc.) is not routing/handling Upgrade requests for that path, or it’s pointing /api/events/in to the wrong upstream. - Another possibility: your server is an older Prefect Server (2.x) that doesn’t provide that events endpoint, while your client is 3.6.4. That mismatch would also 404. What to do 1) Verify your proxy forwards WebSockets on /api/events/in (and /api/events/out) - Ensure HTTP/1.1, Upgrade/Connection headers, and long timeouts. Minimal nginx example:
Copy code
location /api/events/in {
    proxy_pass <http://your_prefect_api_upstream>;
    proxy_http_version 1.1;

    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";

    proxy_set_header Host $host;
    proxy_set_header X-Forwarded-Proto https;
    proxy_buffering off;
    proxy_request_buffering off;
    proxy_read_timeout 7d;
    proxy_send_timeout 7d;
}
If you’re using Cloudflare/ALB/Traefik: - Enable WebSockets, keep HTTP/1.1, and set sufficient idle timeouts. 2) Turn on debug to see the full handshake failure
Copy code
export PREFECT_DEBUG_MODE=1
Then reproduce the warning and check the detailed log. 3) Temporarily suppress the warnings (while you fix the proxy) - Disable event emission in-process:
Copy code
from prefect.events.worker import EventsWorker
from prefect.events.clients import NullEventsClient
EventsWorker.set_client_override(NullEventsClient)
- Or reduce the log level just for the events client:
Copy code
import logging
logging.getLogger("prefect.events.clients").setLevel(logging.ERROR)
4) Confirm server version matches client - If you’re self-hosting Prefect Server, ensure it’s 3.x to match your client (3.6.4). A 2.x server won’t have the same events endpoints and will 404. Why the warnings appear between your prints - 
Block.load(...)
emits an event like 
prefect.block.<type>.loaded
. The event client attempts a WebSocket connection and logs a warning when it’s rejected (404). Since you load multiple blocks, you see a warning after each value print. If you can share: - Your reverse proxy/load balancer type and config for /api (especially /api/events/in and /api/events/out) - Prefect Server version you’re running - The output with 
PREFECT_DEBUG_MODE=1
I can help pinpoint the exact fix. For background, see: - Events overview: Prefect 3 Events - Settings and profiles (for API URL and debug mode): Settings and Profiles
g
@Marvin can I create a flow with this signature?
Copy code
@flow(name=DAG_NAME)
def flow_entrypoint(dag_name: str, parameters: dict):
having a dictionary as one of the parameters
m
✅ thought for 183.9 seconds
Short answer: yes. In Prefect 3.x you can define a flow that takes a dict parameter. Just make sure the values you pass are JSON-serializable since parameters are sent to the API. Your example works:
Copy code
from prefect import flow

@flow(name=DAG_NAME)  # DAG_NAME must be a defined string at import time
def flow_entrypoint(dag_name: str, parameters: dict):
    # use parameters like a normal dict
    ...
Notes and best practices: - Dicts are supported; keep values JSON-serializable (str, int, float, bool, None, lists, nested dicts). - There’s a default parameter size limit (~512 KB). For large configs, store them externally (file/object store) and pass a reference (path/URL). - If you want validation and nicer UI forms, prefer a Pydantic model instead of a bare dict:
Copy code
from typing import Dict, Any
from pydantic import BaseModel, Field
from prefect import flow

class Params(BaseModel):
    settings: Dict[str, Any] = Field(default_factory=dict, description="Run settings")

@flow
def flow_entrypoint(dag_name: str, parameters: Params):
    ...
Docs: - Flows and parameters overview: Flows concept docs - Using Pydantic for structured parameters and UI forms: Form building
g
@Marvin how do I run another flow deployment from another deployment as a subflow which I can wait for from the triggering deployment?
m
✅ thought for 163.5 seconds
Yes—trigger another deployment from a flow and wait for it using run_deployment (sync) or arun_deployment (async). By default it links as a subflow in the UI (as_subflow=True), and it waits until the run reaches a final state unless you set timeout=0. Synchronous example (most common):
Copy code
from prefect import flow
from prefect.deployments import run_deployment

@flow
def orchestrator():
    sub_run = run_deployment(
        "child-flow/production",      # "flow_name/deployment_name" or a deployment UUID
        parameters={"x": 1, "y": 2},
        timeout=600,                  # wait up to 10 minutes (None waits indefinitely)
        poll_interval=5,              # seconds between polls
        as_subflow=True,              # show it as a subflow in the UI (default)
        flow_run_name="child-from-parent"
    )
    if sub_run.state.is_failed():
        raise RuntimeError(f"Child failed: {sub_run.state.message}")
    return sub_run.id
Async example:
Copy code
import asyncio
from prefect import flow
from prefect.deployments import arun_deployment

@flow
async def orchestrator_async():
    sub_run = await arun_deployment(
        "child-flow/production",
        parameters={"x": 1},
        timeout=300,
        poll_interval=2
    )
    return sub_run.id
Fire-and-forget (don’t wait):
Copy code
run_deployment("child-flow/production", timeout=0)
If you need the child flow’s return value, ensure the child persists its result and then fetch it via a future:
Copy code
from prefect import flow
from prefect.deployments import run_deployment
from prefect.futures import PrefectFlowRunFuture

@flow
def orchestrator():
    fr = run_deployment("child-flow/production", timeout=0)  # create run, return immediately
    future = PrefectFlowRunFuture(fr.id)
    result = future.result(raise_on_failure=True)  # requires the child to persist its result
    return result
And set the child to persist:
Copy code
from prefect import flow

@flow(persist_result=True)
def child_flow(...):
    return {"ok": True}
Docs: - API reference: run_deployment - Async API: arun_deployment
g
@Marvin what’s the difference between sync and async versions? The sync one waits and the async does not wait?
m
✅ thought for 16.0 seconds
Not quite—both versions can wait. The difference is how they wait. - run_deployment (sync): - Regular blocking function. - Use in normal (sync) flows/tasks. - Blocks the current thread until the child run reaches a final state or the timeout elapses. - Set timeout=0 to “fire-and-forget” and return immediately. - arun_deployment (async): - Coroutine you must await. - Use in async flows. - Non-blocking for the event loop while it waits. - Great for concurrent triggers with asyncio.gather. - Also supports timeout=0 to return immediately. Key point: Waiting behavior is controlled by timeout for both APIs: - timeout=None (default): wait until a final state - timeout=0: return immediately (don’t wait) - timeout=N: wait up to N seconds Examples:
Copy code
# Sync: blocks until done (or timeout)
from prefect import flow
from prefect.deployments import run_deployment

@flow
def parent():
    fr = run_deployment("child/production", parameters={"x": 1}, timeout=600)
    return fr.state
Copy code
# Async: await, and can run several at once
import asyncio
from prefect import flow
from prefect.deployments import arun_deployment

@flow
async def parent_async():
    runs = await asyncio.gather(
        arun_deployment("child/a", parameters={"x": 1}),
        arun_deployment("child/b", parameters={"x": 2}),
    )
    return [r.state for r in runs]
Docs: - run_deployment: API reference - arun_deployment: API reference
g
@Marvin so it is not the parameter as_subflow which makes it wait. I can run it also with as_subflow=False and it would wait too
m
✅ thought for 14.0 seconds
Correct. 
as_subflow
only controls the parent/child linking and UI lineage, not the waiting behavior. - 
timeout
controls waiting for both sync and async APIs: - 
timeout=None
(default) waits until the child run reaches a final state - 
timeout=0
returns immediately (don’t wait) - 
timeout=N
waits up to N seconds - 
as_subflow=True
(default) just links the created run as a subflow of the caller so it appears under the parent in the UI and lineage. - 
as_subflow=False
creates an independent run. You can still wait on it the same way. Example (wait without subflow linking):
Copy code
from prefect import flow
from prefect.deployments import run_deployment

@flow
def parent():
    fr = run_deployment(
        "child-flow/prod",
        parameters={"x": 1},
        as_subflow=False,   # no UI subflow link
        timeout=600         # still waits up to 10 minutes
    )
    return fr.state
API refs: - run_deployment: API reference - arun_deployment: API reference
g
@Marvin if I run it using as_subflow=False, is there a way from the subflow to know that it was triggered using run_deployment functions?
m
✅ thought for 75.2 seconds
Short answer: there’s no built‑in flag you can read in the child to know it was started via run_deployment when as_subflow=False. What you can detect at runtime: - If as_subflow=True (default): the child run will have a parent_task_run_id/parent_flow_run_id, so you can tell it’s linked as a subflow. - If as_subflow=False: there’s no parent linkage, and from inside the child it looks the same as any other deployment-triggered run (API/CLI/schedule/manual). You’ll see a deployment_id, but no way to tell the exact trigger source. Practical workarounds: - Tags: set a marker tag when triggering and check it inside the child.
Copy code
# parent
run_deployment("child/deploy", as_subflow=False, tags=["via-run-deployment"])

# child
from prefect.runtime import flow_run
if "via-run-deployment" in flow_run.tags:
    ...
- Sentinel parameter: pass an internal flag parameter.
Copy code
run_deployment("child/deploy", as_subflow=False, parameters={"_via_run_deployment": True})
- Custom naming convention: set flow_run_name with a recognizable prefix and inspect it in the child. These are reliable ways to differentiate, since the runtime context itself does not expose the trigger source when not linked as a subflow.
g
@Marvin if I use the tags parameter in run_deployment would it override the default tags or just append the tags I indicate in run_deployment? From a flow run, how do I get the flow run tags?
m
✅ thought for 77.7 seconds
- Passing tags to run_deployment/arun_deployment merges them with the deployment’s tags (set union). They do not override. - tags=None or tags=[] -> only the deployment’s tags are used - tags=["custom"] -> final tags = deployment.tags ∪ {"custom"} - Get the current flow run’s tags inside a flow:
Copy code
from prefect import flow
from prefect.runtime.flow_run import tags

@flow
def my_flow():
    current_tags = tags  # list[str]
    print(current_tags)
- Alternative (advanced):
Copy code
from prefect.context import FlowRunContext

@flow
def my_flow():
    ctx = FlowRunContext.get()
    tags_set = ctx.flow_run.tags  # set[str]
g
@Marvin runnin temp = await client.read_block_documents(offset=offset, limit=limit) gives json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0) in the terminal I’ve set PREFECT_API_URL as well
m
✅ thought for 118.5 seconds
That error usually means the client got something other than JSON back — often HTML or an empty response. For 
read_block_documents
, Prefect is calling the API endpoint for block documents and then doing 
response.json()
. If the response body is empty or is an HTML page, you’ll get:
Copy code
text
json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)
Most likely causes 1. 
PREFECT_API_URL
is pointing at the UI, not the API - Cloud: - wrong: 
<https://app.prefect.cloud>
- correct: 
<https://api.prefect.cloud/api/accounts/<account-id>/workspaces/<workspace-id>>
- Self-hosted: - wrong: 
<http://localhost:4200>
or 
<http://localhost:8080>
- correct: 
<http://localhost:4200/api>
or 
<http://localhost:8080/api>
2. Missing/invalid auth - Especially in Cloud, if 
PREFECT_API_KEY
is missing or bad, you may get a non-JSON response depending on what’s in front of the API. 3. A proxy / ingress / auth layer is returning HTML - For example a login page, gateway error page, etc. Quick checks First, print what Prefect is actually using:
Copy code
python
from prefect import get_client

async with get_client() as client:
    print(client.api_url)
Then test health / hello:
Copy code
python
from prefect import get_client

async with get_client() as client:
    print("API URL:", client.api_url)
    print(await client.api_healthcheck())
Or from the terminal:
Copy code
bash
curl "$PREFECT_API_URL/health"
If that returns HTML, an empty body, or a redirect page, the URL is wrong or something is intercepting the request. What 
read_block_documents
does It hits the block documents API (
/block_documents/filter
) and expects JSON back. So if the base URL is off, this will fail exactly the way you’re seeing. What to verify - 
PREFECT_API_URL
ends with 
/api
for server/self-hosted - For Cloud, 
PREFECT_API_URL
uses 
api.prefect.cloud/.../api/accounts/.../workspaces/...
- 
PREFECT_API_KEY
is set if you’re using Cloud - You’re not accidentally using the browser/UI URL Docs - Configure API access - Prefect API concepts - Prefect Cloud UI If you want, paste your 
PREFECT_API_URL
value here with the IDs/token redacted, and I can tell you if the format is correct.
g
@Marvin I’m reading block_document table from the databaset in Prefect and I see data is a string. How do I get the actual content of the block? Blocks I’m trying to read are not credentials
m
✅ thought for 84.1 seconds
Yes — if you read 
block_document
directly from the Prefect DB, the 
data
column is not plain JSON. In Prefect 3.x, block document data is stored encrypted at rest, so what you’re seeing as a string is expected. What’s happening - 
block_document.data
is stored as encrypted JSON - Prefect uses a Fernet key to encrypt/decrypt it - This applies to all block documents, not just credentials blocks So even for non-credential blocks, raw DB reads will show an encrypted string rather than the decoded content. How to get the actual content You need the Prefect server encryption key, then decrypt the string and parse the JSON. The key comes from one of these places: - 
PREFECT_SERVER_ENCRYPTION_KEY
environment variable, if set - otherwise the 
configuration
table entry for 
ENCRYPTION_KEY
High-level process 1. Read the encrypted 
data
value from 
block_document
2. Read the Fernet key 3. Decrypt the string 4. JSON-decode the result Example:
Copy code
import json
from cryptography.fernet import Fernet

encrypted_data = "<value from block_document.data>"

fernet_key = "<your fernet key>"
fernet = Fernet(fernet_key.encode())

decrypted_bytes = fernet.decrypt(encrypted_data.encode())
block_data = json.loads(decrypted_bytes.decode())

print(block_data)
Where the key is stored If Prefect is not using 
PREFECT_SERVER_ENCRYPTION_KEY
, check the 
configuration
table for the 
ENCRYPTION_KEY
record. The value is stored like:
Copy code
{
  "fernet_key": "<base64-key>"
}
So you’d use the nested 
fernet_key
value. Important caveat If you want to know which fields are considered secret vs non-secret for API-style masking, that comes from the block schema (
block_schema.fields.secret_fields
). But for raw DB decryption, you can still decrypt the whole payload without needing the schema. Better option than reading the DB directly If possible, use the Prefect API/SDK to read the block document instead of querying the table yourself. Prefect will handle decryption for you. If you want, I can also give you: - a SQL query to pull both the encrypted block row and the encryption key, or - a complete Python script that reads from Postgres and decrypts 
block_document.data
end-to-end.
g
@Marvin I’m trying to use patch /block_documents/{id} endpoint without success. How should it be used?
m
✅ thought for 89.3 seconds
PATCH /block_documents/{id}
is for updating the block document's stored data and optionally its 
block_schema_id
. The most important thing is: the body is not the full block document object. It should be an update payload like this:
Copy code
{
  "data": {
    "some_field": "new value"
  },
  "merge_existing_data": true
}
Supported fields in the PATCH body - `data`: object of block field values to update - `merge_existing_data`: boolean, default 
true
- `block_schema_id`: optional, only for migrating to another schema of the same block type What 
merge_existing_data
does - 
true
(default): partial update / merge - `false`: replace existing stored data with exactly what you send Example: Current block data:
Copy code
{
  "a": 1,
  "b": 2,
  "c": 3
}
PATCH body:
Copy code
{
  "data": {
    "b": 99
  },
  "merge_existing_data": true
}
Result:
Copy code
{
  "a": 1,
  "b": 99,
  "c": 3
}
If instead:
Copy code
{
  "data": {
    "b": 99
  },
  "merge_existing_data": false
}
Result becomes:
Copy code
{
  "b": 99
}
Very common reason it “doesn’t work” People often send the wrong shape, such as: - the full block document returned by GET - top-level fields that belong inside 
data
- a body missing the 
data
wrapper For example, this is wrong:
Copy code
{
  "some_field": "new value"
}
This is right:
Copy code
{
  "data": {
    "some_field": "new value"
  }
}
Success response On success, Prefect returns:
Copy code
204 No Content
So you won’t get an updated block back in the response body. Secrets behavior If the block has secret fields and you previously fetched it with obfuscated secrets, Prefect tries to preserve those secrets rather than overwrite them with masked values. So patching a block you got from the API is generally safe. Things you cannot update with this endpoint You cannot patch: - 
name
- 
is_anonymous
- 
block_type_id
- metadata like 
id
, 
created
, 
updated
Possible failure cases - `404`: block document ID does not exist - `400`: invalid payload or invalid schema migration - `409`: invalid referenced block if your data includes block references Example curl
Copy code
curl -X PATCH "http://<prefect-api>/block_documents/<BLOCK_DOCUMENT_ID>" \
  -H "Content-Type: application/json" \
  -d '{
    "data": {
      "my_field": "updated value"
    },
    "merge_existing_data": true
  }'
If you want, paste the exact payload you’re sending to 
PATCH /block_documents/{id}
and the response/error you get, and I can tell you exactly what’s wrong.
g
@Marvin passing just {“data”: <my dict config>} returns 204
m
✅ thought for 9.6 seconds
That means it is succeeding. For this endpoint, 
204 No Content
is the expected success response for a PATCH. It does not return the updated block in the response body. So if you're sending:
Copy code
{
  "data": {
    ...
  }
}
and getting 
204
, Prefect accepted the update. How to verify the change Do a follow-up GET for that block document and inspect the returned data. If you're using the API directly, read the same block document again after the PATCH. A couple of important details: - With only:
Copy code
{
    "data": {...}
  }
Prefect uses 
merge_existing_data=true
by default - So your payload is merged into the existing block data, not a full replacement - If you want to fully replace the stored config, send:
Copy code
{
  "data": {
    ...
  },
  "merge_existing_data": false
}
Why it may look like nothing changed A few common reasons: 1. You expected the PATCH response to include the updated block - It won’t; 
204
means success with no body 2. You are re-reading from the DB directly - 
block_document.data
in the database is encrypted, so you won’t see plain values there 3. You expected omitted keys to be removed - They are preserved unless you set 
merge_existing_data
to 
false
4. You are looking at secret fields - GET responses may obfuscate secret values In short - 
204
= success - Your payload shape is valid - Re-fetch the block to confirm - Use 
merge_existing_data: false
if you want replacement instead of merge If you want, I can show you the exact 
GET
endpoint to call afterward, or help confirm whether your update is being merged vs replaced.
g
@Marvin I don’t see the changes in the block in Prefect
m
✅ thought for 9.6 seconds
If 
PATCH
returns 
204
but the block still looks unchanged in the UI, the most likely cause is this: You are patching the block document directly, but the UI may be showing a hydrated/typed view or you're updating only a subset while merge behavior preserves the old values. A few concrete things to check: 1. Confirm you are patching the correct block document ID It’s easy to patch one block document and inspect another in the UI. If you can, first GET the block document by ID, then PATCH that same ID, then GET it again. 2. Check whether your update is being merged instead of replaced If you send:
Copy code
{
  "data": {
    "x": 2
  }
}
that defaults to:
Copy code
{
  "data": {
    "x": 2
  },
  "merge_existing_data": true
}
So only 
x
changes; everything else remains. If the UI form derives values from other fields, it may look unchanged unless you replace the full config:
Copy code
{
  "data": {
    "x": 2,
    "y": 3
  },
  "merge_existing_data": false
}
3. Make sure your fields belong under 
data
exactly as stored by the block schema The keys in 
data
must match the block schema fields. If the UI uses different labels than the stored field names, you may be updating a field the UI doesn’t render the way you expect. 4. Remember the DB value is encrypted If you’re checking 
block_document.data
in Postgres, that won’t be human-readable. You need to verify via the API/UI, not raw DB contents. 5. Secret fields may appear masked If you update a secret-like field, the UI may still display it as masked, so it can be hard to tell whether it changed. 6. Some blocks may involve nested block references If part of the visible config in the UI comes from a referenced block, patching this block document’s local 
data
may not affect what you expect. Best way to debug Compare the block before and after with GET calls. Before:
Copy code
GET /block_documents/{id}
Patch:
Copy code
PATCH /block_documents/{id}
{
  "data": {
    ...
  }
}
After:
Copy code
GET /block_documents/{id}
If the GET-after response is unchanged, then one of these is true: - wrong block ID - wrong field names - values are being merged and not replacing what you expect - the visible UI values come from referenced/nested data If the GET-after response does show the change, then it’s more likely a UI interpretation/caching issue rather than the PATCH failing. What I’d try next Send a very obvious change to a non-secret top-level field and force replacement:
Copy code
{
  "data": {
    "your_field": "TEST_PATCH_VALUE_123"
  },
  "merge_existing_data": false
}
Then immediately GET the same block document ID. If you want, paste: - the block type - your PATCH payload - the GET response before/after and I can tell you exactly whether the payload shape is correct for that block.
g
@Marvin the endpoint /api/deployments/{deployment_id}/create_flow_run has a parameter called parameters. If I set it to empty list, is it the same as not passing at all? If I set it to None, is it the same as not passing at all?
m
✅ thought for 229.9 seconds
Short answer: - 
parameters = []
→ no, not the same; it's invalid - 
parameters = None
→ no, not the same in raw REST; it's invalid - omitting 
parameters
entirely → valid - 
parameters = {}
→ valid, and functionally the same as omitting it What the endpoint expects 
parameters
must be a JSON object / dictionary, not a list. Valid:
Copy code
{
  "parameters": {
    "x": 1
  }
}
Also valid:
Copy code
{
  "parameters": {}
}
Invalid:
Copy code
{
  "parameters": []
}
Invalid:
Copy code
{
  "parameters": null
}
Semantics For `POST /deployments/{deployment_id}/create_flow_run`: - if 
parameters
is omitted, Prefect treats it like 
{}
and just uses the deployment’s saved parameters - if 
parameters
is 
{}
, same practical result: no overrides, deployment parameters are used - if 
parameters
contains keys, those keys override the deployment’s parameters So: Equivalent in practice
Copy code
{}
and
Copy code
{
  "parameters": {}
}
Not equivalent / invalid
Copy code
{
  "parameters": null
}
Copy code
{
  "parameters": []
}
One subtle note about the Python client If you use the Python client and pass 
parameters=None
, the client typically normalizes that to 
{}
before sending. So in Python client usage, 
None
may appear to work. But in the raw API, 
null
is not valid for this field. If you want, I can also show the exact request/response examples for: - raw REST - Python client - what happens when deployment defaults are present.
g
@Marvin I have a deployment of a flow which is attached to a work queue. The queue has a limit. Is it possible to discard new flow runs when the queue is full instead of putting in schedule state?
m
✅ thought for 136.4 seconds
Short answer: not via work queue limits alone. In Prefect 3.x, a work queue limit is a soft polling/execution limit, not a creation-time rejection limit. So when the queue is “full”: - new flow runs can still be created - they remain in 
Scheduled
- workers just won’t pull more runs until slots open up So there is no built-in “drop if queue full” behavior for work queues. What you can do instead Option 1: Use a deployment concurrency limit with 
CANCEL_NEW
 If your goal is “don’t let excess runs wait; discard them”, this is the closest native feature. With deployment concurrency limits: - 
ENQUEUE
→ extra runs wait - 
CANCEL_NEW
→ extra runs are canceled instead of waiting This is usually the best fit if the behavior should apply per deployment. Option 2: Add your own pre-check before creating the run If you control the caller that creates flow runs, you can: 1. inspect queue/load 2. decide not to call 
create_flow_run
3. return your own “queue full” behavior That prevents creation entirely, but it’s custom logic, not server-enforced. Option 3: Use an automation to cancel newly scheduled runs You could create an automation that watches for new scheduled runs and cancels them when your capacity condition is met. This works, but it’s: - more indirect - eventually consistent - less clean than deployment concurrency with 
CANCEL_NEW
Important distinction - Work queue concurrency limit = controls how many runs workers receive - Deployment concurrency limit = controls whether new runs are allowed to proceed So if you need reject/discard semantics, look at deployment concurrency, not queue concurrency. There weren’t any matching GitHub issues/discussions I found on a built-in work-queue “discard overflow” mode. If helpful, I can show you: - how to configure a deployment concurrency limit with 
CANCEL_NEW
, or - how to model this when multiple deployments share one queue.
g
@Marvin Is there a concurrency limit tag to apply on deployment level instead on task level?
m
✅ thought for 141.2 seconds
Not with tags in the same way as task tag concurrency limits. Current distinction in Prefect 3.x - Task-level tag concurrency limits apply to tasks with matching tags - Deployments / flow runs use deployment concurrency limits, not tag concurrency limits So if you want to limit concurrent flow runs for a deployment, use the deployment’s concurrency settings. Use deployment concurrency for per-deployment limits This is the native deployment-level mechanism. You can configure: - 
concurrency_limit
- 
collision_strategy
like: - 
ENQUEUE
- 
CANCEL_NEW
That gives you deployment-level control over flow run concurrency. If you want one shared limit across multiple deployments There is not a “deployment tag concurrency limit” feature analogous to task tags. Instead, the usual alternatives are: 1. Global concurrency limits - good when several deployments share one resource/API/db - enforced in flow/task code using concurrency primitives 2. Separate deployment concurrency limits - if each deployment should have its own cap 3. Work pool / work queue limits - if you want infrastructure-side throttling - note these do not reject new runs; they just delay execution Important gotcha Flow/deployment tags are mainly for organization/filtering. They do not create flow-run concurrency enforcement the way task tags do. Rule of thumb - limit tasks by tag → task tag concurrency limits - limit one deployment’s flow runs → deployment concurrency limit - limit multiple deployments sharing a resource → global concurrency limit pattern If you want, I can show: - an example deployment concurrency config, or - how to implement a shared cross-deployment limit with global concurrency.
g
@Marvin the thing is that I have multiple deployments which I need to limit and the only way to have a common queue looks to be by tagging them. Also, such deployments are limited by their queue and I cannot use one queue
m
✅ thought for 8.4 seconds
Yes — if you need one shared limit across multiple deployments, a deployment concurrency limit won’t help by itself because that limit is per deployment. And you’re also right that if those deployments can’t all share a single work queue, then queue concurrency is not a usable common limiter. In that situation, the best fit is usually: use a shared global concurrency limit inside the flow (or task), not queue limits and not deployment limits. That gives you one common cap across many deployments, regardless of which queue they run on. Why this works All those deployments can acquire the same named concurrency limit when they start work. For example, if these deployments all hit the same external system, you can make them all share 
"my-shared-limit"
. Conceptually:
Copy code
Deployment A  \
Deployment B   > all acquire "shared-limit"
Deployment C  /
Only 
N
runs can proceed at once. Pattern Inside the flow or an early task, acquire a named global concurrency slot. Example:
Copy code
from prefect import flow
from prefect.concurrency.sync import concurrency

@flow
def my_flow():
    with concurrency("shared-deployment-limit", occupy=1):
        # protected section
        do_the_work()
Then create the global limit once in Prefect. This is the usual way to express: - many deployments - different queues - one shared cap Important behavior difference This does not prevent flow runs from being created. It prevents more than the allowed number from entering the protected section at the same time. So the extra runs will wait there, rather than being rejected at queue creation time. If you need discard instead of wait, you’d need custom logic, because shared queue/tag-level “cancel new” behavior does not exist out of the box across multiple deployments. If you want discard semantics across multiple deployments Your options are more custom: 1. Client-side pre-check before creating runs - check current occupancy - skip creating new flow runs if full 2. Automation that cancels new runs - create runs normally - cancel them if the shared-cap condition is exceeded 3. Custom first-step logic in the flow - inspect current shared occupancy - immediately exit/cancel if above threshold But for shared throttling across deployments, global concurrency is the cleanest built-in option. Bottom line Given: - multiple deployments - cannot use one queue - need one common limit Use global concurrency limits in flow/task code. If you want, I can give you a concrete example of: - how to create the shared global limit, and - how to structure the flow so the whole run is effectively gated by that limit.
g
@Marvin how do I get a tag concurrency limit via api? If I go in Concurrency -> Global concurrency limit, I see the name is tag:my_tag
m
✅ thought for 8.8 seconds
Yes — in current Prefect 3.x, tag concurrency limits are backed by global concurrency limits, so a tag limit appears as a global concurrency limit named:
Copy code
tag:my_tag
So to get a tag concurrency limit via API, query the global concurrency limit by that name. What to look up For tag 
my_tag
, use:
Copy code
tag:my_tag
API shape You want the global concurrency limit endpoints, not a separate “tag limit” endpoint. If you're calling the REST API directly, the flow is generally: 1. list/filter global concurrency limits 2. find the one whose name is 
tag:my_tag
If you want to query by exact name, use 
tag:my_tag
. Important note This is an implementation detail that is now effectively the real storage model for tag limits in Prefect 3.x. So: - task tag concurrency limit 
my_tag
- underlying global concurrency limit name 
tag:my_tag
Practical answer If you see it in the UI under Concurrency → Global concurrency limits as 
tag:my_tag
, that is the object you should read through the API. If you want, I can give you: - the exact REST endpoint/body to read it, or - the Python client code to fetch 
tag:my_tag
.
g
@Marvin is it too much effort for you to show a request example? Oh my
m
✅ thought for 6.9 seconds
Use the global concurrency limit API and look up 
tag:my_tag
. Example request:
Copy code
POST /api/v2/concurrency_limits/filter
Content-Type: application/json

{
  "concurrency_limit": {
    "name": {
      "any_": ["tag:my_tag"]
    }
  },
  "limit": 10,
  "offset": 0
}
Example `curl`:
Copy code
curl -X POST "https://<your-prefect-api>/api/v2/concurrency_limits/filter" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <API_KEY>" \
  -d '{
    "concurrency_limit": {
      "name": {
        "any_": ["tag:my_tag"]
      }
    },
    "limit": 10,
    "offset": 0
  }'
You should get back the global concurrency limit object for that tag. If you already know the ID, you can also read it directly by ID from the global concurrency limit endpoints. Sorry — you were right to ask for the concrete request.
g
@Marvin I created a function with this decorator
Copy code
@task(name="run_task", retries=1, retry_delay_seconds=30, retry_jitter_factor=1)
def run_task(run_options: dict, wait_for_running_task: bool = True, stop_running_tasks: bool = True):
I’m running it like this
m
✅ thought for 10.6 seconds
Can you paste the actual call site too? For example, I’d need to see something like:
Copy code
run_task(...)
or
Copy code
run_task.submit(...)
or however you’re invoking it, plus what behavior/error you’re seeing. A few likely things depending on how you’re calling it: If you call it inside a flow normally
Copy code
run_task(run_options=my_dict)
that’s valid. If you submit it asynchronously
Copy code
run_task.submit(run_options=my_dict)
also valid. If you pass positional args incorrectly your signature is:
Copy code
def run_task(
    run_options: dict,
    wait_for_running_task: bool = True,
    stop_running_tasks: bool = True,
)
So: - first positional arg = 
run_options
- second positional arg = 
wait_for_running_task
- third positional arg = 
stop_running_tasks
Example:
Copy code
run_task(my_dict, False, True)
If you’re trying to set retries dynamically at call time that won’t happen through the function arguments; retries are on the decorator/task config, not normal parameters. One important gotcha If this is being called outside a flow, then 
run_task()
will execute, but 
.submit()
requires a flow/task-run context. Please send: 1. the invocation code 2. whether it’s inside a 
@flow
3. the exact error or unexpected behavior and I’ll pinpoint it quickly.
g
@Marvin I created a function with this decorator
Copy code
@task(name="run_task", retries=1, retry_delay_seconds=30, retry_jitter_factor=1)
def run_task(run_options: dict, wait_for_running_task: bool = True, stop_running_tasks: bool = True):
I’m running it like this
Copy code
run_task.submit(run_options, wait_for=[second], tags=tags)
but I’ve got
Copy code
an unexpected keyword argument 'tags'
how can I pass tags programmatically?
m
✅ thought for 113.7 seconds
tags
is not a supported argument to 
.submit()
. So this:
Copy code
run_task.submit(run_options, wait_for=[second], tags=tags)
fails because Prefect treats 
tags=...
as if you're passing a keyword argument to the underlying Python function, and your task function does not have a 
tags
parameter. How to pass tags programmatically Use the 
tags()
context manager:
Copy code
from prefect import tags

with tags(*tags):
    run_task.submit(run_options, wait_for=[second])
If 
tags
is a list like:
Copy code
tags = ["foo", "bar"]
then:
Copy code
with tags(*tags):
    run_task.submit(run_options, wait_for=[second])
will apply those tags to that task run. If you want tags fixed on the task object itself You can create a task variant with `with_options()`:
Copy code
tagged_run_task = run_task.with_options(tags=tags)
tagged_run_task.submit(run_options, wait_for=[second])
Difference between the two approaches - 
with tags(...):
→ dynamic runtime context, usually best for per-call tagging - 
with_options(tags=...)
→ creates a modified task object with those tags Important note If your task already has tags from the decorator, runtime tags from 
with tags(...)
are combined with them. Working example
Copy code
from prefect import flow, task, tags

@task(name="run_task", retries=1, retry_delay_seconds=30, retry_jitter_factor=1)
def run_task(run_options: dict, wait_for_running_task: bool = True, stop_running_tasks: bool = True):
    return run_options

@flow
def my_flow():
    second = run_task.submit({"step": 2})

    dynamic_tags = ["team-a", "special-run"]

    with tags(*dynamic_tags):
        third = run_task.submit({"step": 3}, wait_for=[second])

    return third.result()
So the direct answer is:
Copy code
with tags(*tags):
    run_task.submit(run_options, wait_for=[second])
If you want, I can also show how to apply different tags to each mapped/submitted task in a loop.
g
@Marvin why wait_for works? I don’t have wait_for parameter in my function and it works
m
✅ thought for 7.4 seconds
Because 
wait_for
is a special argument to Prefect’s 
.submit()
, not an argument to your task function. So in:
Copy code
run_task.submit(run_options, wait_for=[second])
Prefect interprets: - 
run_options
→ argument for your function - 
wait_for=[second]
→ submission metadata telling Prefect about an upstream dependency It does not pass 
wait_for
into:
Copy code
def run_task(run_options: dict, wait_for_running_task: bool = True, stop_running_tasks: bool = True):
That’s why it works even though your function has no 
wait_for
parameter. Why 
tags
fails but 
wait_for
works Because 
.submit()
explicitly supports 
wait_for
, but does not explicitly support 
tags
. So: - 
wait_for
is consumed by Prefect itself - 
tags
falls through as a normal keyword argument - then Prefect tries to pass 
tags
to your task function - your function does not accept 
tags
- result: 
unexpected keyword argument 'tags'
Conceptually 
.submit()
behaves like:
Copy code
task.submit(
    *task_args,
    wait_for=[...],      # Prefect-controlled
    return_state=False,  # Prefect-controlled
    **task_kwargs        # passed to your function
)
So your call:
Copy code
run_task.submit(run_options, wait_for=[second], tags=tags)
is effectively treated like: - special submit option: 
wait_for=[second]
- task kwarg: 
tags=tags
and that causes the error. Rule of thumb When using `.submit()`: - 
wait_for
and 
return_state
are special Prefect submission options - everything else should match your task function signature unless documented otherwise If you want, I can show the mental model for 
.submit()
vs direct function call vs 
.map()
too.
g
@Marvin a post on /api/v2/concurrency_limits/filter with this payload { “limit”: 2, “offset”: 0, “concurrency_limit”: { “name”: { “any_“: [“tag:my_tag”] } } } is returning all concurrency limit tag set. It is not filtering by name
m
✅ thought for 100.0 seconds
You’re right — that payload does not filter, because this endpoint doesn’t support a name filter in the request body. For Prefect 3.x, 
POST /api/v2/concurrency_limits/filter
only takes pagination fields like:
Copy code
{
  "limit": 10,
  "offset": 0
}
So this part:
Copy code
"concurrency_limit": {
  "name": {
    "any_": ["tag:my_tag"]
  }
}
is ignored, which is why you get all limits back. What to use instead Use the direct read endpoint by name:
Copy code
GET /api/v2/concurrency_limits/tag:my_tag
Example with `curl`:
Copy code
curl -X GET "https://<your-prefect-api>/api/v2/concurrency_limits/tag:my_tag" \
  -H "Authorization: Bearer <API_KEY>"
That’s the correct way to fetch the global concurrency limit for a tag-backed limit like 
tag:my_tag
. Two extra notes 1. Your example also uses typographic quotes (
“ ”
) instead of normal JSON quotes (
"
). If that was copied literally, it would not be valid JSON. 2. Even with valid JSON quotes, the filter body still would not work for this endpoint because name filtering is simply not supported there. So the short answer is: - 
POST /api/v2/concurrency_limits/filter
→ list/paginate all - 
GET /api/v2/concurrency_limits/tag:my_tag
→ fetch the specific tag limit If you want, I can also show the equivalent Python client call.
g
@Marvin how do I get the flow name of a deployment?
m
✅ thought for 85.6 seconds
In Prefect 3.x, a deployment gives you the 
flow_id
, not the flow name directly. So the usual way is: 1. 
GET /api/deployments/{deployment_id}
2. read 
flow_id
3. 
GET /api/flows/{flow_id}
4. read 
name
Example Get the deployment:
Copy code
curl -X GET "https://<your-prefect-api>/api/deployments/<deployment_id>" \
  -H "Authorization: Bearer <API_KEY>"
That response includes something like:
Copy code
{
  "id": "...",
  "name": "my-deployment",
  "flow_id": "12345678-1234-1234-1234-123456789abc"
}
Then fetch the flow:
Copy code
curl -X GET "https://<your-prefect-api>/api/flows/12345678-1234-1234-1234-123456789abc" \
  -H "Authorization: Bearer <API_KEY>"
Response:
Copy code
{
  "id": "12345678-1234-1234-1234-123456789abc",
  "name": "my-flow"
}
So the flow name is:
Copy code
my-flow
Important note The deployment response does not include 
flow_name
directly, only 
flow_id
. If you want, I can also show: - how to do this in one small Python snippet, or - how to get a deployment by 
flow_name/deployment_name
instead.
g
@Marvin how do I get the flow name given I have only the deployment name?
m
✅ thought for 6.6 seconds
If you only have the deployment name, you need to know whether by “deployment name” you mean: 1. the full deployment name key in the form 
flow_name/deployment_name
, or 2. just the deployment’s local name like 
prod
That distinction matters. If you have the full name 
flow_name/deployment_name
 Then you already have the flow name — it’s the part before the 
/
. Example: - full deployment name: 
my-flow/prod
- flow name: 
my-flow
- deployment name: 
prod
If you only have the local deployment name Example: just 
prod
Then there is not a direct unique lookup by deployment name alone, because deployment names are scoped to flows. Multiple flows can each have a deployment named 
prod
. So in that case you need to: 1. list/filter deployments 2. find the matching deployment(s) by 
name
3. read the 
flow_id
4. fetch the flow(s) by 
flow_id
5. read the flow 
name
REST approach List deployments:
Copy code
POST /api/deployments/filter
Example:
Copy code
curl -X POST "https://<your-prefect-api>/api/deployments/filter" \
  -H "Authorization: Bearer <API_KEY>" \
  -H "Content-Type: application/json" \
  -d '{
    "limit": 100,
    "offset": 0
  }'
Then filter client-side for:
Copy code
.name == "prod"
For each match, take its 
flow_id
, then call:
Copy code
GET /api/flows/{flow_id}
Example:
Copy code
curl -X GET "https://<your-prefect-api>/api/flows/<flow_id>" \
  -H "Authorization: Bearer <API_KEY>"
Why this is necessary In Prefect 3.x: - deployment objects have 
name
- deployment objects have 
flow_id
- deployment objects do not include 
flow_name
- and deployment 
name
alone is not guaranteed globally unique So the short answer - if you have 
flow_name/deployment_name
, split on 
/
- if you only have 
deployment_name
, you must search deployments, then follow 
flow_id
to the flow If you want, I can give you: - a 
jq
one-liner for this, or - a Python snippet that takes a deployment name and returns all matching flow names.
g
@Marvin I have only deployment_name of flow_name/deployment_name, how do I list all deployments which have a given deployment_name without the flow_name part?
m
✅ thought for 100.4 seconds
Yes — you can do this with 
POST /api/deployments/filter
by filtering on the deployment’s 
name
field only. If you want all deployments whose deployment name part is 
my_deployment
, use:
Copy code
curl -X POST "https://<your-prefect-api>/api/deployments/filter" \
  -H "Authorization: Bearer <API_KEY>" \
  -H "Content-Type: application/json" \
  -d '{
    "deployments": {
      "name": {
        "any_": ["my_deployment"]
      }
    },
    "limit": 100,
    "offset": 0
  }'
That filters on 
Deployment.name
, not 
flow_name/deployment_name
as a combined string. If you want partial matching instead Use `like_`:
Copy code
curl -X POST "https://<your-prefect-api>/api/deployments/filter" \
  -H "Authorization: Bearer <API_KEY>" \
  -H "Content-Type: application/json" \
  -d '{
    "deployments": {
      "name": {
        "like_": "my_deployment"
      }
    },
    "limit": 100,
    "offset": 0
  }'
Useful note The response will include each matching deployment’s 
flow_id
, so if you want the flow name too, you then call:
Copy code
GET /api/flows/{flow_id}
So the direct answer is: yes, server-side filtering by deployment name alone is supported. If you want, I can also give you a 
jq
pipeline that takes the filtered deployments and prints 
flow_name/deployment_name
.
g
@Marvin I’m getting this error when I run Prefect 3.6.4 Jun 18 103732 ip-172-30-1-85 start_prefect.sh[27545]: prefect_agent | Traceback (most recent call last): Jun 18 103732 ip-172-30-1-85 start_prefect.sh[27545]: prefect_agent | File “/usr/local/bin/prefect”, line 3, in <module> Jun 18 103732 ip-172-30-1-85 start_prefect.sh[27545]: prefect_agent | from prefect.cli import app Jun 18 103732 ip-172-30-1-85 start_prefect.sh[27545]: prefect_agent | File “/usr/local/lib/python3.11/site-packages/prefect/cli/__init__.py”, line 29, in <module> Jun 18 103732 ip-172-30-1-85 start_prefect.sh[27545]: prefect_agent | import prefect.cli.work_pool Jun 18 103732 ip-172-30-1-85 start_prefect.sh[27545]: prefect_agent | File “/usr/local/lib/python3.11/site-packages/prefect/cli/work_pool.py”, line 43, in <module> Jun 18 103732 ip-172-30-1-85 start_prefect.sh[27545]: prefect_agent | from prefect.workers.utilities import ( Jun 18 103732 ip-172-30-1-85 start_prefect.sh[27545]: prefect_agent | File “/usr/local/lib/python3.11/site-packages/prefect/workers/__init__.py”, line 1, in <module> Jun 18 103732 ip-172-30-1-85 start_prefect.sh[27545]: prefect_agent | from .process import ProcessWorker Jun 18 103732 ip-172-30-1-85 start_prefect.sh[27545]: prefect_agent | File “/usr/local/lib/python3.11/site-packages/prefect/workers/process.py”, line 42, in <module> Jun 18 103732 ip-172-30-1-85 start_prefect.sh[27545]: prefect_agent | from prefect.workers.base import ( Jun 18 103732 ip-172-30-1-85 start_prefect.sh[27545]: prefect_agent | File “/usr/local/lib/python3.11/site-packages/prefect/workers/base.py”, line 26, in <module> Jun 18 103732 ip-172-30-1-85 start_prefect.sh[27545]: prefect_agent | from importlib_metadata import ( Jun 18 103732 ip-172-30-1-85 start_prefect.sh[27545]: prefect_agent | ModuleNotFoundError: No module named ‘importlib_metadata’
m
✅ thought for 119.4 seconds
This is a Prefect 3.6.x bug/packaging issue, not something wrong with your code. In Prefect 
3.6.4
, 
prefect.workers.base
imports:
Copy code
from importlib_metadata import distributions
but 
importlib_metadata
may not actually be installed in your environment, which causes:
Copy code
ModuleNotFoundError: No module named 'importlib_metadata'
Best fix Upgrade Prefect to a version where this was fixed. In newer versions, Prefect uses the stdlib import instead.
Copy code
pip install --upgrade "prefect>=3.7.2"
If you need to stay on 3.6.4 Install the missing package explicitly:
Copy code
pip install importlib_metadata
That should unblock the CLI immediately. Why this happens on Python 3.11 Python 3.11 already has 
importlib.metadata
in the standard library, so the backport package should not be necessary. But Prefect 3.6.4 still imports the backport package name in this code path. Good next steps 1. If possible, upgrade Prefect:
Copy code
pip install --upgrade prefect
2. If you must stay pinned to 
3.6.4
, install:
Copy code
pip install importlib_metadata
3. If this is inside a service/container, rebuild or restart after installing so the 
prefect
executable sees the updated environment. I didn’t find an existing GitHub issue for this exact error, but I did verify the import behavior and that it was changed in later Prefect versions. If you want, I can help you determine the safest exact upgrade target from 
3.6.4
, or help check whether your service is installing Prefect into a different Python environment than the one you updated.
g
what’s fastapi version compatible with prefect 3.7.4?
@Marvin what’s fastapi version compatible with prefect 3.7.4?
m
✅ thought for 71.1 seconds
For Prefect 3.7.4, the declared FastAPI compatibility is:
Copy code
fastapi>=0.111.0,<1.0.0
So any FastAPI version in that range is compatible by Prefect’s package constraints. Related constraints Prefect 3.7.4 also declares - 
pydantic>=2.10.1,<3.0.0
- excluding 
2.11.0
through 
2.11.4
- 
starlette>=1.0.1
Recommendation If possible, don’t pin FastAPI separately unless you have to. Let 
prefect==3.7.4
resolve its own compatible FastAPI version:
Copy code
pip install prefect==3.7.4
If you do need to pin FastAPI manually, stay within:
Copy code
fastapi>=0.111.0,<1.0.0
If you want, I can also help you find a specific FastAPI + Starlette + Pydantic combination that’s safest with Prefect 3.7.4.
g
@Marvin Prefect 3.7.4 is giving refect_orion | File “/usr/local/lib/python3.11/site-packages/starlette/routing.py”, line 677, in app prefect_orion | match, child_scope = route.matches(scope) prefect_orion | ^^^^^^^^^^^^^^^^^^^^ prefect_orion | File “/usr/local/lib/python3.11/site-packages/fastapi/routing.py”, line 1641, in matches prefect_orion | match, _ = self.original_router.matches(scope) prefect_orion | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ prefect_orion | File “/usr/local/lib/python3.11/site-packages/fastapi/routing.py”, line 2112, in matches prefect_orion | match, child_scope, _, _ = included_router._match(scope) prefect_orion | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ prefect_orion | File “/usr/local/lib/python3.11/site-packages/fastapi/routing.py”, line 1602, in _match prefect_orion | for candidate in self.effective_candidates(): prefect_orion | ^^^^^^^^^^^^^^^^^^^^^^^^^^^ prefect_orion | File “/usr/local/lib/python3.11/site-packages/fastapi/routing.py”, line 1508, in effective_candidates prefect_orion | routes_version = self.original_router._get_routes_version() prefect_orion | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ prefect_orion | File “/usr/local/lib/python3.11/site-packages/fastapi/routing.py”, line 2047, in _get_routes_version prefect_orion | for route in self.routes: prefect_orion | ^^^^^^^^^^^ prefect_orion | AttributeError: ‘PrefectRouter’ object has no attribute ‘routes’
m
✅ thought for 185.8 seconds
This is a known incompatibility between Prefect 3.7.4 and FastAPI 0.137.0+. Your error:
Copy code
AttributeError: 'PrefectRouter' object has no attribute 'routes'
matches that exact breakage. What’s happening Prefect 3.7.4 assumes older FastAPI router behavior. FastAPI 
0.137.0+
changed router inclusion internals, and that breaks Prefect 3.7.4 at request time. What to do Best option: upgrade Prefect Upgrade to a version that contains the compatibility fix:
Copy code
pip install --upgrade "prefect>=3.7.5"
If you must stay on Prefect 3.7.4 Pin FastAPI below 
0.137.0
, for example:
Copy code
pip install "prefect==3.7.4" "fastapi==0.136.3"
That is the safest short-term workaround. Recommended version pairings - 
prefect==3.7.4
+ 
fastapi==0.136.3
→ should work - 
prefect>=3.7.5
+ 
fastapi>=0.137.0
→ should work - 
prefect==3.7.4
+ 
fastapi==0.137.0+
→ broken Check what you currently have
Copy code
pip show prefect fastapi starlette
If this is in Docker / container build Make sure your image is not installing a newer FastAPI after Prefect. If you install both explicitly, pin them together. Example:
Copy code
pip install "prefect==3.7.4" "fastapi==0.136.3"
or better:
Copy code
pip install --upgrade "prefect>=3.7.5"
Related issue There is a matching GitHub issue here: fastapi 0.137 compatibility: AttributeError: 'PrefectRouter' object has no attribute 'routes' And another related report: Broken test harness due to missing routes attribute in PrefectRouter If you want, I can help you build a fully pinned compatible set for: - Prefect - FastAPI - Starlette - Pydantic
g
@Marvin what’s this? Jun 18 132531 ip-172-30-1-157 start_prefect.sh[2057631]: prefect_orion | 132531.044 | ERROR | prefect.server.events.triggers - Error in automation change listener: Invalid IPv6 URL. Reconnecting in 10s... Jun 18 132531 ip-172-30-1-157 start_prefect.sh[2057631]: prefect_orion | Traceback (most recent call last): Jun 18 132531 ip-172-30-1-157 start_prefect.sh[2057631]: prefect_orion | File “/usr/local/lib/python3.11/site-packages/prefect/server/events/triggers.py”, line 1188, in listen_for_automation_changes Jun 18 132531 ip-172-30-1-157 start_prefect.sh[2057631]: prefect_orion | conn = await get_pg_notify_connection() Jun 18 132531 ip-172-30-1-157 start_prefect.sh[2057631]: prefect_orion | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ Jun 18 132531 ip-172-30-1-157 start_prefect.sh[2057631]: prefect_orion | File “/usr/local/lib/python3.11/site-packages/prefect/server/utilities/postgres_listener.py”, line 118, in get_pg_notify_connection Jun 18 132531 ip-172-30-1-157 start_prefect.sh[2057631]: prefect_orion | original_scheme = urlsplit(db_url_str).scheme # e.g. “postgresql+asyncpg” Jun 18 132531 ip-172-30-1-157 start_prefect.sh[2057631]: prefect_orion | ^^^^^^^^^^^^^^^^^^^^ Jun 18 132531 ip-172-30-1-157 start_prefect.sh[2057631]: prefect_orion | File “/usr/local/lib/python3.11/urllib/parse.py”, line 514, in urlsplit Jun 18 132531 ip-172-30-1-157 start_prefect.sh[2057631]: prefect_orion | raise ValueError(“Invalid IPv6 URL”) Jun 18 132531 ip-172-30-1-157 start_prefect.sh[2057631]: prefect_orion | ValueError: Invalid IPv6 URL
m
✅ thought for 130.3 seconds
This usually means your Prefect server database connection URL is malformed, most likely because it contains an IPv6 address without square brackets. The failing code is in Prefect’s Postgres listener for automations/events, and it tries to parse your DB URL:
Copy code
urlsplit(db_url_str)
Python raises:
Copy code
ValueError: Invalid IPv6 URL
Most likely cause If your DB host is IPv6, this is invalid:
Copy code
<postgresql+asyncpg://user:pass@2001:db8::1:5432/prefect>
and this is valid:
Copy code
<postgresql+asyncpg://user:pass@[2001:db8::1]:5432/prefect>
IPv6 addresses in URLs must be wrapped in 
[
and 
]
. What to check Look at your Prefect server DB setting, typically: - 
PREFECT_API_DATABASE_CONNECTION_URL
or whatever is setting the server database URL in your container/service. Examples Invalid:
Copy code
<postgresql://user:password@::1:5432/prefect>
<postgresql+asyncpg://user:password@2001:db8::1:5432/prefect>
Valid:
Copy code
<postgresql://user:password@[::1]:5432/prefect>
<postgresql+asyncpg://user:password@[2001:db8::1]:5432/prefect>
Why it shows up in automations That log line is from the automation change listener, which uses a Postgres notification connection. So the main symptom appears there, but the real issue is the DB URL format. What to do Update your DB connection string so any IPv6 host is bracketed. If your host is not IPv6 and is instead a hostname, then something else may be malformed in the URL. In that case, paste the URL with credentials redacted, like:
Copy code
<postgresql+asyncpg://USER:***@HOST:5432/DBNAME>
and I can sanity-check it. No matching GitHub issue/discussion showed up for this exact message, but the root cause is clear from the parser behavior and Prefect’s DB listener code.
g
@Marvin I’m not passing any ipv6, I’m passing aws rds host
m
✅ thought for 9.6 seconds
If it’s an AWS RDS hostname, then the error is probably not because you intended to use IPv6 — it usually means the URL is malformed in a way that makes Python’s parser think the host portion looks like IPv6. Most likely this is happening in your 
PREFECT_API_DATABASE_CONNECTION_URL
. Common causes with an RDS hostname 1. Password contains special characters and is not URL-encoded This is the most common cause. For example, if your password contains characters like: - 
@
- 
:
- 
/
- 
#
- 
?
then a URL like this can break parsing:
Copy code
<postgresql+asyncpg://user:pa:ss@word@mydb.xxxxx.us-east-1.rds.amazonaws.com:5432/prefect>
because the parser can no longer reliably tell what is password vs host vs port. 2. Extra colon or slash in the host portion For example:
Copy code
<postgresql+asyncpg://user:pass@mydb.xxxxx.us-east-1.rds.amazonaws.com:5432:5432/prefect>
3. The URL was templated incorrectly For example, missing username/password separation or accidental concatenation. 4. Smart quotes / copied formatting If the value was copied from somewhere with transformed punctuation, parsing can fail in strange ways. What a valid RDS URL should look like
Copy code
<postgresql+asyncpg://USERNAME:PASSWORD@mydb.xxxxx.us-east-1.rds.amazonaws.com:5432/prefect>
If the password has special characters, it must be URL-encoded. Example: - raw password:
Copy code
p@ss:word/123
- encoded password:
Copy code
p%40ss%3Aword%2F123
Then the URL becomes:
Copy code
<postgresql+asyncpg://USERNAME:p%40ss%3Aword%2F123@mydb.xxxxx.us-east-1.rds.amazonaws.com:5432/prefect>
What I’d check first Print or inspect the exact value of 
PREFECT_API_DATABASE_CONNECTION_URL
being used by the server process, with the password redacted. It should look like:
Copy code
<postgresql+asyncpg://USER:***@HOST:5432/DBNAME>
Quick sanity checklist - exactly one 
@
separating credentials from host - exactly one 
:5432
port section - no unescaped special characters in password - no surrounding quotes accidentally included in the environment variable - no whitespace/newline in the value Very likely fix If your password contains special characters, URL-encode it. In Python, for example:
Copy code
from urllib.parse import quote_plus
print(quote_plus("p@ss:word/123"))
Or if you build it manually, encode the password before inserting it into the URL. If you want, paste the connection URL with the password redacted but preserving structure, for example:
Copy code
<postgresql+asyncpg://myuser:***@mydb.xxxxx.us-east-1.rds.amazonaws.com:5432/prefect>
and I can tell you whether the shape is valid.
g
@Marvin that env var is the following postgresql+asyncpg://postgres:aaaaaaaa*[{%a&lt;=*aaa@rdsname.part1.part2.rds.amazonaws.com:5432/my_db I don’t see any of the chars you mentioned
m
✅ thought for 8.3 seconds
You do have the problem in the password. This part of your password is breaking URL parsing:
Copy code
aaaaaaaa*[{%a<=*aaa
The key issue is the 
[
character. In a URL, 
[
and 
]
are special because they’re used for IPv6 host notation. When they appear unescaped inside the credential section, Python’s URL parser can get confused and raise:
Copy code
Invalid IPv6 URL
You also have other characters that should be encoded in a URL password, including: - 
[
- 
%
- 
<
- 
=
So the fix is: URL-encode the password before putting it in the connection string. Your current URL
Copy code
<postgresql+asyncpg://postgres:aaaaaaaa*[{%a<=*aaa@rdsname.part1.part2.rds.amazonaws.com:5432/my_db>
Correct approach Encode only the password portion. In Python:
Copy code
from urllib.parse import quote

password = "aaaaaaaa*[{%a<=*aaa"
print(quote(password, safe=""))
That will produce an encoded password you can safely place in the URL. It will look something like:
Copy code
aaaaaaaa%2A%5B%7B%25a%3C%3D%2Aaaa
So your fixed connection URL would be:
Copy code
<postgresql+asyncpg://postgres:aaaaaaaa%2A%5B%7B%25a%3C%3D%2Aaaa@rdsname.part1.part2.rds.amazonaws.com:5432/my_db>
Why this happens Even though the host is an RDS hostname, the parser sees special URL characters in the password and misinterprets the whole string structure before it ever gets to the host correctly. What to do now Replace the password in 
PREFECT_API_DATABASE_CONNECTION_URL
with the URL-encoded version and restart Prefect. If you want, I can encode that exact password for you directly in the full DSN format.
g
@Marvin it was working up until 3.6.4, why 3.7.4 is breaking?
m
✅ thought for 7.8 seconds
Because Prefect 3.7.4 started exercising a stricter URL parsing path for the automation/event Postgres listener. In your traceback, the failing code path is:
Copy code
prefect.server.utilities.postgres_listener.get_pg_notify_connection()
and specifically:
Copy code
urlsplit(db_url_str)
That path is used by the automations/events listener, and in 3.7.4 it’s parsing the DB URL in a way that now trips over the unescaped 
[
in your password. So the short version is: - your connection string was already technically invalid as a URL - older Prefect versions happened not to hit this parsing path the same way - 3.7.4 does, so the bad password encoding is now exposed Why it may have seemed fine before A few reasons this can happen: 1. Different code path - normal DB engine creation may have tolerated or differently parsed the URL - the automation listener now separately parses it for PG notify 2. Stricter parser behavior - 
urllib.parse.urlsplit()
is less forgiving about URL structure - especially around 
[
/ 
]
, which are reserved for IPv6 literals 3. New automations/events behavior - if the listener wasn’t active before, or wasn’t using the same helper path, you may never have hit this failure So this is less “3.7.4 broke a valid URL” and more “3.7.4 exposed an invalid DSN that previously slipped by.” The important takeaway Your password must be URL-encoded in the DSN. The 
[
is enough to cause exactly this issue. So even if 3.6.4 happened to work, I would still recommend fixing the connection string to:
Copy code
<postgresql+asyncpg://postgres:<url-encoded-password>@rdsname.part1.part2.rds.amazonaws.com:5432/my_db>
If you want, I can give you the exact corrected DSN with your password encoded based on the string you pasted.
g
@Marvin I noticed in Prefect 3.7.4 (I’m no sure if this does not exist in prefect > 3.7) that when a job is executed it goes into submitting state. I don’t see such state in the enumeration StateType. Why?
m
✅ thought for 147.5 seconds
Because 
Submitting
is a state name, not a new 
StateType
. In Prefect 3.x, states have two layers: 1. 
type
 — one of the enum values in 
StateType
2. 
name
 — a display/semantic label, which can be more specific So when you see 
Submitting
, it is typically:
Copy code
type = PENDING
name = "Submitting"
That’s why you don’t see 
SUBMITTING
in the 
StateType
enum. Why Prefect does this 
StateType
stays small and represents the fundamental orchestration categories: - 
SCHEDULED
- 
PENDING
- 
RUNNING
- 
COMPLETED
- 
FAILED
- 
CANCELLED
- 
CRASHED
- 
PAUSED
- 
CANCELLING
Then Prefect uses named sub-states for more detail. So for example: - 
Pending
→ type 
PENDING
- 
Submitting
→ type 
PENDING
- 
InfrastructurePending
→ type 
PENDING
Similarly for scheduled states, you may see names like: - 
Scheduled
- 
AwaitingRetry
- 
AwaitingConcurrencySlot
all of which can still be 
type = SCHEDULED
. Why you see it during job execution 
Submitting
is used when Prefect is in the phase of handing the run off to infrastructure / worker execution machinery. So the flow is roughly:
Copy code
Scheduled -> Pending(name="Submitting") -> Pending(name="InfrastructurePending") -> Running
The exact path can vary, but 
Submitting
is part of that transitional handoff. Why the UI shows it The UI/API expose both 
state.type
and 
state.name
, and the UI generally shows the name, because it’s more informative than just 
PENDING
. So you may see:
Copy code
{
  "type": "PENDING",
  "name": "Submitting"
}
Short answer 
Submitting
is not missing from `StateType`; it was never meant to be there. It is a named PENDING state. If you want, I can also give you a quick table of the common Prefect named states and their underlying 
StateType
.
g
@Marvin so it is good I’m basing my logics on StateType instead of state names
m
✅ thought for 6.4 seconds
Yes — that is generally the right thing to do. If your logic depends on orchestration semantics, you should usually key off 
StateType
, not the human-readable state 
name
. Why State names are more specific and can change or expand over time: - 
Submitting
- 
InfrastructurePending
- 
AwaitingRetry
- 
AwaitingConcurrencySlot
But their underlying 
StateType
remains the stable orchestration category: - 
PENDING
- 
SCHEDULED
- etc. So for most programmatic logic, 
StateType
is the more robust choice. Good use cases for 
StateType
 - “is this run terminal?” - “is this run failed?” - “is this run still waiting to execute?” - “is this run actively running?” When state names are useful Use 
state.name
only if you specifically care about a finer-grained sub-state. For example: - distinguish 
AwaitingRetry
from generic 
Scheduled
- distinguish 
Submitting
from other 
PENDING
states - detect 
AwaitingConcurrencySlot
So: - business/orchestration logic → prefer 
StateType
- UI messaging / special-case operational branching → maybe inspect 
state.name
Practical advice If you do rely on 
state.name
, treat it as a narrower, more fragile condition than 
StateType
. So yes: basing your core logic on 
StateType
is the safer design.
43 Views
Open in Slack
PreviousNext
Powered by