Great question — what you’re seeing is the UI treating your input as a JSON string instead of an object. What the

__prefect_kind: "json"

means - That wrapper shows up when the UI captures a value as a JSON string literal (i.e., text that looks like JSON), not as a structured object. In your example,

"value": "{\n\"tracking\": true, ...}"

is a string, not a dict. How to fix it Pick one of these approaches: 1) Paste parameters as a full JSON object in the run/schedule modal - Open the “Edit JSON” (or “Parameters”) editor for the run/schedule. - Enter the full parameters object, not a quoted string. Example:

{
  "kwargs": {
    "tracking": true,
    "queue_maxsize": 16,
    "candidates_method": "local_queues"
  }
}

- Common mistake: entering

"kwargs": "{ \"tracking\": true }"

(note the quotes) — that creates the

__prefect_kind: "json"

wrapper and passes a string. 2) If you’re using the per-parameter form fields in the UI - Find the field for

kwargs

and set its type to “Object” (not “JSON” or “String”). - Paste the dict without quotes:

{
  "tracking": true,
  "queue_maxsize": 16,
  "candidates_method": "local_queues"
}

3) Preferably: type your flow parameter so the UI renders the right editor - If you control the flow, annotate the parameter as a dict or a Pydantic model. This lets the UI render an object editor automatically.

from typing import Any
  from prefect import flow
  from pydantic import BaseModel

  class Kwargs(BaseModel):
      tracking: bool
      queue_maxsize: int
      candidates_method: str

  @flow
  def my_flow(kwargs: Kwargs):
      ...

  # or, if you want a free-form dict
  @flow
  def my_flow(kwargs: dict[str, Any]):
      ...

- After redeploying, the UI will give you a structured “Object” editor for

kwargs

. References - Form building and parameter types (Prefect 3) If this still shows up as

__prefect_kind: "json"

, can you tell me which screen you’re using (Run modal vs. Schedule vs. Deployment default parameters) and share a quick screenshot of the parameter field for

kwargs

? I can tailor the exact clicks for that view.

Thanks for confirming. When that wrapper shows up even with valid object input, it almost always means the deployment’s parameter schema says that field is a string/JSON text, not an object. The UI then only has a “JSON text” editor available and will wrap it as:

{
  "value": "<stringified json>",
  "__prefect_kind": "json"
}

How to fix it for real (Prefect 3.x): 1) Type-hint the parameter as an object (or a Pydantic model) - Make sure your flow has a named parameter (not varargs like **kwargs) and is annotated as a dict or model.

from typing import Any, Dict
from prefect import flow
from pydantic import BaseModel

# Option A: Pydantic model (best UI/editor/validation)
class Kwargs(BaseModel):
    tracking: bool
    queue_maxsize: int
    candidates_method: str

@flow
def my_flow(kwargs: Kwargs):
    ...

# Option B: free-form dict
@flow
def my_flow(kwargs: dict[str, Any]):  # or Dict[str, Any]
    ...

2) Redeploy so the schema updates - The parameter schema is captured at deployment time. After changing the signature, run your deploy again. - Verify the schema shows

kwargs

as an object (not string):

prefect deployment inspect "your-flow/your-deployment"

Look for

parameter_openapi_schema

→

properties

→

kwargs

→

type: object

. 3) Use the Object editor in the UI - After the schema is correct, the UI will render an “Object” editor for

kwargs

. Paste your dict without quotes:

{
  "tracking": true,
  "queue_maxsize": 16,
  "candidates_method": "local_queues"
}

Temporary workaround (if you can’t redeploy yet) - Accept a string and parse it inside your flow:

import json

@flow
def my_flow(kwargs):
    if isinstance(kwargs, str):
        kwargs = json.loads(kwargs)
    ...

But the better fix is updating the type hints + redeploy. If this still happens after redeploy, can you share: - Your flow signature for the parameter (just the def line) - The output of

prefect deployment inspect "flow/deployment"

for the

parameter_openapi_schema

of

kwargs

- Which screen you’re using (Run modal vs Schedule vs Deployment defaults) Docs: Form building and parameter types in Prefect 3

Short answer: Using

**kwargs

is valid Python, but it prevents Prefect from generating a proper parameter schema for the UI. The UI will treat it as a single “JSON text” field and wrap it as

{"value": "...", "__prefect_kind": "json"}

. So it’s “allowed”, but it won’t behave the way you want in the UI. What to do instead: - Best practice: Replace

**kwargs

with a single, named object parameter.

