Yes — this is a good fit for event-driven deployments in Prefect 3.x, including on self-hosted Prefect server with a process worker. Your setup would typically look like this:

external DB check -> emit Prefect event -> automation/deployment trigger -> deployment run -> process worker executes local process

There are two main ways to do it: 1. Direct event-driven: some external script/service checks the DB and emits a custom Prefect event when needed 2. Polling + event-driven: a scheduled Prefect flow checks the DB, and if the condition is met, it emits a custom event that triggers another deployment For your case, option 1 or 2 both work well. If your “external event” is really just “a periodic check against a database,” then option 2 is often the simplest. --- ## High-level steps 1. Run Prefect server with automations/events enabled 2. Create a process work pool 3. Start a process worker 4. Create a deployment for the flow that should react to the event 5. Attach an event trigger to that deployment 6. Emit a custom event with a payload when your DB check says automation is needed 7. Map event payload fields to flow parameters --- ## 1. Start your self-hosted Prefect server If you are using self-hosted Prefect, make sure the server and background services are running. Automations depend on the server-side services. Typical local setup:

bash
prefect server start

If your worker or event emitter is not on the same host, also make sure they point to your API:

bash
export PREFECT_API_URL="http://<your-prefect-host>:4200/api"

Relevant docs: - Events - Automations - Self-hosting Prefect --- ## 2. Create a process work pool Verified CLI family:

prefect work-pool ...

Create a process pool:

bash
prefect work-pool create my-process-pool --type process

Then start a worker for it:

bash
prefect worker start --pool my-process-pool --type process

Docs: - Work pools - Workers --- ## 3. Create the reactive flow Suppose your flow runs a local process and uses payload values from the triggering event. Example:

python
from prefect import flow
import subprocess

@flow(log_prints=True)
def react_to_db_event(job_id: str, action: str, raw_payload: dict | None = None):
    print(f"Starting local process for job_id={job_id}, action={action}")
    print(f"Payload: {raw_payload}")

    result = subprocess.run(
        ["./my_local_script.sh", job_id, action],
        capture_output=True,
        text=True,
        check=False,
    )

    print("stdout:", result.stdout)
    print("stderr:", result.stderr)

    if result.returncode != 0:
        raise RuntimeError(f"Local process failed with code {result.returncode}")

--- ## 4. Attach an event trigger to the deployment You can do this in Python or in

prefect.yaml

. ## Option A: Python SDK deployment with trigger This is the clearest way if you want everything in code. ```python from prefect import flow import subprocess from prefect.events.schemas.deployment_triggers import DeploymentEventTrigger @flow(log_prints=True) def react_to_db_event(job_id: str, action: str, raw_payload: dict | None = None): print(f"Starting local process for job_id={job_id}, action={action}") result = subprocess.run( ["./my_local_script.sh", job_id, action], capture_output=True, text=True, check=False, ) print(result.stdout) if result.returncode != 0: raise RuntimeError(f"Process failed with code {result.returncode}") if name == "__main__": react_to_db_event.serve( name="db-event-handler", work_pool_name="my-process-pool", triggers=[ DeploymentEventTrigger(

expect={"external.db.condition-met"}, match={ "prefect.resource.id": "external.db.monitor" }, parameters={ "job_id": "{{ event.payload.job_id }}", "action": "{{ event.payload.action }}", "raw_payload": { "__prefect_kind": "json", "value": { "__prefect_kind": "jinja", "template": "{{ event.payload | tojson }}" } }, }, ) ], )

Run it:

