Short answer - Yes: you can launch flows in Prefect 3.x on filesystem events. - Common pattern: run a small watchdog process that emits Prefect events or directly calls a deployment. The flow then starts automatically via an event trigger (or your script) as soon as a file is created/updated. - The background watcher can start automatically (systemd, Docker/K8s sidecar, or baked into your worker image/entrypoint). Two solid approaches 1) Watchdog emits Prefect events → Deployment event trigger starts your flow - Watchdog script (long‑running) sends events to Prefect with

emit_event

. - Your deployment has an event trigger (with Jinja mapping) to pass the file path. Watcher

# pip install watchdog prefect
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from prefect.events import emit_event

WATCH_DIR = "/path/to/watch"

class FileChangeHandler(FileSystemEventHandler):
    def on_created(self, event):
        self._maybe_emit(event, "file.created")

    def on_modified(self, event):
        self._maybe_emit(event, "file.modified")

    def _maybe_emit(self, event, name):
        if event.is_directory:
            return
        path = Path(event.src_path)
        emit_event(
            event=name,
            resource={
                "prefect.resource.id": f"file.system.{path.name}",
                "prefect.resource.name": path.name,
                "path": str(path),
            },
            payload={"size": path.stat().st_size if path.exists() else None},
        )

def start():
    obs = Observer()
    obs.schedule(FileChangeHandler(), WATCH_DIR, recursive=True)
    obs.start()
    try:
        obs.join()
    finally:
        obs.stop()
        obs.join()

if __name__ == "__main__":
    start()

Flow + deployment trigger

from prefect import flow
from prefect.events import DeploymentEventTrigger

@flow(log_prints=True)
def process_file(path: str):
    print(f"Processing {path}")

if __name__ == "__main__":
    process_file.serve(
        name="file-processor",
        triggers=[
            DeploymentEventTrigger(
                enabled=True,
                match={"prefect.resource.id": "file.system.*"},
                expect=["file.created", "file.modified"],
                parameters={"path": "{{ event.resource.path }}"},
            )
        ],
    )

Docs: - Events overview: Events - Deployment event triggers: Creating deployment triggers 2) Watchdog calls your deployment directly (no events) - Simpler to start: the watcher just calls

run_deployment(...)

with the path.

# pip install watchdog prefect
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from prefect.deployments import run_deployment

DEPLOYMENT = "process-file/file-processor"  # <FLOW_NAME>/<DEPLOYMENT_NAME>

class FileChangeHandler(FileSystemEventHandler):
    def on_created(self, event):
        self._run(event)

    def on_modified(self, event):
        self._run(event)

    def _run(self, event):
        if event.is_directory:
            return
        path = str(Path(event.src_path))
        # Optional: idempotency_key to avoid duplicates for rapid repeats
        run_deployment(
            name=DEPLOYMENT,
            parameters={"path": path},
            timeout=0,  # don't block the watcher
            idempotency_key=f"{path}",
        )

def start():
    obs = Observer()
    obs.schedule(FileChangeHandler(), "/path/to/watch", recursive=True)
    obs.start()
    try:
        obs.join()
    finally:
        obs.stop()
        obs.join()

if __name__ == "__main__":
    start()

CLI option (if you prefer shelling out):

prefect deployment run "process-file/file-processor" --param path=/path/to/file

We verified this command: Run deployments guide Auto-starting the background watcher - Systemd on a VM/bare metal: run your watcher as a service so it restarts on boot/crash. Example guide: Daemonize processes - Docker: run the watcher as a sidecar container in the same Compose/stack as your Prefect worker. - Kubernetes: run the watcher as a sidecar or a separate Deployment/DaemonSet; both can emit events or call the API. - Bake it into your worker image and start via a supervisor (e.g., s6, supervisord) so worker and watcher start together. - You can also run a long‑lived “monitor” flow, but keeping a small external watcher process is usually simpler and more robust. Notes and best practices - You’re triggering flows, not individual tasks. The flow will run your tasks. - Debounce to avoid duplicate triggers on rapid successive writes (e.g., track mtimes or add a short delay). - For idempotency, set

idempotency_key

keyed to file path + last modified time when calling

run_deployment

. - Make sure your watcher has Prefect connectivity (PREFECT_API_URL; PREFECT_API_KEY for Cloud). - If you can receive events over HTTP instead of watching local disk, consider Prefect Cloud Webhooks: Webhooks + an event trigger to start the flow. Would you share your environment so I can tailor a drop‑in setup? - Where does the filesystem live (local VM, Docker, K8s, network share)? - Are you on Prefect Cloud or Prefect Server? - Do you prefer the “events + trigger” model or the “watcher calls deployment directly” model?

Great context — running on Windows Services does not block this pattern. Here’s what changes and what to do: What’s the same - Both approaches still work on Windows: 1) Watchdog emits Prefect events → your deployment uses an event trigger. 2) Watchdog directly calls your deployment with

run_deployment(...)

