Yes. In Prefect 3.x you can start a Process worker from Python, so you can run any init logic before it begins polling. The simplest pattern is: run your initialization, then construct and start the worker. Example (blocking in the foreground):

from prefect.workers.process import ProcessWorker

def initialize_stuff():
    # your custom init (logging, DB connections, env setup, etc.)
    ...

if __name__ == "__main__":
    initialize_stuff()

    worker = ProcessWorker(
        work_pool_name="my-pool",
        work_queues=["default"],
        name="my-proc-worker",
        # heartbeat_interval_seconds=..., limit=..., prefetch_seconds=...,
    )
    # Blocks this process, polls for work until stopped (Ctrl+C or SIGTERM)
    worker.start()

If you want your init to run as part of the worker’s lifecycle (after it’s constructed but before it polls), subclass and override setup():

from prefect.workers.process import ProcessWorker

class MyProcessWorker(ProcessWorker):
    async def setup(self):
        await super().setup()
        # async initialization here
        # e.g., warm up caches, validate credentials, etc.

if __name__ == "__main__":
    MyProcessWorker(work_pool_name="my-pool", work_queues=["default"]).start()

Run the worker in the background (so your program can keep doing other things):

import threading
from prefect.workers.process import ProcessWorker

worker = ProcessWorker(work_pool_name="my-pool", work_queues=["default"])

t = threading.Thread(target=worker.start, kwargs={"with_healthcheck": True}, daemon=True)
t.start()

# Do other app logic here...
# To stop gracefully, send SIGINT/SIGTERM to the process. The worker handles a clean shutdown.
# Alternatively, exit the process or manage your own stop condition.

Useful flags: - run_once=True lets the worker do a single polling pass then exit (good for batch-style starts) - with_healthcheck=True starts a healthcheck server for monitoring Docs and references: - Workers overview - ProcessWorker API - Developing a custom worker (lifecycle methods like setup) A couple of notes: - ProcessWorker.start() is synchronous and blocking; use a thread if you need it in the background. - For per-flow-run customization (env, command, working dir), use deployment/job configuration; for “once at startup” behavior, use code before start() or override setup(). If you can share what kind of initialization you want to run (e.g., register secrets, configure logging, run migrations), I can suggest the cleanest hook and a tailored example.