from typing import Any
  from prefect import flow

  @flow
  def my_flow(params: dict[str, Any]):
      # use params["tracking"], etc.
      ...

Or use a Pydantic model (you can still allow arbitrary keys):

from pydantic import BaseModel, ConfigDict
  from prefect import flow

  class Params(BaseModel):
      model_config = ConfigDict(extra='allow')  # accept extra keys
      tracking: bool | None = None
      queue_maxsize: int | None = None
      candidates_method: str | None = None

  @flow
  def my_flow(params: Params):
      ...

Then redeploy and confirm the schema shows

params

as an object:

prefect deployment inspect "your-flow/your-deployment"

In the UI, paste:

{
  "params": {
    "tracking": true,
    "queue_maxsize": 16,
    "candidates_method": "local_queues"
  }
}

This will not be wrapped as

__prefect_kind: "json"

. - If you must keep `**kwargs`: - Don’t pass a nested object named

kwargs

in the UI. Instead, provide the fields at the top level:

{
  "tracking": true,
  "queue_maxsize": 16,
  "candidates_method": "local_queues"
}

Inside your flow, you’ll access

kwargs["tracking"]

, etc. - If you’re already getting the wrapper, add a small shim to parse it:

import json
    from prefect import flow

    @flow
    def my_flow(**kwargs):
        # If the UI sent a single top-level "kwargs" with the json wrapper:
        if "kwargs" in kwargs and isinstance(kwargs["kwargs"], dict):
            inner = kwargs["kwargs"]
            if inner.get("__prefect_kind") == "json" and "value" in inner:
                kwargs = json.loads(inner["value"])
        ...

Reference: Form building and parameter types in Prefect 3 If you can share your exact flow signature and the JSON you’re entering in the UI, I can point out the minimal change to make this work without the wrapper.

Got it — thanks for the snippet. Using a named parameter typed as

dict[str, Any] | None

is the right direction, but there are a couple gotchas that can still cause the UI to wrap it as `{"value": "...", "__prefect_kind": "json"}`: What typically causes this - The deployment’s parameter schema wasn’t updated. The UI uses the schema baked into the deployment. If the deployment still thinks

kwargs

is a string/JSON text, the UI will only offer the JSON-text editor and wrap it. - A previously saved default parameter (on the deployment or on the schedule) is still stored as the JSON-wrapper, and the UI keeps using it. - In some UI contexts, “JSON editor” vs “Object editor” can get sticky; if the schema is ambiguous, the UI may default to JSON text. What to do 1) Redeploy so the schema updates - Make sure you redeploy after changing the flow signature. - Then inspect the deployment to confirm

kwargs

is an object:

prefect deployment inspect "your-flow-name/your-deployment-name"

Look under

parameter_openapi_schema.properties.kwargs

. You want to see something like:

"kwargs": {
    "type": "object",
    "additionalProperties": true
  }

If it’s

type: string

or anything else, the UI will wrap it. 2) Clear any old defaults and schedule overrides - In the deployment settings, clear any saved default value for

kwargs

that shows the

__prefect_kind

wrapper. - If you created a schedule with parameter overrides, clear/re-enter those too (sometimes the schedule keeps an old value even after redeploy). 3) Use the Object editor in the UI - After the schema is correct, the UI should render an object field for

kwargs

. Paste without quotes:

{
  "kwargs": {
    "tracking": true,
    "queue_maxsize": 16,
    "candidates_method": "local_queues"
  }
}

4) If it still persists, try renaming the parameter - Rarely, naming the param

kwargs

can be confusing in the UI. Try:

from typing import Any
  from prefect import flow

  @flow
  def add_inference_job_to_queue(
      model_path: str,
      output_dir: str | None = None,
      update_db: bool = True,
      max_jobs: int | None = None,
      params: dict[str, Any] | None = None,
  ):
      ...

Redeploy and try again with

params

instead of

kwargs

. 5) Prefer Pydantic when possible (even with extra keys) - This tends to produce the most reliable object editors and validation in the UI:

from typing import Any
  from pydantic import BaseModel, ConfigDict
  from prefect import flow

  class InferenceParams(BaseModel):
      model_config = ConfigDict(extra='allow')  # allow arbitrary extra keys
      tracking: bool | None = None
      queue_maxsize: int | None = None
      candidates_method: str | None = None

  @flow
  def add_inference_job_to_queue(
      model_path: str,
      output_dir: str | None = None,
      update_db: bool = True,
      max_jobs: int | None = None,
      params: InferenceParams | None = None,
  ):
      ...

Redeploy, then use:

{
  "params": {
    "tracking": true,
    "queue_maxsize": 16,
    "candidates_method": "local_queues"
  }
}

Fallback (if you must keep going right now) - Shim to accept the JSON-wrapper if it appears:

import json
  from typing import Any
  from prefect import flow

  @flow
  def add_inference_job_to_queue(
      model_path: str,
      output_dir: str | None = None,
      update_db: bool = True,
      max_jobs: int | None = None,
      kwargs: dict[str, Any] | None = None,
  ):
      if isinstance(kwargs, dict) and kwargs.get("__prefect_kind") == "json" and "value" in kwargs:
          kwargs = json.loads(kwargs["value"])
      ...

Helpful reference: Form building and parameter types in Prefect 3 If you can share: - The output of

prefect deployment inspect "flow/deployment"

for the

parameter_openapi_schema

of

kwargs

- Which screen you’re using (Deployment default parameters vs Schedule vs Run modal) I can pinpoint why the UI is still wrapping it.

Thanks for sharing the inspect output — that schema is correct:

kwargs

is an object (or null). So this is not a typing/deployment-schema issue. What’s happening - The UI has multiple “input kinds” for object fields. If a prior value for

kwargs

was saved in “JSON text” mode, the UI will keep rendering it with the

__prefect_kind: "json"

wrapper and will continue to save it that way even if you paste a proper object. - This is an intentional UI behavior; see the schema form notes where

__prefect_kind

selects the editor/input type: Schema form component PR. How to fix it 1) Clear any existing saved value for

kwargs

at the level you’re editing: - If you’re editing Deployment defaults: open the deployment parameters, remove the entire

kwargs

field, save, then re-open. - If you’re editing a Schedule override: remove the

kwargs

override there, save, then re-add it. - If you’re starting a manual Run: click “Reset” or clear

kwargs

entirely before re-entering. 2) Switch the field back to the “Object” editor - In the parameter form, for the

kwargs

field, look for the input mode toggle (often a small control near the field) and switch from JSON to Object. Then enter:

{
  "tracking": true,
  "queue_maxsize": 16,
  "candidates_method": "local_queues"
}

- If you don’t see a per-field toggle, click “Edit JSON” for the whole parameters object and paste the full top-level parameters:

{
  "model_path": "…",
  "output_dir": null,
  "update_db": true,
  "max_jobs": null,
  "kwargs": {
    "tracking": true,
    "queue_maxsize": 16,
    "candidates_method": "local_queues"
  }
}

3) Verify nothing stored has the wrapper - After saving, re-open and confirm

kwargs

appears as an object (no

__prefect_kind

). Alternative, deterministic ways to set it - From CLI (forces JSON to be parsed into an object on the server):

prefect deployment run "your-flow/your-deployment" \
    -p model_path="..." \
    -p kwargs='{"tracking": true, "queue_maxsize": 16, "candidates_method": "local_queues"}'

The

-p

value is interpreted as JSON for that key. - In code when deploying:

flow.from_source(...).deploy(
    name="your-deployment",
    parameters={
      "model_path": "...",
      "kwargs": {
        "tracking": true,
        "queue_maxsize": 16,
        "candidates_method": "local_queues",
      },
    },
  )

If it still flips back to the JSON wrapper: - Tell me which exact screen you’re using (Deployment defaults vs Schedule vs Run modal) and whether you see an input-mode toggle on the

kwargs

field. - If possible, paste the raw stored parameter JSON from that screen so I can pinpoint where the

__prefect_kind

is being introduced.

Short answer: not with the CLI today. In Prefect 3.x,

prefect deployment schedule create

does not expose flags to set or update schedule-level parameter overrides. You can create/pause/resume/delete schedules via CLI, but parameter overrides must be set in the UI or via the Python API. Options: - One-off runs: use CLI parameter overrides when triggering a run (not the schedule):

prefect deployment run "flow-name/deployment-name" --param key=value --param other='{"nested": 1}'

- Programmatic update via API (create or replace the schedule with parameters):

from prefect.client.orchestration import get_client
  from prefect.client.schemas.schedules import CronSchedule

  async def set_schedule_params():
      async with get_client() as client:
          dep = await client.read_deployment_by_name("flow-name/deployment-name")

          # Optional: list existing schedules and delete/replace the one you want
          schedules = await client.read_deployment_schedules(dep.id)
          # await client.delete_deployment_schedule(schedules[0].id)

          await client.create_deployment_schedule(
              deployment_id=dep.id,
              schedule=CronSchedule(cron="0 8 * * *", timezone="America/New_York"),
              active=True,
              parameters={"tracking": True, "queue_maxsize": 16},
          )

