Great question. Short answer: - If your app already reacts to DB insert events and you need to decide which deployment to run based on a field value, use the Prefect SDK (

prefect.deployments.run_deployment

). - If you want your database/CDC tool or another service to just POST an event and let Prefect react declaratively, use a Webhook + Automation. A quick decision guide - Use SDK (

run_deployment

) when: - Your app is already handling the DB insert trigger - You want synchronous control, custom logic, and error handling in your app - You need easy parameter passing and deduplication with idempotency keys - Use Webhooks + Automations when: - You prefer “push an HTTP event and forget” without writing app-side API code - A third-party/CDC tool can POST directly to Prefect - You want a declarative, event-driven rule (Automation) that runs a deployment when a specific event arrives Note: Webhooks require Prefect Cloud (publicly reachable endpoint). If you’re not on Cloud, the SDK is the simpler path. Are you on Prefect Cloud or self-hosted Server? Option 1: Trigger runs from your app (SDK) - Works well for “on insert, if field == X then run deployment Y with params Z”. - You can return immediately or wait for completion. - Use idempotency keys to avoid duplicate runs on retries. Example:

import os
from prefect.deployments import run_deployment

# Make sure these are set in your environment if you're on Prefect Cloud:
# PREFECT_API_URL and PREFECT_API_KEY
# e.g., <https://api.prefect.cloud/api/accounts/<account_id>/workspaces/<workspace_id>>

def map_field_to_deployment(field_value: str) -> str:
    # return "flow_name/deployment_name"
    return {
        "alpha": "ingest_flow/alpha-deployment",
        "beta": "ingest_flow/beta-deployment",
    }.get(field_value, "ingest_flow/default-deployment")

def on_row_insert(row: dict):
    deployment_name = map_field_to_deployment(row["field_name"])
    params = {
        "record_id": row["id"],
        "payload": row,
    }
    flow_run = run_deployment(
        name=deployment_name,                 # format: FLOW_NAME/DEPLOYMENT_NAME
        parameters=params,
        flow_run_name=f"db-insert-{row['id']}",
        idempotency_key=f"db-insert:{row['id']}",  # dedupe if retried
        tags=["db-trigger"],
        timeout=0,                            # return immediately; omit to wait for completion
    )
    return flow_run

Notes: - Name format is

FLOW_NAME/DEPLOYMENT_NAME

. You can also pass a deployment UUID. - By default,

run_deployment

waits for the flow run to finish; set

timeout=0

to return immediately. - Use

idempotency_key

tied to a stable DB identifier to avoid duplicates. - Docs: Run deployments (SDK) Option 2: Webhook + Automation (no app code) - External system POSTs to a Prefect Cloud Webhook URL. - Webhook turns the request into a Prefect event using a Jinja template. - An Automation with a “Run Deployment” action fires when the event matches. Setup sketch: 1) Create a webhook in Prefect Cloud with a template that captures fields from the request body:

{
  "event": "db.row_inserted",
  "resource": {"prefect.resource.id": "db.table.my_table"},
  "payload": {
    "id": "{{ body.id }}",
    "field_name": "{{ body.field_name }}",
    "row": {{ body | tojson }}
  }
}

2) Create an Automation (UI or YAML) using a Deployment Event Trigger: - Match on event name:

db.row_inserted

- Optionally filter by

payload.field_name

- Action: Run Deployment - Parameter template:

{
  "record_id": "{{ event.payload.id }}",
  "payload": {{ event.payload.row | tojson }}
}

Tips: - You can validate a shared secret by checking a custom header in your template/filters (e.g., only proceed when

headers['X-Webhook-Token'] == 'your-secret'

). - Webhook URLs are unique; rotate them if needed.

- Docs: Webhooks concept, Create a webhook, Create deployment triggers, Event triggers Which should you choose? - If you already have app code firing on DB inserts and need dynamic mapping by field name with tight control, the SDK is simpler and more direct. - If you’d rather have the database/CDC or another service simply POST an event and let Prefect handle the rest declaratively, use Webhooks + Automations. Follow-ups to tailor the recommendation: - Are you on Prefect Cloud? - Do you already have an app or CDC process that receives the DB insert and can make API calls? - Any need to wait for completion vs fire-and-forget?

Short answer: - Prefect Cloud webhooks are highly available and will 2xx as soon as the event is accepted. They’re reliable for triggering Automations, but duplicates are possible if the sender retries or sends duplicates—so add dedup logic if needed. - Extra webhook traffic won’t “kill” your workers. Prefect queues runs; workers only pick up work within your configured limits. Use concurrency controls to add backpressure. More detail Reliability of webhooks - Availability: Webhooks are served by Prefect Cloud and are HA. The endpoint returns 2xx when the request is accepted and converted into a Prefect event. If the request/template is invalid, you’ll get a non-2xx and see details in your event logs. - Delivery guarantees: They’re as reliable as your sender’s retry policy. Prefect does not automatically de-duplicate inbound webhook requests—if your source retries without idempotency, you can get duplicate events and thus duplicate runs unless you add safeguards. - Validation: Use a shared secret/signature header and check it in your webhook template and/or trigger filters so only valid requests create events. - Docs: Webhooks concept, Create a webhook, Create deployment triggers Preventing overload and keeping workers healthy - Work pool concurrency limit: Caps how many flow runs across the pool can run at once. Extra runs queue; workers don’t crash due to excess demand. - CLI:

