< Marvin> how to properly initialize worker using Prefect s Prefect Community #ask-marvin

Join Slack

<@ULVA73B9P> how to properly initialize worker usi...

# ask-marvin

Muhammad Ridwan

05/22/2026, 6:12 AM

@Marvin how to properly initialize worker using Prefect's ProcessWorker, and subsequently flow.deploy

Marvin

05/22/2026, 6:12 AM

✅ thought for 126.1 seconds

Marvin

05/22/2026, 6:15 AM

In Prefect 3.x, the usual pattern is: 1. Create a process work pool 2. Start a worker that polls that pool 3. Deploy your flow to that pool with
flow.deploy(...)
4. Trigger runs Important distinction:

ProcessWorker

is the Python worker implementation, but in normal usage you usually start it via the CLI with

prefect worker start

. You generally do not need to instantiate

ProcessWorker

directly unless you're embedding worker behavior in your own Python process. Recommended setup 1. Create a process work pool

Copy code

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

2. Start a worker for that pool

Copy code

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

You can also add a name and concurrency limit:

Copy code

bash
prefect worker start --pool my-process-pool --type process --name local-worker --limit 4

3. Deploy your flow to that pool

Copy code

python
from prefect import flow

@flow
def my_flow(name: str = "world"):
    print(f"hello {name}")

if __name__ == "__main__":
    my_flow.deploy(
        name="my-process-deployment",
        work_pool_name="my-process-pool",
    )

Then trigger it from the UI or CLI. Programmatic worker initialization If you specifically want to initialize

ProcessWorker

in Python, use the async context manager pattern:

Copy code

python
import asyncio
from prefect.workers.process import ProcessWorker

async def main():
    worker = ProcessWorker(
        work_pool_name="my-process-pool",
        name="embedded-process-worker",
        limit=4,
        create_pool_if_not_found=True,
    )

    async with worker:
        await worker.start()

if __name__ == "__main__":
    asyncio.run(main())

That works, but for most users the CLI worker is simpler and more reliable operationally. How
flow.deploy()
fits in

flow.deploy(...)

creates a deployment associated with a work pool. For a process pool, the worker will execute the flow run as a local subprocess on the machine where the worker is running. Minimal example:

Copy code

python
from prefect import flow

@flow
def my_flow(x: int = 1, y: int = 2):
    print(x + y)

if __name__ == "__main__":
    my_flow.deploy(
        name="addition",
        work_pool_name="my-process-pool",
        parameters={"x": 10, "y": 20},
    )

Common gotcha with
flow.deploy()
on process pools If your flow code is only on your laptop and your worker runs somewhere else, the worker needs access to that code. A good Prefect 3.x pattern is to deploy from source:

Copy code

python
from prefect import flow

@flow
def my_flow():
    print("running")

if __name__ == "__main__":
    flow.from_source(
        source="<https://github.com/your-org/your-repo.git>",
        entrypoint="flows.py:my_flow",
    ).deploy(
        name="my-process-deployment",
        work_pool_name="my-process-pool",
    )

This is the preferred 3.x approach instead of older removed methods like

Deployment.build_from_flow()

. When to use image/build/push For a process work pool, you often don't need Docker image settings. If you're just running on the same machine/environment as the worker, this is enough:

Copy code

python
my_flow.deploy(
    name="my-process-deployment",
    work_pool_name="my-process-pool",
)

If your deployment setup requires packaging code differently, you can still use

.from_source(...)

. Useful job variables for process workers You can override process worker job settings per deployment:

Copy code

python
my_flow.deploy(
    name="my-process-deployment",
    work_pool_name="my-process-pool",
    job_variables={
        "stream_output": True,
        "working_dir": "/tmp/prefect-work",
    },
)

Best practice summary - Use

prefect work-pool create ... --type process

- Use

prefect worker start --pool ... --type process

- Use

flow.deploy(..., work_pool_name=...)

- Use

flow.from_source(...).deploy(...)

if workers need to fetch code

Marvin

05/22/2026, 6:15 AM

- Only instantiate

ProcessWorker

directly in Python if you have a special embedded use case Docs / references - Prefect docs - Deployments docs - Work pools docs - Workers docs If you want, I can also give you a complete end-to-end example for either: - local-only process worker, or - Git-based deployment with
from_source(...).deploy(...)

Muhammad Ridwan

05/22/2026, 6:20 AM

@Marvin I already use create_pool_if_not_found=True for the worker and started it, but still returns ValueError: Could not find work pool 'work-pool'. Please create it before deploying this flow. when deploying flow

Marvin

05/22/2026, 6:20 AM

✅ thought for 159.8 seconds

Marvin

05/22/2026, 6:23 AM

That error makes sense even if the worker was started with

create_pool_if_not_found=True

. Why this happens

create_pool_if_not_found=True

only affects the worker startup path. It means: - when the worker connects to the API, - it may create the pool at that time if it doesn't exist. But

flow.deploy(...)

does its own immediate validation and checks whether the work pool already exists. If the pool is not yet present in the API when deploy runs, deploy fails with:

Copy code

text
ValueError: Could not find work pool 'work-pool'. Please create it before deploying this flow.

So the key point is: worker auto-creation does not guarantee the pool exists before
flow.deploy()
checks for it. This is usually a timing/order issue. Recommended fix Create the work pool explicitly before deploying.

Copy code

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

Then:

Copy code

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

Then run your deploy script. Best-practice order