```bash
python react_to_db_event.py

That creates the deployment and keeps it synced. --- ## Option B:

prefect.yaml

You can also declare the deployment trigger in

prefect.yaml

. Example:

yaml
deployments:
  - name: db-event-handler
    entrypoint: react_to_db_event.py:react_to_db_event
    work_pool:
      name: my-process-pool
    triggers:
      - type: event
        enabled: true
        expect:
          - external.db.condition-met
        match:
          prefect.resource.id: external.db.monitor
        parameters:
          job_id: "{{ event.payload.job_id }}"
          action: "{{ event.payload.action }}"
          raw_payload:
            __prefect_kind: json
            value:
              __prefect_kind: jinja
              template: "{{ event.payload | tojson }}"

Deploy it with:

bash
prefect deploy -n db-event-handler

Verified command family: -

prefect deploy

Docs: - prefect.yaml deployments - Deployments --- ## 5. Emit the event when the DB condition is met Now you need the external trigger source. Since your trigger condition is “check an external database and decide whether the automation is needed,” you have a few options. # Option 1: External Python script emits event directly This is ideal if the DB check happens outside Prefect.

python
from prefect.events import emit_event

def check_database():
    # Replace with your DB logic
    should_run = True
    payload = {
        "job_id": "12345",
        "action": "refresh_cache",
        "source": "db-checker"
    }
    return should_run, payload

if __name__ == "__main__":
    should_run, payload = check_database()

    if should_run:
        event = emit_event(
            event="external.db.condition-met",
            resource={
                "prefect.resource.id": "external.db.monitor",
                "prefect.resource.name": "external-db-monitor",
            },
            payload=payload,
        )
        print(f"Emitted event: {event.id}")

This event will be evaluated by the trigger, and the deployment run will be created. Docs: - Events --- # Option 2: Scheduled Prefect flow checks DB, then emits event This is often the best pattern when the external system does not naturally push events. You create: - a scheduled polling flow - a reactive flow triggered by events Example checker flow: ```python from prefect import flow from prefect.events import emit_event @flow(log_prints=True) def db_checker(): # Replace with real DB query logic rows_needing_action = [ {"job_id": "12345", "action": "refresh_cache"}, {"job_id": "67890", "action": "send_notification"}, ] for row in rows_needing_action: emit_event( event="external.db.condition-met", resource={ "prefect.resource.id": "external.db.monitor", "prefect.resource.name": "external-db-monitor", }, payload=row, ) print(f"Emitted event for {row['job_id']}") if name == "__main__": db_checker.serve( name="db-checker", work_pool_name="my-process-pool",

cron="*/5 * * * *", )

This gives you:
- one deployment running every 5 minutes
- one event-driven deployment reacting only when needed

This separation is usually cleaner than embedding all logic in one flow.

---

# Option 3: Emit via HTTP to Prefect’s events API

If you want a non-Python external service to trigger Prefect, you can send events to the API.

The exact auth/setup depends on your self-hosted configuration, but the pattern is:

```bash
curl -X POST "http://<your-prefect-host>:4200/api/events" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <YOUR_API_KEY>" \
  -d '[
    {
      "event": "external.db.condition-met",
      "resource": {
        "prefect.resource.id": "external.db.monitor",
        "prefect.resource.name": "external-db-monitor"
      },
      "payload": {
        "job_id": "12345",
        "action": "refresh_cache"
      }
    }
  ]'

If you want, I can help you tailor this to your self-hosted auth model. --- ## 6. How payloads become flow parameters The key mechanism is the

parameters

mapping in the trigger. Example:

yaml
parameters:
  job_id: "{{ event.payload.job_id }}"
  action: "{{ event.payload.action }}"

If the event payload is:

json
{
  "job_id": "12345",
  "action": "refresh_cache"
}

then the flow run will start as if you invoked:

python
react_to_db_event(job_id="12345", action="refresh_cache")

You can also pass the whole payload as a dict using the JSON/Jinja form shown earlier. --- ## 7. Recommended design for your use case Based on your description, I’d recommend this architecture: ### Pattern A: simple and robust - Deployment 1:

db_checker

- scheduled every N minutes - queries external DB - emits

external.db.condition-met

for each actionable item - Deployment 2:

react_to_db_event

- event-driven deployment - triggered by

external.db.condition-met

- runs your local process on the process worker machine This is nice because: - DB polling logic is isolated - event processing logic is isolated - easier to retry and monitor each piece separately ### Pattern B: if DB system can push out - external service checks DB or receives DB notifications - external service emits Prefect event directly - event-driven deployment handles execution Use this if you already have an external service or listener. --- ## 8. Example full setup ###

react_to_db_event.py

python
from prefect import flow
import subprocess
from prefect.events.schemas.deployment_triggers import DeploymentEventTrigger

@flow(log_prints=True)
def react_to_db_event(job_id: str, action: str, raw_payload: dict | None = None):
    print(f"Reacting to event for job_id={job_id}, action={action}")
    print(f"Raw payload: {raw_payload}")

    result = subprocess.run(
        ["./my_local_script.sh", job_id, action],
        capture_output=True,
        text=True,
        check=False,
    )

    print("stdout:", result.stdout)
    print("stderr:", result.stderr)

    if result.returncode != 0:
        raise RuntimeError(f"Process failed with code {result.returncode}")


if __name__ == "__main__":
    react_to_db_event.serve(
        name="db-event-handler",
        work_pool_name="my-process-pool",
        triggers=[
            DeploymentEventTrigger(
                expect={"external.db.condition-met"},
                match={"prefect.resource.id": "external.db.monitor"},
                parameters={
                    "job_id": "{{ event.payload.job_id }}",
                    "action": "{{ event.payload.action }}",
                    "raw_payload": {
                        "__prefect_kind": "json",
                        "value": {
                            "__prefect_kind": "jinja",
                            "template": "{{ event.payload | tojson }}"
                        }
                    },
                },
            )
        ],
    )

###

db_checker.py

```python from prefect import flow

