Short answer: call your system’s SSH client from prefect-shell. prefect-shell runs commands locally on the worker; to run something on a remote host without prefect-ssh, you execute

ssh user@host "<remote command>"

. Key points for Windows (Win32-OpenSSH): - Use the Windows ssh.exe (installed with “OpenSSH Client”). - PowerShell can mangle quotes; either: - Use

shell="cmd"

to avoid PowerShell quoting, or - Use PowerShell’s stop-parsing operator

--%

so everything after it is passed verbatim to ssh. Examples 1) Windows worker -> Linux remote, using PowerShell with stop-parsing:

from prefect import flow
from prefect_shell.commands import shell_run_command

@flow
def run_remote():
    # Add -i <key> if needed, and ensure the host is in known_hosts to avoid prompts
    result = shell_run_command(
        command='ssh --% ubuntu@1.2.3.4 "bash -lc \'hostname && whoami && ls -la /tmp\'"',
        shell="powershell",
        return_all=True,
    )
    return result

2) Windows worker -> Linux remote, using cmd.exe to keep quoting simple:

from prefect import flow
from prefect_shell.commands import shell_run_command

@flow
def run_remote():
    result = shell_run_command(
        command='ssh ubuntu@1.2.3.4 "bash -lc \\"hostname && whoami && ls -la /tmp\\""',
        shell="cmd",
        return_all=True,
    )
    return result

3) Windows worker -> Windows remote (Win32-OpenSSH server defaults to PowerShell on many systems):

from prefect import flow
from prefect_shell.commands import shell_run_command

@flow
def run_remote_windows():
    result = shell_run_command(
        command='ssh --% Administrator@win-host "powershell -NoProfile -Command \\"Get-ComputerInfo | Select-Object CsName,OsName,OsVersion -First 1\\""',
        shell="powershell",
        return_all=True,
    )
    return result

Tips - Authentication: include

-i C:\Path\to\id_ed25519

if you’re not using ssh-agent. Avoid passwords in commands. - Host key prompts: in non-interactive runs, pre-populate

known_hosts

or (only if acceptable) add

-o StrictHostKeyChecking=no

and a throwaway

UserKnownHostsFile

to avoid prompts. - Multiple commands: for Linux remotes,

bash -lc "<commands>"

is a reliable way to run a compound command with a login-like environment. - Observability:

return_all=True

returns all lines of output. You can also stream output to logs via

stream_level

. Docs - Prefect 3.x prefect-shell integration: prefect-shell docs If you share: - the OS/shell of your worker (PowerShell vs cmd) - the remote OS (Windows vs Linux) - your auth method (key vs password) …I can tailor the exact command and quoting for your setup.

Thanks for the details. Today there isn’t an official, maintained

prefect-ssh

integration for Prefect 3.x. The common pattern is to use

prefect-shell

and call your system’s

ssh

client directly from the worker. Since your worker runs on macOS and your target is a Windows host (Win32-OpenSSH), you can run PowerShell remotely like this: - Prefer key-based auth (recommended). Password prompts won’t work in non-interactive runs. - If you must use username/password, see options below. Example: macOS worker -> Windows host over SSH (key-based)

from prefect import flow
from prefect_shell.commands import shell_run_command

@flow
def run_remote_windows():
    # Replace host, user, and key path
    cmd = (
        'ssh -o BatchMode=yes '
        '-o StrictHostKeyChecking=accept-new '
        '-i ~/.ssh/id_ed25519 '
        'Administrator@win-host '
        'powershell -NoProfile -NonInteractive -ExecutionPolicy Bypass '
        '-Command "Get-ComputerInfo | Select-Object CsName,OsName,OsVersion -First 1 '
        '| ConvertTo-Json -Compress"'
    )

    result = shell_run_command(
        command=cmd,
        shell="bash",         # Explicitly use bash/zsh quoting on macOS
        return_all=True,      # Capture all lines
    )
    return result

Notes and tips - Auth: - Key-based is best for unattended workers: add your macOS worker’s public key to the Windows user’s

~\.ssh\authorized_keys

on the server and use

-i

or your ssh-agent. - Use

-o BatchMode=yes

to fail immediately if the key won’t work (prevents hanging on a password prompt). - Host keys: - First connection prompts can block execution. Use

-o StrictHostKeyChecking=accept-new

(OpenSSH 8.0+) or pre-populate

known_hosts

. - PowerShell command packaging: - Use

-NoProfile -NonInteractive -ExecutionPolicy Bypass -Command "<ps command>"

. - If your command contains double quotes, escape them or wrap carefully. Converting to JSON (

ConvertTo-Json

) is handy for parsing results. If you must use username/password - OpenSSH will prompt for a password interactively;

