From 1caba6946cc6be32fab5a27aab52c0e3d52653f9 Mon Sep 17 00:00:00 2001
From: xavix-yo <daniele.guiducci@gmail.com>
Date: Thu, 16 Jul 2026 22:29:33 +0100
Subject: [PATCH] Add Email (IMAP/SMTP) and SSH Remote Access connectors
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- CLAUDE.md: developer guide for the marketplace repo
- email/: generic IMAP+SMTP connector (stdlib only, config via env)
  - env[] array, auth.type: password, verify.py (IMAP+SMTP probe)
- ssh/: SSH Remote Access connector ported from Skald, modified
  - ALIASES_FILE → ~/.ssh_aliases.json (was ./secrets/...)
  - optional env[] (TTL/timeout tunables)
  - auth.type: none (per-alias auth at runtime via elicitation)
- tavily/: restructured with env[] (tavilyApiKey), verify.py, auth block
- connectors.json: added email, ssh, and tavily verify.py entries
- SKALD.md: documented env[], verify, password auth, SSH connector
---
 CLAUDE.md                            |   77 ++
 SKALD.md                             |  148 ++-
 connectors/connectors.json           |   91 ++
 connectors/email/connector.json      |  131 +++
 connectors/email/email_mcp_server.py | 1398 ++++++++++++++++++++++++++
 connectors/email/icon_lg.svg         |    1 +
 connectors/email/icon_sm.svg         |    1 +
 connectors/email/verify.py           |   89 ++
 connectors/ssh/connector.json        |  113 +++
 connectors/ssh/icon_lg.svg           |    6 +
 connectors/ssh/icon_sm.svg           |    4 +
 connectors/ssh/requirements.txt      |    1 +
 connectors/ssh/ssh_mcp_server.py     | 1284 +++++++++++++++++++++++
 connectors/tavily/connector.json     |   19 +-
 connectors/tavily/verify.py          |   65 ++
 15 files changed, 3420 insertions(+), 8 deletions(-)
 create mode 100644 CLAUDE.md
 create mode 100644 connectors/email/connector.json
 create mode 100644 connectors/email/email_mcp_server.py
 create mode 100644 connectors/email/icon_lg.svg
 create mode 100644 connectors/email/icon_sm.svg
 create mode 100644 connectors/email/verify.py
 create mode 100644 connectors/ssh/connector.json
 create mode 100644 connectors/ssh/icon_lg.svg
 create mode 100644 connectors/ssh/icon_sm.svg
 create mode 100644 connectors/ssh/requirements.txt
 create mode 100644 connectors/ssh/ssh_mcp_server.py
 create mode 100644 connectors/tavily/verify.py

diff --git a/CLAUDE.md b/CLAUDE.md
new file mode 100644
index 0000000..22f42c0
--- /dev/null
+++ b/CLAUDE.md
@@ -0,0 +1,77 @@
+# CLAUDE.md
+
+This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
+
+## What this is
+
+The **Skald Connectors Marketplace** — a catalog of tested connectors (adapters that let Skald agents talk to external services). It is **not an application**: there is no build system, package manager, or test suite. The repo is a set of JSON manifests, static HTML, icons, and standalone Python MCP scripts that are served as-is over HTTP.
+
+`SKALD.md` is the authoritative spec (written in Italian) — read it before changing any schema or manifest.
+
+## Architecture
+
+Two-tier manifest model with a single trust root:
+
+- **`connectors/connectors.json`** — the **index** and single root of trust. Lists every connector and, per connector, a `files[]` array with the `sha256` + `size` of each shipped file. It has **no hash of itself**, and the `files[]` arrays **do not include `connector.json`** — both are treated as unsigned manifests. Intended to be digitally signed in the future.
+- **`connectors/<id>/connector.json`** — per-connector **technical manifest** for activation (launch command, transport, `mcp_config`, `auth`, `dependencies`, `docs`). One folder per connector; folder name matches the connector `id`.
+- **`connectors/index.html`** — catalog UI. Fetches `/connectors.json` at absolute path and links to `/<folder>/`, so it only works when served from the site root (not opened as a `file://`).
+
+Connector taxonomy (see SKALD.md for full enums):
+- `type`: `mcp_remote` (hosted MCP server reached by URL, e.g. Tavily) vs `mcp_local` (script run client-side, e.g. Gmail).
+- `scope`: `global` (one shared config) vs `user` (per-user auth/instance).
+- `requires`: prerequisites like `API_KEY`, `OAUTH`, `PYTHON`, `NODE`, `DOCKER`, `ENV`. (`SECRETS_DIR` is **deprecated** — the `secrets/` folder is gone from the model; connectors must declare their inputs as `ENV`/`SECRET`.)
+
+**Unified placeholder syntax.** Every value skald must fill at runtime uses one of two tokens, anywhere it appears (URL, `mcp_config.env`, `verify.command`):
+- `{ENV:NAME}` — a non-sensitive value (host, port, username…), declared in the top-level `env[]` array.
+- `{SECRET:NAME}` — a sensitive value (password, API key, token), also declared in `env[]` with `secret: true`.
+
+Legacy tokens `{key}`, `{env:NAME}`, `{secrets}/…` are **deprecated** and skald no longer substitutes them. The api-key of an `auth.type = "api_key"` connector is exposed as `{SECRET:<auth.param>}` (e.g. Tavily: `?tavilyApiKey={SECRET:tavilyApiKey}`). See SKALD.md § Sintassi placeholder.
+
+**Config via `env[]` (`env` manifest field):** a connector whose `requires` includes `ENV` declares its required environment variables in a top-level `env` array in `connector.json` — each entry has `name`, `label`, `description`, `required`, `secret`, and optional `default`/`example`. skald renders a real form from this schema (masking `secret: true` fields, marking `required`, using `example` as placeholder) and injects the collected values into the process environment at launch; the server reads `os.environ`. Nothing is written to disk. The `email` connector is the reference example (see SKALD.md § Campo env). **`tavily` now also declares `env[]`** (its API key as a `secret` field).
+
+**Verify-before-save (`verify` manifest field).** A connector may declare a `verify` step — a shell command skald runs *after* the user fills the form and *before* persisting the activation, to confirm the credentials actually work:
+```json
+"verify": { "command": "python3 verify.py", "timeout_secs": 20 }
+```
+The command runs in the same sandbox as the server (container `skald-{userid}` for `mcp_local` user, host for `mcp_remote` global), with the collected env/secret injected. It must print one JSON object on stdout — `{"ok": bool, "message": string, "details"?: object}` — and exit 0 on success. If `verify` is absent, no test runs and the activation is direct. See SKALD.md § Campo verify. `email` and `tavily` ship a `verify.py`; `gmail` will get one in the OAuth phase.
+
+**MCP servers** (`connectors/gmail/gmail_mcp_server.py`, `connectors/email/email_mcp_server.py`) are hand-rolled JSON-RPC 2.0 servers over stdio — no MCP SDK / FastMCP. Shared conventions: **stdout is reserved for JSON-RPC**, all logging goes to stderr, a lock guards stdout writes, a background thread emits `event/new_email` push notifications, and each server exposes the same tool shape (`TOOLS` manifest + `TOOL_DISPATCH` + `handle_request`). Mirror this structure for new local connectors.
+- **Gmail** uses the Gmail API with OAuth; today it still reads its token from `./secrets/gmail_creds.json` (override `GMAIL_CREDS_PATH`) generated by `gmail_oauth_setup.py`. The `secrets/` path is **deprecated** — the OAuth phase (skald Fase 2) will migrate Gmail to `{ENV:}`/`{SECRET:}` and a loopback-listener flow. Push = History API polling. `verify` is not yet wired for Gmail.
+- **Email** is a generic IMAP+SMTP connector, **stdlib-only (no dependencies)**, configured entirely from `{ENV:}`/`{SECRET:}` env vars (no `secrets/`), works with any provider. Push = IMAP IDLE with a 60s polling fallback; the request thread and the watcher thread each hold their own IMAP connection (imaplib is not thread-safe). Note: imaplib does not quote SEARCH arguments, so values with spaces must be wrapped via `_q()`. Ships a `verify.py` (IMAP+SMTP login probe).
+- **Tavily** is `mcp_remote`/`global`; its API key is declared as an `env[]` secret (`tavilyApiKey`) and templated into the URL as `{SECRET:tavilyApiKey}`. Ships a `verify.py` (minimal `/search` POST probe).
+- **SSH** is `mcp_local`/`user`; stores aliases in `~/.ssh_aliases.json` (auto-managed, 0600). Auth per-alias: key/agent (default) or elicited password. Sudo via `nopasswd` or elicited password (`sudo -S`). No setup-time credentials: `auth.type: "none"` with no `verify` (activation is direct). All env vars (`SSH_MCP_*`) are optional with defaults. Ships `requirements.txt` (paramiko).
+
+## Critical invariants
+
+- **Editing any shipped connector file (script, icon, requirements, `verify.py`) requires updating its `sha256` AND `size` in `connectors.json`.** A stale hash breaks integrity verification on the client. `connector.json` and `connectors.json` themselves are not hashed, so editing them needs no hash update.
+- **Adding a connector** means changes in two places: a new entry in `connectors.json` (with all file hashes) *and* a new `connectors/<id>/connector.json`. Keep `id`, `scope`, `type`, `tags`, and `requires` consistent between the two.
+- **A new `verify.py` must be added to `files[]`** (with `sha256`+`size`) exactly like any other shipped file — skald refuses to run an unverified script.
+- The `secrets/` directory is **deprecated** in the model (credentials flow through `{ENV:}`/`{SECRET:}`). Gmail still uses it locally pending the OAuth migration; that path is gitignored — never commit credentials or OAuth tokens.
+
+## Commands
+
+Recompute a file hash (do this before deploy for every changed file; SKALD.md mentions a `scripts/update_hashes.py` helper that does not exist in the repo yet):
+```bash
+python3 -c "import hashlib; print(hashlib.sha256(open('connectors/gmail/gmail_mcp_server.py','rb').read()).hexdigest())"
+```
+
+Preview the catalog locally (must serve from the `connectors/` root so `/connectors.json` resolves):
+```bash
+cd connectors && python3 -m http.server 8000   # then open http://localhost:8000/
+```
+
+Set up and run the Gmail connector:
+```bash
+pip install -r connectors/gmail/requirements.txt
+python3 connectors/gmail/gmail_oauth_setup.py    # opens browser for OAuth, writes token
+python3 connectors/gmail/gmail_mcp_server.py      # speaks JSON-RPC on stdin/stdout
+```
+
+Deploy to the production server (Caddy, path `/var/www/connectors.skaldagent.net/`):
+```bash
+rsync -avz --delete ./connectors/ dguiducci@skald-server:/var/www/connectors.skaldagent.net/
+# then fix ownership/permissions on the server:
+sudo chown -R caddy:caddy /var/www/connectors.skaldagent.net/
+sudo find /var/www/connectors.skaldagent.net/ -type f -exec chmod 644 {} \;
+```
+Remote git: `https://git.skaldagent.net/dguiducci/skald-connectors.git` (branch `main`). Live site: `https://connectors.skaldagent.net/`.
diff --git a/SKALD.md b/SKALD.md
index c8ad096..abf9dbc 100644
--- a/SKALD.md
+++ b/SKALD.md
@@ -2,6 +2,8 @@
 
 _Updated: 2026-07-16_
 
+**Remote**: `https://git.skaldagent.net/dguiducci/skald-connectors.git` (branch: `main`)
+
 ## Cos'è
 
 Il **marketplace** è il catalogo dei connector testati per Skald. Ogni connector è un adattatore che permette a Skald di interfacciarsi con un servizio esterno (API, email, calendario, ricerca, messaggistica, ecc.).
@@ -149,7 +151,8 @@ Configurazione tecnica per l'attivazione del connector.
 | `icon_large` | ✅ | Nome file icona nella cartella locale |
 | `launch_command` | solo `mcp_local` | Comando per avviare il server MCP |
 | `transport` | solo `mcp_local` | `stdio` (default) |
-| `dependencies` | consigliato | Dipendenze Python/Node |
+| `dependencies` | consigliato | Dipendenze Python/Node (array vuoto se solo stdlib) |
+| `env` | se `requires` include `ENV` | Variabili d'ambiente che l'utente deve fornire (schema per la UI) — vedi § Campo env |
 | `setup_instructions` | consigliato | Passi per configurare il connector |
 | `mcp_config` | solo `mcp_local` | Configurazione per l'MCP client |
 | `homepage` | opzionale | URL del servizio |
@@ -180,7 +183,8 @@ Configurazione tecnica per l'attivazione del connector.
 | `DOCKER` | Richiede Docker Engine |
 | `NODE` | Richiede Node.js runtime |
 | `PYTHON` | Richiede Python 3 |
-| `SECRETS_DIR` | Richiede file di credenziali in `secrets/` |
+| `SECRETS_DIR` | ❌ **Deprecato** — la cartella `secrets/` è rimossa dal modello; i connector devono usare `ENV`/`SECRET` (vedi § Sintassi placeholder) |
+| `ENV` | Richiede variabili d'ambiente (dichiarate nel campo `env` del manifest) |
 
 ## Campo auth
 
@@ -196,10 +200,137 @@ Struttura che descrive come il connector gestisce l'autenticazione:
 // OAuth2
 {"type": "oauth2",  "provider": "google", "scopes": ["...", "..."]}
 