from prefect.events import emit_event @flow(log_prints=True) def db_checker(): # Replace this with real DB query logic records = [ {"job_id": "12345", "action": "refresh_cache"}, {"job_id": "67890", "action": "send_notification"}, ] for record in records: emit_event( event="external.db.condition-met", resource={ "prefect.resource.id": "external.db.monitor", "prefect.resource.name": "external-db-monitor", }, payload=record, ) print(f"Emitted event for job {record['job_id']}") if name == "__main__": db_checker.serve( name="db-checker", work_pool_name="my-process-pool", cron="*/5 * * * *", )

### Start everything

```bash
export PREFECT_API_URL="http://<your-prefect-host>:4200/api"

prefect work-pool create my-process-pool --type process
prefect worker start --pool my-process-pool --type process
python react_to_db_event.py
python db_checker.py

--- ## 9. Important self-hosted notes A few things to watch for on self-hosted: - Automations/events must be enabled and processed by server services - The worker machine must have access to: - your Prefect API - the local executable/script you want to run - the necessary environment variables and credentials - A process worker runs the flow as a local subprocess on the worker host, so if your “local process” must run on a specific machine, run the process worker there - If events seem to be received but no runs are created, check: - server services/background services are running - trigger

expect

and

match

values actually align with emitted event content - deployment is active and in the expected work pool - worker is polling and healthy --- ## 10. What I’d suggest for you specifically Since your trigger source is “a simple check to an external database,” I’d start with: 1. a small scheduled flow that queries the DB 2.

emit_event(...)

when the condition is true 3. a separate deployment with

DeploymentEventTrigger

4. process worker on the host that needs to execute the local process That gives you clean observability and avoids building a separate daemon unless you really need one. --- ## 11. If you want, I can provide a tailored example If you share just these 3 details, I can give you a more exact setup: 1. How you deploy today: -

serve(...)

-

prefect deploy

-

prefect.yaml

2. How your DB check works: - periodic polling query - DB notification/listener - external app/service 3. What payload you want passed into the flow: - example JSON / row shape If you send that, I can generate: - a ready-to-run

prefect.yaml

- Python flow code - event emitter code - the exact CLI steps for your environment

Perfect — given your setup, here is the recommended pattern: - Deployment A: scheduled polling deployment - runs every N minutes - checks the external DB - emits a custom event when

refresh_cache

should happen - Deployment B: event-driven deployment - listens for that custom event - receives payload like

{"refresh_cache": true}

- runs on your process worker - launches your local process This works well with

prefect.yaml

and self-hosted Prefect. --- # Recommended architecture

text
scheduled polling deployment
    -> query external DB
    -> if condition met, emit event: external.db.refresh-cache-requested
    -> event trigger on second deployment fires
    -> process worker executes local process

--- # Files I’ll show a minimal working example with: -

flows/db_poll.py

-

flows/cache_refresh.py

-

deployments/my_deployment.yaml

--- ## 1. Reactive flow: run the local process

flows/cache_refresh.py

python
from prefect import flow
import subprocess


@flow(log_prints=True)
def cache_refresh_handler(refresh_cache: bool = False):
    if not refresh_cache:
        print("No refresh requested; exiting.")
        return

    print("Refresh requested; starting local process.")

    result = subprocess.run(
        ["./refresh_cache.sh"],
        capture_output=True,
        text=True,
        check=False,
    )

    print("stdout:", result.stdout)
    print("stderr:", result.stderr)

    if result.returncode != 0:
        raise RuntimeError(f"refresh_cache.sh failed with code {result.returncode}")

    print("Cache refresh completed successfully.")

If you want, this flow can also accept the entire payload instead of only

refresh_cache

. --- ## 2. Polling flow: query DB and emit event

flows/db_poll.py

python
from prefect import flow
from prefect.events import emit_event


def check_database() -> bool:
    # Replace this with your real polling query.
    # Return True when a cache refresh should be triggered.
    return True


@flow(log_prints=True)
def poll_database_and_emit():
    should_refresh = check_database()

    print(f"DB check result: should_refresh={should_refresh}")

    if should_refresh:
        event = emit_event(
            event="external.db.refresh-cache-requested",
            resource={
                "prefect.resource.id": "external.db.poller",
                "prefect.resource.name": "external-db-poller",
            },
            payload={
                "refresh_cache": True
            },
        )
        print(f"Emitted event: {event.id if event else 'None'}")
    else:
        print("No event emitted.")

--- ## 3.

prefect.yaml

deployment definitions Since you deploy with a separate YAML file, here is a concrete example for

deployments/my_deployment.yaml

.

yaml
prefect-version: "3.*"

deployments:
  - name: db_poll_prod
    entrypoint: flows/db_poll.py:poll_database_and_emit
    work_pool:
      name: my-process-pool
    schedules:
      - interval: 300

  - name: cache_refresh_prod
    entrypoint: flows/cache_refresh.py:cache_refresh_handler
    work_pool:
      name: my-process-pool
    triggers:
      - enabled: true
        type: event
        expect:
          - external.db.refresh-cache-requested
        match:
          prefect.resource.id: external.db.poller
        parameters:
          refresh_cache: "{{ event.payload.refresh_cache }}"

Notes: -

db_poll_prod

runs every 300 seconds -

cache_refresh_prod

is event-driven - when the event payload contains

refresh_cache: true

, that value is passed into the flow parameter --- # Deploy commands Since you already use

prefect deploy --prefect-file ... -n ...

, keep doing that. Example:

bash
prefect deploy --prefect-file deployments/my_deployment.yaml -n db_poll_prod
prefect deploy --prefect-file deployments/my_deployment.yaml -n cache_refresh_prod

Then ensure your process worker is running: ```bash