. - Your Prefect worker will pick up the scheduled run as usual. What’s different on Windows - Webhooks are Cloud-only. Since you’re self-hosting, prefer the watchdog approach (emit events or call the deployment). - Run your watcher as its own Windows Service so it starts automatically and restarts on failure. - Ensure the service account has: - Read access to the watched directory - Network access to your Prefect Server - Environment variables set for Prefect connectivity (e.g. `PREFECT_API_URL`;

PREFECT_API_KEY

only if you front Prefect with auth). Recommended setup on Windows A) Watchdog → emit events → event-triggered deployment - Watcher (same as Linux/macOS):

# pip install watchdog prefect
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from prefect.events import emit_event

WATCH_DIR = r"C:\data\inbox"

class FileChangeHandler(FileSystemEventHandler):
    def on_created(self, event):
        self._maybe_emit(event, "file.created")

    def on_modified(self, event):
        self._maybe_emit(event, "file.modified")

    def _maybe_emit(self, event, name):
        if event.is_directory:
            return
        p = Path(event.src_path)
        emit_event(
            event=name,
            resource={
                "prefect.resource.id": f"file.system.{p.name}",
                "prefect.resource.name": p.name,
                "path": str(p),
            },
            payload={"size": p.stat().st_size if p.exists() else None},
        )

def start():
    obs = Observer()
    obs.schedule(FileChangeHandler(), WATCH_DIR, recursive=True)
    obs.start()
    try:
        obs.join()
    finally:
        obs.stop()
        obs.join()

if __name__ == "__main__":
    start()

- Flow + event trigger:

from prefect import flow
from prefect.events import DeploymentEventTrigger

@flow(log_prints=True)
def process_file(path: str):
    print(f"Processing {path}")

if __name__ == "__main__":
    process_file.serve(
        name="file-processor",
        triggers=[
            DeploymentEventTrigger(
                enabled=True,
                match={"prefect.resource.id": "file.system.*"},
                expect=["file.created", "file.modified"],
                parameters={"path": "{{ event.resource.path }}"},
            )
        ],
    )

Refs: - Events: Events - Deployment event triggers: Creating deployment triggers B) Watchdog directly calls your deployment (no events) ``` # pip install watchdog prefect from pathlib import Path from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler from prefect.deployments import run_deployment DEPLOYMENT = "process-file/file-processor" # <FLOW_NAME>/<DEPLOYMENT_NAME> WATCH_DIR = r"C:\data\inbox" class FileChangeHandler(FileSystemEventHandler): def on_created(self, event): self._run(event) def on_modified(self, event): self._run(event) def _run(self, event): if event.is_directory: return path = str(Path(event.src_path)) run_deployment( name=DEPLOYMENT, parameters={"path": path}, timeout=0, # don’t block the service # optional: reduce duplicates if the same file fires multiple times idempotency_key=path, ) def start(): obs = Observer() obs.schedule(FileChangeHandler(), WATCH_DIR, recursive=True) obs.start() try: obs.join() finally: obs.stop() obs.join()

if name == "__main__": start() ``` Run deployments guide: Run deployments Installing the watcher as a Windows Service - Easiest path: use NSSM (Non-Sucking Service Manager). - Download NSSM and run: - nssm install PrefectFileWatcher - Application:

C:\Path\To\Python\python.exe

- Arguments:

C:\path\to\file_watcher.py

- Startup directory:

C:\path\to

- Environment (add): -

PREFECT_API_URL=http://<prefect-server-host>:4200/api

(adjust for your server) - (If applicable)

PREFECT_API_KEY=...

- Set “Startup type” to Automatic. - Optional dependency: set DependOnService to e.g. Tcpip or your “PrefectServer” service so the watcher starts after networking/server. - nssm start PrefectFileWatcher - Alternative: Windows Service Wrapper (WinSW) with an XML config calling python + your script. - Native sc.exe alone typically requires a service wrapper; that’s why NSSM or WinSW is recommended. Operational tips - Debounce: Windows editors often cause multiple modify events; consider a short delay or track last mtime to avoid duplicate runs. - Permissions: run the service under an account that can read the watched directory and reach the Prefect Server. - Logging: route stdout/stderr to a file (NSSM I/O tab) so you can troubleshoot. - Connectivity: confirm the service environment has

PREFECT_API_URL

set correctly; worker and watcher should point at the same API. If you share: - Your Prefect version (assuming 3.x), - Your Prefect Server URL, - The directory you want to watch, - Whether you prefer “events + trigger” or “direct call”, I’ll send you a ready-to-run watcher script and exact NSSM settings for your host.

