Absolutely — here’s a small, reusable Python helper that creates a Prefect 3.x deployment against a local Prefect server on Ubuntu. It will: - Point Prefect at your local server - Ensure a process work pool exists (creates it if missing) - Build a schedule (optional) - Create or update the deployment via Flow.from_source(...).deploy(...) Code:

import os
from datetime import timedelta
from typing import Optional, Dict, List

from prefect import get_client
from prefect.exceptions import ObjectNotFound
from prefect.flows import Flow
from prefect.schedules import Cron, Interval


def create_local_deployment(
    *,
    source: str,
    entrypoint: str,
    deployment_name: str,
    work_pool_name: str = "local-process",
    work_queue_name: Optional[str] = None,
    parameters: Optional[Dict] = None,
    tags: Optional[List[str]] = None,
    description: Optional[str] = None,
    cron: Optional[str] = None,
    interval_seconds: Optional[int] = None,
    timezone: Optional[str] = "UTC",
    api_url: str = "<http://127.0.0.1:4200/api>",
) -> str:
    """
    Create or update a Prefect 3.x deployment on a local Prefect server.

    Args:
        source: Local path or VCS URL to your project root (e.g., ".")
        entrypoint: "<path/to/file.py>:<flow_name>" to your flow
        deployment_name: Name for the deployment
        work_pool_name: Name of the work pool (process pool is best for local)
        work_queue_name: Optional work queue name within the pool
        parameters: Default parameters for the deployment’s flow runs
        tags: Optional list of tags
        description: Optional deployment description
        cron: Optional cron string for scheduling (e.g., "0 * * * *")
        interval_seconds: Optional interval in seconds for scheduling
        timezone: Timezone for the schedule (default "UTC")
        api_url: URL to the local Prefect server API (default "<http://127.0.0.1:4200/api>")

    Returns:
        The deployment ID (UUID as a string).
    """
    # Point Prefect at your local server
    os.environ["PREFECT_API_URL"] = api_url

    # Ensure the work pool exists (create if it doesn't)
    with get_client(sync_client=True) as client:
        try:
            client.read_work_pool_by_name(work_pool_name)
        except ObjectNotFound:
            client.create_work_pool(name=work_pool_name, type="process")

    # Build an optional schedule
    schedule = None
    if cron:
        schedule = Cron(cron, timezone=timezone)
    elif interval_seconds:
        schedule = Interval(timedelta(seconds=interval_seconds))

    # Load the flow from source and create/update the deployment
    flow = Flow.from_source(source=source, entrypoint=entrypoint)

    deployment_id = flow.deploy(
        name=deployment_name,
        work_pool_name=work_pool_name,
        work_queue_name=work_queue_name,
        parameters=parameters or {},
        tags=tags or [],
        description=description,
        schedule=schedule,
        # No image build/push for local process work; keep defaults
        build=False,
        push=False,
        print_next_steps=False,
    )

    return str(deployment_id)

Example usage:

if __name__ == "__main__":
    dep_id = create_local_deployment(
        source=".",  # your repo/project root
        entrypoint="flows/my_flow.py:my_flow",  # file:flow_name
        deployment_name="my-local-deployment",
        work_pool_name="local-process",
        work_queue_name="default",
        parameters={"name": "world"},
        tags=["local", "dev"],
        description="Local deployment for testing",
        cron="0 * * * *",  # every hour (optional)
        # or interval_seconds=300,
        api_url="<http://127.0.0.1:4200/api>",
    )
    print("Created/updated deployment:", dep_id)

How to run locally: - Start the local Prefect server:

prefect server start

- Start a worker for your work pool (in another terminal):

prefect worker start -p local-process

Notes:

- This assumes Prefect 3.x. In 3.x, programmatic deployments use

Flow.from_source(...).deploy(...)

— not the old

Deployment.build_from_flow()

. - If you pass a

work_queue_name

, it will be created on first use under that work pool if it doesn’t exist. If you share your flow path and desired schedule/parameters, I can tailor the function call for your exact setup.

Thanks for the error detail — that usually happens when either: - You’re importing/using the decorator

from prefect import flow

instead of the class

from prefect.flows import Flow

, or - Your environment returns an awaitable from