prefect worker start --pool my-process-pool --type process

---

# End-to-end setup steps

## Step 1: start self-hosted Prefect
Make sure your server is running and your worker can reach it.

```bash
export PREFECT_API_URL="http://<your-prefect-host>:4200/api"

## Step 2: create the work pool if needed

bash
prefect work-pool create my-process-pool --type process

## Step 3: start the worker

bash
prefect worker start --pool my-process-pool --type process

## Step 4: deploy both deployments

bash
prefect deploy --prefect-file deployments/my_deployment.yaml -n db_poll_prod
prefect deploy --prefect-file deployments/my_deployment.yaml -n cache_refresh_prod

## Step 5: verify behavior - the polling deployment runs on schedule - when the DB check says yes, it emits

external.db.refresh-cache-requested

- the event-driven deployment gets triggered - the process worker runs

refresh_cache.sh

--- # Slightly richer payload option If later you want more than just a boolean, you can emit:

python
payload={
    "refresh_cache": True,
    "cache_name": "customer_cache",
    "reason": "stale_entries",
}

and map those in the trigger:

yaml
parameters:
  refresh_cache: "{{ event.payload.refresh_cache }}"
  cache_name: "{{ event.payload.cache_name }}"
  reason: "{{ event.payload.reason }}"

Then your flow could be:

python
@flow(log_prints=True)
def cache_refresh_handler(
    refresh_cache: bool = False,
    cache_name: str | None = None,
    reason: str | None = None,
):
    ...

That’s usually a better long-term design. --- # If you want to pass the full payload too You can also pass the whole event payload as a dict:

yaml
parameters:
  refresh_cache: "{{ event.payload.refresh_cache }}"
  raw_payload:
    __prefect_kind: json
    value:
      __prefect_kind: jinja
      template: "{{ event.payload | tojson }}"

Then:

python
from prefect import flow
import subprocess

@flow(log_prints=True)
def cache_refresh_handler(refresh_cache: bool = False, raw_payload: dict | None = None):
    print(f"raw_payload={raw_payload}")
    ...

--- # Best practice for your use case For your specific case, I’d recommend: - keep the payload simple - emit one event per actionable DB result - include at least: -

prefect.resource.id

- a stable custom event name - minimal payload fields A good event name here is:

text
external.db.refresh-cache-requested

That makes it obvious this is a custom external event. --- # Troubleshooting checklist If the second deployment does not trigger: 1. Confirm the polling flow actually emitted the event - check flow logs - look for the

Emitted event:

line 2. Confirm the event name matches exactly - emitted:

external.db.refresh-cache-requested

- trigger `expect`: same exact string 3. Confirm the resource match matches exactly - emitted:

python
     "prefect.resource.id": "external.db.poller"

- trigger:

yaml
     match:
       prefect.resource.id: external.db.poller

4. Confirm server-side automations/events are running - especially important on self-hosted 5. Confirm the worker is polling the right pool -

my-process-pool

6. Confirm the event-driven deployment was actually applied - redeploy after changing trigger config --- # Minimal version if you want the simplest possible config If you want the smallest version possible, use this YAML: ```yaml prefect-version: "3.*" deployments: - name: db_poll_prod entrypoint: flows/db_poll.py:poll_database_and_emit work_pool: name: my-process-pool schedules: - interval: 300 - name: cache_refresh_prod entrypoint: flows/cache_refresh.py:cache_refresh_handler work_pool: name: my-process-pool triggers: - enabled: true type: event expect: - external.db.refresh-cache-requested match: prefect.resource.id: external.db.poller

