Short answer: yes—with a couple of caveats. In Prefect 3.x you can point a deployment’s source at a directory reachable by IP if: - it’s exposed over a supported protocol (e.g., smb://, sftp://, http://), or - it’s mounted locally on the worker (e.g., an NFS mount like

/mnt/nfs/...

) Options: - SMB/CIFS (by IP) - Use an fsspec URL:

<smb://192.168.1.10/SHARE/flows>

- Or use the SMB block for auth/connection handling - Example:

from prefect import flow
    from prefect.filesystems import SMB

    # Ideally create and save this as a block in the UI, then load it
    smb = SMB(
        smb_host="192.168.1.10",
        share_path="/SHARE/flows",
        smb_username="myuser",
        smb_password="mypassword"
    )

    flow.from_source(
        source=smb,
        entrypoint="my_flow.py:my_flow",
    ).deploy(
        name="smb-deploy",
        work_pool_name="my-pool",
    )

- SFTP/HTTP/WebDAV (by IP) - Point

source

at the URL:

flow.from_source(
        source="<sftp://user:pass@192.168.1.20/path/to/flows>",
        entrypoint="my_flow.py:my_flow",
    ).deploy(name="sftp-deploy", work_pool_name="my-pool")

- NFS - Not an fsspec URL—mount the share onto the worker, then reference the local mount:

flow.from_source(
        source="/mnt/nfs/flows",
        entrypoint="my_flow.py:my_flow",
    ).deploy(name="nfs-deploy", work_pool_name="my-pool")

Notes: - Your worker must have network access to the share at run time. - For protocol URLs (SMB/SFTP/etc.), ensure the appropriate fsspec driver is installed in the worker environment (e.g.,

smbfs

for SMB). - Prefer using blocks and Secret blocks for credentials over hardcoding. Docs: - How to store flow code for deployments: Guide: Store flow code - Filesystem blocks (LocalFileSystem, RemoteFileSystem, SMB): API: prefect.filesystems - Deployments overview: Concepts: Deployments - Flow.from_source reference: API: Flow and from_source Which protocol/IP setup are you targeting (SMB, NFS, SFTP, HTTP)? And where will your workers run (VM, Kubernetes, Docker)? I can tailor an exact example and dependency list.

Absolutely. Here are two common ways to point your deployment at a directory served over HTTP/HTTPS. 1) Quick inline URL - Works if your HTTP server allows directory listing (or behaves like a filesystem, e.g., WebDAV-like listing).

from prefect import flow

# entrypoint is relative to the HTTP base path
flow.from_source(
    source="<https://files.example.com/flows/>",   # or http://<ip>/flows/
    entrypoint="my_project/flow.py:my_flow",
).deploy(
    name="http-deploy",
    work_pool_name="my-pool",
)

2) Using RemoteFileSystem (handy for headers/auth) - Useful when you need to pass headers (e.g., bearer token) or tweak connection settings.

from prefect import flow
from prefect.filesystems import RemoteFileSystem

remote_fs = RemoteFileSystem(
    basepath="<https://10.0.0.5/flows/>",   # your HTTP base path
    settings={
        # Example: add auth header via environment variable
        "headers": {"Authorization": "Bearer ${PREFECT_HTTP_TOKEN}"},
        # Other options can be passed through to fsspec if needed
    },
)

flow.from_source(
    source=remote_fs,
    entrypoint="my_project/flow.py:my_flow",
).deploy(
    name="http-remote-fs",
    work_pool_name="my-pool",
)

Important notes: - The worker must be able to reach the HTTP server at run time. - HTTP directory sources work best when the server supports directory listing or WebDAV-style APIs. If your server only serves individual files without listing, consider: - using WebDAV explicitly, SFTP/SMB, or - mounting the directory locally on the worker, or - hosting your code in Git and using a Git URL. Docs: - Store flow code for deployments: Guide: Store flow code - Filesystem storage (LocalFileSystem, RemoteFileSystem, SMB): API: prefect.filesystems If you can share how your HTTP server is configured (plain HTTP vs WebDAV, directory listing enabled, auth), I can tailor the example and settings.

Absolutely—here are two ways to use SFTP as your deployment’s code source. 1) Simple SFTP URL (quickest) - Good for testing; credentials are in the URL (avoid in production).

from prefect import flow

flow.from_source(
    source="<sftp://myuser:mypassword@10.0.0.5/home/myuser/flows/>",
    entrypoint="my_project/flow.py:my_flow",
).deploy(
    name="sftp-deploy",
    work_pool_name="my-pool",
)