Copy code

bash
prefect work-pool create work-pool --type process
prefect worker start --pool work-pool --type process
python deploy.py

If you want to rely on worker auto-creation You need to wait until the worker has fully connected and the pool is actually visible to the API before calling

flow.deploy(...)

. For example: 1. start worker 2. wait a few seconds 3. verify the pool exists 4. deploy You can verify with:

Copy code

bash
prefect work-pool ls

work-pool

is not listed yet,

flow.deploy(...)

will still fail. What I would recommend instead For deployments, treat work pools as infrastructure that should already exist. Use worker auto-create only as a convenience for development, not as something deploy depends on. Minimal working example

Copy code

python
from prefect import flow

@flow
def my_flow():
    print("hello")

if __name__ == "__main__":
    my_flow.deploy(
        name="test-deployment",
        work_pool_name="work-pool",
    )

And before running that:

Copy code

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

If you want to create the pool in Python first You can create it before deploying, instead of depending on the worker:

Copy code

python
import asyncio
from prefect import flow
from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import WorkPoolCreate
from prefect.workers.process import ProcessWorker

@flow
def my_flow():
    print("hello")

async def ensure_pool():
    async with get_client() as client:
        try:
            await client.read_work_pool("work-pool")
        except Exception:
            await client.create_work_pool(
                work_pool=WorkPoolCreate(
                    name="work-pool",
                    type="process",
                    base_job_template=ProcessWorker.get_default_base_job_template(),
                )
            )

async def main():
    await ensure_pool()
    await my_flow.adeploy(
        name="test-deployment",
        work_pool_name="work-pool",
    )

if __name__ == "__main__":
    asyncio.run(main())

Docs - Workers - Deployments - Work pools If you want, I can give you a single complete script that: - ensures the process work pool exists, - starts a

ProcessWorker

, - and deploys the flow correctly.

Muhammad Ridwan

05/22/2026, 9:01 AM

@Marvin would you recommend to not use additional FastAPI API layer? I see prefect client also use FastAPI under the hood

Marvin

05/22/2026, 9:01 AM

✅ thought for 23.7 seconds

Marvin

05/22/2026, 9:02 AM

Not necessarily. I wouldn't recommend adding a FastAPI layer just because Prefect uses FastAPI internally. That's an implementation detail of Prefect, not a reason for you to mirror it. The better question is: What job would your FastAPI layer do? *When I would not add a FastAPI layer* If your app just needs to: - define flows - deploy flows - trigger runs - inspect state/results - use Prefect Cloud / server as the orchestration API then an extra FastAPI service is usually unnecessary complexity. You can often do everything with: - Prefect flows/tasks - the Prefect Python client - deployments / schedules / automations - workers In that case, another API layer gives you: - more code to maintain - another auth boundary - more deployment overhead - duplicated orchestration logic *When I would add a FastAPI layer* A FastAPI service can make sense if you need a product-facing API that does things like: - authenticate your own users - expose a simplified endpoint like

POST /run-report

- validate business-specific request payloads - apply rate limits / quotas / tenancy - orchestrate Prefect runs on behalf of external systems - hide Prefect details from frontend or customer integrations - combine Prefect with other internal services/databases In that architecture, FastAPI is not replacing Prefect — it's a control plane / facade in front of Prefect. Typical pattern:

Copy code

text
Client/UI -> FastAPI -> Prefect client -> create flow run / read status
                          |
                          -> Prefect API / Cloud
Worker -> picks up run from work pool

Rule of thumb - If only internal engineers use it: probably skip FastAPI - If other apps/users/customers need a stable business API: FastAPI can be a good idea Important distinction Prefect's API is an orchestration API. Your FastAPI app, if you build one, should be a business API. Those are different responsibilities. Good lightweight alternative If all you want is to trigger flows from Python code, scripts, or another backend, use the Prefect client directly instead of wrapping it immediately in FastAPI. Example: no extra FastAPI layer

Copy code

python
import asyncio
from prefect import flow
from prefect.client.orchestration import get_client

@flow
def my_flow(x: int):
    return x * 2

async def trigger():
    async with get_client() as client:
        flow_run = await client.create_flow_run_from_deployment(
            deployment_id="YOUR_DEPLOYMENT_ID",
            parameters={"x": 21},
        )
        print(flow_run.id)

asyncio.run(trigger())

Example: with FastAPI layer

Copy code

python
from fastapi import FastAPI
from pydantic import BaseModel
from prefect.client.orchestration import get_client

app = FastAPI()

class RunRequest(BaseModel):
    x: int

@app.post("/run")
async def run_flow(req: RunRequest):
    async with get_client() as client:
        flow_run = await client.create_flow_run_from_deployment(
            deployment_id="YOUR_DEPLOYMENT_ID",
            parameters={"x": req.x},
        )
        return {"flow_run_id": str(flow_run.id)}

That can be useful if you want a clean application API, but it's optional. My recommendation For most cases: start without the extra FastAPI layer. Add it only if you clearly need: - external triggering via HTTP - custom auth/business validation - multi-tenant/API product behavior - an abstraction over Prefect for other teams or clients If you describe your intended architecture a bit more — for example: - who will call this API, - whether it's internal or customer-facing, - and whether you just need to trigger deployments or also manage metadata/results — I can tell you whether FastAPI is worth it in your specific case and sketch the cleanest design.

2 Views

Open in Slack

Previous Next