prefect work-pool set-concurrency-limit "my-pool" 10
    prefect work-pool clear-concurrency-limit "my-pool"

- Docs: Work pools - Deployment concurrency limit: Limits how many concurrent runs of a single deployment can execute. Extras are enqueued by default. - In code when deploying:

from prefect import flow

    @flow
    def my_flow(): ...

    my_flow.deploy(
      name="db-triggered",
      work_pool_name="my-pool",
      concurrency_limit=5,  # per-deployment limit
    )

- Tag-based concurrency (task-level): Throttle tasks sharing a tag across flows. - CLI:

prefect concurrency-limit create db-trigger --concurrency-limit 20

- Docs: Tag-based concurrency - Global concurrency limits / rate limiting: Name-based slots you can use in code for cross-flow throttling. - Docs: Global concurrency limits - Emergency brake: You can pause a work pool (manually or via an Automation action) to halt new starts while the queue drains. - Infrastructure: Use resilient infra for workers (e.g., containers with restart policies, Kubernetes/ECS with autoscaling) so a busy queue doesn’t affect process health. Handling duplicates from webhooks - If duplicates are a concern, you have two good options: - Keep using webhooks + Automations, but add filtering/templating to ignore unwanted events and/or pause/resume pools on surges. Note: Automations’ “Run Deployment” action does not currently expose an idempotency key—duplicates will create duplicate runs unless filtered. - Route through a tiny service and trigger via SDK with idempotency:

from prefect.deployments import run_deployment

    run_deployment(
      "flow_name/deployment_name",
      parameters={"record_id": event_id, "payload": payload},
      idempotency_key=f"db-insert:{event_id}",
      timeout=0  # return immediately
    )

Docs: Run deployments (SDK) If you share: - Whether you’re on Prefect Cloud or self-hosted - Your expected peak rate of inserts/webhooks - Your worker infrastructure (process/Docker/Kubernetes)

…I can recommend concrete limits and a pattern (webhook-only vs small mediator service with idempotency) tailored to your setup.

Great question. - Flow: a Python function decorated with

@flow

. Running it calls your code directly (local process) or as a subflow. It’s your workflow logic, parameters, and tasks. - Deployment: a saved, versioned “run configuration” for a flow. It points to your code (local path or repo), defines infrastructure via a work pool, default parameters, schedules/triggers, tags, concurrency limits, etc. Deployments live in the Prefect API and can be triggered from the UI, API/SDK, CLI, webhooks, or automations. Which runs a workflow remotely? - A deployment does. Remote execution happens when a worker attached to a work pool picks up a run for a deployment and starts it on your chosen infrastructure (process, Docker, Kubernetes, ECS, etc.). Calling a flow function directly just runs it where the code is executing. Typical remote setup (Prefect 3.x) 1) Create or use a work pool (e.g., Kubernetes, Docker, Process) and start a worker that polls it:

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

2) Create a deployment for your flow (code or CLI): - In code:

from prefect import flow

     @flow
     def my_flow():
         ...

     my_flow.deploy(
       name="prod",
       work_pool_name="your-pool",
       # optionally: cron="0 * * * *", parameters={...}, tags=[...]
     )

- Or run

prefect deploy

in your project and follow prompts (it will help configure storage, pool, etc.). 3) Trigger runs remotely via UI, CLI/SDK, webhooks, or schedules:

from prefect.deployments import run_deployment
   run_deployment("my_flow/prod", parameters={"x": 1})

Docs: - Concepts: Deployments - How to run deployments: Run deployments - Work pools/workers: Work pools If you share your target runtime (process, Docker, Kubernetes, etc.), I can give a minimal deployment example tailored to it.

Short answer: No. In Prefect 3.x, a worker can only execute remote runs that come from a deployment. Just creating a flow run and setting `work_pool_name`/`work_queue_name` is not enough for a worker to pick it up—the worker needs the deployment’s context (how to load the code, storage/source, entrypoint, infrastructure/job variables). What those API fields are for -

work_pool_name

and

work_queue_name

on flow runs help with routing/scheduling, but they do not replace a deployment. Without a deployment, there’s no instruction for a worker to fetch and run your code. - To run remotely, create a deployment first, then create runs from that deployment (via SDK/CLI/API). Recommended patterns - One-time (or at startup) create a deployment, then trigger runs dynamically:

from prefect import flow

@flow
def my_flow(x: int = 0):
    ...