Perfect — here’s a Windows-friendly, events-based setup with placeholders. What you’ll build - A Windows Service running a small watchdog that emits Prefect events on file create/modify. - A Prefect deployment with an Event Trigger that starts your flow and passes the file path from the event. 1) Watcher service (emits events) Save as file_watcher.py. Replace placeholders: - WATCH_DIR = r"C:\YOUR\WATCH\DIRECTORY" - Set PREFECT_API_URL to your self-hosted Server URL (e.g. http://YOUR_SERVER:4200/api)

# pip install watchdog prefect

import time
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from prefect.events import emit_event

WATCH_DIR = r"C:\YOUR\WATCH\DIRECTORY"

class FileChangeHandler(FileSystemEventHandler):
    def on_created(self, event):
        self._emit(event, "file.created")

    def on_modified(self, event):
        # Optional: skip very small files or add debounce here if you get dupes
        self._emit(event, "file.modified")

    def _emit(self, event, name):
        if event.is_directory:
            return
        p = Path(event.src_path)
        emit_event(
            event=name,
            resource={
                "prefect.resource.id": f"file.system.{p.name}",
                "prefect.resource.name": p.name,
                "path": str(p),
                "directory": str(p.parent),
            },
            payload={
                "size": p.stat().st_size if p.exists() else None,
                "mtime": p.stat().st_mtime if p.exists() else None,
            },
        )

def start():
    obs = Observer()
    obs.schedule(FileChangeHandler(), WATCH_DIR, recursive=True)
    obs.start()
    try:
        while True:
            time.sleep(1)
    finally:
        obs.stop()
        obs.join()

if __name__ == "__main__":
    start()

Required environment for this process: - PREFECT_API_URL=http://YOUR_PREFECT_SERVER:4200/api - If you added auth to your Server behind a proxy, also set PREFECT_API_KEY. 2) Your flow Save as process_file.py

from prefect import flow

@flow(log_prints=True)
def process_file(path: str):
    print(f"Processing: {path}")
    # TODO: your logic here

3) Create a deployment for the flow - Run: - prefect deploy .\process_file.py:process_file - Follow the prompts to choose a work pool, name (e.g., file-processor), etc. - Or create via UI (Deployments → +). 4) Add an Event Trigger to the deployment (UI steps) - Go to your deployment (e.g., process_file/file-processor) - Triggers → Add Trigger → Event - Enabled: On - Match (resource): prefect.resource.id = file.system.* - Expect (events): file.created, file.modified - Parameters mapping: - path = {{ event.resource.path }} - Save That’s it: whenever the watcher emits file.created/modified events, the deployment will start and pass the file path. Install the watcher as a Windows Service (NSSM) - Download NSSM - Open an elevated CMD/PowerShell and run: - nssm install PrefectFileWatcher - Application: C:\Path\To\Python\python.exe - Arguments: C:\path\to\file_watcher.py - Startup directory: C:\path\to - I/O: (optional) redirect stdout/stderr to log files for troubleshooting - Environment (add): - PREFECT_API_URL=http://YOUR_PREFECT_SERVER:4200/api - (Optional) PREFECT_API_KEY=YOUR_KEY_IF_APPLICABLE - Start it: - nssm start PrefectFileWatcher - Ensure the service account has: - Read access to C:\YOUR\WATCH\DIRECTORY - Network access to YOUR_PREFECT_SERVER Quick tests - Manually trigger a run to confirm the deployment: - prefect deployment run "process_file/file-processor" --param path=C:\test\file.txt --watch - Drop a file in the watched directory and confirm a run is created. Docs - Events: Events - Deployment event triggers: Creating deployment triggers

- Run deployments: Run deployments - Daemonizing (general guidance): Daemonize processes Want me to generate the exact NSSM config and a prefilled watcher script if you share: - YOUR_PREFECT_SERVER (URL/host) - C:\YOUR\WATCH\DIRECTORY - Your deployment name and work pool name?

Absolutely — here are ready-to-paste JSON trigger definitions for a deployment that will start on file create/modify events. Use these in the Prefect UI under your Deployment → Triggers → Add Trigger → JSON. 1) File created or modified (any file)

{
  "type": "event",
  "enabled": true,
  "match": {
    "prefect.resource.id": "file.system.*"
  },
  "expect": [
    "file.created",
    "file.modified"
  ],
  "parameters": {
    "path": "{{ event.resource.path }}"
  }
}

2) Only new files (creations only)

{
  "type": "event",
  "enabled": true,
  "match": {
    "prefect.resource.id": "file.system.*"
  },
  "expect": [
    "file.created"
  ],
  "parameters": {
    "path": "{{ event.resource.path }}"
  }
}

3) Filter to a specific directory (Windows path) and pass extra fields - Note: backslashes must be escaped in JSON

{
  "type": "event",
  "enabled": true,
  "match": {
    "prefect.resource.id": "file.system.*",
    "directory": "C:\\\\YOUR\\\\WATCH\\\\DIRECTORY"
  },
  "expect": [
    "file.created",
    "file.modified"
  ],
  "parameters": {
    "path": "{{ event.resource.path }}",
    "directory": "{{ event.resource.directory }}",
    "size": "{{ event.payload.size }}",
    "mtime": "{{ event.payload.mtime }}"
  }
}

These assume your watcher emits events like: - event: file.created / file.modified - resource fields: path, directory, prefect.resource.id=f"file.system.{filename}" - payload fields: size, mtime Docs for reference: - Deployment event triggers: Creating deployment triggers - Events overview: Events