2) Using RemoteFileSystem (recommended; supports secrets and extra settings) - Pass credentials via Prefect Secret blocks or environment variables.

from prefect import flow
from prefect.filesystems import RemoteFileSystem
from prefect.blocks.system import Secret

sftp_user = Secret.load("sftp-username").get()
sftp_pass = Secret.load("sftp-password").get()

remote_fs = RemoteFileSystem(
    basepath="<sftp://10.0.0.5/home/myuser/flows/>",
    settings={
        "username": sftp_user,
        "password": sftp_pass,
        # Optional extras passed through to fsspec’s SFTPFileSystem:
        # "port": 22,
        # "key_filename": "/home/prefect/.ssh/id_ed25519",  # if using SSH key auth
        # "known_hosts": "/home/prefect/.ssh/known_hosts",
    },
)

flow.from_source(
    source=remote_fs,
    entrypoint="my_project/flow.py:my_flow",
).deploy(
    name="sftp-remote-fs",
    work_pool_name="my-pool",
)

Notes: - The worker running your flow must be able to reach the SFTP server. - SFTP support comes via fsspec; ensure the worker environment has the SFTP extra installed (paramiko). For example:

pip install paramiko

. - The

entrypoint

path is relative to the base directory you point

source

at. Docs: - Store flow code for deployments: Guide: Store flow code - Filesystem storage (RemoteFileSystem, SMB, Local): API: prefect.filesystems If you share whether you’re using password or SSH key auth (and your directory layout), I can tailor the exact settings.

Yes—here are two working patterns for WebDAV in Prefect 3.x using RemoteFileSystem. You’ll need the webdav4 package installed in the worker environment. Install requirement:

pip install webdav4

1) Basic auth (username/password)

from prefect import flow
from prefect.filesystems import RemoteFileSystem

# WebDAV over HTTPS; note the scheme 'webdav' (not 'webdavs')
webdav_storage = RemoteFileSystem(
    basepath="<webdav://dav.example.com/flows/>",   # base directory for your code
    settings={
        "base_url": "<https://dav.example.com>",    # server base URL
        "auth": ("myuser", "mypassword"),         # basic auth
        # Optional:
        # "verify": True,                         # or path to CA bundle
        # "timeout": 30.0,
        # "headers": {"X-Custom": "value"},
    },
)

@flow
def my_flow():
    return "Hello from WebDAV!"

my_flow.from_source(
    source=webdav_storage,
    entrypoint="my_project/flow.py:my_flow",      # relative to basepath
).deploy(
    name="webdav-deploy",
    work_pool_name="default",
)

2) Token-based auth via headers

from prefect import flow
from prefect.filesystems import RemoteFileSystem

webdav_storage = RemoteFileSystem(
    basepath="<webdav://api.example.org/dav/flows/>",
    settings={
        "base_url": "<https://api.example.org/dav>",
        "auth": None,  # skip basic auth
        "headers": {
            "Authorization": "Bearer ${WEBDAV_TOKEN}"  # resolve from env at runtime
        },
        # "verify": "/etc/ssl/certs/company-ca.crt",
    },
)

@flow
def my_flow():
    return "Token-protected WebDAV!"

my_flow.from_source(
    source=webdav_storage,
    entrypoint="my_project/flow.py:my_flow",
).deploy(
    name="webdav-token-deploy",
    work_pool_name="default",
)

Notes: - Use the scheme

webdav://

and set

base_url

to the HTTPS origin. Prefect passes

settings

straight to fsspec’s WebDAV filesystem (webdav4). - Workers must have network access to the server and webdav4 installed. - For self-signed certs in non-prod, you can set

"verify": False

(not recommended for production). - The

entrypoint

is relative to the directory you pointed

basepath

at. Docs: - Guide: Store flow code for deployments - API: prefect.filesystems If you share your server’s auth method (basic vs. token) and directory layout, I can tailor the exact settings.

A few common gotchas can cause “Authentication failed” for SFTP even when plain SSH works: Most frequent causes - SFTP subsystem not enabled on the server - SSH auth can succeed but the SFTP channel fails to open. - Check on the server:

Subsystem sftp /usr/lib/openssh/sftp-server

(or internal-sftp) is present in

/etc/ssh/sshd_config

, and that any Match blocks don’t disable SFTP for your user. - Test explicitly:

sftp -v user@host

(you’ll see SFTP-specific errors if it’s disabled). - You’re authenticating differently than your terminal - Paramiko (what fsspec uses) tries auth in a specific order and may not be using the same key/agent as your shell. - If your terminal uses an SSH agent or a specific key, pass the same details to Prefect/fsspec: -

username="..."