parameters: refresh_cache: "{{ event.payload.refresh_cache }}" ``` --- If you want, I can next give you a single copy-paste-ready deployment YAML plus both Python files adjusted to your exact folder layout.

Seeing the Automations menu in the UI is not enough to confirm the backend automation/event services are actually running. It tells you the UI supports the feature, but not that the self-hosted server is actively processing events and triggers. ## Best ways to confirm ### 1. Check enabled server services from the CLI Run:

bash
prefect server services ls

What you want to see is that the services related to events/automations are enabled, especially: -

EventPersister

-

ReactiveTriggers

-

Actions

Those are the important ones for: - receiving/persisting events - evaluating automation triggers - executing automation actions If those are disabled, event-driven deployments will not work correctly. --- ### 2. Verify the relevant settings/env vars Check whether these are set to false anywhere in your environment:

bash
echo $PREFECT_SERVER_SERVICES_TRIGGERS_ENABLED
echo $PREFECT_SERVER_SERVICES_EVENT_PERSISTER_ENABLED

In normal cases they should either be: - unset, or - explicitly

true

If either is

false

, that’s a problem. --- ### 3. Check server logs If you run Prefect server in the foreground, inspect its logs for service startup. If you run it with Docker / Compose / systemd, inspect those logs instead. You’re looking for evidence that services started successfully, and no errors around triggers/events. Examples of useful places:

bash
docker logs <prefect-container>
docker compose logs prefect-server
journalctl -u prefect-server -f

--- ### 4. Do the most reliable test: emit a test event and see if it triggers a deployment This is the best real-world confirmation. If your event-driven deployment is already set up, emit a matching event manually. For your case, that would be something like:

python
from prefect.events import emit_event

emit_event(
    event="external.db.refresh-cache-requested",
    resource={
        "prefect.resource.id": "external.db.poller",
        "prefect.resource.name": "external-db-poller",
    },
    payload={"refresh_cache": True},
)

If this creates a deployment run, then: - events are being accepted - triggers are being evaluated - actions are being executed That’s stronger proof than the UI alone. --- ### 5. Check in the UI for evidence of event activity In the UI, if available on your version/setup, look for: - automation activity - event feed / event history - deployment runs created by automations If you emit a test event and nothing at all appears, that suggests backend processing may not be functioning. --- ## What I recommend you do first Run this first:

bash
prefect server services ls

If you paste the output here, I can tell you immediately whether the needed automation/event services are enabled. ## Short answer to your direct question No, seeing the Automations menu on the self-hosted dashboard does not by itself confirm that server-side automations/events are running. You need to confirm via: -

prefect server services ls

- logs - or a real emitted-event test If you want, I can give you a small one-off test flow + emit script specifically for your self-hosted instance so you can verify the whole chain end to end.