Flow.from_source(...)

and you didn’t await it Here’s a drop-in fix. It uses the Flow class, and safely handles cases where

from_source

returns a coroutine by awaiting it before calling

deploy

. I’ve also renamed the local variable to

flow_obj

to avoid confusion with the

flow

decorator.

import os
import asyncio
import inspect
from datetime import timedelta
from typing import Optional, Dict, List

from prefect import get_client
from prefect.exceptions import ObjectNotFound
from prefect.flows import Flow
from prefect.schedules import Cron, Interval


def create_local_deployment(
    *,
    source: str,
    entrypoint: str,
    deployment_name: str,
    work_pool_name: str = "local-process",
    work_queue_name: Optional[str] = None,
    parameters: Optional[Dict] = None,
    tags: Optional[List[str]] = None,
    description: Optional[str] = None,
    cron: Optional[str] = None,
    interval_seconds: Optional[int] = None,
    timezone: Optional[str] = "UTC",
    api_url: str = "<http://127.0.0.1:4200/api>",
) -> str:
    """
    Create or update a Prefect 3.x deployment on a local Prefect server.
    """

    # Point Prefect at your local server
    os.environ["PREFECT_API_URL"] = api_url

    # Ensure the work pool exists (create if it doesn't)
    with get_client(sync_client=True) as client:
        try:
            client.read_work_pool_by_name(work_pool_name)
        except ObjectNotFound:
            client.create_work_pool(name=work_pool_name, type="process")

    # Build an optional schedule
    schedule = None
    if cron:
        schedule = Cron(cron, timezone=timezone)
    elif interval_seconds:
        schedule = Interval(timedelta(seconds=interval_seconds))

    # Load the flow from source
    flow_obj = Flow.from_source(source=source, entrypoint=entrypoint)
    if inspect.iscoroutine(flow_obj):
        # Some environments may return a coroutine from from_source; await it
        flow_obj = asyncio.run(flow_obj)

    # Create/update the deployment
    deployment_id = flow_obj.deploy(
        name=deployment_name,
        work_pool_name=work_pool_name,
        work_queue_name=work_queue_name,
        parameters=parameters or {},
        tags=tags or [],
        description=description,
        schedule=schedule,
        build=False,
        push=False,
        print_next_steps=False,
    )

    return str(deployment_id)

Quick checklist: - Ensure you have

from prefect.flows import Flow

(not

from prefect import flow

) - Don’t call your flow function when deploying (e.g., don’t do

my_flow().deploy(...)

) If you can share the few lines you’re using to call

from_source

and

deploy

, I can pinpoint the exact cause in your script.

You’re hitting this because

asyncio.run()

can’t be called when you’re already inside an event loop (e.g., Jupyter/IPython/async app). Here’s a version that supports both sync and async contexts: - Use create_local_deployment(...) in a normal Python script - Use await create_local_deployment_async(...) inside notebooks or other async code

import os
import asyncio
import inspect
from datetime import timedelta
from typing import Optional, Dict, List

from prefect import get_client
from prefect.exceptions import ObjectNotFound
from prefect.flows import Flow
from prefect.schedules import Cron, Interval


async def create_local_deployment_async(
    *,
    source: str,
    entrypoint: str,
    deployment_name: str,
    work_pool_name: str = "local-process",
    work_queue_name: Optional[str] = None,
    parameters: Optional[Dict] = None,
    tags: Optional[List[str]] = None,
    description: Optional[str] = None,
    cron: Optional[str] = None,
    interval_seconds: Optional[int] = None,
    timezone: Optional[str] = "UTC",
    api_url: str = "<http://127.0.0.1:4200/api>",
) -> str:
    """
    Async: Create or update a Prefect 3.x deployment on a local Prefect server.
    Call this with `await ...` if you are in Jupyter/IPython or any async context.
    """
    os.environ["PREFECT_API_URL"] = api_url

    # Ensure the work pool exists (create if it doesn't)
    async with get_client() as client:
        try:
            await client.read_work_pool_by_name(work_pool_name)
        except ObjectNotFound:
            await client.create_work_pool(name=work_pool_name, type="process")

    # Optional schedule
    schedule = None
    if cron:
        schedule = Cron(cron, timezone=timezone)
    elif interval_seconds:
        schedule = Interval(timedelta(seconds=interval_seconds))

    # Load the flow from source; handle cases where it returns an awaitable
    flow_obj = Flow.from_source(source=source, entrypoint=entrypoint)
    if inspect.isawaitable(flow_obj):
        flow_obj = await flow_obj

    # Deploy (this call is synchronous internally)
    deployment_id = flow_obj.deploy(
        name=deployment_name,
        work_pool_name=work_pool_name,
        work_queue_name=work_queue_name,
        parameters=parameters or {},
        tags=tags or [],
        description=description,
        schedule=schedule,
        build=False,
        push=False,
        print_next_steps=False,
    )
    return str(deployment_id)


