diff --git a/README.md b/README.md
index affecb7..b7bc6c4 100644
--- a/README.md
+++ b/README.md
@@ -7,6 +7,7 @@ Automatischer Import von Belegen (Rechnungen, Gutschriften) aus verschiedenen Qu
 - **Scan-Upload**: PDF hochladen, automatische Trennung per QR-Code-Trennseiten
 - **IMAP**: Automatischer Abruf von Belegen aus Email-Postfachern
 - **SMB/Netzlaufwerk**: Automatischer Abruf von Belegen aus Netzwerkordnern
+- **FTP / SFTP**: Automatischer Abruf von Belegen via FTP (passiv, unverschluesselt) oder SFTP (SSH)
 - **Amazon Business**: Automatischer Abruf von Amazon-Rechnungen per API
 - **Eingangs-/Ausgangsbelege**: Getrennte Import-Adressen fur Einkauf und Verkauf
 - **Scheduler**: Automatischer Abruf in konfigurierbaren Intervallen
diff --git a/app/database.py b/app/database.py
index 67adfb1..358abbf 100644
--- a/app/database.py
+++ b/app/database.py
@@ -4,13 +4,13 @@ import aiosqlite
 from cryptography.fernet import Fernet
 
 DB_PATH = os.environ.get("DB_PATH", "/data/belegimport.db")
-SCHEMA_VERSION = 8
+SCHEMA_VERSION = 9
 
 logger = logging.getLogger(__name__)
 
 _fernet = None
 