# one-time deployment
my_flow.deploy(
  name="dynamic",
  work_pool_name="your-pool",
)

# later, trigger runs remotely
from prefect.deployments import run_deployment
run_deployment("my_flow/dynamic", parameters={"x": 42})

- If your code lives in Git (no prebuilt image needed), deploy from source:

from prefect import flow

flow = flow.from_source(
  source="<https://github.com/org/repo>",
  entrypoint="flows/my_flow.py:my_flow",
)
flow.deploy(name="dynamic", work_pool_name="your-pool")

Routing to “any” pool - You can target a specific work queue when creating a run from a deployment (

run_deployment(..., work_queue_name="...")

), but you can’t override the work pool at run time. If you truly need to run on multiple pools, create one deployment per pool (same flow, different

work_pool_name

). Docs - Deployments concept: Deployments - Run deployments (SDK/CLI/API): Run deployments - REST: Create run from deployment: POST /deployments/{id}/flow_runs - REST: Create flow run (general): POST /flow_runs (note: not sufficient for worker execution without a deployment) If you share your target infrastructure (process, Docker, Kubernetes, etc.), I can give a minimal deployment snippet tailored to it.

Short answer: A deployment is required. Automations don’t run “bare” flows; the Run Deployment action creates a new flow run from an existing deployment. How it works - You define an Automation with a trigger (e.g., a webhook event, metric threshold, or another flow’s state change). - The action you choose to start work is Run Deployment, which points at a specific deployment and can pass parameters via templates. - There is no Automation action that executes a flow directly without a deployment. Docs - Automations overview: Automations - Deployment-triggered automations: Run Deployment from events - Chaining deployments with events: Chaining deployments If you’d like, I can sketch a minimal deployment + automation that takes an event field and maps it to deployment parameters.

Short answers - Yes: you can pass webhook request fields into a deployment’s flow run parameters using Jinja templates in the Automation’s Run Deployment action. - Deciding which deployment to run: today, the deployment you run is selected per action. To “route” based on webhook fields, either create multiple Automations (each with filters matching different payload fields) or use a small “router” deployment that chooses which downstream deployment to run in code. How to pass webhook payload into parameters 1) Create a webhook that maps the incoming request into a Prefect event (Jinja available:

body

,

headers

,

query

,

method

):

{
  "event": "db.row_inserted",
  "resource": {
    "prefect.resource.id": "db.table.my_table",
    "op": "{{ body.op|default('insert') }}",
    "env": "{{ headers['x-env']|default('prod') }}",
    "target": "{{ body.target|default('default') }}"
  },
  "payload": {{ body | tojson }}
}

2) Create an Automation with an Event Trigger that matches the webhook event (e.g.,

event == db.row_inserted

) and a Run Deployment action. In the action, pass flow parameters by templating from the event:

{
  "parameters": {
    "record_id": {
      "__prefect_kind": "jinja",
      "template": "{{ event.payload.id }}"
    },
    "op": {
      "__prefect_kind": "jinja",
      "template": "{{ event.resource.op }}"
    },
    "raw": {
      "__prefect_kind": "jinja",
      "template": "{{ event.payload | tojson }}"
    }
  }
}

Routing to different deployments based on webhook fields Option A: Multiple Automations (no code) - Define separate Automations with different match filters, each running a different deployment. For example: - Automation 1: match

event.resource.target == 'alpha'

→ Run Deployment

ingest_flow/alpha

- Automation 2: match

event.resource.target == 'beta'

→ Run Deployment

ingest_flow/beta

This is the simplest purely declarative approach. Option B: One Automation + “router” deployment (code) - Have the Automation always run a small “router” deployment that decides which downstream deployment to run based on the event fields. - Example router flow:

from prefect.deployments import run_deployment
from prefect import flow

@flow
def router(event_payload: dict):
    target = event_payload.get("target", "default")
    deployment = {
        "alpha": "ingest_flow/alpha",
        "beta":  "ingest_flow/beta",
    }.get(target, "ingest_flow/default")

    run_deployment(
        deployment,
        parameters={"payload": event_payload},
        idempotency_key=f"webhook:{event_payload.get('id')}",
        timeout=0  # fire-and-forget
    )

- In the Automation, pass

event.payload

to the

router

parameters (as shown above). Notes and caveats - Automations run deployments; they don’t execute bare flows directly. - The Run Deployment action currently selects a specific deployment; dynamic selection within a single action isn’t supported, hence the “multiple automations” or “router flow” patterns. - If you need deduplication/idempotency, the router pattern gives you access to

idempotency_key

via

run_deployment(...)

. - Webhooks and Automations are Prefect Cloud features. Docs - Webhooks concept: Webhooks - Create a webhook: Create a webhook - Automations: Creating automations - Template parameters in actions: Templating action parameters - Chaining deployments with events: Chaining deployments If you share the exact webhook payload shape and your target deployments, I can draft the webhook template and either a set of Automations or a router flow tailored to your case.