,

key_filename="/path/to/key"

, optional

passphrase="..."

- Or

password="..."

if that’s what you use. - You can also control agent/key search with

allow_agent=True/False

,

look_for_keys=True/False

. - Wrong username or unexpected default user - SSH may default to your local username while your code uses a different one. Explicitly set

username

. - Jump host/proxy required - If you normally reach the target through a jump host (

ssh -J jump ...

), you’ll need to replicate it with Paramiko’s ProxyCommand:

from paramiko.proxy import ProxyCommand
    fs = fsspec.filesystem(
        "sftp",
        host="final.target",
        username="user",
        sock=ProxyCommand("ssh jump.host nc final.target 22"),
    )

- Key passphrase or permissions - Encrypted keys need

passphrase=...

. - Ensure the key has correct perms (chmod 600) and is a supported type (rsa/ecdsa/ed25519). - Server allows SSH but denies SFTP for your account - Check sshd Match blocks (AllowUsers, ChrootDirectory, ForceCommand internal-sftp, etc.) that can restrict SFTP. How to wire this into Prefect 3.x - With flow.from_source() + RemoteFileSystem:

from prefect import flow
from prefect.filesystems import RemoteFileSystem

remote_fs = RemoteFileSystem(
    basepath="<sftp://host.example.com/home/myuser/flows/>",
    settings={
        "username": "myuser",
        # One of the following:
        # "password": "mypassword",
        # "key_filename": "/home/prefect/.ssh/id_ed25519",
        # "passphrase": "my-key-passphrase",
        # Optional knobs:
        # "port": 22,
        # "allow_agent": True,
        # "look_for_keys": True,
        # "timeout": 30,
        # "auth_timeout": 15,
        # "banner_timeout": 10,
    },
)

@flow
def my_flow():
    ...

my_flow.from_source(
    source=remote_fs,
    entrypoint="my_project/flow.py:my_flow",
).deploy(name="sftp-deploy", work_pool_name="default")

Debug tips - Verify SFTP directly: -

sftp -v user@host

- If using a key:

sftp -v -i /path/to/key -o IdentitiesOnly=yes user@host

- Increase Paramiko logging to see auth order and failures:

import logging
  logging.getLogger("paramiko").setLevel(logging.DEBUG)

- If you need a jump host:

from paramiko.proxy import ProxyCommand
  RemoteFileSystem(
      basepath="<sftp://final.target/home/user/flows/>",
      settings={
          "username": "user",
          "sock": ProxyCommand("ssh jump.host nc final.target 22"),
      },
  )

If you can share: - how you log in from the terminal (password vs key, any -J jump host, which key), - and the error snippet you see in Prefect, I can suggest the exact

settings

to mirror your working SSH setup.

That error means you’re passing an unsaved storage block to

from_source

. For deployments, Prefect needs a saved block (so the worker can look it up by ID). Save the block first, or load a previously saved one. Two easy fixes: Option A – save then use the same object

from prefect import flow
from prefect.filesystems import RemoteFileSystem

sftp_block = RemoteFileSystem(
    basepath="<sftp://host.example.com/home/myuser/flows/>",
    settings={
        "username": "myuser",
        "password": "mypassword",
        # or key_filename/passphrase, etc.
    },
)

# Save the block to your Prefect workspace
sftp_block.save("sftp-code-storage", overwrite=True)

@flow
def my_flow():
    ...

my_flow.from_source(
    source=sftp_block,  # now a saved block
    entrypoint="my_project/flow.py:my_flow",
).deploy(
    name="sftp-deploy",
    work_pool_name="default",
)

Option B – load a saved block by name

from prefect import flow
from prefect.filesystems import RemoteFileSystem

# Load the block you saved earlier
sftp_block = RemoteFileSystem.load("sftp-code-storage")

@flow
def my_flow():
    ...

my_flow.from_source(
    source=sftp_block,
    entrypoint="my_project/flow.py:my_flow",
).deploy(
    name="sftp-deploy",
    work_pool_name="default",
)

Alternative (no blocks): pass a URL string instead of a block

my_flow.from_source(
    source="<sftp://myuser:mypassword@host.example.com/home/myuser/flows/>",
    entrypoint="my_project/flow.py:my_flow",
).deploy(name="sftp-deploy", work_pool_name="default")

Notes: - Make sure you’re logged into the same Prefect workspace when saving the block and when your worker runs, so it can resolve the block. - If you change block config later, call `.save(..., overwrite=True)`; your deployment will use the updated block on the next run. Docs: - Guide: Store flow code for deployments - Filesystem blocks API: prefect.filesystems