def create_local_deployment(
    **kwargs
) -> str:
    """
    Sync wrapper: Use in normal Python scripts.
    If already in an event loop, instructs the caller to use the async version.
    """
    try:
        asyncio.get_running_loop()
        # If we get here, we're in an event loop (e.g., Jupyter)
        raise RuntimeError(
            "create_local_deployment() was called from a running event loop. "
            "Use: await create_local_deployment_async(...) instead."
        )
    except RuntimeError:
        # No running loop — safe to use asyncio.run
        return asyncio.run(create_local_deployment_async(**kwargs))

Examples: - In a normal Python script:

if __name__ == "__main__":
    dep_id = create_local_deployment(
        source=".",
        entrypoint="flows/my_flow.py:my_flow",
        deployment_name="my-local-deployment",
        work_pool_name="local-process",
        work_queue_name="default",
        parameters={"name": "world"},
        tags=["local", "dev"],
        cron="0 * * * *",
        api_url="<http://127.0.0.1:4200/api>",
    )
    print("Deployment ID:", dep_id)

- In a notebook or async app:

dep_id = await create_local_deployment_async(
    source=".",
    entrypoint="flows/my_flow.py:my_flow",
    deployment_name="my-local-deployment",
    work_pool_name="local-process",
    work_queue_name="default",
    parameters={"name": "world"},
    tags=["local", "dev"],
    interval_seconds=300,
    api_url="<http://127.0.0.1:4200/api>",
)
print(dep_id)

Also verify:

- Import the class

Flow

via

from prefect.flows import Flow

(not the

flow

decorator) - Don’t call your flow function; deploy the flow object loaded from source Are you running this from a notebook or another async context? If so, switch your call site to the async function and it should resolve the error.

Short answer: if you want “local code only,” don’t pass an image and don’t use a Docker/Kubernetes work pool. Use a Process work pool and set build=False, push=False. If you try image=None on a Docker/K8s pool, Prefect will error because those pools require an image. Here’s a minimal “local code only” version that enforces a Process work pool and uses your local project files (source path) at run time:

import os
import inspect
from datetime import timedelta
from typing import Optional, Dict, List

from prefect import get_client
from prefect.exceptions import ObjectNotFound
from prefect.flows import Flow
from prefect.schedules import Cron, Interval