-ENCRYPTED_KEYS = {"imap_password", "smtp_password", "smb_password", "amazon_password", "amazon_client_secret", "amazon_refresh_token"}
+ENCRYPTED_KEYS = {"imap_password", "smtp_password", "smb_password", "amazon_password", "amazon_client_secret", "amazon_refresh_token", "ftp_password"}
 
 DEFAULT_SETTINGS = {
     "imap_server": "",
@@ -46,6 +46,18 @@ DEFAULT_SETTINGS = {
     "smb_source_path_ausgang": "",
     "smb_processed_path_ausgang": "",
     "smb_mode": "forward",
+    # FTP / SFTP
+    "ftp_enabled": "false",
+    "ftp_protocol": "sftp",  # "sftp" or "ftp"
+    "ftp_server": "",
+    "ftp_port": "22",
+    "ftp_username": "",
+    "ftp_password": "",
+    "ftp_source_path": "",
+    "ftp_processed_path": "Verarbeitet",
+    "ftp_source_path_ausgang": "",
+    "ftp_processed_path_ausgang": "",
+    "ftp_mode": "forward",
     # Amazon
     "amazon_enabled": "false",
     "amazon_email": "",
@@ -241,6 +253,12 @@ async def _run_migrations(db: aiosqlite.Connection, current_version: int):
         await db.commit()
         await _set_schema_version(db, 8)
 
+    if current_version < 9:
+        logger.info("Migration v9: FTP/SFTP-Settings hinzugefuegt (defaults werden eingefuegt)")
+        # No table changes needed - new settings are added via DEFAULT_SETTINGS loop in init_db
+        await db.commit()
+        await _set_schema_version(db, 9)
+
     await db.commit()
 
 
diff --git a/app/ftp_processor.py b/app/ftp_processor.py
new file mode 100644
index 0000000..e4be213
--- /dev/null
+++ b/app/ftp_processor.py
@@ -0,0 +1,614 @@
+"""FTP / SFTP file import processor.
+
+Same design as smb_processor but for FTP (passive, unencrypted) and SFTP (SSH).
+Reads PDF files from a remote source folder, forwards them via SMTP, then moves
+them to a processed folder.
+"""
+
+import asyncio
+import ftplib
+import io
+import logging
+import os
+import posixpath
+import tempfile
+
+import paramiko
+
+from app.database import get_settings, add_log_entry, get_import_email
+from app.mail_processor import _connect_smtp, _build_forward_email, _send_with_log
+from app.scanner import detect_separator_pages, split_pdf
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Generic adapter interface
+# ---------------------------------------------------------------------------
+
+class _FtpAdapter:
+    """Common interface for FTP and SFTP backends."""
+
+    def list_pdfs(self, path: str) -> list[str]:
+        raise NotImplementedError
+
+    def list_dirs(self, path: str, max_depth: int = 5) -> list[str]:
+        raise NotImplementedError
+
+    def read_file(self, path: str) -> bytes:
+        raise NotImplementedError
+
+    def ensure_dir(self, path: str):
+        raise NotImplementedError
+
+    def stat_exists(self, path: str) -> bool:
+        raise NotImplementedError
+
+    def rename(self, src: str, dst: str):
+        raise NotImplementedError
+
+    def close(self):
+        pass
+
+
+# ---------------------------------------------------------------------------
+# FTP (passive, unencrypted)
+# ---------------------------------------------------------------------------
+
+class _PlainFtpAdapter(_FtpAdapter):
+    def __init__(self, server: str, port: int, username: str, password: str):
+        self.ftp = ftplib.FTP()
+        self.ftp.connect(server, port, timeout=15)
+        self.ftp.login(username or "anonymous", password or "")
+        self.ftp.set_pasv(True)
+        # Remember initial CWD - all subsequent operations should resolve relative to this
+        try:
+            self._initial_cwd = self.ftp.pwd()
+        except Exception:
+            self._initial_cwd = None
+        logger.debug(f"FTP initial CWD: {self._initial_cwd}")
+
+    def _reset_cwd(self):
+        """Reset CWD to initial directory after stateful operations."""
+        if self._initial_cwd:
+            try:
+                self.ftp.cwd(self._initial_cwd)
+            except Exception:
+                pass
+
+    def list_pdfs(self, path: str) -> list[str]:
+        """List PDF files via LIST (more reliable than NLST across FTP servers)."""
+        self._reset_cwd()
+        target = path if path else "."
+        lines = []
+        try:
+            self.ftp.retrlines(f"LIST {target}", lines.append)
+        except ftplib.error_perm:
+            return []
+        files = []
+        for line in lines:
+            if not line or line[0:1] == "d":
+                continue  # skip directories
+            parts = line.split(maxsplit=8)
+            if len(parts) < 9:
+                continue
+            name = parts[-1]
+            if name.lower().endswith(".pdf") and not name.startswith("."):
+                files.append(name)
+        return sorted(files)
+
+    def list_dirs(self, path: str, max_depth: int = 5) -> list[str]:
+        self._reset_cwd()
+        base = path or ""
+        logger.debug(f"FTP list_dirs: base={base!r}, max_depth={max_depth}")
+        return self._list_dirs_rec(base, max_depth, 0, "")
+
+    def _list_dirs_rec(self, base: str, max_depth: int, depth: int, prefix: str) -> list[str]:
+        result = []
+        try:
+            entries = []
+            target = base if base else "."
+            self.ftp.retrlines(f"LIST {target}", entries.append)
+            logger.debug(f"FTP LIST {target!r} -> {len(entries)} entries")
+        except ftplib.error_perm as e:
+            logger.warning(f"FTP LIST {base!r} failed: {e}")
+            return []
+        for line in entries:
+            # Try to detect dirs (line starts with 'd') - works for unix-style listings
+            if not line or not line[0:1] == "d":
+                continue
+            parts = line.split(maxsplit=8)
+            if len(parts) < 9:
+                continue
+            name = parts[-1]
+            if name in (".", "..") or name.startswith("."):
+                continue
+            rel = f"{prefix}/{name}" if prefix else name
+            result.append(rel)
+            if depth < max_depth - 1:
+                sub = posixpath.join(base, name) if base else name
+                result.extend(self._list_dirs_rec(sub, max_depth, depth + 1, rel))
+        return result
+
+    def read_file(self, path: str) -> bytes:
+        self._reset_cwd()
+        buf = io.BytesIO()
+        self.ftp.retrbinary(f"RETR {path}", buf.write)
+        return buf.getvalue()
+
+    def ensure_dir(self, path: str):
+        """Create directory tree, walking step by step from initial CWD.
+
+        IMPORTANT: FTP's cwd() is stateful - it changes the current working
+        directory for ALL subsequent operations. We must walk the path one
+        segment at a time relative to the current position, not concatenate
+        and re-cwd from initial each iteration.
+        """
+        if not path:
+            return
+        try:
+            self._reset_cwd()
+            parts = [p for p in path.split("/") if p]
+            for p in parts:
+                try:
+                    self.ftp.cwd(p)
+                except ftplib.error_perm:
+                    # Doesn't exist - create and enter
+                    try:
+                        self.ftp.mkd(p)
+                        self.ftp.cwd(p)
+                    except ftplib.error_perm as e:
+                        logger.warning(f"FTP mkd({p}) failed: {e}")
+                        return
+        finally:
+            self._reset_cwd()
+
+    def stat_exists(self, path: str) -> bool:
+        """Check if a file or directory exists at path."""
+        try:
+            self.ftp.size(path)
+            return True
+        except (ftplib.error_perm, ftplib.error_temp):
+            pass
+        # Try as directory - cwd then immediately reset
+        try:
+            self.ftp.cwd(path)
+            self._reset_cwd()
+            return True
+        except ftplib.error_perm:
+            self._reset_cwd()
+            return False
+
+    def rename(self, src: str, dst: str):
+        self._reset_cwd()
+        self.ftp.rename(src, dst)
+
+    def close(self):
+        try:
+            self.ftp.quit()
+        except Exception:
+            try:
+                self.ftp.close()
+            except Exception:
+                pass
+
+
+# ---------------------------------------------------------------------------
+# SFTP (paramiko)
+# ---------------------------------------------------------------------------
+
+class _SftpAdapter(_FtpAdapter):
+    def __init__(self, server: str, port: int, username: str, password: str):
+        self.transport = paramiko.Transport((server, port))
+        self.transport.connect(username=username, password=password)
+        self.sftp = paramiko.SFTPClient.from_transport(self.transport)
+
+    def _resolve(self, path: str) -> str:
+        """Resolve path - empty/None means user's home/root directory."""
+        if not path:
+            try:
+                return self.sftp.normalize(".")
+            except IOError:
+                return "."
+        return path
+
+    def list_pdfs(self, path: str) -> list[str]:
+        try:
+            entries = self.sftp.listdir(self._resolve(path))
+        except IOError:
+            return []
+        return sorted(
+            e for e in entries if e.lower().endswith(".pdf") and not e.startswith(".")
+        )
+
+    def list_dirs(self, path: str, max_depth: int = 5) -> list[str]:
+        base = self._resolve(path)
+        logger.debug(f"SFTP list_dirs: base={base!r}, max_depth={max_depth}")
+        return self._list_dirs_rec(base, max_depth, 0, "")
+
+    def _list_dirs_rec(self, base: str, max_depth: int, depth: int, prefix: str) -> list[str]:
+        from stat import S_ISDIR
+        result = []
+        try:
+            entries = self.sftp.listdir_attr(base)
+            logger.debug(f"SFTP listdir_attr({base!r}) -> {[e.filename for e in entries]}")
+        except IOError as e:
+            logger.warning(f"SFTP listdir_attr({base!r}) failed: {e}")
+            return result
+        for entry in entries:
+            if entry.filename.startswith(".") or entry.filename in ("..", "."):
+                continue
+            if entry.st_mode and S_ISDIR(entry.st_mode):
+                rel = f"{prefix}/{entry.filename}" if prefix else entry.filename
+                result.append(rel)
+                if depth < max_depth - 1:
+                    sub = posixpath.join(base, entry.filename)
+                    result.extend(self._list_dirs_rec(sub, max_depth, depth + 1, rel))
+        return result
+
+    def read_file(self, path: str) -> bytes:
+        with self.sftp.open(path, "rb") as f:
+            return f.read()
+
+    def ensure_dir(self, path: str):
+        if not path:
+            return
+        parts = [p for p in path.split("/") if p]
+        cur = ""
+        for p in parts:
+            cur = f"{cur}/{p}" if cur else p
+            try:
+                self.sftp.stat(cur)
+            except IOError:
+                try:
+                    self.sftp.mkdir(cur)
+                except IOError:
+                    pass
+
+    def stat_exists(self, path: str) -> bool:
+        try:
+            self.sftp.stat(path)
+            return True
+        except IOError:
+            return False
+
+    def rename(self, src: str, dst: str):
+        self.sftp.rename(src, dst)
+
+    def close(self):
+        try:
+            self.sftp.close()
+        except Exception:
+            pass
+        try:
+            self.transport.close()
+        except Exception:
+            pass
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _make_adapter(settings: dict) -> _FtpAdapter:
+    protocol = settings.get("ftp_protocol", "sftp").lower()
+    server = settings["ftp_server"]
+    username = settings.get("ftp_username", "")
+    password = settings.get("ftp_password", "")
+    if protocol == "sftp":
+        port = int(settings.get("ftp_port") or 22)
+        return _SftpAdapter(server, port, username, password)
+    else:
+        port = int(settings.get("ftp_port") or 21)
+        return _PlainFtpAdapter(server, port, username, password)
+
+
+def _join_path(*parts: str) -> str:
+    """Join FTP/SFTP path segments using forward slash."""
+    result = ""
+    for p in parts:
+        if not p:
+            continue
+        p = p.replace("\\", "/").strip("/")
+        if not p:
+            continue
+        result = f"{result}/{p}" if result else p
+    return result
+
+
+def _move_with_dedup(adapter: _FtpAdapter, src: str, dest_dir: str, filename: str):
+    """Move file to dest_dir, renaming if a duplicate exists."""
+    dest = _join_path(dest_dir, filename)
+    if adapter.stat_exists(dest):
+        name, ext = os.path.splitext(filename)
+        counter = 1
+        while True:
+            new_name = f"{name}_{counter}{ext}"
+            new_dest = _join_path(dest_dir, new_name)
+            if not adapter.stat_exists(new_dest):
+                dest = new_dest
+                break
+            counter += 1
+    adapter.rename(src, dest)
+
+
+# ---------------------------------------------------------------------------
+# Processing pipeline
+# ---------------------------------------------------------------------------
+
+async def _process_ftp_folder(
+    smtp_conn, settings: dict, adapter: _FtpAdapter,
+    source_path: str, processed_path: str,
+    import_email: str, beleg_type: str, mode: str,
+) -> dict:
+    """Process one FTP folder pair. Returns counts dict."""
+    smtp_from = settings.get("smtp_username", "")
+    protocol = settings.get("ftp_protocol", "sftp").upper()
+    processed = 0
+    skipped = 0
+    errors = 0
+
+    await asyncio.to_thread(adapter.ensure_dir, processed_path)
+
+    pdf_files = await asyncio.to_thread(adapter.list_pdfs, source_path)
+    if not pdf_files:
+        logger.info(f"Keine PDF-Dateien im {protocol}-Ordner '{source_path}' ({beleg_type})")
+        return {"processed": 0, "skipped": 0, "errors": 0}
+
+    logger.info(f"{len(pdf_files)} PDF-Datei(en) im {protocol}-Ordner '{source_path}' ({beleg_type})")
+
+    for filename in pdf_files:
+        file_path = _join_path(source_path, filename)
+        try:
+            pdf_data = await asyncio.to_thread(adapter.read_file, file_path)
+
+            if mode == "separator":
+                with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp:
+                    tmp.write(pdf_data)
+                    tmp_path = tmp.name
+                try:
+                    separator_pages = await asyncio.to_thread(detect_separator_pages, tmp_path, None)
+                    documents = await asyncio.to_thread(split_pdf, tmp_path, separator_pages)
+                finally:
+                    os.unlink(tmp_path)
+
+                if not documents:
+                    skipped += 1
+                    continue
+
+                smtp_log_parts = []
+                for i, doc_bytes in enumerate(documents):
+                    doc_filename = f"{os.path.splitext(filename)[0]}_Teil_{i + 1}.pdf"
+                    subject = f"{protocol}-Import: {filename} (Dokument {i + 1}/{len(documents)})"
+                    msg = _build_forward_email(
+                        from_addr=smtp_from,
+                        to_addr=import_email,
+                        original_subject=subject,
+                        original_from=f"{protocol}-Import",
+                        attachments=[(doc_filename, doc_bytes)],
+                    )
+                    smtp_log_parts.append(_send_with_log(smtp_conn, msg))
+
+                await add_log_entry(
+                    email_subject=f"{protocol}: {filename}",
+                    email_from=f"{protocol}-Import",
+                    attachments_count=len(documents),
+                    status="success",
+                    sent_to=import_email,
+                    smtp_log="\n---\n".join(smtp_log_parts),
+                    beleg_type=beleg_type,
+                )
+                logger.info(f"{protocol} verarbeitet ({beleg_type}): {filename} -> {len(documents)} Dokument(e)")
+            else:
+                msg = _build_forward_email(
+                    from_addr=smtp_from,
+                    to_addr=import_email,
+                    original_subject=f"{protocol}-Import: {filename}",
+                    original_from=f"{protocol}-Import",
+                    attachments=[(filename, pdf_data)],
+                )
+                smtp_log = _send_with_log(smtp_conn, msg)
+
+                await add_log_entry(
+                    email_subject=f"{protocol}: {filename}",
+                    email_from=f"{protocol}-Import",
+                    attachments_count=1,
+                    status="success",
+                    sent_to=import_email,
+                    smtp_log=smtp_log,
+                    beleg_type=beleg_type,
+                )
+                logger.info(f"{protocol} verarbeitet ({beleg_type}): {filename}")
+
+            await asyncio.to_thread(_move_with_dedup, adapter, file_path, processed_path, filename)
+            processed += 1
+
+        except Exception as e:
+            errors += 1
+            logger.error(f"Fehler bei {protocol}-Datei {filename}: {e}")
+            try:
+                await add_log_entry(
+                    email_subject=f"{protocol}: {filename}",
+                    email_from=f"{protocol}-Import",
+                    attachments_count=0,
+                    status="error",
+                    error_message=str(e),
+                    beleg_type=beleg_type,
+                )
+            except Exception:
+                pass
+
+    return {"processed": processed, "skipped": skipped, "errors": errors}
+
+
+async def process_ftp() -> dict:
+    """Process PDF files from FTP/SFTP server - main pipeline."""
+    settings = await get_settings()
+
+    if settings.get("ftp_enabled") != "true":
+        return {"processed": 0, "skipped": 0, "errors": 0}
+
+    if not settings.get("ftp_server"):
+        return {"processed": 0, "skipped": 0, "errors": 0, "error": "FTP nicht konfiguriert"}
+
+    import_email_eingang = get_import_email(settings, "eingang")
+    if not import_email_eingang:
+        return {"processed": 0, "skipped": 0, "errors": 0, "error": "Import-Email nicht konfiguriert"}
+
+    mode = settings.get("ftp_mode", "forward")
+    protocol = settings.get("ftp_protocol", "sftp").upper()
+    total = {"processed": 0, "skipped": 0, "errors": 0}
+    smtp_conn = None
+    adapter = None
+
+    try:
+        adapter = await asyncio.to_thread(_make_adapter, settings)
+        smtp_conn = _connect_smtp(settings)
+
+        # Eingangsbelege
+        source = settings.get("ftp_source_path", "")
+        processed_path = settings.get("ftp_processed_path", "Verarbeitet")
+        result = await _process_ftp_folder(
+            smtp_conn, settings, adapter,
+            source, processed_path,
+            import_email_eingang, "eingang", mode,
+        )
+        for k in total:
+            total[k] += result[k]
+
+        # Ausgangsbelege (optional)
+        import_email_ausgang = get_import_email(settings, "ausgang")
+        source_ausgang = settings.get("ftp_source_path_ausgang", "")
+        processed_ausgang = settings.get("ftp_processed_path_ausgang", "")
+        if import_email_ausgang and source_ausgang:
+            if not processed_ausgang:
+                processed_ausgang = source_ausgang + "/Verarbeitet"
+            result = await _process_ftp_folder(
+                smtp_conn, settings, adapter,
+                source_ausgang, processed_ausgang,
+                import_email_ausgang, "ausgang", mode,
+            )
+            for k in total:
+                total[k] += result[k]
+
+    except Exception as e:
+        logger.error(f"{protocol}-Verbindungsfehler: {e}")
+        try:
+            await add_log_entry(
+                email_subject="",
+                email_from=f"{protocol}-Import",
+                attachments_count=0,
+                status="error",
+                error_message=f"{protocol}-Verbindungsfehler: {e}",
+            )
+        except Exception:
+            pass
+        return {**total, "errors": total["errors"] + 1, "error": str(e)}
+
+    finally:
+        if adapter:
+            try:
+                await asyncio.to_thread(adapter.close)
+            except Exception:
+                pass
+        if smtp_conn:
+            try:
+                smtp_conn.quit()
+            except Exception:
+                pass
+
+    logger.info(f"{protocol} fertig: {total['processed']} verarbeitet, {total['skipped']} übersprungen, {total['errors']} Fehler")
+    return total
+
+
+async def test_ftp_connection() -> dict:
+    """Test FTP/SFTP connection and return TOP-LEVEL folders only (lazy loading)."""
+    settings = await get_settings()
+    if not settings.get("ftp_server"):
+        return {"success": False, "error": "FTP-Server nicht konfiguriert", "folders": []}
+
+    adapter = None
+    try:
+        adapter = await asyncio.to_thread(_make_adapter, settings)
+        folders = await asyncio.to_thread(adapter.list_dirs, "", 1)
+        return {"success": True, "folders": sorted(folders)}
+    except Exception as e:
+        logger.error(f"FTP-Test fehlgeschlagen: {e}")
+        return {"success": False, "error": str(e), "folders": []}
+    finally:
+        if adapter:
+            try:
+                await asyncio.to_thread(adapter.close)
+            except Exception:
+                pass
+
+
+async def create_ftp_folder(folder_path: str) -> dict:
+    """Create a folder on the FTP/SFTP server."""
+    settings = await get_settings()
+    if not settings.get("ftp_server"):
+        return {"success": False, "error": "FTP nicht konfiguriert"}
+    if not folder_path or not folder_path.strip():
+        return {"success": False, "error": "Ordnername darf nicht leer sein"}
+
+    folder_path = folder_path.strip().replace("\\", "/")
+    adapter = None
+    try:
+        adapter = await asyncio.to_thread(_make_adapter, settings)
+        await asyncio.to_thread(adapter.ensure_dir, folder_path)
+        return {"success": True}
+    except Exception as e:
+        logger.error(f"FTP-Ordner erstellen fehlgeschlagen: {e}")
+        return {"success": False, "error": str(e)}
+    finally:
+        if adapter:
+            try:
+                await asyncio.to_thread(adapter.close)
+            except Exception:
+                pass
+
+
+async def list_ftp_folders() -> dict:
+    """Return TOP-LEVEL folder list from FTP/SFTP server (lazy loading)."""
+    settings = await get_settings()
+    if not settings.get("ftp_server"):
+        return {"folders": []}
+    adapter = None
+    try:
+        adapter = await asyncio.to_thread(_make_adapter, settings)
+        folders = await asyncio.to_thread(adapter.list_dirs, "", 1)
+        return {"folders": sorted(folders)}
+    except Exception:
+        return {"folders": []}
+    finally:
+        if adapter:
+            try:
+                await asyncio.to_thread(adapter.close)
+            except Exception:
+                pass
+
+
+async def list_ftp_subfolders(parent_path: str) -> dict:
+    """List direct subfolders of a path (one level deep, for lazy tree expansion)."""
+    settings = await get_settings()
+    if not settings.get("ftp_server"):
+        return {"success": False, "error": "FTP nicht konfiguriert", "folders": []}
+    adapter = None
+    try:
+        adapter = await asyncio.to_thread(_make_adapter, settings)
+        rel_folders = await asyncio.to_thread(adapter.list_dirs, parent_path, 1)
+        # Prefix with parent_path so the frontend has full paths
+        if parent_path:
+            folders = [f"{parent_path}/{f}" for f in rel_folders]
+        else:
+            folders = rel_folders
+        return {"success": True, "folders": sorted(folders)}
+    except Exception as e:
+        logger.error(f"FTP-Subfolder-Liste fehlgeschlagen: {e}")
+        return {"success": False, "error": str(e), "folders": []}
+    finally:
+        if adapter:
+            try:
+                await asyncio.to_thread(adapter.close)
+            except Exception:
+                pass
diff --git a/app/main.py b/app/main.py
index 8e52f0f..f3bc250 100644
--- a/app/main.py
+++ b/app/main.py
@@ -15,7 +15,8 @@ from app.database import init_db, get_settings, save_settings, get_log_entries,
 from app.mail_processor import process_mailbox, send_test_email, test_imap_connection, create_imap_folder
 from app.scheduler import start_scheduler, configure_job, get_scheduler_status
 from app.scanner import process_scanned_pdf, generate_separator_pdf, UPLOAD_DIR
-from app.smb_processor import process_smb_share, test_smb_connection, create_smb_folder, list_smb_folders
+from app.smb_processor import process_smb_share, test_smb_connection, create_smb_folder, list_smb_folders, list_smb_subfolders
+from app.ftp_processor import process_ftp, test_ftp_connection, create_ftp_folder, list_ftp_folders, list_ftp_subfolders
 from app.amazon_processor import (
     start_login as amazon_start_login,
     submit_otp as amazon_submit_otp,
@@ -122,6 +123,18 @@ async def _save_form_settings(request: Request) -> dict:
         "smb_source_path_ausgang": form.get("smb_source_path_ausgang", ""),
         "smb_processed_path_ausgang": form.get("smb_processed_path_ausgang", ""),
         "smb_mode": form.get("smb_mode", "forward"),
+        # FTP / SFTP
+        "ftp_enabled": form.get("ftp_enabled", "false"),
+        "ftp_protocol": form.get("ftp_protocol", "sftp"),
+        "ftp_server": form.get("ftp_server", ""),
+        "ftp_port": form.get("ftp_port", "22"),
+        "ftp_username": form.get("ftp_username", ""),
+        "ftp_password": form.get("ftp_password") or current.get("ftp_password", ""),
+        "ftp_source_path": form.get("ftp_source_path", ""),
+        "ftp_processed_path": form.get("ftp_processed_path", "Verarbeitet"),
+        "ftp_source_path_ausgang": form.get("ftp_source_path_ausgang", ""),
+        "ftp_processed_path_ausgang": form.get("ftp_processed_path_ausgang", ""),
+        "ftp_mode": form.get("ftp_mode", "forward"),
         # Debug
         "debug_save_amazon_pdfs": form.get("debug_save_amazon_pdfs", "false"),
     }
@@ -219,6 +232,45 @@ async def api_create_smb_folder(request: Request):
     return JSONResponse(result)
 
 
+@app.get("/api/list-smb-subfolders")
+async def api_list_smb_subfolders(request: Request):
+    parent = request.query_params.get("path", "")
+    result = await list_smb_subfolders(parent)
+    return JSONResponse(result)
+
+
+@app.post("/api/test-ftp")
+async def api_test_ftp(request: Request):
+    await _save_form_settings(request)
+    result = await test_ftp_connection()
+    return JSONResponse(result)
+
+
+@app.post("/api/process-ftp")
+async def api_process_ftp(request: Request):
+    await _save_form_settings(request)
+    result = await process_ftp()
+    return JSONResponse(result)
+
+
+@app.post("/api/create-ftp-folder")
+async def api_create_ftp_folder(request: Request):
+    body = await request.json()
+    folder_name = body.get("folder_name", "")
+    result = await create_ftp_folder(folder_name)
+    if result["success"]:
+        folders_result = await list_ftp_folders()
+        result["folders"] = folders_result.get("folders", [])
+    return JSONResponse(result)
+
+
+@app.get("/api/list-ftp-subfolders")
+async def api_list_ftp_subfolders(request: Request):
+    parent = request.query_params.get("path", "")
+    result = await list_ftp_subfolders(parent)
+    return JSONResponse(result)
+
+
 @app.get("/log", response_class=HTMLResponse)
 async def log_page(request: Request):
     logs = await get_log_entries(limit=500)
diff --git a/app/scheduler.py b/app/scheduler.py
index d27e68b..ff98833 100644
--- a/app/scheduler.py
+++ b/app/scheduler.py
@@ -5,6 +5,7 @@ from apscheduler.triggers.interval import IntervalTrigger
 
 from app.mail_processor import process_mailbox
 from app.smb_processor import process_smb_share
+from app.ftp_processor import process_ftp
 from app.amazon_processor import process_amazon
 from app.amazon_api import process_amazon_api
 from app.database import get_settings
@@ -33,6 +34,10 @@ async def _run_processor():
         smb_result = await process_smb_share()
         logger.info(f"SMB-Verarbeitung abgeschlossen: {smb_result}")
 
+        logger.info("Starte automatische FTP-Verarbeitung...")
+        ftp_result = await process_ftp()
+        logger.info(f"FTP-Verarbeitung abgeschlossen: {ftp_result}")
+
         # Amazon separately with timeout - must not block next scheduler runs
         logger.info("Starte automatische Amazon-Verarbeitung...")
         try:
diff --git a/app/smb_processor.py b/app/smb_processor.py
index 9352d2a..a0dfa60 100644
--- a/app/smb_processor.py
+++ b/app/smb_processor.py
@@ -99,7 +99,7 @@ def _move_smb_file(source: str, dest_dir: str, filename: str):
 
 
 def _list_smb_folders_recursive(
-    base_path: str, max_depth: int = 3, _current_depth: int = 0, _prefix: str = ""
+    base_path: str, max_depth: int = 5, _current_depth: int = 0, _prefix: str = ""
 ) -> list[str]:
     """Recursively list folders on the SMB share, returning relative paths."""
     folders = []
@@ -307,7 +307,7 @@ async def process_smb_share() -> dict:
 
 
 async def test_smb_connection() -> dict:
-    """Test SMB connection and return folder list."""
+    """Test SMB connection and return TOP-LEVEL folders only (lazy loading)."""
     settings = await get_settings()
 
     if not settings.get("smb_server") or not settings.get("smb_share"):
@@ -315,7 +315,7 @@ async def test_smb_connection() -> dict:
 
     try:
         base_path = await asyncio.to_thread(_smb_register_session, settings)
-        folders = await asyncio.to_thread(_list_smb_folders_recursive, base_path, 3)
+        folders = await asyncio.to_thread(_list_smb_folders_recursive, base_path, 1)
         return {"success": True, "folders": sorted(folders)}
 
     except Exception as e:
@@ -347,13 +347,34 @@ async def create_smb_folder(folder_path: str) -> dict:
 
 
 async def list_smb_folders() -> dict:
-    """Return current folder list from SMB share."""
+    """Return TOP-LEVEL folder list from SMB share (lazy loading)."""
     settings = await get_settings()
     if not settings.get("smb_server") or not settings.get("smb_share"):
         return {"folders": []}
     try:
         base_path = await asyncio.to_thread(_smb_register_session, settings)
-        folders = await asyncio.to_thread(_list_smb_folders_recursive, base_path, 3)
+        folders = await asyncio.to_thread(_list_smb_folders_recursive, base_path, 1)
         return {"folders": sorted(folders)}
     except Exception:
         return {"folders": []}
+
+
+async def list_smb_subfolders(parent_path: str) -> dict:
+    """List direct subfolders of a path (one level deep, for lazy tree expansion)."""
+    settings = await get_settings()
+    if not settings.get("smb_server") or not settings.get("smb_share"):
+        return {"success": False, "error": "SMB nicht konfiguriert", "folders": []}
+    try:
+        base_path = await asyncio.to_thread(_smb_register_session, settings)
+        full_path = _smb_unc_path(base_path, parent_path) if parent_path else base_path
+        # max_depth=1 returns only direct children
+        rel_folders = await asyncio.to_thread(_list_smb_folders_recursive, full_path, 1)
+        # Prefix with parent_path so the frontend has full paths
+        if parent_path:
+            folders = [f"{parent_path}/{f}" for f in rel_folders]
+        else:
+            folders = rel_folders
+        return {"success": True, "folders": sorted(folders)}
+    except Exception as e:
+        logger.error(f"SMB-Subfolder-Liste fehlgeschlagen: {e}")
+        return {"success": False, "error": str(e), "folders": []}
diff --git a/app/static/style.css b/app/static/style.css
index 1bdc0c8..698a317 100644
--- a/app/static/style.css
+++ b/app/static/style.css
@@ -424,6 +424,35 @@ small.text-muted {
     align-items: center;
 }
 
+/* Tree-View Toggle */
+.folder-tree-toggle {
+    flex-shrink: 0;
+    background: none;
+    border: none;
+    cursor: pointer;
+    width: 1.5rem;
+    text-align: center;
+    color: var(--text-muted);
+    font-size: 0.75rem;
+    padding: 0.5rem 0;
+    user-select: none;
+}
+
+.folder-tree-toggle:hover {
+    color: var(--primary);
+}
+
+.folder-tree-toggle.empty {
+    cursor: default;
+    visibility: hidden;
+}
+
+.folder-tree-children {
+    margin-left: 1.5rem;
+    border-left: 1px dashed var(--border);
+    padding-left: 0.25rem;
+}
+
 .folder-item {
     display: flex;
     align-items: center;
diff --git a/app/templates/settings.html b/app/templates/settings.html
index 2359b42..8d1e1fd 100644
--- a/app/templates/settings.html
+++ b/app/templates/settings.html
@@ -227,6 +227,97 @@
         </div>
     </div>
 
+    <div class="card">
+        <h2>FTP / SFTP-Server</h2>
+        <div class="form-grid">
+            <div class="form-group">
+                <label for="ftp_enabled">FTP-Import</label>
+                <select id="ftp_enabled" name="ftp_enabled">
+                    <option value="true" {% if settings.get('ftp_enabled') == 'true' %}selected{% endif %}>Aktiviert</option>
+                    <option value="false" {% if settings.get('ftp_enabled') != 'true' %}selected{% endif %}>Deaktiviert</option>
+                </select>
+            </div>
+            <div class="form-group">
+                <label for="ftp_protocol">Protokoll</label>
+                <select id="ftp_protocol" name="ftp_protocol" onchange="updateFtpDefaultPort()">
+                    <option value="sftp" {% if settings.get('ftp_protocol', 'sftp') == 'sftp' %}selected{% endif %}>SFTP (SSH, verschluesselt)</option>
+                    <option value="ftp" {% if settings.get('ftp_protocol') == 'ftp' %}selected{% endif %}>FTP (passiv, unverschluesselt)</option>
+                </select>
+            </div>
+            <div class="form-group">
+                <label for="ftp_mode">Verarbeitungsmodus</label>
+                <select id="ftp_mode" name="ftp_mode">
+                    <option value="forward" {% if settings.get('ftp_mode', 'forward') == 'forward' %}selected{% endif %}>Direkt weiterleiten</option>
+                    <option value="separator" {% if settings.get('ftp_mode') == 'separator' %}selected{% endif %}>Trennseiten-Erkennung</option>
+                </select>
+                <small class="text-muted">Direkt: jede PDF als ein Beleg. Trennseiten: QR-Splitting wie bei Scan-Upload.</small>
+            </div>
+            <div class="form-group">
+                <label for="ftp_server">Server</label>
+                <input type="text" id="ftp_server" name="ftp_server"
+                       value="{{ settings.get('ftp_server', '') }}" placeholder="ftp.example.com oder 192.168.1.100">
+            </div>
+            <div class="form-group">
+                <label for="ftp_port">Port</label>
+                <input type="number" id="ftp_port" name="ftp_port"
+                       value="{{ settings.get('ftp_port', '22') }}">
+                <small class="text-muted">SFTP=22, FTP=21</small>
+            </div>
+            <div class="form-group">
+                <label for="ftp_username">Benutzername</label>
+                <input type="text" id="ftp_username" name="ftp_username"
+                       value="{{ settings.get('ftp_username', '') }}" placeholder="user">
+            </div>
+            <div class="form-group">
+                <label for="ftp_password">Passwort</label>
+                <input type="password" id="ftp_password" name="ftp_password"
+                       placeholder="{% if settings.get('ftp_password') %}(gespeichert){% else %}Passwort eingeben{% endif %}">
+            </div>
+            <div class="form-group">
+                <label for="ftp_source_path">Quellordner Eingangsbelege</label>
+                <div class="input-with-btn">
+                    <input type="text" id="ftp_source_path" name="ftp_source_path"
+                           value="{{ settings.get('ftp_source_path', '') }}" placeholder="(Wurzel)">
+                    <button type="button" class="btn btn-icon" onclick="openFtpFolderPicker('ftp_source_path')" title="Ordner auswählen">&#128193;</button>
+                </div>
+            </div>
+            <div class="form-group">
+                <label for="ftp_processed_path">Verarbeitet-Ordner Eingangsbelege</label>
+                <div class="input-with-btn">
+                    <input type="text" id="ftp_processed_path" name="ftp_processed_path"
+                           value="{{ settings.get('ftp_processed_path', 'Verarbeitet') }}" placeholder="Verarbeitet">
+                    <button type="button" class="btn btn-icon" onclick="openFtpFolderPicker('ftp_processed_path')" title="Ordner auswählen">&#128193;</button>
+                </div>
+            </div>
+            <div class="form-group">
+                <label for="ftp_source_path_ausgang">Quellordner Ausgangsbelege</label>
+                <div class="input-with-btn">
+                    <input type="text" id="ftp_source_path_ausgang" name="ftp_source_path_ausgang"
+                           value="{{ settings.get('ftp_source_path_ausgang', '') }}" placeholder="(optional)">
+                    <button type="button" class="btn btn-icon" onclick="openFtpFolderPicker('ftp_source_path_ausgang')" title="Ordner auswählen">&#128193;</button>
+                </div>
+            </div>
+            <div class="form-group">
+                <label for="ftp_processed_path_ausgang">Verarbeitet-Ordner Ausgangsbelege</label>
+                <div class="input-with-btn">
+                    <input type="text" id="ftp_processed_path_ausgang" name="ftp_processed_path_ausgang"
+                           value="{{ settings.get('ftp_processed_path_ausgang', '') }}" placeholder="(optional)">
+                    <button type="button" class="btn btn-icon" onclick="openFtpFolderPicker('ftp_processed_path_ausgang')" title="Ordner auswählen">&#128193;</button>
+                </div>
+            </div>
+        </div>
+        <div class="form-actions">
+            <button type="button" class="btn btn-secondary" onclick="testFtp()">
+                <span class="btn-text">Verbindung testen &amp; Ordner laden</span>
+                <span class="btn-spinner" style="display:none;">Verbinde...</span>
+            </button>
+            <button type="button" class="btn btn-secondary" onclick="processFtp()">
+                <span class="btn-text">Jetzt abrufen</span>
+                <span class="btn-spinner" style="display:none;">Verarbeite...</span>
+            </button>
+        </div>
+    </div>
+
     <div class="card">
         <h2>Zeitplan</h2>
         <div class="form-grid">
@@ -341,7 +432,10 @@
     <div class="modal">
         <div class="modal-header">
             <h3 id="smbFolderModalTitle">SMB-Ordner auswählen</h3>
-            <button type="button" class="modal-close" onclick="closeSmbFolderModal()">&times;</button>
+            <div style="display:flex;gap:0.5rem;align-items:center;">
+                <button type="button" class="btn btn-secondary btn-sm" onclick="reloadSmbFolders()" title="Ordner-Liste neu laden">&#8634; Neu laden</button>
+                <button type="button" class="modal-close" onclick="closeSmbFolderModal()">&times;</button>
+            </div>
         </div>
         <div class="modal-body">
             <div id="smbFolderList" class="folder-list"></div>
@@ -353,12 +447,176 @@
     </div>
 </div>
 
+<div id="ftpFolderModal" class="modal-overlay" style="display:none;" onclick="closeFtpFolderModal(event)">
+    <div class="modal">
+        <div class="modal-header">
+            <h3 id="ftpFolderModalTitle">FTP-Ordner auswählen</h3>
+            <div style="display:flex;gap:0.5rem;align-items:center;">
+                <button type="button" class="btn btn-secondary btn-sm" onclick="reloadFtpFolders()" title="Ordner-Liste neu laden">&#8634; Neu laden</button>
+                <button type="button" class="modal-close" onclick="closeFtpFolderModal()">&times;</button>
+            </div>
+        </div>
+        <div class="modal-body">
+            <div id="ftpFolderList" class="folder-list"></div>
+            <div id="ftpFolderLoading" class="text-muted" style="display:none;padding:1rem;">
+                Verbinde und lade Ordner...
+            </div>
+            <div id="ftpFolderError" class="alert alert-error" style="display:none;margin:1rem;"></div>
+        </div>
+    </div>
+</div>
+
 <script>
 let cachedFolders = null;
 let cachedDelimiter = '.';
 let folderTargetField = null;
 let createOpenFor = null;
 
+// --- Generic Tree-View Helpers (with Lazy Loading) ---
+// Used by both SMB and FTP folder pickers
+const expandedTreeNodes = { smb: new Set(), ftp: new Set() };
+const loadedTreeNodes = { smb: new Set(['']), ftp: new Set([''])};  // root is always loaded
+const loadingTreeNodes = { smb: new Set(), ftp: new Set() };
+
+function buildFolderTree(paths) {
+    // Build nested tree from flat paths like ["A", "A/B", "A/B/C", "X"]
+    const root = { name: '', path: '', children: {} };
+    paths.forEach(p => {
+        if (!p) return;
+        const parts = p.split('/');
+        let node = root;
+        let curPath = '';
+        for (const part of parts) {
+            curPath = curPath ? curPath + '/' + part : part;
+            if (!node.children[part]) {
+                node.children[part] = { name: part, path: curPath, children: {} };
+            }
+            node = node.children[part];
+        }
+    });
+    return root;
+}
+
+function renderFolderTree(node, ns, currentValue, selectFn, addBtnFn, depth) {
+    let html = '';
+    const childKeys = Object.keys(node.children).sort((a, b) => a.localeCompare(b));
+
+    childKeys.forEach(key => {
+        const child = node.children[key];
+        const hasChildren = Object.keys(child.children).length > 0;
+        const isLoaded = loadedTreeNodes[ns].has(child.path);
+        const isLoading = loadingTreeNodes[ns].has(child.path);
+        const isExpanded = expandedTreeNodes[ns].has(child.path);
+        const isSelected = child.path === currentValue;
+        const escapedPath = child.path.replace(/'/g, "\\'");
+
+        // Always show toggle button - we don't know yet if there are children until loaded
+        const arrow = isLoading ? '&#8987;'  // ⌛
+                    : isExpanded ? '&#9660;'  // ▼
+                    : '&#9654;';              // ▶
+
+        html += '<div class="folder-row">';
+        html += '<button type="button" class="folder-tree-toggle" onclick="toggleTreeNode(\'' + ns + '\',\'' + escapedPath + '\')">' + arrow + '</button>';
+        html += '<button type="button" class="folder-item' + (isSelected ? ' selected' : '') + '" onclick="' + selectFn + '(\'' + escapedPath + '\')">';
+        html += '<span class="folder-icon">&#128193;</span> ' + esc(child.name);
+        if (isSelected) html += ' <span class="badge badge-success">ausgewählt</span>';
+        html += '</button>';
+        html += '<button type="button" class="folder-add-btn" onclick="event.stopPropagation(); ' + addBtnFn + '(\'' + escapedPath + '\')" title="Unterordner erstellen">&#128193;+</button>';
+        html += '</div>';
+        html += '<div id="' + ns + '-create-row-' + CSS.escape(child.path) + '" class="create-inline" style="display:none;"></div>';
+
+        if (isExpanded && (hasChildren || isLoaded)) {
+            html += '<div class="folder-tree-children">';
+            if (hasChildren) {
+                html += renderFolderTree(child, ns, currentValue, selectFn, addBtnFn, depth + 1);
+            } else if (isLoaded) {
+                html += '<p class="text-muted" style="padding:0.25rem 0.5rem;font-size:0.85rem;">(keine Unterordner)</p>';
+            }
+            html += '</div>';
+        }
+    });
+    return html;
+}
+
+async function toggleTreeNode(ns, path) {
+    if (expandedTreeNodes[ns].has(path)) {
+        // Collapse
+        expandedTreeNodes[ns].delete(path);
+        if (ns === 'smb') showSmbFolderModal(smbFolderTargetField);
+        else if (ns === 'ftp') showFtpFolderModal(ftpFolderTargetField);
+        return;
+    }
+
+    // Expand - load subfolders if not loaded yet
+    if (!loadedTreeNodes[ns].has(path)) {
+        loadingTreeNodes[ns].add(path);
+        if (ns === 'smb') showSmbFolderModal(smbFolderTargetField);
+        else if (ns === 'ftp') showFtpFolderModal(ftpFolderTargetField);
+
+        try {
+            const endpoint = '/api/list-' + ns + '-subfolders?path=' + encodeURIComponent(path);
+            const resp = await fetch(endpoint);
+            const data = await resp.json();
+            if (data.success && data.folders) {
+                // Merge new folders into cache
+                const cacheKey = ns === 'smb' ? 'cachedSmbFolders' : 'cachedFtpFolders';
+                const existingSet = new Set(window[cacheKey] || []);
+                data.folders.forEach(f => existingSet.add(f));
+                window[cacheKey] = Array.from(existingSet).sort();
+                loadedTreeNodes[ns].add(path);
+            }
+        } catch (e) {
+            console.error('Lazy load failed:', e);
+        } finally {
+            loadingTreeNodes[ns].delete(path);
+        }
+    }
+
+    expandedTreeNodes[ns].add(path);
+    if (ns === 'smb') showSmbFolderModal(smbFolderTargetField);
+    else if (ns === 'ftp') showFtpFolderModal(ftpFolderTargetField);
+}
+
+function expandTreePathsForValue(ns, value) {
+    if (!value) return;
+    const parts = value.split('/');
+    let cur = '';
+    for (let i = 0; i < parts.length - 1; i++) {
+        cur = cur ? cur + '/' + parts[i] : parts[i];
+        expandedTreeNodes[ns].add(cur);
+    }
+}
+
+async function preloadTreePathForValue(ns, value) {
+    // Lazy-load all ancestor paths so the tree displays the selected value
+    if (!value) return;
+    const parts = value.split('/');
+    let cur = '';
+    const cacheKey = ns === 'smb' ? 'cachedSmbFolders' : 'cachedFtpFolders';
+    for (let i = 0; i < parts.length - 1; i++) {
+        cur = cur ? cur + '/' + parts[i] : parts[i];
+        if (loadedTreeNodes[ns].has(cur)) continue;
+        try {
+            const endpoint = '/api/list-' + ns + '-subfolders?path=' + encodeURIComponent(cur);
+            const resp = await fetch(endpoint);
+            const data = await resp.json();
+            if (data.success && data.folders) {
+                const existingSet = new Set(window[cacheKey] || []);
+                data.folders.forEach(f => existingSet.add(f));
+                window[cacheKey] = Array.from(existingSet).sort();
+                loadedTreeNodes[ns].add(cur);
+            }
+        } catch (e) { /* ignore */ }
+    }
+}
+
+function resetTreeState(ns) {
+    expandedTreeNodes[ns].clear();
+    loadedTreeNodes[ns].clear();
+    loadedTreeNodes[ns].add('');
+    loadingTreeNodes[ns].clear();
+}
+
 function showAlert(message, type) {
     const el = document.getElementById('jsAlert');
     el.textContent = message;
@@ -682,12 +940,19 @@ async function testSmb() {
 
 function openSmbFolderPicker(targetField) {
     smbFolderTargetField = targetField;
-    if (cachedSmbFolders) {
-        showSmbFolderModal(targetField);
-    } else {
-        showSmbFolderModalLoading(targetField);
-        fetchSmbFolders(targetField);
-    }
+    // Always reload when opening to ensure fresh state
+    cachedSmbFolders = null;
+    window.cachedSmbFolders = null;
+    resetTreeState('smb');
+    showSmbFolderModalLoading(targetField);
+    fetchSmbFolders(targetField);
+}
+
+function reloadSmbFolders() {
+    cachedSmbFolders = null;
+    resetTreeState('smb');
+    showSmbFolderModalLoading(smbFolderTargetField);
+    fetchSmbFolders(smbFolderTargetField);
 }
 
 async function fetchSmbFolders(targetField) {
@@ -696,6 +961,12 @@ async function fetchSmbFolders(targetField) {
         const data = await resp.json();
         if (data.success) {
             cachedSmbFolders = data.folders;
+            window.cachedSmbFolders = data.folders;
+            // Preload ancestor paths if a value is already selected
+            if (targetField) {
+                const currentValue = document.getElementById(targetField).value;
+                await preloadTreePathForValue('smb', currentValue);
+            }
             showSmbFolderModal(targetField);
         } else {
             showSmbFolderModalError('SMB-Verbindung fehlgeschlagen: ' + data.error);
@@ -741,19 +1012,12 @@ function showSmbFolderModal(targetField) {
         html += '</div>';
     }
 
-    if (cachedSmbFolders && cachedSmbFolders.length > 0) {
-        cachedSmbFolders.forEach(folder => {
-            const isSelected = folder === currentValue;
-            const escapedFolder = folder.replace(/'/g, "\\'");
-            html += '<div class="folder-row">';
-            html += '<button type="button" class="folder-item' + (isSelected ? ' selected' : '') + '" onclick="selectSmbFolder(\'' + escapedFolder + '\')">';
-            html += '<span class="folder-icon">&#128193;</span> ' + esc(folder);
-            if (isSelected) html += ' <span class="badge badge-success">ausgewählt</span>';
-            html += '</button>';
-            html += '<button type="button" class="folder-add-btn" onclick="event.stopPropagation(); toggleSmbCreateInput(\'' + escapedFolder + '\')" title="Unterordner erstellen">&#128193;+</button>';
-            html += '</div>';
-            html += '<div id="smb-create-row-' + CSS.escape(folder) + '" class="create-inline" style="display:none;"></div>';
-        });
+    const smbFolders = window.cachedSmbFolders || cachedSmbFolders;
+    if (smbFolders && smbFolders.length > 0) {
+        // Auto-expand path to current value, then render tree
+        expandTreePathsForValue('smb', currentValue);
+        const tree = buildFolderTree(smbFolders);
+        html += renderFolderTree(tree, 'smb', currentValue, 'selectSmbFolder', 'toggleSmbCreateInput', 0);
     } else {
         html += '<p class="text-muted" style="padding:0.5rem;">Keine Ordner gefunden.</p>';
     }
@@ -875,5 +1139,259 @@ function closeSmbFolderModal(event) {
     document.getElementById('smbFolderModal').style.display = 'none';
     cachedSmbFolders = null;
 }
+
+// --- FTP / SFTP ---
+let cachedFtpFolders = null;
+let ftpFolderTargetField = null;
+
+function updateFtpDefaultPort() {
+    const proto = document.getElementById('ftp_protocol').value;
+    const portInput = document.getElementById('ftp_port');
+    const current = portInput.value;
+    if (proto === 'sftp' && (current === '21' || !current)) {
+        portInput.value = '22';
+    } else if (proto === 'ftp' && (current === '22' || !current)) {
+        portInput.value = '21';
+    }
+}
+
+async function testFtp() {
+    const btn = event.currentTarget;
+    setButtonLoading(btn, true);
+    try {
+        const resp = await fetch('/api/test-ftp', { method: 'POST', body: getFormData() });
+        const data = await resp.json();
+        if (data.success) {
+            cachedFtpFolders = data.folders;
+            showAlert('FTP-Verbindung erfolgreich! Einstellungen gespeichert.', 'success');
+            showFtpFolderModal(null);
+        } else {
+            showAlert('FTP-Verbindung fehlgeschlagen: ' + data.error, 'error');
+        }
+    } catch (e) {
+        showAlert('Fehler: ' + e.message, 'error');
+    } finally {
+        setButtonLoading(btn, false);
+    }
+}
+
+async function processFtp() {
+    const btn = event.currentTarget;
+    setButtonLoading(btn, true);
+    try {
+        const resp = await fetch('/api/process-ftp', { method: 'POST', body: getFormData() });
+        const data = await resp.json();
+        if (data.error) {
+            showAlert('FTP-Fehler: ' + data.error, 'error');
+        } else {
+            showAlert(`FTP-Abruf fertig: ${data.processed} verarbeitet, ${data.skipped || 0} uebersprungen, ${data.errors} Fehler`, data.errors > 0 ? 'warning' : 'success');
+        }
+    } catch (e) {
+        showAlert('Fehler: ' + e.message, 'error');
+    } finally {
+        setButtonLoading(btn, false);
+    }
+}
+
+function openFtpFolderPicker(targetField) {
+    ftpFolderTargetField = targetField;
+    // Always reload when opening to ensure fresh state
+    cachedFtpFolders = null;
+    window.cachedFtpFolders = null;
+    resetTreeState('ftp');
+    showFtpFolderModalLoading(targetField);
+    fetchFtpFolders(targetField);
+}
+
+function reloadFtpFolders() {
+    cachedFtpFolders = null;
+    resetTreeState('ftp');
+    showFtpFolderModalLoading(ftpFolderTargetField);
+    fetchFtpFolders(ftpFolderTargetField);
+}
+
+async function fetchFtpFolders(targetField) {
+    try {
+        const resp = await fetch('/api/test-ftp', { method: 'POST', body: getFormData() });
+        const data = await resp.json();
+        if (data.success) {
+            cachedFtpFolders = data.folders;
+            window.cachedFtpFolders = data.folders;
+            if (targetField) {
+                const currentValue = document.getElementById(targetField).value;
+                await preloadTreePathForValue('ftp', currentValue);
+            }
+            showFtpFolderModal(targetField);
+        } else {
+            showFtpFolderModalError('FTP-Verbindung fehlgeschlagen: ' + data.error);
+        }
+    } catch (e) {
+        showFtpFolderModalError('Fehler: ' + e.message);
+    }
+}
+
+function showFtpFolderModalLoading(targetField) {
+    const modal = document.getElementById('ftpFolderModal');
+    document.getElementById('ftpFolderList').innerHTML = '';
+    document.getElementById('ftpFolderLoading').style.display = '';
+    document.getElementById('ftpFolderError').style.display = 'none';
+    modal.style.display = 'flex';
+}
+
+function showFtpFolderModal(targetField) {
+    if (targetField) ftpFolderTargetField = targetField;
+    const modal = document.getElementById('ftpFolderModal');
+    document.getElementById('ftpFolderLoading').style.display = 'none';
+    document.getElementById('ftpFolderError').style.display = 'none';
+
+    const list = document.getElementById('ftpFolderList');
+    const currentValue = ftpFolderTargetField ? document.getElementById(ftpFolderTargetField).value : '';
+
+    let html = '<div class="folder-picker-fields">';
+    html += '<button type="button" class="folder-field-btn ' + (ftpFolderTargetField === 'ftp_source_path' ? 'active' : '') + '" onclick="switchFtpFolderTarget(\'ftp_source_path\')">Eingang Quelle: <strong>' + esc(document.getElementById('ftp_source_path').value || '(Wurzel)') + '</strong></button>';
+    html += '<button type="button" class="folder-field-btn ' + (ftpFolderTargetField === 'ftp_processed_path' ? 'active' : '') + '" onclick="switchFtpFolderTarget(\'ftp_processed_path\')">Eingang Verarbeitet: <strong>' + esc(document.getElementById('ftp_processed_path').value) + '</strong></button>';
+    html += '<button type="button" class="folder-field-btn ' + (ftpFolderTargetField === 'ftp_source_path_ausgang' ? 'active' : '') + '" onclick="switchFtpFolderTarget(\'ftp_source_path_ausgang\')">Ausgang Quelle: <strong>' + esc(document.getElementById('ftp_source_path_ausgang').value || '(nicht gesetzt)') + '</strong></button>';
+    html += '<button type="button" class="folder-field-btn ' + (ftpFolderTargetField === 'ftp_processed_path_ausgang' ? 'active' : '') + '" onclick="switchFtpFolderTarget(\'ftp_processed_path_ausgang\')">Ausgang Verarbeitet: <strong>' + esc(document.getElementById('ftp_processed_path_ausgang').value || '(nicht gesetzt)') + '</strong></button>';
+    html += '</div>';
+    html += '<div class="folder-items">';
+
+    // Root option for source path
+    if (ftpFolderTargetField === 'ftp_source_path') {
+        const isRoot = currentValue === '';
+        html += '<div class="folder-row">';
+        html += '<button type="button" class="folder-item' + (isRoot ? ' selected' : '') + '" onclick="selectFtpFolder(\'\')">';
+        html += '<span class="folder-icon">&#128193;</span> (Wurzel des Servers)';
+        if (isRoot) html += ' <span class="badge badge-success">ausgewählt</span>';
+        html += '</button>';
+        html += '</div>';
+    }
+
+    const ftpFolders = window.cachedFtpFolders || cachedFtpFolders;
+    if (ftpFolders && ftpFolders.length > 0) {
+        // Auto-expand path to current value, then render tree
+        expandTreePathsForValue('ftp', currentValue);
+        const tree = buildFolderTree(ftpFolders);
+        html += renderFolderTree(tree, 'ftp', currentValue, 'selectFtpFolder', 'toggleFtpCreateInput', 0);
+    } else {
+        html += '<p class="text-muted" style="padding:0.5rem;">Keine Ordner gefunden.</p>';
+    }
+
+    // Create new root folder option
+    html += '<div class="folder-row" style="border-top:1px solid var(--border);margin-top:0.5rem;padding-top:0.5rem;">';
+    html += '<button type="button" class="folder-item" onclick="event.stopPropagation(); toggleFtpCreateInput(\'\')" style="color:var(--primary);">';
+    html += '<span class="folder-icon">&#128193;+</span> Neuen Ordner erstellen';
+    html += '</button>';
+    html += '</div>';
+    html += '<div id="ftp-create-row-root" class="create-inline" style="display:none;"></div>';
+
+    html += '</div>';
+    list.innerHTML = html;
+    modal.style.display = 'flex';
+}
+
+function switchFtpFolderTarget(field) {
+    ftpFolderTargetField = field;
+    showFtpFolderModal(field);
+}
+
+function selectFtpFolder(folder) {
+    if (ftpFolderTargetField) {
+        document.getElementById(ftpFolderTargetField).value = folder;
+    }
+    showFtpFolderModal(ftpFolderTargetField);
+}
+
+function showFtpFolderModalError(msg) {
+    document.getElementById('ftpFolderLoading').style.display = 'none';
+    document.getElementById('ftpFolderError').textContent = msg;
+    document.getElementById('ftpFolderError').style.display = '';
+}
+
+function toggleFtpCreateInput(parentFolder) {
+    document.querySelectorAll('.create-inline[id^="ftp-create-row"]').forEach(el => {
+        const rowId = parentFolder === '' ? 'ftp-create-row-root' : 'ftp-create-row-' + CSS.escape(parentFolder);
+        if (el.id !== rowId) {
+            el.style.display = 'none';
+            el.innerHTML = '';
+        }
+    });
+
+    const rowId = parentFolder === '' ? 'ftp-create-row-root' : 'ftp-create-row-' + CSS.escape(parentFolder);
+    const row = document.getElementById(rowId);
+    if (!row) return;
+
+    if (row.style.display !== 'none') {
+        row.style.display = 'none';
+        row.innerHTML = '';
+        return;
+    }
+
+    const prefix = parentFolder ? parentFolder + '/' : '';
+    row.innerHTML =
+        '<div class="create-folder-inline">' +
+        '<span class="create-folder-prefix">' + esc(prefix) + '</span>' +
+        '<input type="text" class="create-folder-input" id="newFtpSubfolderInput" placeholder="Name" autofocus>' +
+        '<button type="button" class="btn btn-primary btn-sm" onclick="doCreateFtpFolder(\'' + parentFolder.replace(/'/g, "\\'") + '\')">OK</button>' +
+        '<button type="button" class="btn btn-secondary btn-sm" onclick="toggleFtpCreateInput(\'' + parentFolder.replace(/'/g, "\\'") + '\')">Abbrechen</button>' +
+        '</div>' +
+        '<div id="ftpCreateError" class="text-error" style="display:none;"></div>';
+    row.style.display = '';
+
+    const input = document.getElementById('newFtpSubfolderInput');
+    if (input) {
+        input.focus();
+        input.addEventListener('keydown', function(e) {
+            if (e.key === 'Enter') { e.preventDefault(); doCreateFtpFolder(parentFolder); }
+            if (e.key === 'Escape') { toggleFtpCreateInput(parentFolder); }
+        });
+    }
+}
+
+async function doCreateFtpFolder(parentFolder) {
+    const input = document.getElementById('newFtpSubfolderInput');
+    const errorEl = document.getElementById('ftpCreateError');
+    if (!input) return;
+
+    const subName = input.value.trim();
+    if (!subName) {
+        errorEl.textContent = 'Bitte einen Namen eingeben.';
+        errorEl.style.display = '';
+        return;
+    }
+
+    const fullName = parentFolder ? parentFolder + '/' + subName : subName;
+    errorEl.style.display = 'none';
+    input.disabled = true;
+
+    try {
+        const resp = await fetch('/api/create-ftp-folder', {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify({ folder_name: fullName })
+        });
+        const data = await resp.json();
+        if (data.success) {
+            cachedFtpFolders = data.folders;
+            if (ftpFolderTargetField) {
+                document.getElementById(ftpFolderTargetField).value = fullName;
+            }
+            showFtpFolderModal(ftpFolderTargetField);
+        } else {
+            errorEl.textContent = data.error;
+            errorEl.style.display = '';
+            input.disabled = false;
+        }
+    } catch (e) {
+        errorEl.textContent = e.message;
+        errorEl.style.display = '';
+        input.disabled = false;
+    }
+}
+
+function closeFtpFolderModal(event) {
+    if (event && event.target !== document.getElementById('ftpFolderModal')) return;
+    document.getElementById('ftpFolderModal').style.display = 'none';
+    cachedFtpFolders = null;
+}
 </script>
 {% endblock %}
diff --git a/requirements.txt b/requirements.txt
index 6e322f1..e563ac6 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -12,6 +12,7 @@ PyMuPDF==1.25.3
 qrcode==8.0
 sse-starlette==2.2.1
 smbprotocol==1.14.0
+paramiko==3.5.0
 playwright==1.49.1
 playwright-stealth==2.0.2
 httpx==0.28.1