<@ULVA73B9P> <@ULVA73B9P>
I am trying to achieve ...
# ask-marvino
oleander-whirlwind
08/10/2026, 9:25 AM@Marvin @Marvin
I am trying to achieve the following architecture in Prefect 3.x:
Requirements:
✅ Prefect deployment orchestration
✅ A long-running flow served via flow.serve()
✅ Deployment runs triggered normally through Prefect
✅ A separate long-lived execution environment
✅ Heavy libraries (xarray, zarr, rasterio, custom analytics code) loaded exactly once
✅ No subprocess startup and no re-import of heavy libraries for each request
✅ No Celery
Current understanding:
- flow.serve() creates a long-lived runner process
- deployment executions are still run in isolated subprocesses
- therefore flow.serve() alone does not provide a warm Python execution environment
What I would like to do instead:
Container A:
- flow.serve()
- accepts deployment runs
- performs orchestration only
Container B:
- Prefect Background Task Worker
- loads xarray/zarr/rasterio/custom code once at startup
- remains alive indefinitely
- executes submitted work repeatedly using the same interpreter
Flow:
flow.serve()
-> deployment run received
-> submits work to a Background Task
-> Background Task Worker executes heavy computation
-> flow waits for or retrieves result
-> flow returns result
Questions:
1. Is this architecture officially supported in Prefect 3.x?
2. Do Background Task Workers actually execute multiple task runs in the same long-lived Python interpreter, allowing module imports and global objects to remain loaded between executions?
3. Can a flow running from flow.serve() submit work to a Background Task Worker and synchronously wait for the result?
4. Are there any Prefect-native patterns recommended by the community for combining:
- flow.serve() deployments
- Background Task Workers
- warm execution runtimes
5. If this architecture is not supported, what is the closest Prefect-native solution that provides:
- deployment-based orchestration
- a permanently warm Python runtime
- no per-execution library re-imports?
m
Marvin
08/10/2026, 9:25 AM✅ thought for 156.7 seconds
Marvin
08/10/2026, 9:28 AMYes — in Prefect 3.x, this is basically a supported pattern, with one important caveat:
Your flow runtime and your background-task runtime have different execution models.
Short answers
1. Officially supported?
Yes. Background tasks and task workers are a real Prefect 3.x feature, and using a flow as the orchestrator plus a separate task worker as the execution runtime is aligned with the product model.
2. Do Background Task Workers reuse the same interpreter?
Yes. Task workers are long-lived processes and execute task runs in-process via threads, not a fresh subprocess per task run. That means module imports, globals, caches, and objects created at worker startup can persist across executions.
3. Can a
flow.serve().result()flow.serve()flow.serve()flow.serve()xarrayzarrrasterio Copy code
python
from prefect import flow, task
# Heavy imports should live in the worker process module
@task(persist_result=True)
def heavy_compute(payload):
import xarray
import zarr
import rasterio
# or initialize these once at module import time in the worker container
return {"status": "done", "payload": payload}
@flow
def orchestrator(payload):
future = heavy_compute.delay(payload)
result = future.result()
return result
if __name__ == "__main__":
orchestrator.serve(name="orchestrator") Copy code
python
from prefect.task_worker import serve
from my_module import heavy_compute
if __name__ == "__main__":
serve(heavy_compute)Marvin
08/10/2026, 9:28 AMreturn {"ready": ANALYTICS_CONTEXT["ready"], "payload": payload}
```
That way they are loaded once when the task worker process starts.
Behavior details that matter
Because task workers execute in a persistent process:
- module-level imports persist
- module globals persist
- singleton-style objects can persist
- connection pools / models / lookup tables can persist
But because concurrency is thread-based, be careful with:
- thread safety of globals
- mutable shared state
- non-thread-safe native libraries
- per-task cleanup
If your workload is CPU-heavy and releases the GIL poorly, thread concurrency may not help much. But it still preserves warm imports.
Answering your questions directly
1. Is this architecture officially supported in Prefect 3.x?
Yes. Using
flow.serve()TaskWorkerflow.serve().delay().result()flow.serve()flow.serve()xarrayzarrrasterioflow.serve().delay().result()o
oleander-whirlwind
08/10/2026, 9:32 AM@Marvin
@Marvin
I need the exact Prefect 3.x APIs and a fully executable example for the following architecture.
Target architecture:
Container A
-----------
flow.serve()
- receives deployment runs
- acts only as orchestrator
Container B
-----------
Background Task Worker
- long-lived process
- xarray imported once
- zarr imported once
- rasterio imported once
- RemoteFileSystem loaded once
- credentials loaded once
- executes repeated requests without re-importing
Requirements:
✅ Prefect deployment orchestration
✅ flow.serve()
✅ warm execution runtime
✅ Background Task Worker
✅ flow waits for task result
✅ no Celery
✅ no external queue
✅ heavy libraries loaded once per worker startup
Questions:
1. What are the EXACT imports for Background Task Workers in current Prefect 3.x?
For example, is it:
from prefect.task_worker import serve
or something else?
2. What is the exact syntax to serve a task worker?
Example wanted:
# worker.py
...
3. What is the exact syntax to submit background work from inside a flow?
Example wanted:
future = my_task.delay(...)
Is .delay() correct?
4. What object type is returned from .delay()?
5. How do I synchronously wait for completion and retrieve the result from inside a flow?
Example wanted:
result = ???
6. Can persisted task results be retrieved directly through the returned future?
7. Please provide a COMPLETE WORKING EXAMPLE consisting of exactly these files:
tasks.py
--------
- module-level warm imports
- module-level cache initialization
- background task definition
worker.py
---------
- exact worker startup code
- exact serve statement
orchestrator.py
---------------
- flow.serve()
- deployment definition
- submit background task
- wait for completion
- return task result
Please use only officially supported Prefect 3.x APIs.
No pseudocode.
No architecture diagrams.
No shortened examples.
Complete executable files only.
m
Marvin
08/10/2026, 9:32 AM✅ thought for 24.1 seconds
Marvin
08/10/2026, 9:33 AMBelow are the exact Prefect 3.x APIs and complete executable files.
Direct answers first
1. Exact import for Background Task Worker serve helper
Yes:
Copy code
python
from prefect.task_worker import serve Copy code
python
from prefect.task_worker import TaskWorkerserve(...) Copy code
python
serve(my_task) Copy code
python
serve(my_task, limit=2, status_server_port=8081) Copy code
python
future = my_task.delay(...).delay()PrefectDistributedFuture Copy code
python
result = future.result() Copy code
python
result = future.result(timeout=300)persist_result=Truefuture.result()from prefect import flow, taskfrom prefect.task_worker import servefuture = some_task.delay(...)result = future.result(timeout=...)my_flow.serve(name="...")RemoteFileSystemRemoteFileSystemNonetasks.pyMarvin
08/10/2026, 9:33 AMdef warm_heavy_compute(dataset_name: str, value: int) -> dict[str, Any]:
with _CACHE_LOCK:
_CACHE["request_count"] += 1
request_count = _CACHE["request_count"]
dataset = xr.Dataset(
data_vars={
"example_value": (("x",), [value, value + 1, value + 2]),
},
coords={"x": [0, 1, 2]},
attrs={"dataset_name": dataset_name},
)
mean_value = float(dataset["example_value"].mean().item())
result = {
"dataset_name": dataset_name,
"input_value": value,
"mean_value": mean_value,
"request_count_seen_by_worker": request_count,
"warm_state": {
"initialized": _CACHE["initialized"],
"xarray_version": _CACHE["xarray_version"],
"zarr_version": _CACHE["zarr_version"],
"rasterio_version": _CACHE["rasterio_version"],
"credentials_loaded": _CACHE["credentials_loaded"],
"remote_fs_loaded": _CACHE["remote_fs_loaded"],
},
"worker_credentials_data_root": WORKER_CREDENTIALS["data_root"],
"remote_filesystem_loaded": REMOTE_FS is not None,
}
print(f"Handled request #{request_count} in warm worker process")
return result
Copy code
---
*File: `worker.py`*
```python
from prefect.task_worker import serve
from tasks import warm_heavy_compute
if __name__ == "__main__":
serve(
warm_heavy_compute,
limit=2,
status_server_port=8081,
)orchestrator.py Copy code
python
from __future__ import annotations
from prefect import flow
from tasks import warm_heavy_compute
@flow(name="background-task-orchestrator", log_prints=True)
def orchestrate(dataset_name: str = "demo-dataset", value: int = 10) -> dict:
future = warm_heavy_compute.delay(dataset_name, value)
result = future.result(timeout=300)
print(f"Received result from background task: {result}")
return result
if __name__ == "__main__":
orchestrate.serve(
name="background-task-orchestrator-deployment",
) Copy code
bash
python worker.py Copy code
bash
python orchestrator.py Copy code
bash
prefect deployment run 'background-task-orchestrator/background-task-orchestrator-deployment' --param dataset_name=test --param value=5 Copy code
python
"request_count_seen_by_worker" Copy code
python
from prefect import task Copy code
python
from prefect.task_worker import serve
serve(my_task) Copy code
python
future = my_task.delay(...)PrefectDistributedFuture Copy code
python
future.wait(timeout=300) Copy code
python
result = future.result(timeout=300)o
oleander-whirlwind
08/10/2026, 9:42 AM@Marvin
Show a COMPLETE Prefect 3.x example consisting of:
1. worker.py
- background task worker startup
- exact imports
- exact serve syntax
2. tasks.py
- background task definition
- exact delay() usage
- exact result persistence
3. orchestrator.py
- flow.serve()
- submission of task
- waiting for result
Use only APIs that exist in Prefect 3.x and provide executable code.
m
Marvin
08/10/2026, 9:42 AM✅ thought for 9.0 seconds
Marvin
08/10/2026, 9:42 AMHere is a complete executable Prefect 3.x example using only current 3.x APIs.
worker.py
Copy code
python
from prefect.task_worker import serve
from tasks import heavy_task
if __name__ == "__main__":
serve(
heavy_task,
limit=2,
status_server_port=8081,
) Copy code
python
from __future__ import annotations
from prefect import task
@task(
name="heavy-task",
persist_result=True,
log_prints=True,
)
def heavy_task(x: int, y: int) -> dict:
result = x + y
print(f"Computed {x} + {y} = {result}")
return {
"x": x,
"y": y,
"sum": result,
}
def submit_heavy_task(x: int, y: int):
future = heavy_task.delay(x, y)
return future Copy code
python
from __future__ import annotations
from prefect import flow
from tasks import heavy_task
@flow(name="orchestrator", log_prints=True)
def orchestrator(x: int = 1, y: int = 2) -> dict:
future = heavy_task.delay(x, y)
result = future.result(timeout=300)
print(f"Received result: {result}")
return result
if __name__ == "__main__":
orchestrator.serve(
name="orchestrator-deployment",
) Copy code
python
from prefect.task_worker import serve Copy code
python
from prefect import task Copy code
python
future = heavy_task.delay(x, y) Copy code
python
result = future.result(timeout=300) Copy code
python
orchestrator.serve(name="orchestrator-deployment")2 Views