def create_local_code_deployment(
    *,
    source: str = ".",                    # your local project root
    entrypoint: str,                      # "path/to/file.py:flow_name"
    deployment_name: str,
    work_pool_name: str = "local-process",
    work_queue_name: Optional[str] = None,
    parameters: Optional[Dict] = None,
    tags: Optional[List[str]] = None,
    description: Optional[str] = None,
    cron: Optional[str] = None,
    interval_seconds: Optional[int] = None,
    timezone: Optional[str] = "UTC",
    api_url: str = "<http://127.0.0.1:4200/api>",
) -> str:
    """
    Create/update a Prefect 3.x deployment that runs local code with a Process work pool.
    """
    os.environ["PREFECT_API_URL"] = api_url

    # Ensure the work pool exists and is a 'process' type
    with get_client(sync_client=True) as client:
        try:
            wp = client.read_work_pool_by_name(work_pool_name)
        except ObjectNotFound:
            wp = client.create_work_pool(name=work_pool_name, type="process")

        # Guard against using Docker/K8s pools without an image
        if getattr(wp, "type", None) != "process":
            raise ValueError(
                f"Work pool '{work_pool_name}' is type '{getattr(wp, 'type', None)}'. "
                "Local code-only deployments require a Process work pool. "
                "Create/use a Process pool or provide an image for Docker/K8s pools."
            )

    # Optional schedule
    schedule = None
    if cron:
        schedule = Cron(cron, timezone=timezone)
    elif interval_seconds:
        schedule = Interval(timedelta(seconds=interval_seconds))

    # Load flow from local source; handle awaitable in case environment returns coroutine
    flow_obj = Flow.from_source(source=source, entrypoint=entrypoint)
    if inspect.isawaitable(flow_obj):
        import asyncio
        try:
            loop = asyncio.get_running_loop()
            # If we're in an event loop (e.g., a notebook), require user to await the async version
            raise RuntimeError(
                "Detected a running event loop. In notebooks/async code, use the async version."
            )
        except RuntimeError:
            # No running loop; safe to run
            flow_obj = asyncio.run(flow_obj)

    deployment_id = flow_obj.deploy(
        name=deployment_name,
        work_pool_name=work_pool_name,
        work_queue_name=work_queue_name,
        parameters=parameters or {},
        tags=tags or [],
        description=description,
        schedule=schedule,
        # Critical for local code-only
        build=False,
        push=False,
        # Do not pass `image` at all for process pools
        print_next_steps=False,
    )
    return str(deployment_id)

Async-friendly variant (use this in notebooks or any async context): ``` import os import inspect from datetime import timedelta from typing import Optional, Dict, List from prefect import get_client from prefect.exceptions import ObjectNotFound from prefect.flows import Flow from prefect.schedules import Cron, Interval async def create_local_code_deployment_async( *, source: str = ".", entrypoint: str, deployment_name: str, work_pool_name: str = "local-process", work_queue_name: Optional[str] = None, parameters: Optional[Dict] = None,

tags: Optional[List[str]] = None, description: Optional[str] = None, cron: Optional[str] = None, interval_seconds: Optional[int] = None, timezone: Optional[str] = "UTC", api_url: str = "http://127.0.0.1:4200/api", ) -> str: os.environ["PREFECT_API_URL"] = api_url async with get_client() as client: try: wp = await client.read_work_pool_by_name(work_pool_name) except ObjectNotFound: wp = await client.create_work_pool(name=work_pool_name, type="process") if getattr(wp, "type", None) != "process": raise ValueError( f"Work pool '{work_pool_name}' is type '{getattr(wp, 'type', None)}'. " "Use a Process work pool for local code-only, or provide an image for Docker/K8s pools." ) schedule = None if cron: schedule = Cron(cron, timezone=timezone) elif interval_seconds: schedule = Interval(timedelta(seconds=interval_seconds)) flow_obj = Flow.from_source(source=source, entrypoint=entrypoint) if inspect.isawaitable(flow_obj): flow_obj = await flow_obj deployment_id = flow_obj.deploy( name=deployment_name, work_pool_name=work_pool_name, work_queue_name=work_queue_name, parameters=parameters or {}, tags=tags or [], description=description, schedule=schedule, build=False, push=False, print_next_steps=False, ) return str(deployment_id)

Example call (local code only):

dep_id = create_local_code_deployment( source=".", # your local repo/project root entrypoint="flows/my_flow.py:my_flow", deployment_name="my-local-code-deployment", work_pool_name="local-process", work_queue_name="default", parameters={"name": "world"}, tags=["local", "dev"], # cron="0 * * * *", # or interval_seconds=300 api_url="http://127.0.0.1:4200/api", ) print(dep_id)

Run-time steps:
- Start your local server:

prefect server start

- Start a worker for the Process pool in the same machine/environment that has your code:

prefect worker start -p local-process ``` Why image=None fails: - On Docker or Kubernetes pools, the job template expects an image;

image=None

leaves a required field empty and raises a validation error. - For “local code only,” switch to a Process work pool. If you must use Docker/K8s pools, you need to supply an image (and typically build/push it). Docs you referenced: Deploy via Python If you share your current work pool type and the exact error message, I can pinpoint the minimal change (likely “switch to a Process work pool” or “provide an image for Docker/K8s”).