+// Password / app-password fornita via variabili d'ambiente
+{"type": "password", "delivery": "env"}
+
 // Nessuna autenticazione
 {"type": "none"}
 ```
 
+## Campo env (variabili d'ambiente)
+
+Usato quando `requires` include `ENV`. È un **array** che dichiara le variabili
+d'ambiente che l'utente deve fornire per far funzionare il connector; **nessuna
+credenziale finisce su disco né in `secrets/`** — l'host raccoglie i valori,
+obbliga la compilazione dei campi obbligatori, e li inietta come environment
+nel processo del server MCP al lancio. Il server le legge da `os.environ`.
+
+```json
+"env": [
+  {
+    "name": "EMAIL_IMAP_HOST",   // nome della variabile d'ambiente
+    "label": "IMAP host",         // etichetta per la UI
+    "description": "IMAP server hostname (es. imap.gmail.com)",
+    "required": true,             // se true, l'host deve obbligare la compilazione
+    "secret": false,              // se true, la UI la maschera e la tratta come segreto
+    "example": "imap.gmail.com"   // placeholder/esempio (opzionale)
+  },
+  {
+    "name": "EMAIL_PASSWORD",
+    "label": "Password / app password",
+    "description": "Password o app-password del provider",
+    "required": true,
+    "secret": true,
+    "default": ""                 // valore di default se non obbligatorio (opzionale)
+  }
+]
+```
+
+| Campo | Obbligatorio | Descrizione |
+|-------|-------------|-------------|
+| `name` | ✅ | Nome della variabile d'ambiente (UPPER_SNAKE_CASE) |
+| `label` | ✅ | Etichetta breve per la UI |
+| `description` | ✅ | Testo di aiuto |
+| `required` | ✅ | Se `true`, l'host obbliga l'utente a fornire un valore |
+| `secret` | consigliato | Se `true`, valore sensibile (mascherato, non loggato) |
+| `default` | opzionale | Valore usato se non fornito (solo per non obbligatorie) |
+| `example` | opzionale | Placeholder di esempio per la UI |
+
+## Sintassi placeholder (unificata)
+
+Ogni valore che skald deve riempire a runtime con un dato fornito dall'utente
+usa **uno di due token**, ovunque compaia (URL, `mcp_config.env`, `verify.command`):
+
+| Token | Significato | Esempio |
+|-------|-------------|---------|
+| `{ENV:NAME}` | Variabile non sensibile (hostname, porta, username…) | `{ENV:EMAIL_IMAP_HOST}` |
+| `{SECRET:NAME}` | Variabile sensibile (password, API key, token) | `{SECRET:EMAIL_PASSWORD}` |
+
+`NAME` è il nome dichiarato nell'array `env[]` (campo `name`). skald raccoglie
+i valori tramite un form (maschera i `{SECRET:}`), li inietta come environment
+nel processo del server MCP / verify, e sostituisce i token nel manifest.
+
+Regole:
+- I token non riconosciuti (es. `{secrets}/…`, legacy `{key}`, `{env:NAME}`)
+  sono **deprecati**: skald non li sostituisce e il manifest va aggiornato.
+- `{SECRET:<auth.param>}` è riservato alla chiave primaria quando
+  `auth.type = "api_key"` (es. Tavily: `{SECRET:tavilyApiKey}`). skald tratta
+  quel valore anche come API key per il routing bearer/header.
+- Un token `{ENV:X}` o `{SECRET:X}` la cui `X` non è nell'`env[]` del manifest
+  viene sostituito con stringa vuota (l'host non può indovinarlo).
+
+### Deprecations
+
+| Token | Stato | Sostituzione |
+|-------|-------|--------------|
+| `{key}` | ❌ deprecato | `{SECRET:<auth.param>}` |
+| `{env:NAME}` | ❌ deprecato | `{ENV:NAME}` |
+| `{secrets}/…` | ❌ deprecato | Il connector deve dichiarare il path come `{ENV:…}` (la cartella `secrets/` è rimossa dal modello) |
+
+## Campo verify (test prima del salvataggio)
+
+Dichiara un comando shell che skald esegue **dopo** che l'utente ha compilato
+il form e **prima** di persistere l'attivazione. Serve a verificare che le
+credenziali appena inserite funzionino davvero.
+
+```json
+"verify": {
+  "command": "python3 verify.py",
+  "timeout_secs": 20
+}
+```
+
+| Campo | Obbligatorio | Descrizione |
+|-------|-------------|-------------|
+| `command` | ✅ | Comando shell. Gira nello stesso sandbox del server: container `skald-{userid}` per `mcp_local` user, host per `mcp_remote` global. Le env/secret dichiarati sono iniettate |
+| `timeout_secs` | opzionale | Default 15. skald killa il processo allo scadere |
+
+### Convenzione output
+
+Il comando deve stampare **un singolo oggetto JSON su stdout** e nient'altro:
+
+```json
+{"ok": true,  "message": "IMAP and SMTP authentication successful", "details": {"imap": "...", "smtp": "..."}}
+{"ok": false, "message": "IMAP login failed: INVALID_CREDENTIALS"}
+```
+
+| Campo | Tipo | Descrizione |
+|-------|------|-------------|
+| `ok` | bool | `true` = test passato |
+| `message` | string | Messaggio mostrato all'utente (mai loggare secret qui dentro) |
+| `details` | object | Opzionale, dettagli strutturati mostrati in `<pre>` |
+
+Exit code: 0 su successo, ≠ 0 su fallimento (skald usa l'exit code come
+fallback se il parse JSON fallisce). **Mai stampare credenziali** nel
+`message`/`details`.
+
+### Dove mettere lo script
+
+Se `command` referenzia un file (es. `verify.py`), il file va:
+1. Aggiunto all'array `files[]` in `connectors.json` (con `sha256` e `size`)
+2. Salvato nella cartella del connector (`<id>/verify.py`)
+
+skald lo scarica, ne verifica lo SHA-256 contro l'indice, e lo rende disponibile
+nello stesso path del server principale (container per i `mcp_local`, dir
+`./scripts/<id>/` sull'host per i `mcp_remote`).
+
+### Senza verify
+
+Se `verify` manca, skald **non esegue nessun test** — l'attivazione è diretta
+come oggi. Il connector va in `auth_state='ready'` senza verifica. Per i
+`mcp_remote` non c'è fallback handshake: l'autore del manifest decide se vuole
+il test scrivendo `verify`.
+
 ## Convenzioni icone
 
 - **Formato**: SVG per icone vettoriali (meglio per retina/zoom), PNG per raster
@@ -256,7 +387,12 @@ sudo find /var/www/connectors.skaldagent.net/ -type f -exec chmod 644 {} \;
 
 ## Connector attuali
 
-| ID | Nome | Tipo | Scope |
-|----|------|------|-------|
-| `tavily` | Tavily | `mcp_remote` | `global` |
-| `gmail` | Gmail | `mcp_local` | `user` |
+| ID | Nome | Tipo | Scope | Auth | Verify |
+|----|------|------|-------|------|--------|
+ | `tavily` | Tavily | `mcp_remote` | `global` | api_key (`{SECRET:tavilyApiKey}` in URL) | `verify.py` (HTTP probe `/search`) |
+| `gmail` | Gmail | `mcp_local` | `user` | oauth2 (Google) | ⏳ Fase 2 — il flusso OAuth non è ancora cablato in skald |
+| `email` | Email (IMAP/SMTP) | `mcp_local` | `user` | password (env) | `verify.py` (IMAP+SMTP probe) |
+| `ssh` | SSH Remote Access | `mcp_local` | `user` | none (auth runtime per-alias) | — (nessun setup credential) |
+**Stato del verify-before-save in skald**: `email` e `tavily` hanno `verify` completo
+(script + JSON output); `gmail` aspetta la Fase 2 (OAuth via loopback listener).
+Un connector senza `verify` viene attivato senza test — vedi § Senza verify.
diff --git a/connectors/connectors.json b/connectors/connectors.json
index 98ac64a..8f994ea 100644
--- a/connectors/connectors.json
+++ b/connectors/connectors.json
@@ -48,6 +48,92 @@
         }
       ]
     },
+    {
+      "id": "email",
+      "name": "Email (IMAP/SMTP)",
+      "type": "mcp_local",
+      "scope": "user",
+      "icon_small": "email/icon_sm.svg",
+      "icon_large": "email/icon_lg.svg",
+      "user_description": "Read, search, organize, and send email over IMAP/SMTP — any provider, no OAuth. Real-time push notifications on new mail.",
+      "requires": [
+        "PYTHON",
+        "ENV"
+      ],
+      "tags": [
+        "email",
+        "mcp",
+        "local",
+        "imap",
+        "smtp"
+      ],
+      "folder": "email",
+      "files": [
+        {
+          "path": "email_mcp_server.py",
+          "sha256": "e8e0efc7d07be0d743f73ae80b15c924a763e57fc19000b1ef1e98237465da0f",
+          "size": 54522
+        },
+        {
+          "path": "icon_lg.svg",
+          "sha256": "94128b8ab2a3e701d042b6294c46bf0581152a3fe066e2fb415433e964f6935b",
+          "size": 308
+        },
+        {
+          "path": "icon_sm.svg",
+          "sha256": "05b8d18f8233503bba4b6668dc6c83b2e5d65cb26cfc4901366e2fa43d90737a",
+          "size": 306
+        },
+        {
+          "path": "verify.py",
+          "sha256": "5bd16cf4c1e2fc8b5b58083d12be0999c06dcc48ab52c3db3a531025771f6ff7",
+          "size": 3395
+        }
+      ]
+    },
+    {
+      "id": "ssh",
+      "name": "SSH Remote Access",
+      "type": "mcp_local",
+      "scope": "user",
+      "icon_small": "ssh/icon_sm.svg",
+      "icon_large": "ssh/icon_lg.svg",
+      "user_description": "Full SSH remote server access — files, commands, systemd, upload/download. Aliases manage host config; auth via key/agent or elicited password.",
+      "requires": [
+        "PYTHON"
+      ],
+      "tags": [
+        "ssh",
+        "mcp",
+        "local",
+        "remote",
+        "server",
+        "sysadmin"
+      ],
+      "folder": "ssh",
+      "files": [
+        {
+          "path": "ssh_mcp_server.py",
+          "sha256": "fbd7aa9dfad0da57f4b3d9d437c76fd978145c8364ed5fdfdfc3ba7ccfaf117b",
+          "size": 49764
+        },
+        {
+          "path": "requirements.txt",
+          "sha256": "fa65d44e3d3219c79e5272ef7123352b5d71a35871ede721e66422c897b94e0e",
+          "size": 14
+        },
+        {
+          "path": "icon_sm.svg",
+          "sha256": "57e9e5de2ec56548a8a3821bb29c7f75c8a02f45ae236c8ae3388bb0be9b4a81",
+          "size": 254
+        },
+        {
+          "path": "icon_lg.svg",
+          "sha256": "84dc85685be0fff4cae1c2554b5fe958dd94455e182117c7470838b3cfca6ba1",
+          "size": 432
+        }
+      ]
+    },
     {
       "id": "tavily",
       "name": "Tavily",
@@ -75,6 +161,11 @@
           "path": "icon_sm.png",
           "sha256": "92962ea1d49f272665262d55ecdea929972d3efab5261fc6ffea632803fceaf8",
           "size": 24264
+        },
+        {
+          "path": "verify.py",
+          "sha256": "35a4f3b518c79363893c0438b1932b68e2dc5b4b61303037984488f6967b73a4",
+          "size": 2057
         }
       ]
     }
diff --git a/connectors/email/connector.json b/connectors/email/connector.json
new file mode 100644
index 0000000..7ba2c91
--- /dev/null
+++ b/connectors/email/connector.json
@@ -0,0 +1,131 @@
+{
+  "id": "email",
+  "name": "Email (IMAP/SMTP)",
+  "version": "1.0.0",
+  "type": "mcp_local",
+  "launch_command": "python3 email_mcp_server.py",
+  "transport": "stdio",
+  "requires": [
+    "PYTHON",
+    "ENV"
+  ],
+  "dependencies": [],
+  "env": [
+    {
+      "name": "EMAIL_IMAP_HOST",
+      "label": "IMAP host",
+      "description": "IMAP server hostname for reading mail (e.g. imap.gmail.com, outlook.office365.com, imap.mail.me.com).",
+      "required": true,
+      "secret": false,
+      "example": "imap.gmail.com"
+    },
+    {
+      "name": "EMAIL_IMAP_PORT",
+      "label": "IMAP port",
+      "description": "IMAP-over-SSL port. Almost always 993.",
+      "required": false,
+      "secret": false,
+      "default": "993",
+      "example": "993"
+    },
+    {
+      "name": "EMAIL_SMTP_HOST",
+      "label": "SMTP host",
+      "description": "SMTP server hostname for sending mail (e.g. smtp.gmail.com, smtp.office365.com, smtp.mail.me.com).",
+      "required": true,
+      "secret": false,
+      "example": "smtp.gmail.com"
+    },
+    {
+      "name": "EMAIL_SMTP_PORT",
+      "label": "SMTP port",
+      "description": "SMTP port: 465 for implicit SSL, 587 for STARTTLS.",
+      "required": false,
+      "secret": false,
+      "default": "465",
+      "example": "465"
+    },
+    {
+      "name": "EMAIL_SMTP_SECURITY",
+      "label": "SMTP security",
+      "description": "How to secure the SMTP connection: 'ssl' (port 465), 'starttls' (port 587), or 'plain'. If omitted, inferred from the port.",
+      "required": false,
+      "secret": false,
+      "default": "ssl",
+      "example": "ssl"
+    },
+    {
+      "name": "EMAIL_USERNAME",
+      "label": "Username / email address",
+      "description": "Login username, usually the full email address.",
+      "required": true,
+      "secret": false,
+      "example": "me@example.com"
+    },
+    {
+      "name": "EMAIL_PASSWORD",
+      "label": "Password / app password",
+      "description": "Account password. Providers with 2FA (Gmail, iCloud, Yahoo, Outlook) require an app-specific password, not your normal login password.",
+      "required": true,
+      "secret": true,
+      "example": "abcd efgh ijkl mnop"
+    },
+    {
+      "name": "EMAIL_FROM",
+      "label": "From address",
+      "description": "Address to put in the From header when sending. Defaults to the username if omitted.",
+      "required": false,
+      "secret": false,
+      "example": "me@example.com"
+    }
+  ],
+  "setup_instructions": [
+    "No files on disk and no OAuth: configuration is entirely via environment variables.",
+    "Set EMAIL_IMAP_HOST, EMAIL_SMTP_HOST, EMAIL_USERNAME and EMAIL_PASSWORD (required).",
+    "For providers with 2FA (Gmail, iCloud, Yahoo, Outlook), generate an app-specific password and use it as EMAIL_PASSWORD.",
+    "Optionally set EMAIL_IMAP_PORT (993), EMAIL_SMTP_PORT (465/587), EMAIL_SMTP_SECURITY (ssl/starttls) and EMAIL_FROM.",
+    "Run the 'status' tool to verify IMAP and SMTP both authenticate."
+  ],
+  "docs": [
+    {
+      "lang": "en",
+      "description": "Generic email connector over IMAP + SMTP. Works with any provider (Gmail, Outlook/Office 365, iCloud, Yahoo, Fastmail, self-hosted, corporate) using standard protocols and an app password — no Google Cloud Console, no OAuth. Read, search, organise and send mail, download attachments, and receive real-time push notifications when new mail arrives (IMAP IDLE, with polling fallback). Standard library only, zero dependencies.",
+      "llm_short_description": "Email MCP server (IMAP/SMTP, any provider). Config via env vars. Tools: status, list_messages, get_message, get_thread, list_folders, modify_message, send_message, get_profile, create_folder, download_attachments. Emits event/new_email push notifications on new INBOX mail."
+    }
+  ],
+  "mcp_config": {
+    "command": "python3",
+    "args": [
+      "email_mcp_server.py"
+    ],
+    "env": {
+      "EMAIL_IMAP_HOST": "{ENV:EMAIL_IMAP_HOST}",
+      "EMAIL_IMAP_PORT": "{ENV:EMAIL_IMAP_PORT}",
+      "EMAIL_SMTP_HOST": "{ENV:EMAIL_SMTP_HOST}",
+      "EMAIL_SMTP_PORT": "{ENV:EMAIL_SMTP_PORT}",
+      "EMAIL_SMTP_SECURITY": "{ENV:EMAIL_SMTP_SECURITY}",
+      "EMAIL_USERNAME": "{ENV:EMAIL_USERNAME}",
+      "EMAIL_PASSWORD": "{SECRET:EMAIL_PASSWORD}",
+      "EMAIL_FROM": "{ENV:EMAIL_FROM}"
+    }
+  },
+  "verify": {
+    "command": "python3 verify.py",
+    "timeout_secs": 20
+  },
+  "homepage": "",
+  "icon_small": "icon_sm.svg",
+  "icon_large": "icon_lg.svg",
+  "scope": "user",
+  "tags": [
+    "email",
+    "mcp",
+    "local",
+    "imap",
+    "smtp"
+  ],
+  "auth": {
+    "type": "password",
+    "delivery": "env"
+  }
+}
diff --git a/connectors/email/email_mcp_server.py b/connectors/email/email_mcp_server.py
new file mode 100644
index 0000000..73630ab
--- /dev/null
+++ b/connectors/email/email_mcp_server.py
@@ -0,0 +1,1398 @@
+#!/usr/bin/env python3
+"""Generic Email MCP server — IMAP + SMTP over stdio (JSON-RPC 2.0).
+
+Works with ANY email provider (Gmail, Outlook/Office 365, iCloud, Yahoo,
+Fastmail, self-hosted, corporate…). No Google Cloud Console, no OAuth: just
+standard IMAP for reading/organising and SMTP for sending, with an
+app-password. Standard library only — no pip install required.
+
+Capabilities (callable as `mcp__email__<tool>`):
+  status               — self-check: IMAP + SMTP login and reachability
+  list_messages        — list messages in a folder with a Gmail-like query
+  get_message          — read a single message by UID (body text + attachments)
+  get_thread           — best-effort thread reconstruction (References/Subject)
+  list_folders         — list IMAP folders/mailboxes with total/unread counts
+  modify_message        — flag/unflag, mark read/unread, move, archive, delete
+  send_message         — send an email (in-thread replies + file attachments)
+  get_profile          — configured account + INBOX totals
+  create_folder        — create a new IMAP folder/mailbox
+  download_attachments — save all attachments from a message to disk
+
+Push notifications: a background watcher emits `event/new_email` the moment a
+new message lands in INBOX, using IMAP IDLE when the server advertises it and
+falling back to 60s polling otherwise — mirroring the Gmail connector's event.
+
+Configuration is read entirely from environment variables (nothing on disk):
+  EMAIL_IMAP_HOST      (required)  e.g. imap.gmail.com
+  EMAIL_IMAP_PORT      (default 993, IMAP over SSL)
+  EMAIL_SMTP_HOST      (required)  e.g. smtp.gmail.com
+  EMAIL_SMTP_PORT      (default 465)
+  EMAIL_SMTP_SECURITY  (ssl | starttls | plain; default: ssl if port 465 else starttls)
+  EMAIL_USERNAME       (required)  usually the full email address
+  EMAIL_PASSWORD       (required)  password or provider app-password
+  EMAIL_FROM           (optional)  From address; defaults to EMAIL_USERNAME
+"""
+
+from __future__ import annotations
+
+import email
+import email.utils
+import imaplib
+import json
+import mimetypes
+import os
+import re
+import select
+import smtplib
+import ssl
+import sys
+import threading
+import time
+from email.header import decode_header, make_header
+from email.message import EmailMessage
+from html.parser import HTMLParser
+from typing import Any, Callable
+
+
+# Log to stderr so stdout stays clean for JSON-RPC.
+def log(msg: str) -> None:
+    print(f"[email_mcp] {msg}", file=sys.stderr, flush=True)
+
+
+# Protects all stdout writes (main request thread + push watcher thread).
+_stdout_lock = threading.Lock()
+
+
+# ── Push notifications ──────────────────────────────────────────────────────────
+
+def _emit_notification(method: str, params: dict) -> None:
+    """Write a JSON-RPC notification (no id) to stdout."""
+    msg = json.dumps({"jsonrpc": "2.0", "method": method, "params": params})
+    with _stdout_lock:
+        sys.stdout.write(msg + "\n")
+        sys.stdout.flush()
+
+
+# Re-issue IDLE well within the 29-minute ceiling recommended by RFC 2177.
+_IDLE_REFRESH_SECS = 20 * 60
+_POLL_INTERVAL_SECS = 60
+_RECONNECT_DELAY_SECS = 60
+_watch_thread: threading.Thread | None = None
+
+
+def _start_watching() -> None:
+    """Spin up the background push watcher (its own dedicated IMAP connection)."""
+    global _watch_thread
+    cfg = _get_config()
+    if cfg is None:
+        log(f"Push watcher disabled: {_init_error}")
+        return
+    _watch_thread = threading.Thread(target=_watch_loop, daemon=True, name="email-watch")
+    _watch_thread.start()
+
+
+def _watch_loop() -> None:
+    """Outer reconnect loop: keep a watcher connection alive forever."""
+    cfg = _get_config()
+    if cfg is None:
+        return
+    while True:
+        conn = None
+        try:
+            conn = _connect_imap(cfg)
+            conn.select("INBOX", readonly=True)
+            last_uid = _inbox_uidnext(conn)
+            use_idle = _server_has_idle(conn)
+            log(f"Push watcher started (mode={'IDLE' if use_idle else 'poll'}, "
+                f"next_uid={last_uid}).")
+            while True:
+                if use_idle:
+                    try:
+                        _idle_wait(conn, _IDLE_REFRESH_SECS)
+                    except Exception as e:
+                        log(f"IDLE failed ({e}); falling back to 60s polling.")
+                        use_idle = False
+                        time.sleep(_POLL_INTERVAL_SECS)
+                else:
+                    time.sleep(_POLL_INTERVAL_SECS)
+                    try:
+                        conn.noop()  # refresh the mailbox view before searching
+                    except Exception:
+                        raise  # drop to the reconnect loop
+                last_uid = _emit_new_since(conn, last_uid)
+        except Exception as e:
+            log(f"Push watcher connection error: {_format_imap_error(e)}; "
+                f"reconnecting in {_RECONNECT_DELAY_SECS}s.")
+            if conn is not None:
+                try:
+                    conn.logout()
+                except Exception:
+                    pass
+            time.sleep(_RECONNECT_DELAY_SECS)
+
+
+def _server_has_idle(conn: imaplib.IMAP4) -> bool:
+    try:
+        return any(c.upper() == "IDLE" for c in conn.capabilities)
+    except Exception:
+        return False
+
+
+def _inbox_uidnext(conn: imaplib.IMAP4) -> int:
+    """UID that will be assigned to the next new message (our 'new mail' cursor)."""
+    try:
+        typ, data = conn.status("INBOX", "(UIDNEXT)")
+        if typ == "OK" and data and data[0]:
+            m = re.search(rb"UIDNEXT\s+(\d+)", data[0])
+            if m:
+                return int(m.group(1))
+    except Exception:
+        pass
+    return 1
+
+
+def _idle_wait(conn: imaplib.IMAP4, timeout: int) -> None:
+    """Enter IMAP IDLE and block until the server reports activity or `timeout`.
+
+    Implemented by hand because imaplib only grew a native idle() in Python 3.13.
+    We reuse imaplib's tag counter (_new_tag) so its internal state stays
+    consistent, and always send DONE + drain the tagged completion before
+    returning so the connection is usable for a follow-up UID SEARCH.
+    """
+    tag = conn._new_tag()  # type: ignore[attr-defined]
+    conn.send(tag + b" IDLE\r\n")
+    resp = conn.readline()
+    if not resp.lstrip().startswith(b"+"):
+        raise RuntimeError(f"server refused IDLE: {resp!r}")
+    try:
+        ready, _, _ = select.select([conn.sock], [], [], timeout)
+        if ready:
+            conn.readline()  # consume the untagged EXISTS/RECENT push
+    finally:
+        conn.send(b"DONE\r\n")
+        deadline = time.time() + 10
+        while time.time() < deadline:
+            line = conn.readline()
+            if not line or line.startswith(tag):
+                break
+
+
+def _emit_new_since(conn: imaplib.IMAP4, last_uid: int) -> int:
+    """Emit event/new_email for every INBOX message with UID >= last_uid."""
+    try:
+        typ, data = conn.uid("SEARCH", None, f"UID {last_uid}:*")
+    except Exception as e:
+        log(f"new-mail search failed: {_format_imap_error(e)}")
+        return last_uid
+    if typ != "OK" or not data or not data[0]:
+        return last_uid
+    # "UID n:*" always returns the highest message even if its UID < n, so filter.
+    uids = sorted(u for u in (int(x) for x in data[0].split()) if u >= last_uid)
+    for uid in uids:
+        _fetch_and_emit(conn, uid)
+        last_uid = uid + 1
+    return last_uid
+
+
+def _fetch_and_emit(conn: imaplib.IMAP4, uid: int) -> None:
+    """Fetch a new message's headers and emit an event/new_email notification."""
+    try:
+        typ, data = conn.uid(
+            "FETCH", str(uid),
+            "(BODY.PEEK[HEADER.FIELDS (SUBJECT FROM DATE MESSAGE-ID)])",
+        )
+        if typ != "OK":
+            return
+        headers = _parse_header_bytes(_first_literal(data))
+        _emit_notification("event/new_email", {
+            "message_id": str(uid),
+            "folder":     "INBOX",
+            "thread_id":  headers.get("message-id", ""),
+            "subject":    headers.get("subject", "(no subject)"),
+            "from":       headers.get("from", "?"),
+            "date":       headers.get("date", "?"),
+            "snippet":    "",
+        })
+        log(f"Notification emitted: new email uid={uid} from {headers.get('from', '?')!r}")
+    except Exception as e:
+        log(f"Failed to emit notification for uid {uid}: {_format_imap_error(e)}")
+
+
+# ── Configuration (environment only) ─────────────────────────────────────────────
+
+_config: dict | None = None
+_config_loaded = False
+_init_error: str | None = None
+
+
+def _get_config() -> dict | None:
+    """Load and validate config from environment variables (once)."""
+    global _config, _config_loaded, _init_error
+    if _config_loaded:
+        return _config
+    _config_loaded = True
+
+    missing = [k for k in ("EMAIL_IMAP_HOST", "EMAIL_SMTP_HOST", "EMAIL_USERNAME", "EMAIL_PASSWORD")
+               if not os.environ.get(k)]
+    if missing:
+        _init_error = (
+            "Missing required environment variable(s): " + ", ".join(missing) + ". "
+            "Set EMAIL_IMAP_HOST, EMAIL_SMTP_HOST, EMAIL_USERNAME and EMAIL_PASSWORD "
+            "(plus optional EMAIL_IMAP_PORT / EMAIL_SMTP_PORT / EMAIL_SMTP_SECURITY / EMAIL_FROM)."
+        )
+        log(_init_error)
+        return None
+
+    def _int(name: str, default: int) -> int:
+        raw = os.environ.get(name)
+        if not raw:
+            return default
+        try:
+            return int(raw)
+        except ValueError:
+            log(f"Invalid {name}={raw!r}; using default {default}.")
+            return default
+
+    smtp_port = _int("EMAIL_SMTP_PORT", 465)
+    security = (os.environ.get("EMAIL_SMTP_SECURITY") or "").strip().lower()
+    if security not in ("ssl", "starttls", "plain"):
+        security = "ssl" if smtp_port == 465 else "starttls"
+
+    username = os.environ["EMAIL_USERNAME"]
+    _config = {
+        "imap_host": os.environ["EMAIL_IMAP_HOST"],
+        "imap_port": _int("EMAIL_IMAP_PORT", 993),
+        "smtp_host": os.environ["EMAIL_SMTP_HOST"],
+        "smtp_port": smtp_port,
+        "smtp_security": security,
+        "username": username,
+        "password": os.environ["EMAIL_PASSWORD"],
+        "from_addr": os.environ.get("EMAIL_FROM") or username,
+    }
+    log(f"Config loaded for {username} (imap {_config['imap_host']}:{_config['imap_port']}, "
+        f"smtp {_config['smtp_host']}:{smtp_port}/{security}).")
+    return _config
+
+
+# ── IMAP connection (request thread) ─────────────────────────────────────────────
+
+# Single connection reused by the request-handling thread. The main loop reads
+# stdin sequentially and dispatches synchronously, so this is only ever touched
+# by one thread; the push watcher keeps its OWN separate connection.
+_imap_conn: imaplib.IMAP4 | None = None
+
+
+def _connect_imap(cfg: dict) -> imaplib.IMAP4_SSL:
+    """Open and authenticate a fresh IMAP-over-SSL connection (raises on failure)."""
+    conn = imaplib.IMAP4_SSL(cfg["imap_host"], cfg["imap_port"],
+                             ssl_context=ssl.create_default_context())
+    conn.login(cfg["username"], cfg["password"])
+    return conn
+
+
+def _imap() -> imaplib.IMAP4 | None:
+    """Return a healthy IMAP connection for the request thread, or None (+ _init_error)."""
+    global _imap_conn, _init_error
+    cfg = _get_config()
+    if cfg is None:
+        return None
+    if _imap_conn is not None:
+        try:
+            _imap_conn.noop()
+            return _imap_conn
+        except Exception:
+            try:
+                _imap_conn.logout()
+            except Exception:
+                pass
+            _imap_conn = None
+    try:
+        _imap_conn = _connect_imap(cfg)
+        return _imap_conn
+    except Exception as e:
+        _init_error = _format_imap_error(e)
+        log(_init_error)
+        return None
+
+
+def _select(conn: imaplib.IMAP4, folder: str, readonly: bool = True) -> tuple[bool, str]:
+    """SELECT a folder; return (ok, error_message)."""
+    typ, data = conn.select(_quote_mailbox(folder), readonly=readonly)
+    if typ != "OK":
+        detail = data[0].decode("utf-8", "replace") if data and data[0] else "unknown error"
+        return False, f"Error: cannot open folder {folder!r}: {detail}"
+    return True, ""
+
+
+def _quote_mailbox(name: str) -> str:
+    """Quote a mailbox name for IMAP if it contains spaces/specials."""
+    if name and re.fullmatch(r"[A-Za-z0-9_./\-]+", name):
+        return name
+    return '"' + name.replace("\\", "\\\\").replace('"', '\\"') + '"'
+
+
+# ── Error mapping ────────────────────────────────────────────────────────────────
+
+def _format_imap_error(e: Exception) -> str:
+    """Map an IMAP/socket exception into an actionable Error: string."""
+    text = str(e).strip()
+    low = text.lower()
+    if isinstance(e, imaplib.IMAP4.error):
+        if "authentication" in low or "login" in low or "credentials" in low or "auth" in low:
+            return ("Error: IMAP login was rejected. Check EMAIL_USERNAME / EMAIL_PASSWORD — "
+                    "most providers require an app-specific password (not your normal login "
+                    "password) and IMAP to be enabled in account settings.")
+        return f"Error: IMAP error: {text}"
+    if isinstance(e, (TimeoutError, ConnectionError, OSError)):
+        return (f"Error: could not reach the IMAP server ({text}). Check EMAIL_IMAP_HOST / "
+                "EMAIL_IMAP_PORT and your network.")
+    return f"Error: IMAP call failed: {text}"
+
+
+def _format_smtp_error(e: Exception) -> str:
+    text = str(e).strip()
+    if isinstance(e, smtplib.SMTPAuthenticationError):
+        return ("Error: SMTP login was rejected. Check EMAIL_USERNAME / EMAIL_PASSWORD — an "
+                "app-specific password is usually required for SMTP too.")
+    if isinstance(e, smtplib.SMTPException):
+        return f"Error: SMTP error: {text}"
+    if isinstance(e, (TimeoutError, ConnectionError, OSError)):
+        return (f"Error: could not reach the SMTP server ({text}). Check EMAIL_SMTP_HOST / "
+                "EMAIL_SMTP_PORT / EMAIL_SMTP_SECURITY.")
+    return f"Error: SMTP call failed: {text}"
+
+
+def _unavailable() -> str:
+    """Error string for tools when config/connection isn't ready (no double 'Error:')."""
+    msg = _init_error or "Email connector is not configured."
+    return msg if msg.startswith("Error:") else f"Error: {msg}"
+
+
+def _status_report(icon: str, label: str, kind: str, description: str,
+                   steps: list[str] | None = None) -> str:
+    lines = [f"Status: {label} {icon} ({kind})", description]
+    if steps:
+        lines.append("")
+        lines.append("What to do:")
+        for i, s in enumerate(steps, 1):
+            lines.append(f"{i}. {s}")
+    return "\n".join(lines)
+
+
+# ── MIME / parsing helpers ───────────────────────────────────────────────────────
+
+def _decode_hdr(value: str | None) -> str:
+    """Decode a possibly MIME-encoded header (=?utf-8?...?=) to a plain string."""
+    if not value:
+        return ""
+    try:
+        return str(make_header(decode_header(value)))
+    except Exception:
+        return value
+
+
+def _first_literal(data: Any) -> bytes:
+    """Return the first literal ({N}-prefixed) byte payload from a FETCH response."""
+    if not data:
+        return b""
+    for item in data:
+        if isinstance(item, tuple) and len(item) >= 2 and item[1] is not None:
+            return item[1]
+    return b""
+
+
+def _parse_header_bytes(raw: bytes) -> dict[str, str]:
+    """Parse RFC822 header bytes into a lowercased, MIME-decoded dict."""
+    if not raw:
+        return {}
+    msg = email.message_from_bytes(raw)
+    out: dict[str, str] = {}
+    for key in msg.keys():
+        out[key.lower()] = _decode_hdr(msg.get(key))
+    return out
+
+
+def _part_text(part: email.message.Message) -> str:
+    payload = part.get_payload(decode=True)
+    if payload is None:
+        return ""
+    charset = part.get_content_charset() or "utf-8"
+    try:
+        return payload.decode(charset, errors="replace")
+    except (LookupError, TypeError):
+        return payload.decode("utf-8", errors="replace")
+
+
+def _is_attachment(part: email.message.Message) -> bool:
+    disp = str(part.get("Content-Disposition") or "").lower()
+    if "attachment" in disp:
+        return True
+    return bool(part.get_filename())
+
+
+def _extract_body(msg: email.message.Message) -> tuple[str, bool]:
+    """Return (text, from_html). Prefer text/plain; fall back to text/html→text."""
+    if msg.is_multipart():
+        plain = html = ""
+        for part in msg.walk():
+            if part.is_multipart() or _is_attachment(part):
+                continue
+            ctype = part.get_content_type()
+            if ctype == "text/plain" and not plain:
+                plain = _part_text(part)
+            elif ctype == "text/html" and not html:
+                html = _part_text(part)
+        if plain:
+            return plain, False
+        if html:
+            return _html_to_text(html), True
+        return "", False
+    text = _part_text(msg)
+    if msg.get_content_type() == "text/html":
+        return _html_to_text(text), True
+    return text, False
+
+
+def _iter_attachments(msg: email.message.Message):
+    for part in msg.walk():
+        if part.is_multipart():
+            continue
+        if _is_attachment(part):
+            yield part
+
+
+class _HTMLTextExtractor(HTMLParser):
+    """Collect readable text from HTML, skipping scripts/styles, block newlines."""
+
+    _SKIP = {"script", "style", "head"}
+    _BLOCK = {"p", "div", "br", "li", "tr", "h1", "h2", "h3", "h4", "h5", "h6"}
+
+    def __init__(self) -> None:
+        super().__init__(convert_charrefs=True)
+        self._chunks: list[str] = []
+        self._skip_depth = 0
+
+    def handle_starttag(self, tag: str, attrs: Any) -> None:
+        if tag in self._SKIP:
+            self._skip_depth += 1
+        elif tag == "br":
+            self._chunks.append("\n")
+
+    def handle_endtag(self, tag: str) -> None:
+        if tag in self._SKIP and self._skip_depth:
+            self._skip_depth -= 1
+        elif tag in self._BLOCK:
+            self._chunks.append("\n")
+
+    def handle_data(self, data: str) -> None:
+        if not self._skip_depth:
+            self._chunks.append(data)
+
+    def get_text(self) -> str:
+        return re.sub(r"\n{3,}", "\n\n", "".join(self._chunks)).strip()
+
+
+def _html_to_text(html_str: str) -> str:
+    try:
+        parser = _HTMLTextExtractor()
+        parser.feed(html_str)
+        return parser.get_text()
+    except Exception:
+        return html_str
+
+
+# ── Gmail-like search → IMAP SEARCH criteria ─────────────────────────────────────
+
+def _imap_date(value: str) -> str | None:
+    """Convert YYYY-MM-DD (or DD-Mon-YYYY) into an IMAP date (01-Jan-2024)."""
+    value = value.strip()
+    for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%d-%b-%Y"):
+        try:
+            return time.strftime("%d-%b-%Y", time.strptime(value, fmt))
+        except ValueError:
+            continue
+    return None
+
+
+def _q(s: str) -> str:
+    """IMAP quoted-string. imaplib does NOT quote SEARCH args, so we must — any
+    value with a space (TEXT/SUBJECT phrases) is otherwise split into atoms."""
+    return '"' + s.replace("\\", "\\\\").replace('"', '\\"') + '"'
+
+
+def _build_search(query: str, unread_only: bool) -> list[str]:
+    """Turn a Gmail-ish query into IMAP SEARCH tokens.
+
+    Supported: from:x  to:x  subject:x  since:YYYY-MM-DD  before:YYYY-MM-DD
+    unread / is:unread / read / is:read  has:attachment  and free text (→ TEXT).
+    """
+    criteria: list[str] = []
+    free: list[str] = []
+    for tok in (query or "").split():
+        low = tok.lower()
+        if low in ("unread", "is:unread"):
+            criteria += ["UNSEEN"]
+        elif low in ("read", "is:read"):
+            criteria += ["SEEN"]
+        elif low in ("has:attachment", "is:attachment"):
+            criteria += ["KEYWORD", "attachment"]  # best-effort; not all servers honour it
+        elif low.startswith("from:"):
+            criteria += ["FROM", _q(tok[5:])]
+        elif low.startswith("to:"):
+            criteria += ["TO", _q(tok[3:])]
+        elif low.startswith("subject:"):
+            criteria += ["SUBJECT", _q(tok[8:])]
+        elif low.startswith("since:"):
+            d = _imap_date(tok[6:])
+            if d:
+                criteria += ["SINCE", d]
+        elif low.startswith("before:"):
+            d = _imap_date(tok[7:])
+            if d:
+                criteria += ["BEFORE", d]
+        else:
+            free.append(tok)
+    if free:
+        criteria += ["TEXT", _q(" ".join(free))]
+    if unread_only and "UNSEEN" not in criteria:
+        criteria += ["UNSEEN"]
+    return criteria or ["ALL"]
+
+
+# ── Tool implementations ─────────────────────────────────────────────────────────
+
+def _email_status(args: dict | None = None) -> str:
+    """Self-check: IMAP login + SMTP login both succeed."""
+    cfg = _get_config()
+    if cfg is None:
+        return _status_report("❌", "NOT_CONFIGURED", "action needed",
+            _init_error or "Configuration is incomplete.",
+            ["Set the required environment variables: EMAIL_IMAP_HOST, EMAIL_SMTP_HOST, "
+             "EMAIL_USERNAME, EMAIL_PASSWORD.",
+             "Use an app-specific password if your provider requires one (Gmail, iCloud, Yahoo…)."])
+
+    # IMAP probe.
+    conn = _imap()
+    if conn is None:
+        return _status_report("❌", "IMAP_ERROR", "action needed",
+            _init_error or "Could not connect to IMAP.",
+            ["Verify EMAIL_IMAP_HOST / EMAIL_IMAP_PORT and that IMAP is enabled for the account.",
+             "Verify EMAIL_USERNAME / EMAIL_PASSWORD (app password may be required)."])
+    try:
+        conn.select("INBOX", readonly=True)
+    except Exception as e:
+        return _status_report("❌", "IMAP_ERROR", "action needed",
+            _format_imap_error(e), ["Check IMAP settings and credentials."])
+
+    # SMTP probe (connect + login, then quit).
+    try:
+        smtp = _open_smtp(cfg)
+        smtp.quit()
+    except Exception as e:
+        return _status_report("⚠️", "IMAP_OK_SMTP_ERROR", "partial",
+            f"IMAP works, but SMTP login failed: {_format_smtp_error(e)}",
+            ["Reading works; sending will not until SMTP is fixed.",
+             "Check EMAIL_SMTP_HOST / EMAIL_SMTP_PORT / EMAIL_SMTP_SECURITY and the password."])
+
+    return _status_report("✅", "READY", "ok",
+        "Email integration is operational: IMAP and SMTP both authenticate. All tools "
+        "(list/get/thread/folders/modify/send/download) are usable.\n"
+        f"Account: {cfg['username']} (IMAP {cfg['imap_host']}, SMTP {cfg['smtp_host']})")
+
+
+def _fetch_summary(conn: imaplib.IMAP4, uid: int) -> str:
+    """One-message summary line for list_messages (headers + read/flag state)."""
+    typ, data = conn.uid(
+        "FETCH", str(uid),
+        "(FLAGS BODY.PEEK[HEADER.FIELDS (SUBJECT FROM DATE)])",
+    )
+    if typ != "OK":
+        return f"- uid {uid} (error fetching)"
+    flags = imaplib.ParseFlags(data[0][0]) if data and isinstance(data[0], tuple) else ()
+    flag_names = {f.decode("ascii", "replace").lstrip("\\").lower() for f in flags}
+    headers = _parse_header_bytes(_first_literal(data))
+    unread = "seen" not in flag_names
+    marks = []
+    if unread:
+        marks.append("UNREAD")
+    if "flagged" in flag_names:
+        marks.append("★")
+    mark_str = (" [" + ", ".join(marks) + "]") if marks else ""
+    return (f"- {headers.get('subject', '(no subject)')}{mark_str}\n"
+            f"  From: {headers.get('from', '?')} | Date: {headers.get('date', '?')} | UID: {uid}")
+
+
+def _email_list_messages(args: dict) -> str:
+    conn = _imap()
+    if conn is None:
+        return _unavailable()
+
+    folder = args.get("folder", "INBOX")
+    query = args.get("query", "")
+    unread_only = bool(args.get("unread_only", False))
+    max_results = min(int(args.get("max_results", 20) or 20), 50)
+
+    ok, err = _select(conn, folder, readonly=True)
+    if not ok:
+        return err
+
+    criteria = _build_search(query, unread_only)
+    try:
+        typ, data = conn.uid("SEARCH", None, *criteria)
+    except Exception as e:
+        return _format_imap_error(e)
+    if typ != "OK":
+        return f"Error: IMAP SEARCH failed in {folder!r}."
+
+    uids = [int(x) for x in data[0].split()] if data and data[0] else []
+    if not uids:
+        return f"No messages found in {folder!r}."
+
+    # Newest first, capped.
+    uids = sorted(uids, reverse=True)[:max_results]
+    lines = [f"Messages in {folder!r} ({len(uids)} shown, newest first):"]
+    for uid in uids:
+        try:
+            lines.append(_fetch_summary(conn, uid))
+        except Exception as e:
+            lines.append(f"- uid {uid} (error: {e})")
+    lines.append("\nUse get_message with the UID (and folder if not INBOX) to read a message.")
+    return "\n".join(lines)
+
+
+def _fetch_full(conn: imaplib.IMAP4, uid: int) -> email.message.Message | None:
+    typ, data = conn.uid("FETCH", str(uid), "(BODY.PEEK[])")
+    if typ != "OK":
+        return None
+    raw = _first_literal(data)
+    if not raw:
+        return None
+    return email.message_from_bytes(raw)
+
+
+def _email_get_message(args: dict) -> str:
+    conn = _imap()
+    if conn is None:
+        return _unavailable()
+
+    uid = args.get("message_id")
+    if not uid:
+        return "Error: Missing required parameter 'message_id' (the message UID)."
+    folder = args.get("folder", "INBOX")
+    include_body = args.get("include_body", True)
+
+    ok, err = _select(conn, folder, readonly=True)
+    if not ok:
+        return err
+
+    try:
+        msg = _fetch_full(conn, int(uid))
+    except ValueError:
+        return f"Error: message_id must be a numeric UID, got {uid!r}."
+    except Exception as e:
+        return _format_imap_error(e)
+    if msg is None:
+        return f"Error: message UID {uid} not found in folder {folder!r}."
+
+    lines = [
+        f"UID: {uid}",
+        f"Folder: {folder}",
+        f"From: {_decode_hdr(msg.get('From'))}",
+        f"To: {_decode_hdr(msg.get('To'))}",
+        f"Date: {_decode_hdr(msg.get('Date'))}",
+        f"Subject: {_decode_hdr(msg.get('Subject')) or '(no subject)'}",
+        f"Message-ID: {msg.get('Message-ID', '?')}",
+    ]
+
+    attachments = [_decode_hdr(p.get_filename()) for p in _iter_attachments(msg)]
+    if attachments:
+        lines.append(f"Attachments: {', '.join(a for a in attachments if a)}")
+
+    if include_body:
+        body_text, from_html = _extract_body(msg)
+        if body_text:
+            label = "--- Body (converted from HTML) ---" if from_html else "--- Body ---"
+            lines.append("\n" + label)
+            if len(body_text) > 10000:
+                lines.append(body_text[:10000] + "\n... [truncated at 10000 chars]")
+            else:
+                lines.append(body_text)
+        else:
+            lines.append("\n(no text body found)")
+
+    return "\n".join(lines)
+
+
+def _email_get_thread(args: dict) -> str:
+    """Best-effort thread: group messages by shared Message-ID references / subject."""
+    conn = _imap()
+    if conn is None:
+        return _unavailable()
+
+    uid = args.get("message_id")
+    if not uid:
+        return "Error: Missing required parameter 'message_id' (a message UID in the thread)."
+    folder = args.get("folder", "INBOX")
+
+    ok, err = _select(conn, folder, readonly=True)
+    if not ok:
+        return err
+
+    try:
+        seed = _fetch_full(conn, int(uid))
+    except ValueError:
+        return f"Error: message_id must be a numeric UID, got {uid!r}."
+    except Exception as e:
+        return _format_imap_error(e)
+    if seed is None:
+        return f"Error: message UID {uid} not found in folder {folder!r}."
+
+    msg_id = (seed.get("Message-ID") or "").strip()
+    subject = _decode_hdr(seed.get("Subject"))
+    base_subject = re.sub(r"(?i)^\s*(re|fwd|fw|r|aw|antw|sv)\s*:\s*", "", subject).strip()
+
+    # Collect candidate UIDs: same base subject, plus anything referencing this Message-ID.
+    found: set[int] = {int(uid)}
+    try:
+        if base_subject:
+            typ, data = conn.uid("SEARCH", None, "SUBJECT", _q(base_subject))
+            if typ == "OK" and data and data[0]:
+                found.update(int(x) for x in data[0].split())
+        if msg_id:
+            for field in ("In-Reply-To", "References"):
+                typ, data = conn.uid("SEARCH", None, "HEADER", field, _q(msg_id))
+                if typ == "OK" and data and data[0]:
+                    found.update(int(x) for x in data[0].split())
+    except Exception as e:
+        log(f"thread search partial failure: {_format_imap_error(e)}")
+
+    ordered = sorted(found)
+    lines = [f"Thread (best-effort) around UID {uid} in {folder!r} — {len(ordered)} message(s):"]
+    if base_subject:
+        lines.insert(1, f"Subject: {base_subject}")
+    for muid in ordered:
+        try:
+            typ, data = conn.uid(
+                "FETCH", str(muid),
+                "(BODY.PEEK[HEADER.FIELDS (SUBJECT FROM DATE)])",
+            )
+            h = _parse_header_bytes(_first_literal(data))
+            lines.append(f"\n[UID {muid}] From: {h.get('from', '?')} | Date: {h.get('date', '?')}")
+            lines.append(f"    Subject: {h.get('subject', '(no subject)')}")
+        except Exception as e:
+            lines.append(f"\n[UID {muid}] (error: {e})")
+    return "\n".join(lines)
+
+
+def _email_list_folders(args: dict) -> str:
+    conn = _imap()
+    if conn is None:
+        return _unavailable()
+
+    try:
+        typ, data = conn.list()
+    except Exception as e:
+        return _format_imap_error(e)
+    if typ != "OK" or not data:
+        return "No folders found."
+
+    lines = ["Folders:"]
+    for raw in data:
+        if raw is None:
+            continue
+        name = _parse_list_mailbox(raw)
+        if not name:
+            continue
+        counts = ""
+        try:
+            typ2, st = conn.status(_quote_mailbox(name), "(MESSAGES UNSEEN)")
+            if typ2 == "OK" and st and st[0]:
+                total = re.search(rb"MESSAGES\s+(\d+)", st[0])
+                unseen = re.search(rb"UNSEEN\s+(\d+)", st[0])
+                counts = (f" — {int(total.group(1)) if total else '?'} total, "
+                          f"{int(unseen.group(1)) if unseen else '?'} unread")
+        except Exception:
+            pass
+        lines.append(f"- {name}{counts}")
+    return "\n".join(lines)
+
+
+def _parse_list_mailbox(raw: bytes) -> str:
+    """Extract the mailbox name from a LIST response line."""
+    text = raw.decode("utf-8", "replace") if isinstance(raw, bytes) else str(raw)
+    # Format: (\HasNoChildren) "/" "Folder Name"  — name is the last token, maybe quoted.
+    m = re.search(r'"(?:[^"\\]|\\.)*"\s*$', text)
+    if m:
+        return m.group(0)[1:-1].replace('\\"', '"').replace("\\\\", "\\")
+    parts = text.rsplit(" ", 1)
+    return parts[-1].strip() if parts else ""
+
+
+# Friendly flag name → IMAP system flag.
+_FLAG_MAP = {
+    "read": "\\Seen", "seen": "\\Seen",
+    "flagged": "\\Flagged", "starred": "\\Flagged", "star": "\\Flagged",
+    "answered": "\\Answered", "draft": "\\Draft", "deleted": "\\Deleted",
+}
+
+
+def _map_flags(names: Any) -> list[str]:
+    if isinstance(names, str):
+        names = [names]
+    out = []
+    for n in names or []:
+        out.append(_FLAG_MAP.get(str(n).strip().lower(), n))
+    return out
+
+
+def _expunge_uid(conn: imaplib.IMAP4, uid: str) -> None:
+    """Expunge one message. Uses UID EXPUNGE (UIDPLUS) so we don't remove other
+    \\Deleted messages in the folder; falls back to a full EXPUNGE otherwise."""
+    if any(c.upper() == "UIDPLUS" for c in conn.capabilities):
+        try:
+            conn.uid("EXPUNGE", uid)
+            return
+        except Exception:
+            pass
+    conn.expunge()
+
+
+def _email_modify_message(args: dict) -> str:
+    """Add/remove flags, mark read/unread, move to another folder, or delete."""
+    conn = _imap()
+    if conn is None:
+        return _unavailable()
+
+    uid = args.get("message_id")
+    if not uid:
+        return "Error: Missing required parameter 'message_id' (the message UID)."
+    folder = args.get("folder", "INBOX")
+    add_flags = _map_flags(args.get("add_flags"))
+    remove_flags = _map_flags(args.get("remove_flags"))
+    move_to = args.get("move_to_folder")
+
+    # Convenience booleans.
+    if args.get("mark_read"):
+        add_flags.append("\\Seen")
+    if args.get("mark_unread"):
+        remove_flags.append("\\Seen")
+
+    ok, err = _select(conn, folder, readonly=False)
+    if not ok:
+        return err
+
+    changes = []
+    try:
+        if add_flags:
+            conn.uid("STORE", str(uid), "+FLAGS", "(" + " ".join(add_flags) + ")")
+            changes.append(f"added flags {add_flags}")
+        if remove_flags:
+            conn.uid("STORE", str(uid), "-FLAGS", "(" + " ".join(remove_flags) + ")")
+            changes.append(f"removed flags {remove_flags}")
+
+        if move_to:
+            # Prefer server-side MOVE; fall back to COPY + \Deleted + EXPUNGE.
+            # Gate MOVE on both the server capability AND imaplib knowing the verb
+            # (older Python builds lack MOVE in imaplib.Commands).
+            can_move = ("MOVE" in imaplib.Commands) and any(c.upper() == "MOVE" for c in conn.capabilities)
+            if can_move:
+                conn.uid("MOVE", str(uid), _quote_mailbox(move_to))
+            else:
+                conn.uid("COPY", str(uid), _quote_mailbox(move_to))
+                conn.uid("STORE", str(uid), "+FLAGS", "(\\Deleted)")
+                _expunge_uid(conn, str(uid))
+            changes.append(f"moved to {move_to!r}")
+        elif "\\Deleted" in add_flags:
+            _expunge_uid(conn, str(uid))
+            changes.append("expunged")
+    except Exception as e:
+        return _format_imap_error(e)
+
+    if not changes:
+        return ("Nothing to do: pass add_flags/remove_flags (e.g. 'read', 'flagged', 'deleted'), "
+                "mark_read/mark_unread, or move_to_folder.")
+    return f"✅ Message UID {uid} in {folder!r}: {'; '.join(changes)}"
+
+
+_MAX_ATTACHMENT_BYTES = 25 * 1024 * 1024
+
+
+def _open_smtp(cfg: dict) -> smtplib.SMTP:
+    """Open and authenticate an SMTP connection per the configured security mode."""
+    ctx = ssl.create_default_context()
+    if cfg["smtp_security"] == "ssl":
+        smtp: smtplib.SMTP = smtplib.SMTP_SSL(cfg["smtp_host"], cfg["smtp_port"], context=ctx, timeout=30)
+    else:
+        smtp = smtplib.SMTP(cfg["smtp_host"], cfg["smtp_port"], timeout=30)
+        smtp.ehlo()
+        if cfg["smtp_security"] == "starttls":
+            smtp.starttls(context=ctx)
+            smtp.ehlo()
+    smtp.login(cfg["username"], cfg["password"])
+    return smtp
+
+
+def _append_to_sent(conn: imaplib.IMAP4, raw: bytes) -> None:
+    """Best-effort: save a sent message into the account's Sent folder."""
+    sent = None
+    try:
+        typ, data = conn.list()
+        if typ == "OK":
+            for line in data or []:
+                if line and b"\\Sent" in line:
+                    sent = _parse_list_mailbox(line)
+                    break
+    except Exception:
+        pass
+    candidates = [sent] if sent else []
+    candidates += ["Sent", "Sent Items", "Sent Mail", "[Gmail]/Sent Mail", "INBOX.Sent"]
+    for name in candidates:
+        if not name:
+            continue
+        try:
+            typ, _ = conn.append(_quote_mailbox(name), "(\\Seen)",
+                                 imaplib.Time2Internaldate(time.time()), raw)
+            if typ == "OK":
+                return
+        except Exception:
+            continue
+
+
+def _email_send_message(args: dict) -> str:
+    conn_cfg = _get_config()
+    if conn_cfg is None:
+        return _unavailable()
+
+    to = args.get("to")
+    if not to:
+        return "Error: Missing required parameter 'to'."
+    subject = args.get("subject", "")
+    body_text = args.get("body", "")
+    cc = args.get("cc")
+    bcc = args.get("bcc")
+    in_reply_to = args.get("in_reply_to")
+    attachments = args.get("attachments") or []
+    if isinstance(attachments, str):
+        attachments = [attachments]
+
+    # Resolve attachment paths (absolute or relative to this connector's folder).
+    root = os.path.dirname(os.path.abspath(__file__))
+    resolved: list[str] = []
+    total = 0
+    for raw_path in attachments:
+        path = raw_path if os.path.isabs(raw_path) else os.path.join(root, raw_path)
+        if not os.path.isfile(path):
+            return f"Error: attachment not found: {raw_path}"
+        total += os.path.getsize(path)
+        resolved.append(path)
+    if total > _MAX_ATTACHMENT_BYTES:
+        return (f"Error: attachments total ~{total // (1024 * 1024)} MB, over the "
+                f"{_MAX_ATTACHMENT_BYTES // (1024 * 1024)} MB limit.")
+
+    msg = EmailMessage()
+    msg["From"] = conn_cfg["from_addr"]
+    msg["To"] = to
+    if cc:
+        msg["Cc"] = cc
+    if bcc:
+        msg["Bcc"] = bcc
+    msg["Subject"] = subject
+    msg["Date"] = email.utils.formatdate(localtime=True)
+    msg["Message-ID"] = email.utils.make_msgid()
+    if in_reply_to:
+        ref = in_reply_to if in_reply_to.startswith("<") else f"<{in_reply_to}>"
+        msg["In-Reply-To"] = ref
+        msg["References"] = ref
+    msg.set_content(body_text)
+
+    for path in resolved:
+        ctype, encoding = mimetypes.guess_type(path)
+        if ctype is None or encoding is not None:
+            ctype = "application/octet-stream"
+        maintype, subtype = ctype.split("/", 1)
+        try:
+            with open(path, "rb") as f:
+                data = f.read()
+        except Exception as e:
+            return f"Error: could not read attachment {path}: {e}"
+        msg.add_attachment(data, maintype=maintype, subtype=subtype,
+                           filename=os.path.basename(path))
+
+    # Recipient list includes Cc/Bcc for the envelope.
+    recipients = [a.strip() for a in re.split(r"[,;]", to) if a.strip()]
+    for extra in (cc, bcc):
+        if extra:
+            recipients += [a.strip() for a in re.split(r"[,;]", extra) if a.strip()]
+
+    try:
+        smtp = _open_smtp(conn_cfg)
+    except Exception as e:
+        return _format_smtp_error(e)
+    try:
+        smtp.send_message(msg, from_addr=conn_cfg["from_addr"], to_addrs=recipients)
+    except Exception as e:
+        return _format_smtp_error(e)
+    finally:
+        try:
+            smtp.quit()
+        except Exception:
+            pass
+
+    # Best-effort save-to-Sent (many providers don't do this for SMTP sends).
+    conn = _imap()
+    if conn is not None:
+        try:
+            _append_to_sent(conn, msg.as_bytes())
+        except Exception:
+            pass
+
+    suffix = (f" ({len(resolved)} attachment{'s' if len(resolved) != 1 else ''})"
+              if resolved else "")
+    return f"✅ Message sent to {to}{suffix}."
+
+
+def _email_get_profile(args: dict) -> str:
+    cfg = _get_config()
+    if cfg is None:
+        return _unavailable()
+    conn = _imap()
+    if conn is None:
+        return _unavailable()
+
+    inbox_total = inbox_unread = "?"
+    folder_count = "?"
+    try:
+        typ, st = conn.status("INBOX", "(MESSAGES UNSEEN)")
+        if typ == "OK" and st and st[0]:
+            m = re.search(rb"MESSAGES\s+(\d+)", st[0])
+            u = re.search(rb"UNSEEN\s+(\d+)", st[0])
+            inbox_total = int(m.group(1)) if m else "?"
+            inbox_unread = int(u.group(1)) if u else "?"
+        typ2, data = conn.list()
+        if typ2 == "OK" and data:
+            folder_count = sum(1 for x in data if x)
+    except Exception as e:
+        return _format_imap_error(e)
+
+    return (f"Account: {cfg['username']}\n"
+            f"From address: {cfg['from_addr']}\n"
+            f"IMAP: {cfg['imap_host']}:{cfg['imap_port']}\n"
+            f"SMTP: {cfg['smtp_host']}:{cfg['smtp_port']} ({cfg['smtp_security']})\n"
+            f"Folders: {folder_count}\n"
+            f"INBOX: {inbox_total} total, {inbox_unread} unread")
+
+
+def _email_create_folder(args: dict) -> str:
+    conn = _imap()
+    if conn is None:
+        return _unavailable()
+    name = args.get("name")
+    if not name:
+        return "Error: Missing required parameter 'name'."
+    try:
+        typ, data = conn.create(_quote_mailbox(name))
+    except Exception as e:
+        return _format_imap_error(e)
+    if typ != "OK":
+        detail = data[0].decode("utf-8", "replace") if data and data[0] else "unknown error"
+        return f"Error: could not create folder {name!r}: {detail}"
+    return f"✅ Folder {name!r} created."
+
+
+def _email_download_attachments(args: dict) -> str:
+    conn = _imap()
+    if conn is None:
+        return _unavailable()
+    uid = args.get("message_id")
+    if not uid:
+        return "Error: Missing required parameter 'message_id' (the message UID)."
+    folder = args.get("folder", "INBOX")
+    default_folder = os.path.join(os.path.dirname(os.path.abspath(__file__)),
+                                  "data", "email_attachments")
+    dest = args.get("dest_folder") or default_folder
+
+    ok, err = _select(conn, folder, readonly=True)
+    if not ok:
+        return err
+    try:
+        msg = _fetch_full(conn, int(uid))
+    except ValueError:
+        return f"Error: message_id must be a numeric UID, got {uid!r}."
+    except Exception as e:
+        return _format_imap_error(e)
+    if msg is None:
+        return f"Error: message UID {uid} not found in folder {folder!r}."
+
+    parts = list(_iter_attachments(msg))
+    if not parts:
+        return "No attachments found."
+    os.makedirs(dest, exist_ok=True)
+
+    saved = []
+    for part in parts:
+        filename = _decode_hdr(part.get_filename()) or "attachment.bin"
+        payload = part.get_payload(decode=True)
+        if payload is None:
+            saved.append(f"- {filename}: empty payload")
+            continue
+        safe_name = os.path.basename(filename)
+        path = os.path.join(dest, safe_name)
+        try:
+            with open(path, "wb") as f:
+                f.write(payload)
+        except Exception as e:
+            saved.append(f"- {safe_name}: ERROR writing file: {e}")
+            continue
+        saved.append(f"- {os.path.abspath(path)} ({len(payload)} bytes)")
+    return "\n".join(["✅ Attachments downloaded:"] + saved)
+
+
+# ── Tool manifest ────────────────────────────────────────────────────────────────
+
+TOOLS = [
+    {
+        "name": "status",
+        "description": (
+            "Self-check that the email integration is operational: verifies IMAP and SMTP both "
+            "authenticate with the configured credentials. Call this first whenever another email "
+            "tool fails, or to give the user a quick yes/no on whether email is usable right now."
+        ),
+        "inputSchema": {"type": "object", "properties": {}},
+    },
+    {
+        "name": "list_messages",
+        "description": (
+            "List messages in a folder (default INBOX), newest first. The optional 'query' supports "
+            "a Gmail-like mini-syntax: from:x, to:x, subject:x, since:YYYY-MM-DD, before:YYYY-MM-DD, "
+            "'unread'/'read', 'has:attachment', and free text (matched against headers+body). "
+            "Returns subject, sender, date, read/flag state and the message UID; pass the UID (and "
+            "the same folder) to get_message / modify_message."
+        ),
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "folder": {"type": "string", "description": "IMAP folder to list (default 'INBOX'). See list_folders."},
+                "query": {"type": "string", "description": "Gmail-like query, e.g. 'from:john unread since:2024-01-01'. Empty = all."},
+                "unread_only": {"type": "boolean", "description": "Shortcut to only return unread messages (default false)."},
+                "max_results": {"type": "integer", "description": "Max messages to return (default 20, max 50)."},
+            },
+        },
+    },
+    {
+        "name": "get_message",
+        "description": (
+            "Get the full content of a message by its UID, including body text (truncated at 10000 "
+            "chars; HTML-only emails are converted to readable text). Attachment filenames are "
+            "listed — download them with download_attachments. Pass 'folder' if the message is not "
+            "in INBOX (a UID is only unique within one folder)."
+        ),
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "message_id": {"type": "string", "description": "The message UID (as returned by list_messages)."},
+                "folder": {"type": "string", "description": "Folder the message is in (default 'INBOX')."},
+                "include_body": {"type": "boolean", "description": "Include the full body text (default true)."},
+            },
+            "required": ["message_id"],
+        },
+    },
+    {
+        "name": "get_thread",
+        "description": (
+            "Best-effort thread reconstruction for a message UID: gathers messages in the same "
+            "folder that share the Message-ID reference chain (In-Reply-To/References) or the same "
+            "base subject. IMAP has no native threads, so this is heuristic."
+        ),
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "message_id": {"type": "string", "description": "A message UID that belongs to the thread."},
+                "folder": {"type": "string", "description": "Folder to search (default 'INBOX')."},
+            },
+            "required": ["message_id"],
+        },
+    },
+    {
+        "name": "list_folders",
+        "description": "List all IMAP folders/mailboxes with total and unread message counts. Use to discover folder names for the other tools.",
+        "inputSchema": {"type": "object", "properties": {}},
+    },
+    {
+        "name": "modify_message",
+        "description": (
+            "Change a message's state: add/remove flags, mark read/unread, move to another folder, "
+            "archive or delete. Friendly flag names: 'read'/'seen', 'flagged'/'starred', 'answered', "
+            "'deleted'. mark_read=true marks as read; move_to_folder='Archive' archives (moves); "
+            "add_flags=['deleted'] deletes (and expunges). Pass 'folder' if not INBOX."
+        ),
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "message_id": {"type": "string", "description": "The message UID to modify."},
+                "folder": {"type": "string", "description": "Folder the message is in (default 'INBOX')."},
+                "add_flags": {"type": ["string", "array"], "items": {"type": "string"},
+                              "description": "Flag(s) to add: 'read', 'flagged', 'answered', 'deleted'. String or array."},
+                "remove_flags": {"type": ["string", "array"], "items": {"type": "string"},
+                                 "description": "Flag(s) to remove (e.g. 'read' to mark unread)."},
+                "mark_read": {"type": "boolean", "description": "Convenience: mark the message as read."},
+                "mark_unread": {"type": "boolean", "description": "Convenience: mark the message as unread."},
+                "move_to_folder": {"type": "string", "description": "Move the message to this folder (archive = move to your archive folder)."},
+            },
+            "required": ["message_id"],
+        },
+    },
+    {
+        "name": "send_message",
+        "description": (
+            "Send an email via SMTP. Supports in-thread replies by passing in_reply_to (the "
+            "Message-ID of the message being replied to, with or without angle brackets), which sets "
+            "the In-Reply-To/References headers. Attach files by passing local paths in "
+            "'attachments'; if any path is missing the email is NOT sent. The message is also "
+            "best-effort saved to the account's Sent folder."
+        ),
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "to": {"type": "string", "description": "Recipient address(es), comma-separated."},
+                "subject": {"type": "string", "description": "Subject line."},
+                "body": {"type": "string", "description": "Plain-text body."},
+                "cc": {"type": "string", "description": "CC address(es) (optional)."},
+                "bcc": {"type": "string", "description": "BCC address(es) (optional)."},
+                "in_reply_to": {"type": "string", "description": "Message-ID being replied to, for correct threading (optional)."},
+                "attachments": {"type": "array", "items": {"type": "string"},
+                                "description": "File path(s) to attach (absolute or relative to the connector folder). Total ~25 MB."},
+            },
+            "required": ["to", "subject", "body"],
+        },
+    },
+    {
+        "name": "get_profile",
+        "description": "Show the configured account: username, From address, IMAP/SMTP servers, folder count, and INBOX total/unread.",
+        "inputSchema": {"type": "object", "properties": {}},
+    },
+    {
+        "name": "create_folder",
+        "description": "Create a new IMAP folder/mailbox. Fails if it already exists.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "name": {"type": "string", "description": "Name of the new folder (e.g. 'Archive', 'Receipts')."},
+            },
+            "required": ["name"],
+        },
+    },
+    {
+        "name": "download_attachments",
+        "description": (
+            "Download all attachments from a message to a local folder (default data/email_attachments/). "
+            "Returns the absolute path and size of each saved file. Pass 'folder' if the message is not in INBOX."
+        ),
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "message_id": {"type": "string", "description": "The message UID to download attachments from."},
+                "folder": {"type": "string", "description": "Folder the message is in (default 'INBOX')."},
+                "dest_folder": {"type": "string", "description": "Local folder to save into (default data/email_attachments/)."},
+            },
+            "required": ["message_id"],
+        },
+    },
+]
+
+
+# ── JSON-RPC dispatch ────────────────────────────────────────────────────────────
+
+TOOL_DISPATCH: dict[str, Callable[[dict], str]] = {
+    "status":               _email_status,
+    "list_messages":        _email_list_messages,
+    "get_message":          _email_get_message,
+    "get_thread":           _email_get_thread,
+    "list_folders":         _email_list_folders,
+    "modify_message":       _email_modify_message,
+    "send_message":         _email_send_message,
+    "get_profile":          _email_get_profile,
+    "create_folder":        _email_create_folder,
+    "download_attachments": _email_download_attachments,
+}
+
+
+def _ok(req_id: Any, result: Any) -> str:
+    return json.dumps({"jsonrpc": "2.0", "id": req_id, "result": result})
+
+
+def _text_result(req_id: Any, text: str, is_error: bool = False) -> str:
+    payload: dict = {
+        "jsonrpc": "2.0",
+        "id": req_id,
+        "result": {"content": [{"type": "text", "text": text}]},
+    }
+    if is_error:
+        payload["result"]["isError"] = True
+    return json.dumps(payload)
+
+
+def handle_request(msg: dict) -> str | None:
+    method = msg.get("method", "")
+    req_id = msg.get("id")
+
+    if method == "initialize":
+        return _ok(req_id, {
+            "protocolVersion": "2024-11-05",
+            "capabilities": {"tools": {}},
+            "serverInfo": {"name": "email", "version": "1.0.0"},
+        })
+
+    if method == "notifications/initialized":
+        return None
+
+    if method == "tools/list":
+        return _ok(req_id, {"tools": TOOLS})
+
+    if method == "tools/call":
+        params = msg.get("params", {})
+        tool_name = params.get("name", "")
+        tool_args = params.get("arguments", {}) or {}
+        handler = TOOL_DISPATCH.get(tool_name)
+        if handler is None:
+            return _text_result(req_id, f"Error: Unknown tool: {tool_name}", is_error=True)
+        try:
+            text = handler(tool_args)
+            return _text_result(req_id, text, is_error=text.startswith("Error:"))
+        except Exception as e:
+            log(f"Unhandled exception in tool '{tool_name}': {e}")
+            return _text_result(req_id, f"Error: Internal error in tool '{tool_name}': {e}", is_error=True)
+
+    return json.dumps({
+        "jsonrpc": "2.0",
+        "id": req_id,
+        "error": {"code": -32601, "message": f"Method not found: {method}"},
+    })
+
+
+# ── Main loop ────────────────────────────────────────────────────────────────────
+
+def main() -> None:
+    log("Starting Email MCP server")
+    # Validate config and start the background push watcher (best-effort).
+    _start_watching()
+    try:
+        for line in sys.stdin:
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                msg = json.loads(line)
+            except json.JSONDecodeError as e:
+                log(f"Invalid JSON input: {e}")
+                continue
+            resp = handle_request(msg)
+            if resp is not None:
+                with _stdout_lock:
+                    sys.stdout.write(resp + "\n")
+                    sys.stdout.flush()
+    except KeyboardInterrupt:
+        pass
+
+
+if __name__ == "__main__":
+    main()
diff --git a/connectors/email/icon_lg.svg b/connectors/email/icon_lg.svg
new file mode 100644
index 0000000..6b94229
--- /dev/null
+++ b/connectors/email/icon_lg.svg
@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 96 96"><rect width="96" height="96" rx="20" fill="#2563eb"/><rect x="20" y="30" width="56" height="36" rx="5" fill="#fff"/><path d="M22 34l26 18 26-18" fill="none" stroke="#2563eb" stroke-width="4.5" stroke-linecap="round" stroke-linejoin="round"/></svg>
diff --git a/connectors/email/icon_sm.svg b/connectors/email/icon_sm.svg
new file mode 100644
index 0000000..86e2a7e
--- /dev/null
+++ b/connectors/email/icon_sm.svg
@@ -0,0 +1 @@
+<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 48 48"><rect width="48" height="48" rx="10" fill="#2563eb"/><rect x="10" y="15" width="28" height="18" rx="3" fill="#fff"/><path d="M11 17l13 9 13-9" fill="none" stroke="#2563eb" stroke-width="2.5" stroke-linecap="round" stroke-linejoin="round"/></svg>
diff --git a/connectors/email/verify.py b/connectors/email/verify.py
new file mode 100644
index 0000000..86d2c07
--- /dev/null
+++ b/connectors/email/verify.py
@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+"""Verify the Email connector credentials by attempting IMAP and SMTP logins.
+
+Reads the EMAIL_* environment variables (the same set the MCP server consumes),
+opens one IMAP and one SMTP connection, attempts LOGIN on each, and prints a
+single JSON object on stdout:
+
+    {"ok": true,  "message": "...", "details": {"imap": "...", "smtp": "..."}}
+    {"ok": false, "message": "...", "details": {"imap": "...", "smtp": "..."}}
+
+Exit code is 0 on success, 1 on any failure. No credentials are ever printed.
+
+stdlib only (imaplib, smtplib, ssl) — mirrors the email_mcp_server.py constraint.
+"""
+import imaplib
+import json
+import os
+import smtplib
+import ssl
+import sys
+
+
+def _result(ok, message, **extra):
+    print(json.dumps({"ok": ok, "message": message, **extra}))
+    sys.exit(0 if ok else 1)
+
+
+def _imap():
+    host = os.environ.get("EMAIL_IMAP_HOST", "").strip()
+    port = int(os.environ.get("EMAIL_IMAP_PORT", "993") or "993")
+    user = os.environ.get("EMAIL_USERNAME", "").strip()
+    pw = os.environ.get("EMAIL_PASSWORD", "")
+    if not host or not user or not pw:
+        return False, "EMAIL_IMAP_HOST, EMAIL_USERNAME and EMAIL_PASSWORD are required"
+    try:
+        ctx = ssl.create_default_context()
+        with imaplib.IMAP4_SSL(host, port, ssl_context=ctx) as imap:
+            imap.login(user, pw)
+            imap.select("INBOX", readonly=True)
+        return True, f"IMAP login to {host}:{port} successful"
+    except Exception as e:
+        return False, f"IMAP login to {host}:{port} failed: {e}"
+
+
+def _smtp():
+    host = os.environ.get("EMAIL_SMTP_HOST", "").strip()
+    port = int(os.environ.get("EMAIL_SMTP_PORT", "465") or "465")
+    sec = os.environ.get("EMAIL_SMTP_SECURITY", "").strip().lower()
+    user = os.environ.get("EMAIL_USERNAME", "").strip()
+    pw = os.environ.get("EMAIL_PASSWORD", "")
+    if not host or not user or not pw:
+        return False, "EMAIL_SMTP_HOST, EMAIL_USERNAME and EMAIL_PASSWORD are required"
+    if not sec:
+        sec = "starttls" if port in (587, 25) else "ssl"
+    try:
+        ctx = ssl.create_default_context()
+        if sec == "ssl":
+            with smtplib.SMTP_SSL(host, port, context=ctx, timeout=15) as s:
+                s.login(user, pw)
+        elif sec == "starttls":
+            with smtplib.SMTP(host, port, timeout=15) as s:
+                s.starttls(context=ctx)
+                s.login(user, pw)
+        else:
+            with smtplib.SMTP(host, port, timeout=15) as s:
+                s.login(user, pw)
+        return True, f"SMTP login to {host}:{port} ({sec}) successful"
+    except Exception as e:
+        return False, f"SMTP login to {host}:{port} failed: {e}"
+
+
+def main():
+    imap_ok, imap_msg = _imap()
+    smtp_ok, smtp_msg = _smtp()
+    if imap_ok and smtp_ok:
+        _result(True, "IMAP and SMTP authentication successful",
+                details={"imap": imap_msg, "smtp": smtp_msg})
+    if imap_ok and not smtp_ok:
+        _result(False, f"IMAP ok but SMTP failed: {smtp_msg}",
+                details={"imap": imap_msg, "smtp": smtp_msg})
+    if smtp_ok and not imap_ok:
+        _result(False, f"SMTP ok but IMAP failed: {imap_msg}",
+                details={"imap": imap_msg, "smtp": smtp_msg})
+    _result(False, f"Both IMAP and SMTP failed: {imap_msg} | {smtp_msg}",
+            details={"imap": imap_msg, "smtp": smtp_msg})
+
+
+if __name__ == "__main__":
+    main()
diff --git a/connectors/ssh/connector.json b/connectors/ssh/connector.json
new file mode 100644
index 0000000..992cc1b
--- /dev/null
+++ b/connectors/ssh/connector.json
@@ -0,0 +1,113 @@
+{
+  "id": "ssh",
+  "name": "SSH Remote Access",
+  "version": "1.0.0",
+  "type": "mcp_local",
+  "launch_command": "python3 ssh_mcp_server.py",
+  "transport": "stdio",
+  "requires": [
+    "PYTHON"
+  ],
+  "dependencies": [
+    "paramiko>=3.4"
+  ],
+  "env": [
+    {
+      "name": "SSH_MCP_POOL_TTL",
+      "label": "Connection pool TTL (seconds)",
+      "description": "How long an idle SSH connection stays open before being closed. Default: 300 (5 min).",
+      "required": false,
+      "secret": false,
+      "default": "300",
+      "example": "300"
+    },
+    {
+      "name": "SSH_MCP_COMMAND_TIMEOUT",
+      "label": "Command timeout (seconds)",
+      "description": "Max time a remote command can run before being killed. Default: 120.",
+      "required": false,
+      "secret": false,
+      "default": "120",
+      "example": "120"
+    },
+    {
+      "name": "SSH_MCP_CONNECT_TIMEOUT",
+      "label": "Connect timeout (seconds)",
+      "description": "Max time to wait for the SSH handshake. Default: 15.",
+      "required": false,
+      "secret": false,
+      "default": "15",
+      "example": "15"
+    },
+    {
+      "name": "SSH_MCP_LOGIN_PW_TTL",
+      "label": "Login password cache TTL (seconds)",
+      "description": "How long an elicited login password stays cached in RAM. Default: 300.",
+      "required": false,
+      "secret": false,
+      "default": "300",
+      "example": "300"
+    },
+    {
+      "name": "SSH_MCP_SUDO_PW_TTL",
+      "label": "Sudo password cache TTL (seconds)",
+      "description": "How long an elicited sudo password stays cached in RAM. Default: 300.",
+      "required": false,
+      "secret": false,
+      "default": "300",
+      "example": "300"
+    },
+    {
+      "name": "SSH_MCP_KEY_PASSPHRASE",
+      "label": "SSH key passphrase (non-interactive override)",
+      "description": "Passphrase for an encrypted private key. Overrides elicitation. Use only in automated environments.",
+      "required": false,
+      "secret": true,
+      "example": ""
+    }
+  ],
+  "setup_instructions": [
+    "Install dependencies: pip install -r requirements.txt",
+    "No OAuth or API key needed: host authentication is managed at runtime via add_alias.",
+    "Add an alias: mcp__ssh__add_alias(alias=\"my-server\", hostname=\"...\", username=\"...\", auth=\"key\")",
+    "The server stores aliases in ~/.ssh_aliases.json (auto-managed, never hand-edit).",
+    "Optional: tune TTLs via SSH_MCP_* environment variables (see env[] above)."
+  ],
+  "docs": [
+    {
+      "lang": "en",
+      "description": "Full SSH remote server access: read/write files, grep, execute commands, manage systemd services, upload/download files and directories, and probe system info — all over SFTP and SSH with the same output format as Skald's native tools. Aliases manage host config (hostname, port, user, auth method, sudo policy). Login passwords and sudo passwords are elicited on demand and kept only in RAM with configurable TTL. No OAuth, no API keys, no secrets on disk.",
+      "llm_short_description": "SSH MCP server: 13 tools (list/add/remove_alias, read_file, list_files, grep_files, edit_file, replace_lines, exec, upload, download, sysinfo, systemd). Auth via key/agent or elicited password. Sudo via nopasswd or elicited password. Connection pooling with lazy TTL eviction."
+    }
+  ],
+  "mcp_config": {
+    "command": "python3",
+    "args": [
+      "ssh_mcp_server.py"
+    ],
+    "env": {
+      "SSH_MCP_POOL_TTL": "{ENV:SSH_MCP_POOL_TTL}",
+      "SSH_MCP_COMMAND_TIMEOUT": "{ENV:SSH_MCP_COMMAND_TIMEOUT}",
+      "SSH_MCP_CONNECT_TIMEOUT": "{ENV:SSH_MCP_CONNECT_TIMEOUT}",
+      "SSH_MCP_LOGIN_PW_TTL": "{ENV:SSH_MCP_LOGIN_PW_TTL}",
+      "SSH_MCP_SUDO_PW_TTL": "{ENV:SSH_MCP_SUDO_PW_TTL}",
+      "SSH_MCP_KEY_PASSPHRASE": "{SECRET:SSH_MCP_KEY_PASSPHRASE}"
+    }
+  },
+  "homepage": "",
+  "icon_small": "icon_sm.svg",
+  "icon_large": "icon_lg.svg",
+  "scope": "user",
+  "tags": [
+    "ssh",
+    "mcp",
+    "local",
+    "remote",
+    "server",
+    "terminal",
+    "sysadmin"
+  ],
+  "auth": {
+    "type": "none"
+  }
+}
diff --git a/connectors/ssh/icon_lg.svg b/connectors/ssh/icon_lg.svg
new file mode 100644
index 0000000..a0bd61c
--- /dev/null
+++ b/connectors/ssh/icon_lg.svg
@@ -0,0 +1,6 @@
+<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 96 96" width="96" height="96">
+  <rect width="96" height="96" rx="14" fill="#1e293b"/>
+  <text x="48" y="55" font-family="monospace" font-size="40" fill="#22d3ee" text-anchor="middle" font-weight="bold">SSH</text>
+  <line x1="20" y1="70" x2="76" y2="70" stroke="#334155" stroke-width="2"/>
+  <rect x="28" y="60" width="40" height="2" rx="1" fill="#22d3ee" opacity="0.5"/>
+</svg>
diff --git a/connectors/ssh/icon_sm.svg b/connectors/ssh/icon_sm.svg
new file mode 100644
index 0000000..6e1fb90
--- /dev/null
+++ b/connectors/ssh/icon_sm.svg
@@ -0,0 +1,4 @@
+<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 48 48" width="48" height="48">
+  <rect width="48" height="48" rx="8" fill="#1e293b"/>
+  <text x="24" y="30" font-family="monospace" font-size="22" fill="#22d3ee" text-anchor="middle">SSH</text>
+</svg>
diff --git a/connectors/ssh/requirements.txt b/connectors/ssh/requirements.txt
new file mode 100644
index 0000000..718ab58
--- /dev/null
+++ b/connectors/ssh/requirements.txt
@@ -0,0 +1 @@
+paramiko>=3.4
diff --git a/connectors/ssh/ssh_mcp_server.py b/connectors/ssh/ssh_mcp_server.py
new file mode 100644
index 0000000..9a1585f
--- /dev/null
+++ b/connectors/ssh/ssh_mcp_server.py
@@ -0,0 +1,1284 @@
+#!/usr/bin/env python3
+"""SSH MCP server (JSON-RPC 2.0 over stdio).
+
+Exposes SSH tools that operate on remote hosts with **the same output format**
+as Skald's native filesystem tools (`read_file`, `list_files`, `grep_files`,
+`edit_file`, `replace_lines`, `exec`). The only thing the LLM sees differently
+is the first `alias` argument selecting the host. Tool names here are bare
+(`read_file`, `exec`, …); Skald prepends the `mcp__ssh__` prefix automatically.
+
+Hosts are addressed by alias — hostname and credentials never appear in tool
+calls. Aliases live in ``~/.ssh_aliases.json`` (auto-managed, never edited
+by hand). No secret is ever stored in that file.
+
+Login auth (``auth`` per alias, set on ``add_alias``):
+  * ``key``      — SSH key / ssh-agent only (default). If the chosen private key
+    is encrypted, its passphrase is requested on demand via **MCP elicitation**
+    (lazy: only when paramiko reports the key needs one). ``SSH_MCP_KEY_PASSPHRASE``
+    still works as a non-interactive override.
+  * ``password`` — login password requested on demand via **MCP elicitation**
+    (Skald shows a masked field in the Agent Inbox); agent/key auth is skipped.
+
+Elicited login secrets are kept only in this process's RAM with a short TTL
+(``SSH_MCP_LOGIN_PW_TTL``), never sent to the LLM and never written to disk;
+they are dropped on an authentication failure so the next attempt re-prompts.
+
+sudo (two methods per alias, set on ``add_alias``):
+  * ``nopasswd`` — ``sudo -n``: non-interactive, fails fast if NOPASSWD is not
+    configured on the host (no hung channel). No secret stored anywhere.
+  * ``prompt``  — ``sudo -S``: the password is requested on demand via **MCP
+    elicitation** (Skald shows a masked field in the Agent Inbox), fed to
+    sudo's stdin, kept only in this process's RAM with a short TTL, never sent
+    to the LLM and never written to disk.
+
+Connections are pooled per alias with lazy TTL eviction. Host keys are verified
+against ``~/.ssh/known_hosts`` (unknown hosts are rejected unless the alias was
+added with ``accept_new_host_key=true``).
+
+Run with:
+  python3 scripts/ssh_mcp_server.py
+
+Dependency: paramiko>=3.4 (in requirements.txt; installed into .venv by run.sh).
+"""
+
+from __future__ import annotations
+
+import itertools
+import json
+import os
+import posixpath
+import re
+import shlex
+import socket
+import stat
+import sys
+import time
+from typing import Any
+
+
+# ── Config ───────────────────────────────────────────────────────────────────
+
+ALIASES_FILE = os.path.expanduser("~/.ssh_aliases.json")
+
+POOL_TTL = int(os.environ.get("SSH_MCP_POOL_TTL", "300"))            # idle connection eviction
+SUDO_PW_TTL = int(os.environ.get("SSH_MCP_SUDO_PW_TTL", "300"))      # in-RAM sudo password cache
+LOGIN_PW_TTL = int(os.environ.get("SSH_MCP_LOGIN_PW_TTL", "300"))    # in-RAM login/passphrase cache
+CONNECT_TIMEOUT = int(os.environ.get("SSH_MCP_CONNECT_TIMEOUT", "15"))
+DEFAULT_CMD_TIMEOUT = int(os.environ.get("SSH_MCP_COMMAND_TIMEOUT", "120"))
+
+# Mirror the native list_files skip set so remote listings match local ones.
+SKIP_DIRS = {"target", ".git", "node_modules", ".venv", "__pycache__", "secrets"}
+
+# Match the native read_file cap.
+MAX_READ_LINES = 2000
+
+
+def log(msg: str) -> None:
+    """Log to stderr; stdout is reserved for JSON-RPC."""
+    print(f"[ssh_mcp] {msg}", file=sys.stderr, flush=True)
+
+
+class ToolError(Exception):
+    """Expected, user-facing failure. Surfaced as ``Error: <message>``."""
+
+
+# ── stdio JSON-RPC I/O (single readline path so elicit() can re-enter) ─────────
+
+def send(obj: dict) -> None:
+    sys.stdout.write(json.dumps(obj) + "\n")
+    sys.stdout.flush()
+
+
+def readline() -> dict | None:
+    """Blocking read of one non-empty JSON-RPC message; None on EOF."""
+    while True:
+        line = sys.stdin.readline()
+        if not line:
+            return None
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            return json.loads(line)
+        except json.JSONDecodeError as e:
+            log(f"invalid JSON input: {e}")
+            continue
+
+
+_eid = itertools.count(1)
+
+
+def elicit(message: str, requested_schema: dict) -> dict:
+    """Send an ``elicitation/create`` request and block until the reply arrives.
+
+    Returns the JSON-RPC ``result`` ({"action": ..., "content": {...}}). While
+    waiting, any other inbound message is ignored (v1: serial processing).
+    """
+    eid = f"ssh-elicit-{next(_eid)}"
+    send({
+        "jsonrpc": "2.0",
+        "id": eid,
+        "method": "elicitation/create",
+        "params": {"message": message, "requestedSchema": requested_schema},
+    })
+    while True:
+        msg = readline()
+        if msg is None:
+            return {"action": "cancel"}
+        if msg.get("id") == eid:
+            return msg.get("result", {"action": "cancel"})
+        log(f"ignoring inbound while awaiting elicitation: {msg.get('method') or msg.get('id')}")
+
+
+def _ok(req_id: Any, result: Any) -> dict:
+    return {"jsonrpc": "2.0", "id": req_id, "result": result}
+
+
+def _text_result(req_id: Any, text: str, is_error: bool = False) -> dict:
+    res: dict = {"content": [{"type": "text", "text": text}]}
+    if is_error:
+        res["isError"] = True
+    return {"jsonrpc": "2.0", "id": req_id, "result": res}
+
+
+# ── Alias store (auto-managed, 0600) ───────────────────────────────────────────
+
+def _load_aliases() -> dict:
+    try:
+        with open(ALIASES_FILE) as f:
+            return json.load(f)
+    except FileNotFoundError:
+        return {"aliases": []}
+    except Exception as e:
+        log(f"failed to read aliases: {e}")
+        return {"aliases": []}
+
+
+def _save_aliases(data: dict) -> None:
+    os.makedirs(os.path.dirname(ALIASES_FILE), exist_ok=True)
+    tmp = f"{ALIASES_FILE}.tmp.{os.getpid()}"
+    with open(tmp, "w") as f:
+        json.dump(data, f, indent=2)
+    os.replace(tmp, ALIASES_FILE)
+    try:
+        os.chmod(ALIASES_FILE, 0o600)
+    except OSError:
+        pass
+
+
+def _find_alias(name: str) -> dict | None:
+    for a in _load_aliases().get("aliases", []):
+        if a.get("alias") == name:
+            return a
+    return None
+
+
+# ── Connection pool (paramiko) ─────────────────────────────────────────────────
+
+_pool: dict[str, dict] = {}            # alias -> {client, sftp, last_used}
+_sudo_pw_cache: dict[str, tuple] = {}  # alias -> (password, ts)
+_login_pw_cache: dict[str, tuple] = {} # "alias:login" | "alias:passphrase" -> (secret, ts)
+
+
+def _login_password(alias: str, kind: str = "login") -> str | None:
+    """Return the SSH login password (``kind="login"``) or private-key passphrase
+    (``kind="passphrase"``) for ``alias`` from the RAM cache, or elicit it.
+
+    Never persisted. Returns None if the user declines/cancels/times out.
+    """
+    now = time.time()
+    key = f"{alias}:{kind}"
+    cached = _login_pw_cache.get(key)
+    if cached and (now - cached[1] <= LOGIN_PW_TTL):
+        return cached[0]
+
+    if kind == "passphrase":
+        message = f"Enter the passphrase for the private key of SSH alias '{alias}'."
+        title = f"key passphrase — {alias}"
+    else:
+        message = f"Enter the SSH login password for alias '{alias}'."
+        title = f"SSH password — {alias}"
+
+    result = elicit(
+        message,
+        {
+            "type": "object",
+            "properties": {
+                "password": {"type": "string", "format": "password", "title": title}
+            },
+            "required": ["password"],
+        },
+    )
+    if result.get("action") == "accept":
+        pw = (result.get("content") or {}).get("password", "")
+        _login_pw_cache[key] = (pw, now)
+        return pw
+    return None
+
+
+def _clear_login_pw(alias: str) -> None:
+    """Drop any cached login password / passphrase for ``alias``."""
+    for k in [k for k in _login_pw_cache if k.startswith(f"{alias}:")]:
+        _login_pw_cache.pop(k, None)
+
+
+def _is_auth_failure(paramiko, e: Exception) -> bool:
+    """True if ``e`` is an SSH auth rejection a login password could resolve.
+
+    ``AuthenticationException`` (wrong/refused key) always qualifies. A plain
+    ``SSHException`` qualifies only when its message says paramiko had no method
+    to try — e.g. a password-only host with no key/agent: *"No authentication
+    methods available"*. Other SSH errors (banner, host key, protocol) do not.
+    """
+    if isinstance(e, paramiko.AuthenticationException):
+        return True
+    msg = str(e).lower()
+    return "authentication method" in msg or "no authentication" in msg
+
+
+def _require_paramiko():
+    try:
+        import paramiko  # type: ignore
+        return paramiko
+    except ImportError:
+        raise ToolError(
+            "paramiko not installed — add 'paramiko>=3.4' to requirements.txt "
+            "and reinstall the .venv (uv pip install -r requirements.txt)."
+        )
+
+
+def _connect(cfg: dict, paramiko):
+    alias = cfg.get("alias", "")
+    auth = (cfg.get("auth") or "key").lower()
+
+    identity = cfg.get("identity_file")
+    identity = os.path.expanduser(identity) if identity else None
+
+    password = None
+    if auth == "password":
+        password = _login_password(alias, "login")
+        if password is None:
+            raise ToolError(
+                f"login password required for alias '{alias}' (user declined or timed out)"
+            )
+
+    def attempt(passphrase):
+        # With a password in hand, skip agent/key probing so paramiko goes
+        # straight to password auth instead of failing on keys first.
+        use_pw = password is not None
+        client = paramiko.SSHClient()
+        client.load_system_host_keys()
+        known = os.path.expanduser("~/.ssh/known_hosts")
+        if os.path.exists(known):
+            try:
+                client.load_host_keys(known)
+            except Exception as e:
+                log(f"could not load known_hosts: {e}")
+        if cfg.get("accept_new_host_key"):
+            client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        else:
+            client.set_missing_host_key_policy(paramiko.RejectPolicy())
+        client.connect(
+            hostname=cfg["hostname"],
+            port=int(cfg.get("port", 22)),
+            username=cfg.get("username"),
+            password=password,
+            key_filename=identity,
+            passphrase=passphrase,
+            allow_agent=not use_pw,
+            look_for_keys=not use_pw,
+            timeout=CONNECT_TIMEOUT,
+        )
+        return client
+
+    passphrase = os.environ.get("SSH_MCP_KEY_PASSPHRASE") or None
+    try:
+        return attempt(passphrase)
+    except paramiko.PasswordRequiredException:
+        # Encrypted private key with no passphrase supplied — ask for it (lazy).
+        if passphrase is not None:
+            raise   # we already had one and it was rejected; don't loop
+        passphrase = _login_password(alias, "passphrase")
+        if passphrase is None:
+            raise ToolError(
+                f"key passphrase required for alias '{alias}' (user declined or timed out)"
+            )
+        return attempt(passphrase)
+    except (paramiko.AuthenticationException, paramiko.SSHException) as e:
+        # Key/agent auth was rejected, or the host offers no method paramiko
+        # could try (e.g. a password-only host: "No authentication methods
+        # available"). If we haven't tried a password yet, elicit one and retry.
+        # Declining re-raises the original error. Covers aliases left as the
+        # default auth=key that actually need a login password.
+        if password is not None or not _is_auth_failure(paramiko, e):
+            raise
+        password = _login_password(alias, "login")
+        if password is None:
+            raise
+        return attempt(passphrase)
+
+
+def _close(alias: str) -> None:
+    entry = _pool.pop(alias, None)
+    if not entry:
+        return
+    try:
+        if entry.get("sftp"):
+            entry["sftp"].close()
+    except Exception:
+        pass
+    try:
+        entry["client"].close()
+    except Exception:
+        pass
+
+
+def _get_client(alias: str):
+    cfg = _find_alias(alias)
+    if not cfg:
+        raise ToolError(f"unknown alias '{alias}'")
+    paramiko = _require_paramiko()
+    now = time.time()
+
+    entry = _pool.get(alias)
+    if entry:
+        t = entry["client"].get_transport()
+        if (now - entry["last_used"] <= POOL_TTL) and t is not None and t.is_active():
+            entry["last_used"] = now
+            return entry["client"]
+        _close(alias)
+
+    try:
+        client = _connect(cfg, paramiko)
+    except paramiko.AuthenticationException:
+        _clear_login_pw(alias)   # wrong password/passphrase → re-prompt next time
+        raise ToolError(f"authentication failed for alias '{alias}' (check key/agent/password)")
+    except paramiko.BadHostKeyException:
+        raise ToolError(
+            f"host key mismatch for alias '{alias}' (possible MITM) — fix ~/.ssh/known_hosts"
+        )
+    except paramiko.SSHException as e:
+        if "not found in known_hosts" in str(e):
+            raise ToolError(
+                f"unknown host key for alias '{alias}' — re-add it with "
+                f"accept_new_host_key=true to trust it on first connect"
+            )
+        raise ToolError(f"SSH error for alias '{alias}': {e}")
+    except (OSError, socket.error) as e:
+        raise ToolError(f"connection to alias '{alias}' failed: {e}")
+
+    _pool[alias] = {"client": client, "sftp": None, "last_used": now}
+    return client
+
+
+def _get_sftp(alias: str):
+    client = _get_client(alias)
+    entry = _pool[alias]
+    if entry.get("sftp") is None:
+        entry["sftp"] = client.open_sftp()
+    return entry["sftp"]
+
+
+def _run_with_stdin(client, command: str, timeout: int, stdin_data: str | None = None):
+    """Run a remote command; return (stdout, stderr, exit_code). Raises on timeout."""
+    try:
+        chan_in, chan_out, chan_err = client.exec_command(command, timeout=timeout)
+        if stdin_data is not None:
+            try:
+                chan_in.write(stdin_data)
+                chan_in.flush()
+            except Exception:
+                pass
+        out = chan_out.read().decode("utf-8", "replace")
+        err = chan_err.read().decode("utf-8", "replace")
+        code = chan_out.channel.recv_exit_status()
+        return out, err, code
+    except socket.timeout:
+        raise ToolError(f"command timed out after {timeout}s")
+
+
+# ── sudo ───────────────────────────────────────────────────────────────────────
+
+def _sudo_password(alias: str) -> str | None:
+    """Return the sudo password for ``alias`` from RAM cache, or elicit it.
+
+    Never persisted. Returns None if the user declines/cancels/times out.
+    """
+    now = time.time()
+    cached = _sudo_pw_cache.get(alias)
+    if cached and (now - cached[1] <= SUDO_PW_TTL):
+        return cached[0]
+
+    result = elicit(
+        f"Enter the sudo password for SSH alias '{alias}'.",
+        {
+            "type": "object",
+            "properties": {
+                "password": {
+                    "type": "string",
+                    "format": "password",
+                    "title": f"sudo password — {alias}",
+                }
+            },
+            "required": ["password"],
+        },
+    )
+    if result.get("action") == "accept":
+        pw = (result.get("content") or {}).get("password", "")
+        _sudo_pw_cache[alias] = (pw, now)
+        return pw
+    return None
+
+
+def _sudo_prefix(alias: str, cfg: dict, sudo_user: str | None):
+    """Build the sudo prefix for ``cfg``. Returns (prefix, stdin_password).
+
+    Raises ToolError when sudo is disabled or the password is unavailable.
+    """
+    method = (cfg.get("sudo") or {}).get("method", "prompt")
+    u = f"-u {shlex.quote(sudo_user)} " if sudo_user else ""
+    if method == "none":
+        raise ToolError(f"sudo is disabled for alias '{alias}'")
+    if method == "nopasswd":
+        return f"sudo -n {u}", None
+    pw = _sudo_password(alias)
+    if pw is None:
+        raise ToolError("sudo password required (user declined or timed out)")
+    return f"sudo -S -p '' {u}", pw
+
+
+# ── SFTP helpers ───────────────────────────────────────────────────────────────
+
+def _sftp_read_text(sftp, path: str) -> str:
+    with sftp.open(path, "r") as f:
+        data = f.read()
+    return data.decode("utf-8", "replace") if isinstance(data, (bytes, bytearray)) else data
+
+
+def _sftp_write_atomic(sftp, path: str, content: str) -> None:
+    """Write atomically: temp file in the same dir + posix_rename. Preserve mode."""
+    d = posixpath.dirname(path) or "."
+    base = posixpath.basename(path)
+    tmp = posixpath.join(d, f".{base}.tmp.{os.getpid()}")
+
+    mode = None
+    try:
+        mode = stat.S_IMODE(sftp.stat(path).st_mode)
+    except IOError:
+        pass
+
+    with sftp.open(tmp, "w") as f:
+        f.write(content)
+    if mode is not None:
+        try:
+            sftp.chmod(tmp, mode)
+        except IOError:
+            pass
+    try:
+        sftp.posix_rename(tmp, path)
+    except (IOError, AttributeError):
+        try:
+            sftp.remove(path)
+        except IOError:
+            pass
+        sftp.rename(tmp, path)
+
+
+def _sftp_mkdirs(sftp, d: str) -> None:
+    if not d or d in ("/", "."):
+        return
+    try:
+        sftp.stat(d)
+        return
+    except IOError:
+        pass
+    parent = posixpath.dirname(d)
+    if parent and parent != d:
+        _sftp_mkdirs(sftp, parent)
+    try:
+        sftp.mkdir(d)
+    except IOError:
+        pass
+
+
+def _relpath(root: str, full: str) -> str:
+    r = root.rstrip("/") or "/"
+    return posixpath.relpath(full, r)
+
+
+# ── Tools: aliases ─────────────────────────────────────────────────────────────
+
+def _tool_list_aliases(args: dict) -> str:
+    out = []
+    for a in _load_aliases().get("aliases", []):
+        out.append({
+            "alias": a.get("alias"),
+            "hostname": a.get("hostname"),
+            "port": a.get("port", 22),
+            "username": a.get("username"),
+            "auth": a.get("auth", "key"),
+            "sudo_method": (a.get("sudo") or {}).get("method", "prompt"),
+            "description": a.get("description", ""),
+        })
+    return json.dumps(out, indent=2)
+
+
+def _tool_add_alias(args: dict) -> str:
+    name = args.get("alias")
+    if not name:
+        return "Error: missing required argument: alias"
+    if not args.get("hostname"):
+        return "Error: missing required argument: hostname"
+
+    sudo = args.get("sudo")
+    method = sudo.get("method") if isinstance(sudo, dict) else (sudo or "prompt")
+    if method not in ("nopasswd", "prompt", "none"):
+        return f"Error: invalid sudo method '{method}' (use nopasswd|prompt|none)"
+
+    auth = (args.get("auth") or "key").lower()
+    if auth not in ("key", "password"):
+        return f"Error: invalid auth method '{auth}' (use key|password)"
+
+    entry = {
+        "alias": name,
+        "hostname": args["hostname"],
+        "port": int(args.get("port", 22)),
+        "username": args.get("username"),
+        "identity_file": args.get("identity_file"),
+        "description": args.get("description", ""),
+        "auth": auth,
+        "sudo": {"method": method},
+        "accept_new_host_key": bool(args.get("accept_new_host_key", False)),
+    }
+
+    data = _load_aliases()
+    aliases = data.setdefault("aliases", [])
+    prev = None
+    for i, a in enumerate(aliases):
+        if a.get("alias") == name:
+            prev = a
+            aliases[i] = entry
+            break
+    else:
+        aliases.append(entry)
+    _save_aliases(data)
+    _close(name)          # config may have changed — drop any pooled connection
+    _sudo_pw_cache.pop(name, None)
+    _clear_login_pw(name)
+
+    target = f"{entry.get('username')}@{entry['hostname']}:{entry['port']}"
+    if prev:
+        return f"Updated alias '{name}' → {target} (auth: {auth}, sudo: {method})."
+    return f"Added alias '{name}' → {target} (auth: {auth}, sudo: {method})."
+
+
+def _tool_remove_alias(args: dict) -> str:
+    name = args.get("alias")
+    if not name:
+        return "Error: missing required argument: alias"
+    data = _load_aliases()
+    aliases = data.get("aliases", [])
+    kept = [a for a in aliases if a.get("alias") != name]
+    if len(kept) == len(aliases):
+        return f"Error: alias '{name}' not found"
+    data["aliases"] = kept
+    _save_aliases(data)
+    _close(name)
+    _sudo_pw_cache.pop(name, None)
+    _clear_login_pw(name)
+    return f"Removed alias '{name}'."
+
+
+# ── Tools: filesystem (native output format) ───────────────────────────────────
+
+def _tool_read_file(args: dict) -> str:
+    alias, path = args.get("alias"), args.get("path")
+    if not alias or not path:
+        return "Error: 'alias' and 'path' are required"
+    sftp = _get_sftp(alias)
+    try:
+        content = _sftp_read_text(sftp, path)
+    except IOError as e:
+        raise ToolError(f"cannot read {path}: {e}")
+
+    lines = content.splitlines()
+    total = len(lines)
+
+    limit = args.get("limit")
+    limit = min(int(limit), MAX_READ_LINES) if limit is not None else None
+    start = max(int(args["start_line"]) - 1, 0) if args.get("start_line") is not None else 0
+    if args.get("end_line") is not None:
+        end = min(int(args["end_line"]), total)
+    elif limit is not None:
+        end = min(start + limit, total)
+    else:
+        end = total
+
+    if start >= total and total > 0:
+        return f"(file has only {total} lines; start_line {start + 1} is out of range)"
+    end = max(end, start)
+
+    width = max(len(str(total)), 3)
+    return "\n".join(
+        f"{start + i + 1:>{width}} | {line}" for i, line in enumerate(lines[start:end])
+    )
+
+
+def _tool_list_files(args: dict) -> str:
+    alias, path = args.get("alias"), args.get("path")
+    if not alias or not path:
+        return "Error: 'alias' and 'path' are required"
+    max_depth = int(args.get("depth", 3))
+    dirs_only = bool(args.get("dirs_only", False))
+    sftp = _get_sftp(alias)
+
+    out: list[str] = []
+
+    def walk(d: str, depth: int) -> None:
+        try:
+            entries = sftp.listdir_attr(d)
+        except IOError:
+            return
+        for a in entries:
+            full = posixpath.join(d, a.filename)
+            if stat.S_ISDIR(a.st_mode):
+                if a.filename in SKIP_DIRS:
+                    continue
+                if dirs_only:
+                    out.append(_relpath(path, full))
+                if depth + 1 < max_depth:
+                    walk(full, depth + 1)
+            elif stat.S_ISREG(a.st_mode) and not dirs_only:
+                out.append(_relpath(path, full))
+
+    try:
+        sftp.listdir_attr(path)
+    except IOError as e:
+        raise ToolError(f"cannot list {path}: {e}")
+    walk(path, 0)
+    out.sort()
+    return json.dumps(out)
+
+
+def _grep_flags(args: dict) -> str:
+    flags = ""
+    if not bool(args.get("case_sensitive", False)):
+        flags += "-i "
+    inc = args.get("include_glob")
+    if inc:
+        flags += f"--include={shlex.quote(inc)} "
+    return flags
+
+
+def _tool_grep_files(args: dict) -> str:
+    alias, path, pattern = args.get("alias"), args.get("path"), args.get("pattern")
+    if not alias or not path or pattern is None:
+        return "Error: 'alias', 'path' and 'pattern' are required"
+    mode = args.get("output_mode", "content")
+    ctx = min(int(args.get("context_lines", 0) or 0), 10)
+    maxr = int(args.get("max_results", 100))
+    client = _get_client(alias)
+
+    flags = _grep_flags(args)
+    qpat, qpath = shlex.quote(pattern), shlex.quote(path)
+    root_prefix = path.rstrip("/") + "/"
+
+    def rel(p: str) -> str:
+        return p[len(root_prefix):] if p.startswith(root_prefix) else p
+
+    if mode == "files_only":
+        cmd = f"grep -rlIZ {flags}-E -e {qpat} -- {qpath}"
+        out, err, code = _run_with_stdin(client, cmd, DEFAULT_CMD_TIMEOUT)
+        if code >= 2 and not out:
+            raise ToolError(err.strip() or "grep failed")
+        files = [rel(f) for f in out.split("\0") if f][:maxr]
+        if not files:
+            return f'No files match "{pattern}" in {path}.'
+        return f"{len(files)} file(s):\n" + "\n".join(files)
+
+    if mode == "count":
+        cmd = f"grep -rcI {flags}-E -e {qpat} -- {qpath}"
+        out, err, code = _run_with_stdin(client, cmd, DEFAULT_CMD_TIMEOUT)
+        if code >= 2 and not out:
+            raise ToolError(err.strip() or "grep failed")
+        items = []
+        for line in out.splitlines():
+            f, _, c = line.rpartition(":")     # rpartition: count is numeric at end
+            if f and c.isdigit() and int(c) > 0:
+                items.append((rel(f), int(c)))
+        items = items[:maxr]
+        if not items:
+            return f'No matches for "{pattern}" in {path}.'
+        return f"{len(items)} file(s):\n" + "\n".join(f"{f}: {c}" for f, c in items)
+
+    # content mode
+    cflag = f"-C {ctx} " if ctx else ""
+    cmd = f"grep -rnIZ {cflag}{flags}-E -e {qpat} -- {qpath}"
+    out, err, code = _run_with_stdin(client, cmd, DEFAULT_CMD_TIMEOUT)
+    if code >= 2 and not out:
+        raise ToolError(err.strip() or "grep failed")
+
+    entries: list[str] = []
+    if ctx == 0:
+        for line in out.split("\n"):
+            if not line:
+                continue
+            if "\0" in line:
+                f, _, rest = line.partition("\0")
+            else:
+                f, _, rest = line.partition(":")
+            lineno, _, body = rest.partition(":")
+            entries.append(f"{rel(f)}:{lineno}: {body}")
+            if len(entries) >= maxr:
+                break
+    else:
+        prev_file = None
+        for line in out.split("\n"):
+            if not line:
+                continue
+            if line == "--":
+                if prev_file is not None and len(entries) < maxr:
+                    entries.append(f"{rel(prev_file)}:---")
+                continue
+            if "\0" in line:
+                f, _, rest = line.partition("\0")
+            else:
+                f, _, rest = line.partition(":")
+            m = re.match(r"(\d+)([:-])(.*)$", rest, re.S)
+            if not m:
+                continue
+            lineno, sep, body = m.group(1), m.group(2), m.group(3)
+            marker = ">" if sep == ":" else " "
+            entries.append(f"{marker}{rel(f)}: {lineno}: {body}")
+            prev_file = f
+            if len(entries) >= maxr:
+                break
+
+    if not entries:
+        return f'No matches for "{pattern}" in {path}.'
+    return f"{len(entries)} match(es):\n" + "\n".join(entries)
+
+
+def _tool_edit_file(args: dict) -> str:
+    alias, path = args.get("alias"), args.get("path")
+    old, new = args.get("old"), args.get("new")
+    if not alias or not path:
+        return "Error: 'alias' and 'path' are required"
+    if old is None or new is None:
+        return "Error: 'old' and 'new' are required"
+    replace_all = bool(args.get("replace_all", False))
+    sftp = _get_sftp(alias)
+    try:
+        content = _sftp_read_text(sftp, path)
+    except IOError as e:
+        raise ToolError(f"cannot read {path}: {e}")
+
+    not_found = (
+        f"Error: Text not found in {path}. "
+        f"Call read_file first and copy the text exactly as shown after the '| ' prefix."
+    )
+    if replace_all:
+        if old not in content:
+            return not_found
+        updated = content.replace(old, new)
+    else:
+        cnt = content.count(old)
+        if cnt > 1:
+            return (
+                f"Error: Text found {cnt} times in {path}. "
+                f"Include more surrounding context in `old` to make it unique, "
+                f"or set replace_all=true."
+            )
+        if cnt == 0:
+            return not_found
+        updated = content.replace(old, new, 1)
+
+    _sftp_write_atomic(sftp, path, updated)
+    return f"Edited {path}."
+
+
+def _tool_replace_lines(args: dict) -> str:
+    alias, path = args.get("alias"), args.get("path")
+    if not alias or not path:
+        return "Error: 'alias' and 'path' are required"
+    if args.get("from_line") is None or args.get("to_line") is None or args.get("new") is None:
+        return "Error: 'from_line', 'to_line' and 'new' are required"
+    from_line = int(args["from_line"])
+    to_line = int(args["to_line"])
+    new = args["new"]
+    if from_line < 1:
+        return "Error: from_line must be >= 1"
+    if to_line < from_line:
+        return "Error: to_line must be >= from_line"
+
+    sftp = _get_sftp(alias)
+    try:
+        content = _sftp_read_text(sftp, path)
+    except IOError as e:
+        raise ToolError(f"cannot read {path}: {e}")
+
+    lines = content.splitlines()
+    total = len(lines)
+    if from_line > total:
+        return f"Error: from_line {from_line} exceeds file length ({total} lines)"
+    to_clamped = min(to_line, total)
+    new_lines = new.splitlines()
+    lines[from_line - 1:to_clamped] = new_lines
+
+    updated = "\n".join(lines)
+    if content.endswith("\n"):
+        updated += "\n"
+    _sftp_write_atomic(sftp, path, updated)
+    return f"Replaced lines {from_line}–{to_clamped} in {path} with {len(new_lines)} new lines."
+
+
+# ── Tools: exec / sudo / systemd ────────────────────────────────────────────────
+
+def _tool_exec(args: dict) -> str:
+    alias, command = args.get("alias"), args.get("command")
+    if not alias or command is None:
+        return "Error: 'alias' and 'command' are required"
+    sudo = bool(args.get("sudo", False))
+    sudo_user = args.get("sudo_user")
+    timeout = int(args.get("timeout_sec", DEFAULT_CMD_TIMEOUT))
+    cfg = _find_alias(alias)
+    if not cfg:
+        return f"Error: unknown alias '{alias}'"
+
+    pw = None
+    wrapped = command
+    if sudo:
+        prefix, pw = _sudo_prefix(alias, cfg, sudo_user)
+        wrapped = prefix + command
+
+    client = _get_client(alias)
+    try:
+        chan_in, chan_out, chan_err = client.exec_command(wrapped, timeout=timeout)
+        if pw is not None:
+            try:
+                chan_in.write(pw + "\n")
+                chan_in.flush()
+            except Exception:
+                pass
+        out = chan_out.read().decode("utf-8", "replace")
+        err = chan_err.read().decode("utf-8", "replace")
+        code = chan_out.channel.recv_exit_status()
+    except socket.timeout:
+        return f"Error: command timed out after {timeout}s"
+    return json.dumps({"stdout": out, "stderr": err, "exit_code": code})
+
+
+def _tool_systemd(args: dict) -> str:
+    alias, service, action = args.get("alias"), args.get("service"), args.get("action")
+    if not alias or not service or not action:
+        return "Error: 'alias', 'service' and 'action' are required"
+    allowed = {"status", "start", "stop", "restart", "reload", "enable", "disable"}
+    if action not in allowed:
+        return f"Error: invalid action '{action}' (allowed: {', '.join(sorted(allowed))})"
+    cfg = _find_alias(alias)
+    if not cfg:
+        return f"Error: unknown alias '{alias}'"
+
+    qsvc = shlex.quote(service)
+    client = _get_client(alias)
+
+    parts: list[str] = []
+    if action != "status":
+        prefix, pw = _sudo_prefix(alias, cfg, None)
+        out, err, code = _run_with_stdin(
+            client, f"{prefix}systemctl {action} {qsvc}", DEFAULT_CMD_TIMEOUT,
+            (pw + "\n") if pw else None,
+        )
+        parts.append(f"$ systemctl {action} {service}  (exit {code})")
+        if out.strip():
+            parts.append(out.strip())
+        if err.strip():
+            parts.append(err.strip())
+
+    status, _, _ = _run_with_stdin(
+        client, f"systemctl status {qsvc} --no-pager 2>&1 | head -n 20", DEFAULT_CMD_TIMEOUT)
+    parts.append("── status ──")
+    parts.append(status.strip())
+
+    journal, _, _ = _run_with_stdin(
+        client, f"journalctl -u {qsvc} -n 10 --no-pager 2>&1", DEFAULT_CMD_TIMEOUT)
+    parts.append("── journal (last 10) ──")
+    parts.append(journal.strip())
+    return "\n".join(parts)
+
+
+# ── Tools: transfer / diagnostics ───────────────────────────────────────────────
+
+def _tool_upload(args: dict) -> str:
+    alias = args.get("alias")
+    local_path, remote_path = args.get("local_path"), args.get("remote_path")
+    if not alias or not local_path or not remote_path:
+        return "Error: 'alias', 'local_path' and 'remote_path' are required"
+    if not os.path.exists(local_path):
+        return f"Error: local path not found: {local_path}"
+    sftp = _get_sftp(alias)
+
+    count = total = 0
+    dest_shown = remote_path
+    if os.path.isdir(local_path):
+        for root, _dirs, files in os.walk(local_path):
+            relroot = os.path.relpath(root, local_path)
+            rdir = remote_path if relroot == "." else posixpath.join(
+                remote_path, relroot.replace(os.sep, "/"))
+            _sftp_mkdirs(sftp, rdir)
+            for fn in files:
+                lf = os.path.join(root, fn)
+                sftp.put(lf, posixpath.join(rdir, fn))
+                count += 1
+                total += os.path.getsize(lf)
+    else:
+        # scp/rsync semantics: a trailing-slash or existing-directory remote_path
+        # means "upload the file INTO that directory". paramiko's sftp.put needs a
+        # full destination FILE path — handed a directory path it fails with a
+        # generic "Failure" — so append the local basename in that case.
+        into_dir = remote_path.endswith("/")
+        if not into_dir:
+            try:
+                into_dir = stat.S_ISDIR(sftp.stat(remote_path).st_mode)
+            except IOError:
+                into_dir = False
+        if into_dir:
+            target_dir = remote_path.rstrip("/") or "/"
+            _sftp_mkdirs(sftp, target_dir)
+            dest = posixpath.join(target_dir, os.path.basename(local_path))
+        else:
+            dest = remote_path
+            parent = posixpath.dirname(dest)
+            if parent:
+                _sftp_mkdirs(sftp, parent)
+        sftp.put(local_path, dest)
+        count, total = 1, os.path.getsize(local_path)
+        dest_shown = dest
+    return f"Uploaded {count} file(s), {total} bytes → {dest_shown}"
+
+
+def _tool_download(args: dict) -> str:
+    alias = args.get("alias")
+    remote_path, local_path = args.get("remote_path"), args.get("local_path")
+    if not alias or not remote_path or not local_path:
+        return "Error: 'alias', 'remote_path' and 'local_path' are required"
+    sftp = _get_sftp(alias)
+    try:
+        st = sftp.stat(remote_path)
+    except IOError as e:
+        raise ToolError(f"remote path not found: {remote_path} ({e})")
+
+    count = total = 0
+    if stat.S_ISDIR(st.st_mode):
+        def rec(rdir: str, ldir: str) -> None:
+            nonlocal count, total
+            os.makedirs(ldir, exist_ok=True)
+            for a in sftp.listdir_attr(rdir):
+                rf = posixpath.join(rdir, a.filename)
+                lf = os.path.join(ldir, a.filename)
+                if stat.S_ISDIR(a.st_mode):
+                    rec(rf, lf)
+                elif stat.S_ISREG(a.st_mode):
+                    sftp.get(rf, lf)
+                    count += 1
+                    total += a.st_size or os.path.getsize(lf)
+        rec(remote_path, local_path)
+    else:
+        parent = os.path.dirname(local_path)
+        if parent:
+            os.makedirs(parent, exist_ok=True)
+        sftp.get(remote_path, local_path)
+        count, total = 1, os.path.getsize(local_path)
+    return f"Downloaded {count} file(s), {total} bytes → {local_path}"
+
+
+def _tool_sysinfo(args: dict) -> str:
+    alias = args.get("alias")
+    if not alias:
+        return "Error: 'alias' is required"
+    client = _get_client(alias)
+    cmd = (
+        "echo OS=$(uname -s 2>/dev/null); "
+        "echo KERNEL=$(uname -r 2>/dev/null); "
+        "echo CPU=$(nproc 2>/dev/null); "
+        "echo MEMTOTAL=$(awk '/MemTotal/{print $2}' /proc/meminfo 2>/dev/null); "
+        "echo MEMAVAIL=$(awk '/MemAvailable/{print $2}' /proc/meminfo 2>/dev/null); "
+        "echo DISKTOTAL=$(df -kP / 2>/dev/null | tail -1 | awk '{print $2}'); "
+        "echo DISKAVAIL=$(df -kP / 2>/dev/null | tail -1 | awk '{print $4}'); "
+        "echo UPTIME=$(uptime -p 2>/dev/null || uptime 2>/dev/null)"
+    )
+    out, _, _ = _run_with_stdin(client, cmd, DEFAULT_CMD_TIMEOUT)
+    kv: dict[str, str] = {}
+    for line in out.splitlines():
+        if "=" in line:
+            k, _, v = line.partition("=")
+            kv[k.strip()] = v.strip()
+
+    def gb(key: str):
+        try:
+            return round(int(kv.get(key, "")) / 1024 / 1024, 2)
+        except (ValueError, TypeError):
+            return None
+
+    info = {
+        "os": kv.get("OS", ""),
+        "kernel": kv.get("KERNEL", ""),
+        "cpu_count": int(kv["CPU"]) if kv.get("CPU", "").isdigit() else None,
+        "ram_total_gb": gb("MEMTOTAL"),
+        "ram_free_gb": gb("MEMAVAIL"),
+        "disk_total_gb": gb("DISKTOTAL"),
+        "disk_free_gb": gb("DISKAVAIL"),
+        "uptime": kv.get("UPTIME", ""),
+    }
+    return json.dumps(info, indent=2)
+
+
+# ── Tool registry ────────────────────────────────────────────────────────────────
+
+_ALIAS = {"type": "string", "description": "Host alias registered via add_alias."}
+_SFTP_NOTE = (
+    " Runs as the login user (no sudo): for paths needing root, use exec with "
+    "sudo=true (e.g. tee/install)."
+)
+
+TOOLS = [
+    {
+        "name": "list_aliases",
+        "description": "List configured SSH host aliases (never reveals keys or sudo passwords).",
+        "inputSchema": {"type": "object", "properties": {}},
+    },
+    {
+        "name": "add_alias",
+        "description": "Register or update an SSH host alias. Login via SSH key/agent (default) or login password asked on demand via elicitation.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "alias": {"type": "string", "description": "Short name used to address the host."},
+                "hostname": {"type": "string", "description": "Host or IP."},
+                "port": {"type": "integer", "description": "SSH port (default 22)."},
+                "username": {"type": "string", "description": "Login user."},
+                "identity_file": {"type": "string", "description": "Path to private key (optional; ssh-agent is also tried). An encrypted key's passphrase is asked via elicitation."},
+                "description": {"type": "string", "description": "Free-text note."},
+                "auth": {"type": "string", "enum": ["key", "password"],
+                         "description": "Login auth. key: SSH key/agent (default). password: login password asked on demand via elicitation, kept only in RAM."},
+                "sudo": {"type": "string", "enum": ["nopasswd", "prompt", "none"],
+                         "description": "How sudo authenticates. Use 'prompt' unless you KNOW otherwise — it is the safe default: runs 'sudo -S' and asks the user for the sudo password on demand via elicitation, so it works on any host where the login user is a normal sudoer. Only pick 'nopasswd' when the remote /etc/sudoers actually grants THIS user passwordless sudo (a NOPASSWD: rule): it runs 'sudo -n' and NEVER prompts, so on a normal host every sudo call fails immediately with 'a password is required'. 'none' disables sudo. Default prompt."},
+                "accept_new_host_key": {"type": "boolean", "description": "Trust the host key on first connect (TOFU). Default false."},
+            },
+            "required": ["alias", "hostname", "username"],
+        },
+    },
+    {
+        "name": "remove_alias",
+        "description": "Remove a host alias and close its pooled connection.",
+        "inputSchema": {"type": "object", "properties": {"alias": _ALIAS}, "required": ["alias"]},
+    },
+    {
+        "name": "read_file",
+        "description": "Read a remote file with 1-based line numbers (same format as the local read_file)." + _SFTP_NOTE,
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "alias": _ALIAS,
+                "path": {"type": "string", "description": "Absolute remote path."},
+                "start_line": {"type": "integer", "description": "First line (1-based, inclusive)."},
+                "end_line": {"type": "integer", "description": "Last line (1-based, inclusive)."},
+                "limit": {"type": "integer", "description": "Max lines to read (cap 2000)."},
+            },
+            "required": ["alias", "path"],
+        },
+    },
+    {
+        "name": "list_files",
+        "description": "List files/dirs under a remote path; returns a JSON array of relative paths (same as local list_files).",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "alias": _ALIAS,
+                "path": {"type": "string", "description": "Absolute remote directory."},
+                "depth": {"type": "integer", "description": "Max recursion depth (default 3; 1 = immediate contents)."},
+                "dirs_only": {"type": "boolean", "description": "Only directories (default false)."},
+            },
+            "required": ["alias", "path"],
+        },
+    },
+    {
+        "name": "grep_files",
+        "description": "Search a remote path with a regex; output matches the local grep_files (uses remote grep -E).",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "alias": _ALIAS,
+                "path": {"type": "string", "description": "Remote file or directory."},
+                "pattern": {"type": "string", "description": "Regex (case-insensitive by default)."},
+                "case_sensitive": {"type": "boolean", "description": "Default false."},
+                "include_glob": {"type": "string", "description": "Restrict to files matching this glob, e.g. '*.rs'."},
+                "output_mode": {"type": "string", "enum": ["content", "files_only", "count"], "description": "Default 'content'."},
+                "context_lines": {"type": "integer", "description": "Lines of context per match (default 0, max 10)."},
+                "max_results": {"type": "integer", "description": "Stop after N results (default 100)."},
+            },
+            "required": ["alias", "path", "pattern"],
+        },
+    },
+    {
+        "name": "edit_file",
+        "description": "Find & replace in a remote file (atomic). `old` must match exactly once unless replace_all." + _SFTP_NOTE,
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "alias": _ALIAS,
+                "path": {"type": "string", "description": "Absolute remote path."},
+                "old": {"type": "string", "description": "Exact text to replace."},
+                "new": {"type": "string", "description": "Replacement text."},
+                "replace_all": {"type": "boolean", "description": "Replace every occurrence (default false)."},
+            },
+            "required": ["alias", "path", "old", "new"],
+        },
+    },
+    {
+        "name": "replace_lines",
+        "description": "Replace a 1-based inclusive line range in a remote file (atomic)." + _SFTP_NOTE,
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "alias": _ALIAS,
+                "path": {"type": "string", "description": "Absolute remote path."},
+                "from_line": {"type": "integer", "description": "First line (1-based, inclusive)."},
+                "to_line": {"type": "integer", "description": "Last line (1-based, inclusive)."},
+                "new": {"type": "string", "description": "Replacement text."},
+            },
+            "required": ["alias", "path", "from_line", "to_line", "new"],
+        },
+    },
+    {
+        "name": "exec",
+        "description": "Run a command on the remote host. Set sudo=true to run via sudo (method per alias).",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "alias": _ALIAS,
+                "command": {"type": "string", "description": "Shell command."},
+                "sudo": {"type": "boolean", "description": "Run via sudo (default false)."},
+                "sudo_user": {"type": "string", "description": "Target user for sudo -u (optional)."},
+                "timeout_sec": {"type": "integer", "description": "Kill after N seconds (default 120)."},
+            },
+            "required": ["alias", "command"],
+        },
+    },
+    {
+        "name": "upload",
+        "description": "Upload a local file or directory (recursive) to the remote host via SFTP." + _SFTP_NOTE,
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "alias": _ALIAS,
+                "local_path": {"type": "string", "description": "Local file or directory."},
+                "remote_path": {"type": "string", "description": "Remote destination. For a single file: a trailing '/' (or an existing remote directory) uploads the file INTO that directory keeping its name; otherwise it is the exact destination file path (parent dirs are created)."},
+            },
+            "required": ["alias", "local_path", "remote_path"],
+        },
+    },
+    {
+        "name": "download",
+        "description": "Download a remote file or directory (recursive) to the local host via SFTP.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "alias": _ALIAS,
+                "remote_path": {"type": "string", "description": "Remote file or directory."},
+                "local_path": {"type": "string", "description": "Local destination path."},
+            },
+            "required": ["alias", "remote_path", "local_path"],
+        },
+    },
+    {
+        "name": "sysinfo",
+        "description": "Report OS, kernel, CPU count, RAM and root-disk usage, and uptime.",
+        "inputSchema": {"type": "object", "properties": {"alias": _ALIAS}, "required": ["alias"]},
+    },
+    {
+        "name": "systemd",
+        "description": "Manage a systemd service (status/start/stop/restart/reload/enable/disable) + last 10 journal lines. Mutating actions use sudo.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "alias": _ALIAS,
+                "service": {"type": "string", "description": "Service/unit name."},
+                "action": {"type": "string", "enum": ["status", "start", "stop", "restart", "reload", "enable", "disable"]},
+            },
+            "required": ["alias", "service", "action"],
+        },
+    },
+]
+
+TOOL_DISPATCH = {
+    "list_aliases": _tool_list_aliases,
+    "add_alias": _tool_add_alias,
+    "remove_alias": _tool_remove_alias,
+    "read_file": _tool_read_file,
+    "list_files": _tool_list_files,
+    "grep_files": _tool_grep_files,
+    "edit_file": _tool_edit_file,
+    "replace_lines": _tool_replace_lines,
+    "exec": _tool_exec,
+    "upload": _tool_upload,
+    "download": _tool_download,
+    "sysinfo": _tool_sysinfo,
+    "systemd": _tool_systemd,
+}
+
+
+# ── JSON-RPC dispatch ────────────────────────────────────────────────────────────
+
+def handle_message(msg: dict) -> dict | None:
+    method = msg.get("method", "")
+    req_id = msg.get("id")
+
+    if method == "initialize":
+        return _ok(req_id, {
+            "protocolVersion": "2025-06-18",
+            "capabilities": {"tools": {}},
+            "serverInfo": {"name": "ssh", "version": "1.0.0"},
+        })
+    if method == "notifications/initialized":
+        return None
+    if method == "tools/list":
+        return _ok(req_id, {"tools": TOOLS})
+    if method == "tools/call":
+        params = msg.get("params", {})
+        name = params.get("name", "")
+        targs = params.get("arguments", {}) or {}
+        handler = TOOL_DISPATCH.get(name)
+        if handler is None:
+            return _text_result(req_id, f"Error: Unknown tool: {name}", True)
+        try:
+            text = handler(targs)
+        except ToolError as e:
+            text = f"Error: {e}"
+        except Exception as e:
+            log(f"unhandled exception in tool '{name}': {e}")
+            text = f"Error: internal error in '{name}': {e}"
+        return _text_result(req_id, text, text.startswith("Error:"))
+
+    if req_id is not None:
+        return {"jsonrpc": "2.0", "id": req_id,
+                "error": {"code": -32601, "message": f"Method not found: {method}"}}
+    return None
+
+
+def main() -> None:
+    log("starting SSH MCP server")
+    try:
+        while True:
+            msg = readline()
+            if msg is None:
+                break
+            resp = handle_message(msg)
+            if resp is not None:
+                send(resp)
+    except KeyboardInterrupt:
+        pass
+
+
+if __name__ == "__main__":
+    main()
diff --git a/connectors/tavily/connector.json b/connectors/tavily/connector.json
index 5120008..6080d2f 100644
--- a/connectors/tavily/connector.json
+++ b/connectors/tavily/connector.json
@@ -4,12 +4,27 @@
   "version": "1.0.0",
   "type": "mcp_remote",
   "mcp_config": {
-    "url": "https://mcp.tavily.com/mcp/?tavilyApiKey={key}",
+    "url": "https://mcp.tavily.com/mcp/?tavilyApiKey={SECRET:tavilyApiKey}",
     "transport": "streamable-http"
   },
   "requires": [
-    "API_KEY"
+    "API_KEY",
+    "ENV"
   ],
+  "env": [
+    {
+      "name": "tavilyApiKey",
+      "label": "Tavily API key",
+      "description": "Your Tavily API key (find it at https://app.tavily.com). Used both as the MCP URL query parameter and for the verification request.",
+      "required": true,
+      "secret": true,
+      "example": "tvly-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
+    }
+  ],
+  "verify": {
+    "command": "python3 verify.py",
+    "timeout_secs": 15
+  },
   "docs": [
     {
       "lang": "en",
diff --git a/connectors/tavily/verify.py b/connectors/tavily/verify.py
new file mode 100644
index 0000000..56c5375
--- /dev/null
+++ b/connectors/tavily/verify.py
@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Verify the Tavily API key by making a minimal search request.
+
+Reads TAVILY_API_KEY from the environment (the key skald collected from the
+user and substituted into the {SECRET:tavilyApiKey} placeholder), POSTs a
+minimal search to https://api.tavily.com/search, and prints a single JSON
+object on stdout:
+
+    {"ok": true,  "message": "Tavily API key is valid"}
+    {"ok": false, "message": "Tavily API key is invalid or unauthorized"}
+
+Exit code is 0 on success, 1 on any failure. The API key is never printed.
+
+stdlib only (urllib).
+"""
+import json
+import os
+import sys
+import urllib.error
+import urllib.request
+
+
+def _result(ok, message):
+    print(json.dumps({"ok": ok, "message": message}))
+    sys.exit(0 if ok else 1)
+
+
+def main():
+    api_key = os.environ.get("tavilyApiKey", "").strip()
+    if not api_key:
+        # Also accept the UPPER_SNAKE form some users may type.
+        api_key = os.environ.get("TAVILY_API_KEY", "").strip()
+    if not api_key:
+        _result(False, "No Tavily API key provided (tavilyApiKey env var is empty)")
+
+    body = json.dumps({
+        "api_key": api_key,
+        "query": "skald connectivity check",
+        "max_results": 1,
+        "search_depth": "basic",
+    }).encode("utf-8")
+
+    req = urllib.request.Request(
+        "https://api.tavily.com/search",
+        data=body,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+
+    try:
+        with urllib.request.urlopen(req, timeout=15) as resp:
+            if 200 <= resp.status < 300:
+                _result(True, "Tavily API key is valid")
+            _result(False, f"Tavily returned HTTP {resp.status}")
+    except urllib.error.HTTPError as e:
+        if e.code in (401, 403):
+            _result(False, "Tavily API key is invalid or unauthorized "
+                           f"(HTTP {e.code})")
+        _result(False, f"Tavily returned HTTP {e.code}: {e.reason}")
+    except Exception as e:
+        _result(False, f"Tavily verify request failed: {e}")
+
+
+if __name__ == "__main__":
+    main()