prefect-shell

won’t handle that prompt. You have two main options: 1) Use Paramiko in a Prefect task (no external prompt, supports password):

from prefect import flow, task
  import paramiko

  @task
  def run_ps_over_paramiko(host, username, password, ps_command):
      client = paramiko.SSHClient()
      client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
      client.connect(hostname=host, username=username, password=password)
      full = f'powershell -NoProfile -NonInteractive -ExecutionPolicy Bypass -Command "{ps_command}"'
      _, stdout, stderr = client.exec_command(full)
      out, err = stdout.read().decode(), stderr.read().decode()
      client.close()
      return out, err

  @flow
  def flow():
      return run_ps_over_paramiko(
          host="win-host",
          username="Administrator",
          password="***",
          ps_command='Get-ComputerInfo | Select-Object CsName,OsName,OsVersion -First 1 | ConvertTo-Json -Compress'
      )

2) Use

sshpass

(not recommended for security; may require a Homebrew tap on macOS). If you go this route, pass the password via environment variables or a secret store and be mindful of logs. Docs - Prefect 3.x shell integration: prefect-shell docs If you can switch to keys, I can help you craft the exact

ssh

command for your environment. If you must stay on password auth, I’d recommend the Paramiko task approach above for reliability.

Short answer: there isn’t a built‑in “Shell” block to configure in the UI. In Prefect 3.x you typically just call `ShellOperation`/`shell_run_command` directly in your flow. If you want UI‑managed config, use Blocks (e.g., Secret, String, JSON) to store host/user/key/options and load them in your flow to build the SSH command string. What you can do today - Use

prefect-shell

directly in code (no block needed):

from prefect import flow
from prefect_shell.commands import shell_run_command

@flow
def run_remote_windows():
    cmd = (
        'ssh -o BatchMode=yes '
        '-o StrictHostKeyChecking=accept-new '
        '-i ~/.ssh/id_ed25519 '
        'Administrator@win-host '
        'powershell -NoProfile -NonInteractive -ExecutionPolicy Bypass '
        '-Command "Get-ComputerInfo | Select-Object CsName,OsName,OsVersion -First 1 '
        '| ConvertTo-Json -Compress"'
    )
    return shell_run_command(command=cmd, shell="bash", return_all=True)

- Store configuration/secrets in Blocks and load them: - Create a Secret block in the UI named

winssh-password

(or store a key path, host, user in String/JSON blocks).

from prefect import flow
from prefect.blocks.system import Secret
from prefect_shell.commands import shell_run_command

@flow
def run_remote_with_blocks():
    # Example with key path from a Secret block
    key_path = Secret.load("winssh-key-path").get()
    host = "win-host"  # or load from a String/JSON block
    user = "Administrator"

    cmd = (
        f'ssh -o BatchMode=yes -o StrictHostKeyChecking=accept-new -i {key_path} '
        f'{user}@{host} '
        'powershell -NoProfile -NonInteractive -ExecutionPolicy Bypass '
        '-Command "Get-ComputerInfo | Select-Object CsName,OsName,OsVersion -First 1 | ConvertTo-Json -Compress"'
    )
    return shell_run_command(command=cmd, shell="bash", return_all=True)

- Optional: define your own UI‑configurable “SSH Target” block (host/user/key/options) and register it so your team can fill it in once in the UI:

# ssh_block.py
from prefect.blocks.core import Block

class SshTarget(Block):
    _block_type_name = "SSH Target"
    host: str
    user: str
    key_path: str | None = None
    extra_args: str = ""

# Register once:
# prefect block register -f ssh_block.py
# Then create an instance in the UI. In your flow:
# target = SshTarget.load("prod-windows")
# cmd = f'ssh {target.extra_args} -i {target.key_path} {target.user}@{target.host} ...'

CLI references: - List block types:

prefect block type ls

- Register custom blocks:

prefect block register -f ssh_block.py

- Generate UI creation link:

prefect block create <block_type_slug>

Docs: - Shell integration: prefect-shell docs - Blocks concepts: Blocks overview - CLI for blocks: prefect block CLI - Storing secrets: Store secrets Important note about password auth - System

ssh

will prompt interactively for a password; that won’t work in unattended flows. There’s no “shell block” that can inject a password into that prompt. - Recommended: switch to key‑based auth and keep using

prefect-shell

. - If you must use username/password, run SSH via a library that supports programmatic passwords (e.g., Paramiko in a Prefect task) instead of the system

ssh

client. If you want, I can help you: - set up a small custom “SSH Target” block, - or refactor your flow to load host/user/key from Blocks, - or provide a Paramiko-based task for password auth.