If you want, I can draft a tiny script you can run to update a specific schedule’s parameter overrides given its deployment name and schedule ID.

@Marvin Let's do it through the python API then, How do I define a intervel schedule where it happens every 90 minutes (CRON is fine if you can provide me the format

Yes — use an IntervalSchedule. It’s the simplest and most reliable way to do “every 90 minutes”; cron does not have a single expression for 90-minute cadence. Example: create/update a schedule via Python API with an interval of 90 minutes and set parameter overrides:

from datetime import timedelta
from prefect.client.orchestration import get_client
from prefect.client.schemas.schedules import IntervalSchedule

async def upsert_every_90_min_schedule():
    async with get_client() as client:
        dep = await client.read_deployment_by_name("flow-name/deployment-name")

        schedule = IntervalSchedule(
            interval=timedelta(minutes=90),
            # optional:
            # anchor_date="2025-01-01T00:00:00Z",  # ISO datetime string
            # timezone="America/New_York",
        )

        await client.create_deployment_schedule(
            deployment_id=dep.id,
            schedule=schedule,
            active=True,
            parameters={
                "kwargs": {
                    "tracking": True,
                    "queue_maxsize": 16,
                    "candidates_method": "local_queues",
                }
            },
        )

# Then run this coroutine with your event loop, e.g.:
# import asyncio; asyncio.run(upsert_every_90_min_schedule())

Notes: - anchor_date sets the start reference; if omitted, Prefect picks a reasonable default. Add it if you need the cadence to align to a specific clock. - timezone is optional for interval schedules; if you include it, use an IANA name like “America/New_York”. If you truly need cron, you’d have to approximate 90 minutes with multiple expressions (not recommended). IntervalSchedule is the right tool here.

You’re right — that AttributeError is on me. There isn’t a PrefectClient.create_deployment_schedule method in Prefect 3.x. The supported programmatic path is to redeploy the flow and attach schedules using the schedules API. Here’s how to add a 90‑minute interval schedule (with optional parameter overrides) to an existing deployment: - Preferred: Interval schedule (no tricky cron math)

from datetime import timedelta
from prefect.schedules import Interval

# Every 90 minutes; optionally set timezone and per-schedule parameter overrides
every_90 = Interval(
    interval=timedelta(minutes=90),
    timezone="UTC",                      # or "America/New_York"
    parameters={
        "kwargs": {
            "tracking": True,
            "queue_maxsize": 16,
            "candidates_method": "local_queues",
        }
    },
)

Attach it by redeploying your flow with the same deployment name:

from prefect import flow

# your existing flow
@flow
def add_inference_job_to_queue(...):
    ...

# Update the deployment to include the new schedule
add_inference_job_to_queue.deploy(
    name="your-deployment-name",   # must match your existing deployment
    schedules=[every_90],          # you can pass multiple schedules here
)

Notes: - This “redeploy” updates the deployment in place. If you already have schedules, this adds to them; you can pause/delete old ones in the UI or via CLI if you want to replace instead. - Schedule-level parameters go on the Interval/Cron object via its parameters= argument, not in deployment defaults. If you must use cron to approximate 90 minutes Cron can’t express “every 90 minutes” in a single expression. Use two:

from prefect.schedules import Cron

cron_a = Cron("0 */3 * * *", timezone="UTC", parameters={...})       # 00:00, 03:00, 06:00, ...
cron_b = Cron("30 1-23/3 * * *", timezone="UTC", parameters={...})   # 01:30, 04:30, 07:30, ...
add_inference_job_to_queue.deploy(name="your-deployment-name", schedules=[cron_a, cron_b])

Verify the schedule(s) - List schedules:

prefect deployment schedule ls "your-flow-name/your-deployment-name"

- Inspect deployment (see parameter_openapi_schema and schedules):

prefect deployment inspect "your-flow-name/your-deployment-name"

Docs - Schedules concepts - Create schedules for deployments If you’d prefer to update by pulling from source (e.g., GitHub) instead of importing the flow locally, you can do:

from prefect import flow
from prefect.schedules import Interval

schedule = Interval(interval=90*60, timezone="UTC")  # seconds are allowed

flow.from_source(
    source="<https://github.com/your-org/your-repo>",
    entrypoint="path/to/file.py:add_inference_job_to_queue",
).deploy(
    name="your-deployment-name",
    schedules=[schedule],
)

Want me to draft a small one-off script to add this schedule and pause/delete any existing ones automatically?