feat(mcp): Connectors — catalog + global vs per-user runtimes (§7/§14/§15)

Re-architects MCP from one owner table + agent-written registration into an
admin-curated catalog with two runtimes unioned per session, surfaced in the UI
as "Connectors" (mcp/schema stays neutral, §0.1).

Two runtimes behind one seam (§7):
- Global runtime: shared, stateless connectors (web-search, Tavily…) on the
  HOST, connected at boot from mcp_global_servers, access-filtered per user via
  mcp_global_access.
- Per-user runtime: a user's activated connectors run INSIDE their container,
  started at first login from mcp_user_servers and living until restart (§9);
  docker exec -i children die via kill_on_drop when the UserContext drops.
- McpProvider trait (mcp/provider.rs): the session round-loop never learns which
  runtime owns a server. McpManager implements it directly (inert ownerless
  bundle); UserMcpView implements global ∪ user with an accessible_global
  snapshot. Both share McpManager::connect_all; McpServerSpec +
  global_row_spec/user_row_spec turn a DB row into a connectable spec.
- mcp-client: McpServerConfig.launch_in runs a stdio command inside a container
  via docker exec -i (set at runtime, never parsed from config).

Authorization is a capability on the role, not `if role==admin` (§0.1/§14):
role_capabilities table + db/role_capabilities.rs — register_remote and
register_local_from_catalog are self-service (seeded on every new role), while
register_local_script and manage_catalog are admin-only. admin holds every
capability by construction. This removes the agent-facing register_mcp/delete_mcp
tools and the mcp kinds of list_items/toggle_item, closing the §14 RCE vector.

Schema:
- Registry: mcp_catalog (vetted templates — schema only, no live creds),
  mcp_global_servers + mcp_global_access, role_capabilities.
- Owner: mcp_user_servers (per-user activations; api_key encrypted at rest,
  catalog_name a bare TEXT snapshot, never an owner→registry FK).
- Drops the old owner table mcp_servers.

API + UI: src/frontend/api/mcp.rs (admin catalog/global/access + user
available/activate/activated, all capability-gated via require_cap);
web/components/connectors.js (<connectors-page>) renders the user view always
and the admin view for role_id === 'admin'.

Deferred: interactive per-user auth (OAuth callback / QR / SSH elicitation, §15)
— only none/api_key wired; no boot seed of catalog presets; per-(user, session)
MCP grant model still open.

crates/mcp-client/src/config.rs

@@ -16,6 +16,11 @@ pub struct McpServerConfig {
     pub url:     Option<String>,
     /// http only: API key sent as `Authorization: Bearer <key>` (supports `${VAR}` interpolation).
     pub api_key: Option<String>,
+    /// stdio only: when `Some(container)`, the command runs INSIDE that Docker
+    /// container via `docker exec -i` instead of on the host. Set at runtime by
+    /// the manager (per-user connectors, blueprint §7), never parsed from config.
+    #[serde(skip)]
+    pub launch_in: Option<String>,
 }
 
 #[derive(Debug, Deserialize, Clone)]

crates/mcp-client/src/server.rs

@@ -233,15 +233,42 @@ impl McpServer {
         let command = cfg.command.as_deref()
             .ok_or_else(|| anyhow::anyhow!("stdio server '{}' requires 'command'", cfg.name))?;
 
-        let mut cmd = Command::new(command);
-        if let Some(args) = &cfg.args {
-            cmd.args(args);
-        }
-        if let Some(env_map) = &cfg.env {
-            for (k, v) in env_map {
-                cmd.env(k, interpolate_env(v));
+        let mut cmd = match &cfg.launch_in {
+            // Host transport: spawn the command directly.
+            None => {
+                let mut c = Command::new(command);
+                if let Some(args) = &cfg.args {
+                    c.args(args);
+                }
+                if let Some(env_map) = &cfg.env {
+                    for (k, v) in env_map {
+                        c.env(k, interpolate_env(v));
+                    }
+                }
+                c
             }
-        }
+            // Container transport (per-user connectors, blueprint §7): run the
+            // command INSIDE the user's container via `docker exec -i`. stdin/
+            // stdout/stderr are proxied transparently, so the JSON-RPC read-loop,
+            // the stderr drain and elicitation write-back all work unchanged. Env
+            // is passed with `-e K=V` so it lands inside the container, not on the
+            // `docker` client. Workdir defaults to the image WORKDIR (`/root`, the
+            // bind-mounted home), so no `-w` coupling to skald-core's path layout.
+            Some(container) => {
+                let mut c = Command::new("docker");
+                c.arg("exec").arg("-i");
+                if let Some(env_map) = &cfg.env {
+                    for (k, v) in env_map {
+                        c.arg("-e").arg(format!("{k}={}", interpolate_env(v)));
+                    }
+                }
+                c.arg(container).arg(command);
+                if let Some(args) = &cfg.args {
+                    c.args(args);
+                }
+                c
+            }
+        };
         cmd.stdin(Stdio::piped())
            .stdout(Stdio::piped())
            // Capture the child's stderr instead of inheriting it: many MCP

crates/mcp-client/tests/elicitation.rs

@@ -109,6 +109,7 @@ async fn elicitation_roundtrip_returns_secret_to_server() {
         env:       None,
         url:       None,
         api_key:   None,
+        launch_in: None,
     };
 
     let server = McpServer::start(&cfg, None, None, Some(Arc::new(AcceptHandler)))

crates/mcp-client/tests/logging.rs

@@ -94,6 +94,7 @@ async fn stderr_and_log_records_are_captured_and_diverted() {
         env: None,
         url: None,
         api_key: None,
+        launch_in: None,
     };
 
     let (notif_tx, mut notif_rx) = mpsc::unbounded_channel::<McpNotification>();

crates/mcp-client/tests/pagination.rs

@@ -86,6 +86,7 @@ async fn tools_list_follows_next_cursor_across_pages() {
         env: None,
         url: None,
         api_key: None,
+        launch_in: None,
     };
 
     let server = McpServer::start(&cfg, None, None, None)

crates/mcp-client/tests/tasks.rs

@@ -101,6 +101,7 @@ fn cfg(script: &std::path::Path, mode: &str, marker: Option<&std::path::Path>) -
         env: None,
         url: None,
         api_key: None,
+        launch_in: None,
     }
 }
 

crates/skald-core/src/db/mcp_catalog.rs

@@ -0,0 +1,177 @@
+//! The admin-curated catalog of installable MCP connectors (blueprint §14/§15).
+//!
+//! Registry table in `system.db`: instance-wide, listable without any user key so
+//! the "Connectors" UI can render it. Each entry is a *template* — a per-user
+//! connector is later instantiated from it into a `{userid}.db`
+//! (`mcp_user_servers`), or a global one is enabled by the admin
+//! (`mcp_global_servers`). No live credential ever lands here: `config_schema_json`
+//! only names the env/secret keys an activation must collect.
+
+use std::collections::HashMap;
+
+use anyhow::Result;
+use serde::Serialize;
+use sqlx::SqlitePool;
+
+#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
+pub struct McpCatalogRow {
+    pub id:                 i64,
+    pub name:               String,
+    /// 'per_user' | 'global' — which category (§15) this entry can be activated as.
+    pub scope:              String,
+    /// 'remote' | 'local_script' — the §14 risk axis.
+    pub source:             String,
+    pub transport:          String,
+    pub command:            Option<String>,
+    pub args_json:          Option<String>,
+    pub env_json:           Option<String>,
+    pub url:                Option<String>,
+    /// local_script: the vetted source path under `./scripts`.
+    pub script_path:        Option<String>,
+    /// Names of the env/secret keys the activation UI must collect (never values).
+    pub config_schema_json: Option<String>,
+    /// 'none'|'api_key'|'oauth'|'qr'|'ssh_key'. Only 'none'/'api_key' are wired now.
+    pub auth_kind:          String,
+    /// JSON array of role ids allowed to activate this; NULL = all roles (§15).
+    pub role_filter:        Option<String>,
+    pub friendly_name:      Option<String>,
+    pub description:        Option<String>,
+    pub created_at:         String,
+}
+
+impl McpCatalogRow {
+    pub fn args(&self) -> Vec<String> {
+        self.args_json.as_deref()
+            .and_then(|s| serde_json::from_str(s).ok())
+            .unwrap_or_default()
+    }
+
+    pub fn env(&self) -> HashMap<String, String> {
+        self.env_json.as_deref()
+            .and_then(|s| serde_json::from_str(s).ok())
+            .unwrap_or_default()
+    }
+
+    /// The role ids allowed to activate this entry, or `None` when unrestricted.
+    pub fn allowed_roles(&self) -> Option<Vec<String>> {
+        self.role_filter.as_deref()
+            .and_then(|s| serde_json::from_str::<Vec<String>>(s).ok())
+    }
+
+    /// Whether a user in `role_id` may activate this entry (§15 per-role catalog).
+    pub fn allowed_for_role(&self, role_id: &str) -> bool {
+        match self.allowed_roles() {
+            None => true,
+            Some(roles) => roles.iter().any(|r| r == role_id),
+        }
+    }
+}
+
+const SELECT: &str =
+    "SELECT id, name, scope, source, transport, command, args_json, env_json, url, \
+            script_path, config_schema_json, auth_kind, role_filter, friendly_name, \
+            description, created_at \
+     FROM mcp_catalog";
+
+// ── Reads ────────────────────────────────────────────────────────────────────
+
+pub async fn list(pool: &SqlitePool) -> Result<Vec<McpCatalogRow>> {
+    let rows = sqlx::query_as::<_, McpCatalogRow>(sqlx::AssertSqlSafe(format!("{SELECT} ORDER BY name")))
+        .fetch_all(pool)
+        .await?;
+    Ok(rows)
+}
+
+/// Catalog entries in a given scope ('per_user' | 'global').
+pub async fn list_for_scope(pool: &SqlitePool, scope: &str) -> Result<Vec<McpCatalogRow>> {
+    let rows = sqlx::query_as::<_, McpCatalogRow>(sqlx::AssertSqlSafe(format!("{SELECT} WHERE scope = ? ORDER BY name")))
+        .bind(scope)
+        .fetch_all(pool)
+        .await?;
+    Ok(rows)
+}
+
+pub async fn get_by_name(pool: &SqlitePool, name: &str) -> Result<Option<McpCatalogRow>> {
+    let row = sqlx::query_as::<_, McpCatalogRow>(sqlx::AssertSqlSafe(format!("{SELECT} WHERE name = ?")))
+        .bind(name)
+        .fetch_optional(pool)
+        .await?;
+    Ok(row)
+}
+
+pub async fn get(pool: &SqlitePool, id: i64) -> Result<Option<McpCatalogRow>> {
+    let row = sqlx::query_as::<_, McpCatalogRow>(sqlx::AssertSqlSafe(format!("{SELECT} WHERE id = ?")))
+        .bind(id)
+        .fetch_optional(pool)
+        .await?;
+    Ok(row)
+}
+
+// ── Writes ───────────────────────────────────────────────────────────────────
+
+/// Fields for creating or updating a catalog entry (keyed on `name`).
+pub struct UpsertCatalog<'a> {
+    pub name:               &'a str,
+    pub scope:              &'a str,
+    pub source:             &'a str,
+    pub transport:          &'a str,
+    pub command:            Option<&'a str>,
+    pub args_json:          Option<String>,
+    pub env_json:           Option<String>,
+    pub url:                Option<&'a str>,
+    pub script_path:        Option<&'a str>,
+    pub config_schema_json: Option<String>,
+    pub auth_kind:          &'a str,
+    pub role_filter:        Option<String>,
+    pub friendly_name:      Option<&'a str>,
+    pub description:        Option<&'a str>,
+}
+
+pub async fn upsert(pool: &SqlitePool, e: UpsertCatalog<'_>) -> Result<i64> {
+    let row = sqlx::query_as::<_, (i64,)>(
+        "INSERT INTO mcp_catalog
+            (name, scope, source, transport, command, args_json, env_json, url,
+             script_path, config_schema_json, auth_kind, role_filter, friendly_name, description)
+         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)
+         ON CONFLICT(name) DO UPDATE SET
+             scope              = excluded.scope,
+             source             = excluded.source,
+             transport          = excluded.transport,
+             command            = excluded.command,
+             args_json          = excluded.args_json,
+             env_json           = excluded.env_json,
+             url                = excluded.url,
+             script_path        = excluded.script_path,
+             config_schema_json = excluded.config_schema_json,
+             auth_kind          = excluded.auth_kind,
+             role_filter        = excluded.role_filter,
+             friendly_name      = excluded.friendly_name,
+             description        = excluded.description
+         RETURNING id",
+    )
+    .bind(e.name)
+    .bind(e.scope)
+    .bind(e.source)
+    .bind(e.transport)
+    .bind(e.command)
+    .bind(e.args_json)
+    .bind(e.env_json)
+    .bind(e.url)
+    .bind(e.script_path)
+    .bind(e.config_schema_json)
+    .bind(e.auth_kind)
+    .bind(e.role_filter)
+    .bind(e.friendly_name)
+    .bind(e.description)
+    .fetch_one(pool)
+    .await?;
+    Ok(row.0)
+}
+
+pub async fn delete(pool: &SqlitePool, id: i64) -> Result<()> {
+    sqlx::query("DELETE FROM mcp_catalog WHERE id = ?")
+        .bind(id)
+        .execute(pool)
+        .await?;
+    Ok(())
+}

crates/skald-core/src/db/mcp_global_access.rs

@@ -0,0 +1,91 @@
+//! Which users may use each globally-active MCP connector (blueprint §15).
+//!
+//! Registry junction table in `system.db` — the per-user access filter over
+//! `mcp_global_servers`. Both FKs are registry→registry (allowed), mirroring
+//! `shared_folder_members`. The admin UI's "grant to all / by role" is just a
+//! convenience that inserts rows here.
+
+use anyhow::Result;
+use sqlx::SqlitePool;
+
+// ── Reads ────────────────────────────────────────────────────────────────────
+
+/// The names of the **enabled** global servers a user may use. Feeds the
+/// `accessible_global` snapshot captured when the user's context is built.
+pub async fn server_names_for_user(pool: &SqlitePool, user_id: &str) -> Result<Vec<String>> {
+    let rows = sqlx::query_as::<_, (String,)>(
+        "SELECT s.name
+         FROM   mcp_global_access a
+         JOIN   mcp_global_servers s ON s.id = a.server_id
+         WHERE  a.user_id = ? AND s.enabled = 1
+         ORDER  BY s.name",
+    )
+    .bind(user_id)
+    .fetch_all(pool)
+    .await?;
+    Ok(rows.into_iter().map(|(n,)| n).collect())
+}
+
+/// The ids of the users granted access to a given global server.
+pub async fn users_for_server(pool: &SqlitePool, server_id: i64) -> Result<Vec<String>> {
+    let rows = sqlx::query_as::<_, (String,)>(
+        "SELECT user_id FROM mcp_global_access WHERE server_id = ? ORDER BY user_id",
+    )
+    .bind(server_id)
+    .fetch_all(pool)
+    .await?;
+    Ok(rows.into_iter().map(|(u,)| u).collect())
+}
+
+pub async fn has_access(pool: &SqlitePool, server_id: i64, user_id: &str) -> Result<bool> {
+    let row = sqlx::query_as::<_, (i64,)>(
+        "SELECT 1 FROM mcp_global_access WHERE server_id = ? AND user_id = ?",
+    )
+    .bind(server_id)
+    .bind(user_id)
+    .fetch_optional(pool)
+    .await?;
+    Ok(row.is_some())
+}
+
+// ── Writes ───────────────────────────────────────────────────────────────────
+
+/// Grants a user access to a global server. Idempotent on the PK.
+pub async fn grant(pool: &SqlitePool, server_id: i64, user_id: &str) -> Result<()> {
+    sqlx::query(
+        "INSERT OR IGNORE INTO mcp_global_access (server_id, user_id) VALUES (?, ?)",
+    )
+    .bind(server_id)
+    .bind(user_id)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+pub async fn revoke(pool: &SqlitePool, server_id: i64, user_id: &str) -> Result<()> {
+    sqlx::query("DELETE FROM mcp_global_access WHERE server_id = ? AND user_id = ?")
+        .bind(server_id)
+        .bind(user_id)
+        .execute(pool)
+        .await?;
+    Ok(())
+}
+
+/// Replaces the full access list for a server in one shot (used by the admin UI's
+/// "set who can use this" form, incl. the by-role bulk grant).
+pub async fn set_access(pool: &SqlitePool, server_id: i64, user_ids: &[String]) -> Result<()> {
+    let mut tx = pool.begin().await?;
+    sqlx::query("DELETE FROM mcp_global_access WHERE server_id = ?")
+        .bind(server_id)
+        .execute(&mut *tx)
+        .await?;
+    for user_id in user_ids {
+        sqlx::query("INSERT OR IGNORE INTO mcp_global_access (server_id, user_id) VALUES (?, ?)")
+            .bind(server_id)
+            .bind(user_id)
+            .execute(&mut *tx)
+            .await?;
+    }
+    tx.commit().await?;
+    Ok(())
+}

crates/skald-core/src/db/mcp_global_servers.rs

@@ -0,0 +1,145 @@
+//! Globally-active MCP connectors (blueprint §7/§15b): shared, stateless servers
+//! (web-search, Tavily…) that run on the HOST and are offered to every user the
+//! admin grants access to (`mcp_global_access`).
+//!
+//! Registry table in `system.db`. The global secret (the admin's API key) is fine
+//! here — `system.db` is admin-owned (§4). `catalog_name` is a registry→registry
+//! FK to `mcp_catalog(name)` (both in this file), which is allowed.
+
+use std::collections::HashMap;
+
+use anyhow::Result;
+use serde::Serialize;
+use sqlx::SqlitePool;
+
+#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
+pub struct McpGlobalServerRow {
+    pub id:            i64,
+    pub name:          String,
+    pub catalog_name:  Option<String>,
+    pub transport:     String,
+    pub command:       Option<String>,
+    pub args_json:     Option<String>,
+    pub env_json:      Option<String>,
+    pub url:           Option<String>,
+    pub api_key:       Option<String>,
+    pub friendly_name: Option<String>,
+    pub description:   Option<String>,
+    pub enabled:       bool,
+}
+
+impl McpGlobalServerRow {
+    pub fn args(&self) -> Vec<String> {
+        self.args_json.as_deref()
+            .and_then(|s| serde_json::from_str(s).ok())
+            .unwrap_or_default()
+    }
+
+    pub fn env(&self) -> HashMap<String, String> {
+        self.env_json.as_deref()
+            .and_then(|s| serde_json::from_str(s).ok())
+            .unwrap_or_default()
+    }
+}
+
+const SELECT: &str =
+    "SELECT id, name, catalog_name, transport, command, args_json, env_json, url, \
+            api_key, friendly_name, description, enabled \
+     FROM mcp_global_servers";
+
+// ── Reads ────────────────────────────────────────────────────────────────────
+
+pub async fn all(pool: &SqlitePool) -> Result<Vec<McpGlobalServerRow>> {
+    let rows = sqlx::query_as::<_, McpGlobalServerRow>(sqlx::AssertSqlSafe(format!("{SELECT} ORDER BY name")))
+        .fetch_all(pool)
+        .await?;
+    Ok(rows)
+}
+
+pub async fn all_enabled(pool: &SqlitePool) -> Result<Vec<McpGlobalServerRow>> {
+    let rows = sqlx::query_as::<_, McpGlobalServerRow>(sqlx::AssertSqlSafe(format!("{SELECT} WHERE enabled = 1 ORDER BY name")))
+        .fetch_all(pool)
+        .await?;
+    Ok(rows)
+}
+
+pub async fn get(pool: &SqlitePool, id: i64) -> Result<Option<McpGlobalServerRow>> {
+    let row = sqlx::query_as::<_, McpGlobalServerRow>(sqlx::AssertSqlSafe(format!("{SELECT} WHERE id = ?")))
+        .bind(id)
+        .fetch_optional(pool)
+        .await?;
+    Ok(row)
+}
+
+pub async fn get_by_name(pool: &SqlitePool, name: &str) -> Result<Option<McpGlobalServerRow>> {
+    let row = sqlx::query_as::<_, McpGlobalServerRow>(sqlx::AssertSqlSafe(format!("{SELECT} WHERE name = ?")))
+        .bind(name)
+        .fetch_optional(pool)
+        .await?;
+    Ok(row)
+}
+
+// ── Writes ───────────────────────────────────────────────────────────────────
+
+pub struct UpsertGlobal<'a> {
+    pub name:          &'a str,
+    pub catalog_name:  Option<&'a str>,
+    pub transport:     &'a str,
+    pub command:       Option<&'a str>,
+    pub args_json:     Option<String>,
+    pub env_json:      Option<String>,
+    pub url:           Option<&'a str>,
+    pub api_key:       Option<&'a str>,
+    pub friendly_name: Option<&'a str>,
+    pub description:   Option<&'a str>,
+}
+
+pub async fn upsert(pool: &SqlitePool, p: UpsertGlobal<'_>) -> Result<i64> {
+    let row = sqlx::query_as::<_, (i64,)>(
+        "INSERT INTO mcp_global_servers
+            (name, catalog_name, transport, command, args_json, env_json, url, api_key, friendly_name, description, enabled)
+         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, 1)
+         ON CONFLICT(name) DO UPDATE SET
+             catalog_name  = excluded.catalog_name,
+             transport     = excluded.transport,
+             command       = excluded.command,
+             args_json     = excluded.args_json,
+             env_json      = excluded.env_json,
+             url           = excluded.url,
+             api_key       = excluded.api_key,
+             friendly_name = excluded.friendly_name,
+             description   = excluded.description,
+             enabled       = 1
+         RETURNING id",
+    )
+    .bind(p.name)
+    .bind(p.catalog_name)
+    .bind(p.transport)
+    .bind(p.command)
+    .bind(p.args_json)
+    .bind(p.env_json)
+    .bind(p.url)
+    .bind(p.api_key)
+    .bind(p.friendly_name)
+    .bind(p.description)
+    .fetch_one(pool)
+    .await?;
+    Ok(row.0)
+}
+
+pub async fn set_enabled(pool: &SqlitePool, id: i64, enabled: bool) -> Result<()> {
+    sqlx::query("UPDATE mcp_global_servers SET enabled = ?1 WHERE id = ?2")
+        .bind(enabled as i64)
+        .bind(id)
+        .execute(pool)
+        .await?;
+    Ok(())
+}
+
+pub async fn delete(pool: &SqlitePool, id: i64) -> Result<()> {
+    sqlx::query("DELETE FROM mcp_global_servers WHERE id = ?")
+        .bind(id)
+        .execute(pool)
+        .await?;
+    Ok(())
+}

crates/skald-core/src/db/mcp_servers.rs

@@ -1,129 +0,0 @@
-use std::collections::HashMap;
-
-use anyhow::Result;
-use serde::{Deserialize, Serialize};
-use sqlx::SqlitePool;
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct McpServerRow {
-    pub id:            i64,
-    pub name:          String,
-    pub transport:     String,
-    pub command:       Option<String>,
-    pub args_json:     Option<String>,
-    pub env_json:      Option<String>,
-    pub url:           Option<String>,
-    pub api_key:       Option<String>,
-    pub description:   Option<String>,
-    pub friendly_name: Option<String>,
-    pub enabled:       bool,
-}
-
-impl McpServerRow {
-    pub fn args(&self) -> Vec<String> {
-        self.args_json.as_deref()
-            .and_then(|s| serde_json::from_str(s).ok())
-            .unwrap_or_default()
-    }
-
-    pub fn env(&self) -> HashMap<String, String> {
-        self.env_json.as_deref()
-            .and_then(|s| serde_json::from_str(s).ok())
-            .unwrap_or_default()
-    }
-}
-
-type RawRow = (i64, String, String, Option<String>, Option<String>, Option<String>, Option<String>, Option<String>, Option<String>, Option<String>, i64);
-
-fn from_raw(r: RawRow) -> McpServerRow {
-    McpServerRow {
-        id:            r.0,
-        name:          r.1,
-        transport:     r.2,
-        command:       r.3,
-        args_json:     r.4,
-        env_json:      r.5,
-        url:           r.6,
-        api_key:       r.7,
-        description:   r.8,
-        friendly_name: r.9,
-        enabled:       r.10 != 0,
-    }
-}
-
-const SELECT: &str =
-    "SELECT id, name, transport, command, args_json, env_json, url, api_key, description, friendly_name, enabled \
-     FROM mcp_servers";
-
-pub async fn all(pool: &SqlitePool) -> Result<Vec<McpServerRow>> {
-    let rows = sqlx::query_as::<_, RawRow>(sqlx::AssertSqlSafe(format!("{SELECT} ORDER BY name")))
-        .fetch_all(pool)
-        .await?;
-    Ok(rows.into_iter().map(from_raw).collect())
-}
-
-pub async fn all_enabled(pool: &SqlitePool) -> Result<Vec<McpServerRow>> {
-    let rows = sqlx::query_as::<_, RawRow>(sqlx::AssertSqlSafe(format!("{SELECT} WHERE enabled = 1 ORDER BY name")))
-        .fetch_all(pool)
-        .await?;
-    Ok(rows.into_iter().map(from_raw).collect())
-}
-
-pub struct UpsertParams<'a> {
-    pub name:          &'a str,
-    pub transport:     &'a str,
-    pub command:       Option<&'a str>,
-    pub args_json:     Option<String>,
-    pub env_json:      Option<String>,
-    pub url:           Option<&'a str>,
-    pub api_key:       Option<&'a str>,
-    pub description:   Option<&'a str>,
-    pub friendly_name: Option<&'a str>,
-}
-
-pub async fn upsert(pool: &SqlitePool, p: UpsertParams<'_>) -> Result<i64> {
-    let row = sqlx::query_as::<_, (i64,)>(
-        "INSERT INTO mcp_servers (name, transport, command, args_json, env_json, url, api_key, description, friendly_name, enabled)
-         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 1)
-         ON CONFLICT(name) DO UPDATE SET
-             transport     = excluded.transport,
-             command       = excluded.command,
-             args_json     = excluded.args_json,
-             env_json      = excluded.env_json,
-             url           = excluded.url,
-             api_key       = excluded.api_key,
-             description   = excluded.description,
-             friendly_name = excluded.friendly_name,
-             enabled       = 1
-         RETURNING id",
-    )
-    .bind(p.name)
-    .bind(p.transport)
-    .bind(p.command)
-    .bind(p.args_json)
-    .bind(p.env_json)
-    .bind(p.url)
-    .bind(p.api_key)
-    .bind(p.description)
-    .bind(p.friendly_name)
-    .fetch_one(pool)
-    .await?;
-    Ok(row.0)
-}
-
-pub async fn set_enabled(pool: &SqlitePool, name: &str, enabled: bool) -> Result<()> {
-    sqlx::query("UPDATE mcp_servers SET enabled = ?1 WHERE name = ?2")
-        .bind(enabled as i64)
-        .bind(name)
-        .execute(pool)
-        .await?;
-    Ok(())
-}
-
-pub async fn delete(pool: &SqlitePool, name: &str) -> Result<()> {
-    sqlx::query("DELETE FROM mcp_servers WHERE name = ?1")
-        .bind(name)
-        .execute(pool)
-        .await?;
-    Ok(())
-}

crates/skald-core/src/db/mcp_user_servers.rs

@@ -0,0 +1,156 @@
+//! A user's activated per-user MCP connectors (blueprint §7/§14).
+//!
+//! Owner table in each `{userid}.db` — encrypted at rest (SQLCipher), so `api_key`
+//! (a personal secret / OAuth refresh token) needs no column-level crypto.
+//! `catalog_name` is a BARE `TEXT` snapshot of `mcp_catalog.name`, never a FK: an
+//! owner→registry key would fail every INSERT under `PRAGMA foreign_keys=ON` in an
+//! isolated file. Local-script connectors run INSIDE the user's container against a
+//! script copied into the bind-mounted home (`script_rel_path`).
+
+use std::collections::HashMap;
+
+use anyhow::Result;
+use serde::Serialize;
+use sqlx::SqlitePool;
+
+#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
+pub struct McpUserServerRow {
+    pub id:              i64,
+    pub name:            String,
+    /// Bare snapshot of the originating `mcp_catalog.name`; NULL for a
+    /// self-registered remote.
+    pub catalog_name:    Option<String>,
+    /// 'remote' | 'local_script'.
+    pub source:          String,
+    pub transport:       String,
+    pub command:         Option<String>,
+    pub args_json:       Option<String>,
+    pub env_json:        Option<String>,
+    pub url:             Option<String>,
+    pub api_key:         Option<String>,
+    /// Container path of the copied script, for a `local_script`.
+    pub script_rel_path: Option<String>,
+    /// 'pending' | 'ready' — the interactive-auth gate ('ready' while api-key).
+    pub auth_state:      String,
+    pub enabled:         bool,
+}
+
+impl McpUserServerRow {
+    pub fn args(&self) -> Vec<String> {
+        self.args_json.as_deref()
+            .and_then(|s| serde_json::from_str(s).ok())
+            .unwrap_or_default()
+    }
+
+    pub fn env(&self) -> HashMap<String, String> {
+        self.env_json.as_deref()
+            .and_then(|s| serde_json::from_str(s).ok())
+            .unwrap_or_default()
+    }
+}
+
+const SELECT: &str =
+    "SELECT id, name, catalog_name, source, transport, command, args_json, env_json, url, \
+            api_key, script_rel_path, auth_state, enabled \
+     FROM mcp_user_servers";
+
+// ── Reads ────────────────────────────────────────────────────────────────────
+
+pub async fn all(pool: &SqlitePool) -> Result<Vec<McpUserServerRow>> {
+    let rows = sqlx::query_as::<_, McpUserServerRow>(sqlx::AssertSqlSafe(format!("{SELECT} ORDER BY name")))
+        .fetch_all(pool)
+        .await?;
+    Ok(rows)
+}
+
+/// The enabled, `auth_state='ready'` connectors — the set the per-user runtime
+/// starts at login. A 'pending' one is activated but not yet authenticated.
+pub async fn all_startable(pool: &SqlitePool) -> Result<Vec<McpUserServerRow>> {
+    let rows = sqlx::query_as::<_, McpUserServerRow>(sqlx::AssertSqlSafe(format!(
+        "{SELECT} WHERE enabled = 1 AND auth_state = 'ready' ORDER BY name"
+    )))
+    .fetch_all(pool)
+    .await?;
+    Ok(rows)
+}
+
+pub async fn get(pool: &SqlitePool, id: i64) -> Result<Option<McpUserServerRow>> {
+    let row = sqlx::query_as::<_, McpUserServerRow>(sqlx::AssertSqlSafe(format!("{SELECT} WHERE id = ?")))
+        .bind(id)
+        .fetch_optional(pool)
+        .await?;
+    Ok(row)
+}
+
+pub async fn get_by_name(pool: &SqlitePool, name: &str) -> Result<Option<McpUserServerRow>> {
+    let row = sqlx::query_as::<_, McpUserServerRow>(sqlx::AssertSqlSafe(format!("{SELECT} WHERE name = ?")))
+        .bind(name)
+        .fetch_optional(pool)
+        .await?;
+    Ok(row)
+}
+
+// ── Writes ───────────────────────────────────────────────────────────────────
+
+pub struct InsertUserServer<'a> {
+    pub name:            &'a str,
+    pub catalog_name:    Option<&'a str>,
+    pub source:          &'a str,
+    pub transport:       &'a str,
+    pub command:         Option<&'a str>,
+    pub args_json:       Option<String>,
+    pub env_json:        Option<String>,
+    pub url:             Option<&'a str>,
+    pub api_key:         Option<&'a str>,
+    pub script_rel_path: Option<&'a str>,
+    pub auth_state:      &'a str,
+}
+
+pub async fn insert(pool: &SqlitePool, s: InsertUserServer<'_>) -> Result<i64> {
+    let id = sqlx::query(
+        "INSERT INTO mcp_user_servers
+            (name, catalog_name, source, transport, command, args_json, env_json, url, api_key, script_rel_path, auth_state, enabled)
+         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, 1)",
+    )
+    .bind(s.name)
+    .bind(s.catalog_name)
+    .bind(s.source)
+    .bind(s.transport)
+    .bind(s.command)
+    .bind(s.args_json)
+    .bind(s.env_json)
+    .bind(s.url)
+    .bind(s.api_key)
+    .bind(s.script_rel_path)
+    .bind(s.auth_state)
+    .execute(pool)
+    .await?
+    .last_insert_rowid();
+    Ok(id)
+}
+
+pub async fn set_enabled(pool: &SqlitePool, id: i64, enabled: bool) -> Result<()> {
+    sqlx::query("UPDATE mcp_user_servers SET enabled = ?1 WHERE id = ?2")
+        .bind(enabled as i64)
+        .bind(id)
+        .execute(pool)
+        .await?;
+    Ok(())
+}
+
+pub async fn set_auth_state(pool: &SqlitePool, id: i64, auth_state: &str) -> Result<()> {
+    sqlx::query("UPDATE mcp_user_servers SET auth_state = ?1 WHERE id = ?2")
+        .bind(auth_state)
+        .bind(id)
+        .execute(pool)
+        .await?;
+    Ok(())
+}
+
+pub async fn delete(pool: &SqlitePool, id: i64) -> Result<()> {
+    sqlx::query("DELETE FROM mcp_user_servers WHERE id = ?")
+        .bind(id)
+        .execute(pool)
+        .await?;
+    Ok(())
+}

crates/skald-core/src/db/mod.rs

@@ -11,10 +11,14 @@ pub mod job_runs;
 pub mod known_tools;
 pub mod llm_requests;
 pub mod llm_request_payloads;
+pub mod mcp_catalog;
 pub mod mcp_events;
-pub mod mcp_servers;
+pub mod mcp_global_access;
+pub mod mcp_global_servers;
+pub mod mcp_user_servers;
 pub mod memory_docs;
 pub mod plugins;
+pub mod role_capabilities;
 pub mod roles;
 pub mod scheduled_jobs;
 pub mod scratchpad;
@@ -419,6 +423,86 @@ async fn create_registry_tables(pool: &SqlitePool) -> Result<()> {
     .execute(pool)
     .await?;
 
+    // ── MCP catalog + globally-active instances (blueprint §7/§14/§15) ──────────
+    //
+    // Registry tables: instance-wide MCP config, listable without any user key so
+    // the admin can render the "Connectors" catalog. The catalog is the admin's
+    // vetted set of installable connectors; a user later *instantiates* a per-user
+    // one into their own `{userid}.db` (`mcp_user_servers`, owner bucket) or the
+    // admin *enables* a global one here. Per-user credentials never land here — the
+    // catalog holds only the *schema* of what an activation must supply.
+    sqlx::query(
+        "CREATE TABLE IF NOT EXISTS mcp_catalog (
+            id                 INTEGER PRIMARY KEY AUTOINCREMENT,
+            name               TEXT    NOT NULL UNIQUE,
+            scope              TEXT    NOT NULL,                    -- 'per_user' | 'global'
+            source             TEXT    NOT NULL,                    -- 'remote' | 'local_script'
+            transport          TEXT    NOT NULL DEFAULT 'stdio',
+            command            TEXT,
+            args_json          TEXT,
+            env_json           TEXT,
+            url                TEXT,
+            script_path        TEXT,                               -- local_script: source under ./scripts
+            config_schema_json TEXT,                               -- names of env/secret keys the UI must collect
+            auth_kind          TEXT    NOT NULL DEFAULT 'none',    -- 'none'|'api_key'|'oauth'|'qr'|'ssh_key'
+            role_filter        TEXT,                               -- JSON array of role ids; NULL = all
+            friendly_name      TEXT,
+            description        TEXT,
+            created_at         TEXT    NOT NULL DEFAULT (datetime('now'))
+        )",
+    )
+    .execute(pool)
+    .await?;
+
+    // Concrete globally-active connectors (shared, stateless — web-search etc.).
+    // They run on the HOST. The global secret (admin's API key) is fine here:
+    // `system.db` is admin-owned (§4/§15b). `catalog_name` is a registry→registry
+    // FK (both in this file) — allowed.
+    sqlx::query(
+        "CREATE TABLE IF NOT EXISTS mcp_global_servers (
+            id            INTEGER PRIMARY KEY AUTOINCREMENT,
+            name          TEXT    NOT NULL UNIQUE,
+            catalog_name  TEXT    REFERENCES mcp_catalog(name),
+            transport     TEXT    NOT NULL DEFAULT 'stdio',
+            command       TEXT,
+            args_json     TEXT,
+            env_json      TEXT,
+            url           TEXT,
+            api_key       TEXT,
+            friendly_name TEXT,
+            description   TEXT,
+            enabled       INTEGER NOT NULL DEFAULT 1,
+            created_at    TEXT    NOT NULL DEFAULT (datetime('now'))
+        )",
+    )
+    .execute(pool)
+    .await?;
+
+    // Which users may use each globally-active connector (§15 per-user access).
+    // Mirrors `shared_folder_members`: both FKs are registry→registry, allowed.
+    sqlx::query(
+        "CREATE TABLE IF NOT EXISTS mcp_global_access (
+            server_id INTEGER NOT NULL REFERENCES mcp_global_servers(id) ON DELETE CASCADE,
+            user_id   TEXT    NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+            PRIMARY KEY (server_id, user_id)
+        )",
+    )
+    .execute(pool)
+    .await?;
+
+    // Capability grants per role (blueprint §14). A single indexed lookup instead
+    // of parsing `roles.attrs`. `admin` implicitly holds every capability (checked
+    // in code), so only non-admin roles need rows here.
+    sqlx::query(
+        "CREATE TABLE IF NOT EXISTS role_capabilities (
+            role_id    TEXT NOT NULL REFERENCES roles(id) ON DELETE CASCADE,
+            capability TEXT NOT NULL,
+            PRIMARY KEY (role_id, capability)
+        )",
+    )
+    .execute(pool)
+    .await?;
+
     Ok(())
 }
 
@@ -626,25 +710,6 @@ pub async fn create_owner_tables(pool: &SqlitePool) -> Result<()> {
     .execute(pool)
     .await?;
 
-    sqlx::query(
-        "CREATE TABLE IF NOT EXISTS mcp_servers (
-            id            INTEGER PRIMARY KEY AUTOINCREMENT,
-            name          TEXT    NOT NULL UNIQUE,
-            transport     TEXT    NOT NULL DEFAULT 'stdio',
-            command       TEXT,
-            args_json     TEXT,
-            env_json      TEXT,
-            url           TEXT,
-            api_key       TEXT,
-            description   TEXT,
-            friendly_name TEXT,
-            enabled       INTEGER NOT NULL DEFAULT 1,
-            created_at    TEXT    NOT NULL DEFAULT (datetime('now'))
-        )",
-    )
-    .execute(pool)
-    .await?;
-
     sqlx::query(
         "CREATE TABLE IF NOT EXISTS mcp_events (
             id           INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -667,6 +732,35 @@ pub async fn create_owner_tables(pool: &SqlitePool) -> Result<()> {
     .execute(pool)
     .await?;
 
+    // A user's activated per-user connectors (blueprint §7/§14). Owner table:
+    // encrypted at rest in `{userid}.db`, so `api_key` (a personal secret / OAuth
+    // refresh token) needs no column-level crypto. `catalog_name` is a BARE `TEXT`
+    // snapshot of `mcp_catalog.name`, never a FK — an owner→registry key would pass
+    // CREATE TABLE and fail every INSERT under `PRAGMA foreign_keys=ON` in an
+    // isolated file (guarded by `owner_tables_stand_alone_with_foreign_keys_on`).
+    // Local-script connectors run INSIDE the user's container against a script
+    // copied into the bind-mounted home (`script_rel_path`).
+    sqlx::query(
+        "CREATE TABLE IF NOT EXISTS mcp_user_servers (
+            id              INTEGER PRIMARY KEY AUTOINCREMENT,
+            name            TEXT    NOT NULL UNIQUE,
+            catalog_name    TEXT,                              -- bare ref to mcp_catalog.name; NULL = self-registered remote
+            source          TEXT    NOT NULL,                  -- 'remote' | 'local_script'
+            transport       TEXT    NOT NULL DEFAULT 'stdio',
+            command         TEXT,
+            args_json       TEXT,
+            env_json        TEXT,
+            url             TEXT,
+            api_key         TEXT,                              -- per-user secret / OAuth refresh token
+            script_rel_path TEXT,                              -- container path for a local_script
+            auth_state      TEXT    NOT NULL DEFAULT 'ready',  -- 'pending' | 'ready'
+            enabled         INTEGER NOT NULL DEFAULT 1,
+            created_at      TEXT    NOT NULL DEFAULT (datetime('now'))
+        )",
+    )
+    .execute(pool)
+    .await?;
+
     sqlx::query(
         "CREATE TABLE IF NOT EXISTS sources (
             id                TEXT    PRIMARY KEY,
@@ -841,7 +935,9 @@ mod tests {
         one("INSERT INTO scheduled_jobs (id, title, cron, prompt, session_id) VALUES (1, 't', '* * * * *', 'p', 1)")
             .await.unwrap();
         one("INSERT INTO job_runs (job_id, started_at, status) VALUES (1, 'now', 'completed')").await.unwrap();
-        one("INSERT INTO mcp_servers (name) VALUES ('srv')").await.unwrap();
+        // Owner table with a BARE `catalog_name` ref — proves it stands alone with
+        // FKs on (an owner→registry FK here would die on this INSERT).
+        one("INSERT INTO mcp_user_servers (name, catalog_name, source) VALUES ('u', 'whatsapp', 'local_script')").await.unwrap();
         one("INSERT INTO mcp_events (source, method, payload) VALUES ('s', 'm', '{}')").await.unwrap();
         one("INSERT INTO sources (id, active_session_id) VALUES ('web', 1)").await.unwrap();
         one("INSERT INTO secrets (key, value) VALUES ('k', 'v')").await.unwrap();

crates/skald-core/src/db/role_capabilities.rs

@@ -0,0 +1,88 @@
+//! Capability grants per role (blueprint §14).
+//!
+//! Registry table in `system.db`. Authorization is a **capability on the role**,
+//! not `if role == admin` (§0.1). The MCP registration axis (§14):
+//!
+//! - [`REGISTER_REMOTE`] / [`REGISTER_LOCAL_FROM_CATALOG`] — self-service, any user
+//!   (egress-only / already-vetted code).
+//! - [`REGISTER_LOCAL_SCRIPT`] / [`MANAGE_CATALOG`] — admin-only (RCE / catalog
+//!   curation).
+//!
+//! The built-in `admin` role implicitly holds every capability — [`has`] short-
+//! circuits on it — so only non-admin roles ever need rows here.
+
+use anyhow::Result;
+use sqlx::SqlitePool;
+
+use super::roles::ADMIN_ROLE_ID;
+
+/// Register a remote MCP into one's own scope (egress-only, self-service).
+pub const REGISTER_REMOTE: &str = "mcp.register_remote";
+/// Instantiate an admin-vetted local-script connector from the catalog.
+pub const REGISTER_LOCAL_FROM_CATALOG: &str = "mcp.register_local_from_catalog";
+/// Add a brand-new local script to the catalog (RCE surface — admin only).
+pub const REGISTER_LOCAL_SCRIPT: &str = "mcp.register_local_script";
+/// Curate the connector catalog (admin only).
+pub const MANAGE_CATALOG: &str = "mcp.manage_catalog";
+
+/// The default capabilities of an ordinary (non-admin) user role.
+pub const DEFAULT_USER_CAPABILITIES: &[&str] = &[REGISTER_REMOTE, REGISTER_LOCAL_FROM_CATALOG];
+
+// ── Reads ────────────────────────────────────────────────────────────────────
+
+/// Whether a role holds a capability. `admin` holds everything by construction.
+pub async fn has(pool: &SqlitePool, role_id: &str, capability: &str) -> Result<bool> {
+    if role_id == ADMIN_ROLE_ID {
+        return Ok(true);
+    }
+    let row = sqlx::query_as::<_, (i64,)>(
+        "SELECT 1 FROM role_capabilities WHERE role_id = ? AND capability = ?",
+    )
+    .bind(role_id)
+    .bind(capability)
+    .fetch_optional(pool)
+    .await?;
+    Ok(row.is_some())
+}
+
+pub async fn list_for_role(pool: &SqlitePool, role_id: &str) -> Result<Vec<String>> {
+    let rows = sqlx::query_as::<_, (String,)>(
+        "SELECT capability FROM role_capabilities WHERE role_id = ? ORDER BY capability",
+    )
+    .bind(role_id)
+    .fetch_all(pool)
+    .await?;
+    Ok(rows.into_iter().map(|(c,)| c).collect())
+}
+
+// ── Writes ───────────────────────────────────────────────────────────────────
+
+pub async fn grant(pool: &SqlitePool, role_id: &str, capability: &str) -> Result<()> {
+    sqlx::query("INSERT OR IGNORE INTO role_capabilities (role_id, capability) VALUES (?, ?)")
+        .bind(role_id)
+        .bind(capability)
+        .execute(pool)
+        .await?;
+    Ok(())
+}
+
+pub async fn revoke(pool: &SqlitePool, role_id: &str, capability: &str) -> Result<()> {
+    sqlx::query("DELETE FROM role_capabilities WHERE role_id = ? AND capability = ?")
+        .bind(role_id)
+        .bind(capability)
+        .execute(pool)
+        .await?;
+    Ok(())
+}
+
+/// Seeds the standard non-admin capability set for a newly created role.
+/// Idempotent. No-op for `admin` (which holds everything implicitly).
+pub async fn seed_defaults(pool: &SqlitePool, role_id: &str) -> Result<()> {
+    if role_id == ADMIN_ROLE_ID {
+        return Ok(());
+    }
+    for cap in DEFAULT_USER_CAPABILITIES {
+        grant(pool, role_id, cap).await?;
+    }
+    Ok(())
+}

crates/skald-core/src/mcp/mod.rs

@@ -25,6 +25,9 @@ pub use mcp_client::{
 use mcp_client::McpTransport;
 
 mod logs;
+mod provider;
+
+pub use provider::{McpProvider, UserMcpView};
 
 const SERVER_START_TIMEOUT_SECS: u64 = 120;
 
@@ -116,22 +119,6 @@ impl McpManager {
         }
     }
 
-    fn cfg_from_row(row: &crate::db::mcp_servers::McpServerRow) -> McpServerConfig {
-        McpServerConfig {
-            name:      row.name.clone(),
-            transport: match row.transport.as_str() {
-                "http" => McpTransport::Http,
-                "sse"  => McpTransport::Sse,
-                _      => McpTransport::Stdio,
-            },
-            command: row.command.clone(),
-            args:    Some(row.args()).filter(|v| !v.is_empty()),
-            env:     Some(row.env()).filter(|m| !m.is_empty()),
-            url:     row.url.clone(),
-            api_key: row.api_key.clone(),
-        }
-    }
-
     async fn start_one(
         cfg: &McpServerConfig,
         notification_tx: Option<mpsc::UnboundedSender<McpNotification>>,
@@ -154,29 +141,49 @@ impl McpManager {
         }
     }
 
+    /// Connects the GLOBAL runtime at boot: reads the enabled globally-active
+    /// connectors (`mcp_global_servers`, host transport) and connects to them.
+    /// The per-user runtime (blueprint §7/§9) is built separately at login and
+    /// shares [`connect_all`] rather than this table-bound entry point.
     pub async fn initialize(&self) {
-        let rows = match crate::db::mcp_servers::all_enabled(&self.pool).await {
+        let rows = match crate::db::mcp_global_servers::all_enabled(&self.pool).await {
             Ok(r) => r,
             Err(e) => { warn!("McpManager::initialize: failed to read DB: {e}"); return; }
         };
 
         if rows.is_empty() {
-            info!("No enabled MCP servers in DB — MCP disabled.");
+            info!("No enabled global MCP servers in DB — global MCP disabled.");
             crate::boot::section("MCP servers — none enabled");
             return;
         }
 
-        let cfgs: Vec<_> = rows.iter().map(Self::cfg_from_row).collect();
+        let specs = rows.iter().map(global_row_spec).collect();
+        self.connect_all(specs, true).await;
+    }
+
+    /// Connects to a batch of servers in parallel (each bounded by
+    /// `SERVER_START_TIMEOUT_SECS`), recording their tools, errors and prompt
+    /// descriptions. The reusable core shared by the global runtime
+    /// ([`initialize`]) and the per-user runtime (built at login, §7). `boot`
+    /// gates the curated boot-console lines, which only make sense at startup —
+    /// a login-time per-user connect passes `false`.
+    pub async fn connect_all(&self, specs: Vec<McpServerSpec>, boot: bool) {
+        if specs.is_empty() {
+            return;
+        }
         {
             let mut descs = self.descriptions.write().unwrap();
-            for row in &rows {
-                descs.insert(row.name.clone(), row.description.clone());
+            for spec in &specs {
+                descs.insert(spec.config.name.clone(), spec.description.clone());
             }
         }
-        crate::boot::section(format!(
-            "MCP servers — connecting to {} in background", cfgs.len()
-        ));
-        let handles: Vec<_> = cfgs.into_iter().map(|cfg| {
+        if boot {
+            crate::boot::section(format!(
+                "MCP servers — connecting to {} in background", specs.len()
+            ));
+        }
+        let handles: Vec<_> = specs.into_iter().map(|spec| {
+            let cfg    = spec.config;
             let tx     = self.notification_tx.clone();
             let log_tx = self.log_tx.clone();
             let eh = self.elicitation_handler();
@@ -186,30 +193,33 @@ impl McpManager {
                     Duration::from_secs(SERVER_START_TIMEOUT_SECS),
                     Self::start_one(&cfg, Some(tx), Some(log_tx), eh),
                 ).await;
-                (cfg.name, cfg.transport, result)
+                (cfg.name, result)
             })
         }).collect();
 
         for handle in handles {
             match handle.await {
-                Ok((name, _, Ok(Ok(s)))) => {
-                    let tool_names: Vec<_> = s.tools().iter().map(|t| t.name.as_str()).collect();
-                    info!("MCP server '{}' ready — {} tool(s): {}", name, tool_names.len(), tool_names.join(", "));
+                Ok((name, Ok(Ok(s)))) => {
+                    let tool_names: Vec<_> = s.tools().iter().map(|t| t.name.clone()).collect();
                     let n = tool_names.len();
-                    crate::boot::ok(format!("{name} ({n} tool{})", if n == 1 { "" } else { "s" }));
+                    info!("MCP server '{}' ready — {n} tool(s): {}", name, tool_names.join(", "));
+                    if boot {
+                        crate::boot::ok(format!("{name} ({n} tool{})", if n == 1 { "" } else { "s" }));
+                    }
                     self.log_lifecycle(&name, format!("connected — {n} tool(s)"));
+                    self.errors.write().unwrap().remove(&name);
                     self.servers.write().unwrap().insert(name, s);
                 }
-                Ok((name, _, Ok(Err(e)))) => {
+                Ok((name, Ok(Err(e)))) => {
                     warn!("MCP server '{}' failed to start: {e}", name);
-                    crate::boot::fail(format!("{name} — {e}"));
+                    if boot { crate::boot::fail(format!("{name} — {e}")); }
                     self.log_lifecycle(&name, format!("failed to start: {e}"));
                     self.errors.write().unwrap().insert(name, e.to_string());
                 }
-                Ok((name, _, Err(_))) => {
+                Ok((name, Err(_))) => {
                     let msg = format!("startup timed out after {SERVER_START_TIMEOUT_SECS}s");
                     warn!("MCP server '{}' {msg}", name);
-                    crate::boot::fail(format!("{name} — {msg}"));
+                    if boot { crate::boot::fail(format!("{name} — {msg}")); }
                     self.log_lifecycle(&name, &msg);
                     self.errors.write().unwrap().insert(name, msg);
                 }
@@ -218,77 +228,38 @@ impl McpManager {
         }
     }
 
-    pub async fn register(&self, p: crate::db::mcp_servers::UpsertParams<'_>) -> Result<Vec<String>> {
-        let name = p.name.to_string();
-
-        crate::db::mcp_servers::upsert(&self.pool, p).await?;
-
-        let rows = crate::db::mcp_servers::all_enabled(&self.pool).await?;
-        let row = rows.into_iter().find(|r| r.name == name)
-            .ok_or_else(|| anyhow::anyhow!("register: server '{}' not found after upsert", name))?;
-        let cfg = Self::cfg_from_row(&row);
-
+    /// Starts (or restarts) a single server from a spec and records it in the
+    /// runtime maps. The DB write is the caller's job (the Connectors activation
+    /// API) — this only touches the live connections. Returns the tool names.
+    pub async fn start_server(&self, spec: McpServerSpec) -> Result<Vec<String>> {
+        let name = spec.config.name.clone();
         let client = tokio::time::timeout(
             Duration::from_secs(SERVER_START_TIMEOUT_SECS),
-            Self::start_one(&cfg, Some(self.notification_tx.clone()), Some(self.log_tx.clone()), self.elicitation_handler()),
+            Self::start_one(&spec.config, Some(self.notification_tx.clone()), Some(self.log_tx.clone()), self.elicitation_handler()),
         ).await
         .map_err(|_| {
             self.log_lifecycle(&name, "timed out during connection");
-            anyhow::anyhow!("MCP server '{}' timed out during connection", name)
+            anyhow::anyhow!("MCP server '{name}' timed out during connection")
         })?
         .map_err(|e| {
             self.log_lifecycle(&name, format!("failed to start: {e}"));
-            anyhow::anyhow!("MCP server '{}' failed to start: {e}", name)
+            anyhow::anyhow!("MCP server '{name}' failed to start: {e}")
         })?;
 
         let tool_names: Vec<String> = client.tools().iter().map(|t| t.name.clone()).collect();
         self.log_lifecycle(&name, format!("connected — {} tool(s)", tool_names.len()));
         self.errors.write().unwrap().remove(&name);
-        self.descriptions.write().unwrap().insert(name.clone(), row.description.clone());
+        self.descriptions.write().unwrap().insert(name.clone(), spec.description);
         self.servers.write().unwrap().insert(name, client);
-
         Ok(tool_names)
     }
 
-    pub async fn unregister(&self, name: &str) -> Result<()> {
-        crate::db::mcp_servers::delete(&self.pool, name).await?;
+    /// Stops a running server (dropping the client → `kill_on_drop`) and forgets
+    /// it. DB removal is the caller's responsibility.
+    pub fn stop_server(&self, name: &str) {
         self.servers.write().unwrap().remove(name);
         self.errors.write().unwrap().remove(name);
         self.descriptions.write().unwrap().remove(name);
-        Ok(())
-    }
-
-    pub async fn set_enabled(&self, name: &str, enabled: bool) -> Result<()> {
-        crate::db::mcp_servers::set_enabled(&self.pool, name, enabled).await
-    }
-
-    pub async fn list(&self) -> Result<Vec<McpServerInfo>> {
-        let rows = crate::db::mcp_servers::all(&self.pool).await?;
-        let servers = self.servers.read().unwrap();
-        let errors  = self.errors.read().unwrap();
-
-        let infos = rows.into_iter().map(|row| {
-            let status = if !row.enabled {
-                McpServerStatus::Disabled
-            } else if let Some(s) = servers.get(&row.name) {
-                McpServerStatus::Running {
-                    tools: s.tools().iter().map(|t| t.name.clone()).collect(),
-                }
-            } else if let Some(e) = errors.get(&row.name) {
-                McpServerStatus::Error { message: e.clone() }
-            } else {
-                McpServerStatus::Error { message: "not connected".to_string() }
-            };
-            McpServerInfo {
-                name: row.name,
-                transport: row.transport,
-                description: row.description,
-                friendly_name: row.friendly_name,
-                status,
-            }
-        }).collect();
-
-        Ok(infos)
     }
 
     pub fn tools(&self) -> Vec<McpTool> {
@@ -405,6 +376,68 @@ impl McpManager {
     }
 }
 
+/// A server to connect: its transport config plus the description shown in the
+/// "Available MCP servers" prompt section. Decouples [`McpManager`] from any DB
+/// table — the global and per-user runtimes each build these from their own rows
+/// (`global_row_spec` / `user_row_spec`).
+pub struct McpServerSpec {
+    pub config:      McpServerConfig,
+    pub description: Option<String>,
+}
+
+fn transport_of(s: &str) -> McpTransport {
+    match s {
+        "http" => McpTransport::Http,
+        "sse"  => McpTransport::Sse,
+        _      => McpTransport::Stdio,
+    }
+}
+
+/// Builds a spec for a globally-active connector — host transport (`launch_in`
+/// = None), so it runs in the Skald process, not in any container (§7).
+pub fn global_row_spec(row: &crate::db::mcp_global_servers::McpGlobalServerRow) -> McpServerSpec {
+    McpServerSpec {
+        config: McpServerConfig {
+            name:      row.name.clone(),
+            transport: transport_of(&row.transport),
+            command:   row.command.clone(),
+            args:      Some(row.args()).filter(|v| !v.is_empty()),
+            env:       Some(row.env()).filter(|m| !m.is_empty()),
+            url:       row.url.clone(),
+            api_key:   row.api_key.clone(),
+            launch_in: None,
+        },
+        description: row.description.clone(),
+    }
+}
+
+/// Builds a spec for a user's per-user connector — container transport: a
+/// `local_script` (or any stdio server) runs INSIDE the user's container
+/// (`launch_in = Some(container)`), against the script copied into the
+/// bind-mounted home. Remote (HTTP) connectors ignore `launch_in`.
+pub fn user_row_spec(
+    row:       &crate::db::mcp_user_servers::McpUserServerRow,
+    container: &str,
+) -> McpServerSpec {
+    let transport = transport_of(&row.transport);
+    let launch_in = matches!(transport, McpTransport::Stdio).then(|| container.to_string());
+    McpServerSpec {
+        config: McpServerConfig {
+            name:      row.name.clone(),
+            transport,
+            command:   row.command.clone(),
+            args:      Some(row.args()).filter(|v| !v.is_empty()),
+            env:       Some(row.env()).filter(|m| !m.is_empty()),
+            url:       row.url.clone(),
+            api_key:   row.api_key.clone(),
+            launch_in,
+        },
+        // A per-user connector's description falls back to its catalog name; the
+        // catalog's friendly description can be injected by the caller if richer.
+        description: row.catalog_name.clone(),
+    }
+}
+
 /// Generates a 32-char alphanumeric id for a persisted media filename
 /// (mirrors `ImageGeneratorManager`).
 fn random_id() -> String {

crates/skald-core/src/mcp/provider.rs

@@ -0,0 +1,109 @@
+//! The MCP tool surface a session sees, behind one trait.
+//!
+//! A logged-in user's tools are the union of two runtimes (blueprint §7): the
+//! access-filtered GLOBAL runtime (host, shared) and their own PER-USER runtime
+//! (in their container). [`McpProvider`] is the seam the session code talks to,
+//! so the round-loop (`all_tool_defs`, `render_mcp_list`, `ActivateTools`) never
+//! has to know which runtime owns a server. [`McpManager`] implements it directly
+//! (used as-is for the inert ownerless bundle); [`UserMcpView`] implements it as
+//! the union.
+
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+
+use anyhow::Result;
+use async_trait::async_trait;
+use serde_json::Value;
+
+use crate::tools::ToolResult;
+
+use super::{McpManager, McpTool};
+
+#[async_trait]
+pub trait McpProvider: Send + Sync {
+    fn tools(&self) -> Vec<McpTool>;
+    fn tools_for(&self, names: &[String]) -> Vec<McpTool>;
+    fn server_descriptions(&self) -> HashMap<String, Option<String>>;
+    fn server_infos(&self) -> Vec<Value>;
+    async fn call(&self, server: &str, tool: &str, args: Value) -> Result<ToolResult>;
+}
+
+#[async_trait]
+impl McpProvider for McpManager {
+    fn tools(&self) -> Vec<McpTool> { McpManager::tools(self) }
+    fn tools_for(&self, names: &[String]) -> Vec<McpTool> { McpManager::tools_for(self, names) }
+    fn server_descriptions(&self) -> HashMap<String, Option<String>> { McpManager::server_descriptions(self) }
+    fn server_infos(&self) -> Vec<Value> { McpManager::server_infos(self) }
+    async fn call(&self, server: &str, tool: &str, args: Value) -> Result<ToolResult> {
+        McpManager::call(self, server, tool, args).await
+    }
+}
+
+/// One logged-in user's MCP view: the access-filtered global runtime unioned with
+/// their per-user container runtime. A per-user server wins on a name collision
+/// (which activation prevents anyway — see the uniqueness check at activation).
+pub struct UserMcpView {
+    pub global: Arc<McpManager>,
+    pub user:   Arc<McpManager>,
+    /// Names of the global servers this user may use — a snapshot of
+    /// `mcp_global_access`, captured when the user's context is built.
+    pub accessible_global: HashSet<String>,
+}
+
+impl UserMcpView {
+    fn accessible_names(&self) -> Vec<String> {
+        self.accessible_global.iter().cloned().collect()
+    }
+}
+
+#[async_trait]
+impl McpProvider for UserMcpView {
+    fn tools(&self) -> Vec<McpTool> {
+        let mut out = self.global.tools_for(&self.accessible_names());
+        out.extend(self.user.tools());
+        out
+    }
+
+    fn tools_for(&self, names: &[String]) -> Vec<McpTool> {
+        // A granted name belongs to exactly one runtime (unique per user); route
+        // the accessible-global ones to the global runtime and the rest to the
+        // per-user one, which filters to its own server map.
+        let global_names: Vec<String> = names.iter()
+            .filter(|n| self.accessible_global.contains(*n))
+            .cloned()
+            .collect();
+        let mut out = self.global.tools_for(&global_names);
+        out.extend(self.user.tools_for(names));
+        out
+    }
+
+    fn server_descriptions(&self) -> HashMap<String, Option<String>> {
+        let mut m: HashMap<String, Option<String>> = self.global.server_descriptions()
+            .into_iter()
+            .filter(|(name, _)| self.accessible_global.contains(name))
+            .collect();
+        m.extend(self.user.server_descriptions());
+        m
+    }
+
+    fn server_infos(&self) -> Vec<Value> {
+        let mut v: Vec<Value> = self.global.server_infos()
+            .into_iter()
+            .filter(|info| info["name"].as_str()
+                .map(|n| self.accessible_global.contains(n))
+                .unwrap_or(false))
+            .collect();
+        v.extend(self.user.server_infos());
+        v
+    }
+
+    async fn call(&self, server: &str, tool: &str, args: Value) -> Result<ToolResult> {
+        if self.accessible_global.contains(server) {
+            self.global.call(server, tool, args).await
+        } else {
+            // A per-user server, or an unknown/forbidden one — the per-user
+            // runtime returns a "not found" error for the latter.
+            self.user.call(server, tool, args).await
+        }
+    }
+}

crates/skald-core/src/session/handler/interface_tools.rs

@@ -3,7 +3,7 @@ use std::sync::{Arc, RwLock};
 
 use serde_json::Value;
 
-use crate::mcp::McpManager;
+use crate::mcp::McpProvider;
 use crate::tools::Tool;
 use crate::tools::tool_names as tn;
 
@@ -51,8 +51,9 @@ pub struct AgentRunConfig {
     pub memory_tools: Vec<Arc<dyn Tool>>,
     /// Image generation tools — present only when at least one provider is registered.
     pub image_tools: Vec<Arc<dyn Tool>>,
-    /// MCP manager — used by `all_tool_defs()` to resolve which tools to include.
-    pub mcp: Arc<McpManager>,
+    /// MCP provider (global ∪ per-user) — used by `all_tool_defs()` to resolve
+    /// which tools to include.
+    pub mcp: Arc<dyn McpProvider>,
     /// Set of MCP server names currently granted (activated) for this agent run.
     ///
     /// - Root agents: pre-populated from `session_mcp_grants` DB at config-build time;

crates/skald-core/src/session/handler/message_builder.rs

@@ -7,7 +7,7 @@ use sqlx::SqlitePool;
 use crate::compactor::{ContextCompactor, SUMMARY_PREFIX};
 use crate::config::DatetimeConfig;
 use crate::db::{chat_history, chat_llm_tools, chat_summaries};
-use crate::mcp::McpManager;
+use crate::mcp::McpProvider;
 use crate::tools::tool_names as tn;
 
 /// Registry of installed skills, relative to Skald's process cwd. Injected into agents
@@ -38,7 +38,7 @@ pub struct MessageBuilder {
     /// owner `pool` above backs `user-memory/`.
     pub shared_pool:           Arc<SqlitePool>,
     pub session_id:            i64,
-    pub mcp:                   Arc<McpManager>,
+    pub mcp:                   Arc<dyn McpProvider>,
     pub datetime_config:       DatetimeConfig,
     pub max_history_messages:  usize,
     pub max_tool_result_chars: Option<usize>,

crates/skald-core/src/session/handler/mod.rs

@@ -22,7 +22,7 @@ use crate::events::ServerEvent;
 use core_api::message_meta::MessageMetadata;
 use core_api::user_fs::UserFs;
 use crate::llm::LlmManager;
-use crate::mcp::McpManager;
+use crate::mcp::McpProvider;
 use crate::image_generate::ImageGeneratorManager;
 use crate::memory::MemoryManager;
 use crate::tool_discovery::ToolDiscovery;
@@ -292,7 +292,7 @@ pub struct ChatSessionHandler {
     /// True for short-lived automated sessions (cron, tic).
     pub(super) is_ephemeral:     bool,
     pub(super) tools:            Arc<ToolRegistry>,
-    pub(super) mcp:              Arc<McpManager>,
+    pub(super) mcp:              Arc<dyn McpProvider>,
     /// Records tools offered to the LLM each round so the Security-groups UI can
     /// list/gate dynamically-injected tools (interface/plugin/provider tools).
     pub(super) tool_discovery:   Arc<ToolDiscovery>,
@@ -354,7 +354,7 @@ impl ChatSessionHandler {
         is_interactive:        bool,
         is_ephemeral:          bool,
         tools:                 Arc<ToolRegistry>,
-        mcp:                   Arc<McpManager>,
+        mcp:                   Arc<dyn McpProvider>,
         approval:              Arc<ApprovalManager>,
         clarification:         Arc<ClarificationManager>,
         event_bus:             Arc<ChatEventBus>,

crates/skald-core/src/session/manager.rs

@@ -13,7 +13,7 @@ use crate::compactor::ContextCompactor;
 use crate::config::DatetimeConfig;
 use crate::db::{chat_sessions, chat_sessions_stack};
 use crate::llm::LlmManager;
-use crate::mcp::McpManager;
+use crate::mcp::McpProvider;
 use crate::image_generate::ImageGeneratorManager;
 use crate::memory::MemoryManager;
 use crate::run_context::{RunContext, RunContextManager};
@@ -38,7 +38,9 @@ pub struct ChatSessionManager {
     max_tool_result_chars: Option<usize>,
     datetime_config:       DatetimeConfig,
     tools:                 Arc<ToolRegistry>,
-    mcp:                   Arc<McpManager>,
+    /// The MCP tools visible to this owner: the access-filtered global runtime
+    /// unioned with their per-user runtime (blueprint §7), behind one trait.
+    mcp:                   Arc<dyn McpProvider>,
     approval:              Arc<ApprovalManager>,
     clarification:         Arc<ClarificationManager>,
     event_bus:             Arc<ChatEventBus>,
@@ -66,7 +68,7 @@ impl ChatSessionManager {
         max_tool_result_chars: Option<usize>,
         datetime_config:       DatetimeConfig,
         tools:                 Arc<ToolRegistry>,
-        mcp:                   Arc<McpManager>,
+        mcp:                   Arc<dyn McpProvider>,
         approval:              Arc<ApprovalManager>,
         clarification:         Arc<ClarificationManager>,
         event_bus:             Arc<ChatEventBus>,

crates/skald-core/src/skald/bundles.rs

@@ -211,13 +211,13 @@ impl Tools {
         tool_registry.register(crate::tools::exec::ExecuteCmd);
         tool_registry.register(crate::tools::read_notification::ReadNotification);
         tool_registry.register(crate::tools::restart::Restart);
-        // Unified listing / toggling across mcp, plugins, cron (+ agents for list).
+        // Unified listing / toggling across plugins, cron (+ agents for list). MCP
+        // is no longer agent-managed (blueprint §14): connectors are curated by the
+        // admin and activated by the user via the Connectors UI/API, not tools.
         tool_registry.register(crate::tools::list_items::ListItems::new(
-            Arc::clone(&integrations.mcp), Arc::clone(&integrations.plugin_manager), Arc::clone(&tasks.cron)));
+            Arc::clone(&integrations.plugin_manager), Arc::clone(&tasks.cron)));
         tool_registry.register(crate::tools::toggle_item::ToggleItem::new(
-            Arc::clone(&integrations.mcp), Arc::clone(&integrations.plugin_manager), Arc::clone(&tasks.cron)));
-        tool_registry.register(crate::tools::register_mcp::RegisterMcp::new(Arc::clone(&integrations.mcp)));
-        tool_registry.register(crate::tools::register_mcp::DeleteMcp::new(Arc::clone(&integrations.mcp)));
+            Arc::clone(&integrations.plugin_manager), Arc::clone(&tasks.cron)));
         tool_registry.register(crate::tools::cron_jobs::DeleteCronJob);
         tool_registry.register(crate::tools::set_secret::SetSecret(Arc::clone(&models.secrets)));
         tool_registry.register(crate::tools::list_secrets::ListSecrets(Arc::clone(&models.secrets)));
@@ -358,7 +358,9 @@ impl Conversation {
             config.llm.max_tool_result_chars,
             DatetimeConfig { timezone: config.timezone.clone(), ..config.llm.datetime },
             Arc::clone(&tools.tools),
-            Arc::clone(&integrations.mcp),
+            // Inert ownerless bundle (§19): the global runtime as a provider,
+            // unfiltered — never actually exercised (no loops, no consumers).
+            Arc::clone(&integrations.mcp) as Arc<dyn crate::mcp::McpProvider>,
             Arc::clone(&interaction.approval),
             Arc::clone(&interaction.clarification),
             Arc::clone(&rt.event_bus),

crates/skald-core/src/skald/mod.rs

@@ -88,7 +88,7 @@ impl Skald {
         // Per-user context factory: captures the global capability managers, so a
         // per-user chat/hub/cron/interaction stack can be stamped out on demand.
         let user_contexts = UserContextRegistry::new(UserContextFactory::new(
-            &rt, &models, &media, &tools, &integrations, &conversation, config,
+            &rt, &models, &media, &tools, &integrations, &conversation, &container, config,
         ));
 
         // Build the runtime image and reconcile a container for every active user.

crates/skald-core/src/skald/user_context.rs

@@ -43,12 +43,13 @@ use crate::chat_hub::ChatHub;
 use crate::clarification::ClarificationManager;
 use crate::compactor::ContextCompactor;
 use crate::config::{CompactionConfig, CoreConfig, DatetimeConfig};
+use crate::container::ContainerManager;
 use crate::cron::TaskManager;
 use crate::elicitation::ElicitationManager;
 use crate::image_generate::ImageGeneratorManager;
 use crate::inbox::Inbox;
 use crate::llm::LlmManager;
-use crate::mcp::McpManager;
+use crate::mcp::{McpManager, McpProvider, UserMcpView};
 use crate::memory::MemoryManager;
 use crate::projects::tickets::ProjectTicketManager;
 use crate::run_context::RunContextManager;
@@ -76,6 +77,11 @@ pub struct UserContext {
     pub clarification: Arc<ClarificationManager>,
     pub elicitation:   Arc<ElicitationManager>,
     pub inbox:         Inbox,
+    /// This user's own MCP runtime (blueprint §7/§9): connectors that run inside
+    /// their container, started at first login and living until restart. Held
+    /// here so its lifetime equals the pool's; its `docker exec -i` children die
+    /// via `kill_on_drop` when the context is dropped at shutdown.
+    pub user_mcp:      Arc<McpManager>,
     /// Per-user server→client push channel. WS handlers subscribe here (via the
     /// hub) so a user's `ServerEvent`s never reach another user's socket.
     pub global_tx:     broadcast::Sender<GlobalEvent>,
@@ -87,7 +93,12 @@ pub(super) struct UserContextFactory {
     registry_pool:           Arc<SqlitePool>,
     llm_manager:             Arc<LlmManager>,
     tools:                   Arc<ToolRegistry>,
+    /// The GLOBAL MCP runtime (host, shared). Unioned per-user with the per-user
+    /// runtime built at login (`UserMcpView`).
     mcp:                     Arc<McpManager>,
+    /// Container lifecycle — used to ensure a user's container is up before their
+    /// per-user (container-hosted) MCP connectors start.
+    container:               ContainerManager,
     memory_manager:          Arc<MemoryManager>,
     image_generator_manager: Arc<ImageGeneratorManager>,
     run_context_manager:     Arc<RunContextManager>,
@@ -111,6 +122,7 @@ impl UserContextFactory {
         tools:        &Tools,
         integrations: &Integrations,
         conversation: &Conversation,
+        container:    &ContainerManager,
         config:       &CoreConfig,
     ) -> Self {
         let cron_tz = config.timezone.as_deref().and_then(|s| s.parse::<Tz>().ok());
@@ -119,6 +131,7 @@ impl UserContextFactory {
             llm_manager:             Arc::clone(&models.llm_manager),
             tools:                   Arc::clone(&tools.tools),
             mcp:                     Arc::clone(&integrations.mcp),
+            container:               container.clone(),
             memory_manager:          Arc::clone(&models.memory_manager),
             image_generator_manager: Arc::clone(&media.image_generator_manager),
             run_context_manager:     Arc::clone(&conversation.run_context_manager),
@@ -165,6 +178,58 @@ impl UserContextFactory {
             ))
         });
 
+        // Per-user MCP runtime (blueprint §7/§9): the connectors this user has
+        // activated, run INSIDE their container. Started here on first login and
+        // living until restart — its `docker exec -i` children die via
+        // `kill_on_drop` when this context (holding `user_mcp`) is dropped at
+        // shutdown. Ensure the container is up first (idempotent: boot
+        // reconciliation and user-create already do this; the belt-and-braces call
+        // recovers a container stopped since). Non-fatal — a container hiccup
+        // degrades MCP/exec but must not block login.
+        if let Err(e) = self.container.ensure(user_id).await {
+            tracing::warn!(user = %user_id, error = %e, "failed to ensure container before per-user MCP start");
+        }
+        let user_mcp = Arc::new(McpManager::new(
+            Arc::clone(&pool),
+            self.shutdown_token.clone(),
+            "data",
+        ));
+        // NOTE: per-user MCP elicitation (interactive connector login, §15) is
+        // deferred — api-key connectors don't need it. Wire the user's
+        // ElicitationBridge here when interactive auth lands.
+        {
+            let um        = Arc::clone(&user_mcp);
+            let upool     = Arc::clone(&pool);
+            let container = crate::container::container_name(user_id);
+            let mname: &'static str = Box::leak(format!("mcp:{user_id}").into_boxed_str());
+            self.supervisor.adopt_one(mname, tokio::spawn(async move {
+                match crate::db::mcp_user_servers::all_startable(&upool).await {
+                    Ok(rows) => {
+                        let specs = rows.iter()
+                            .map(|r| crate::mcp::user_row_spec(r, &container))
+                            .collect();
+                        um.connect_all(specs, false).await;
+                    }
+                    Err(e) => tracing::warn!(error = %e, "per-user MCP init: failed to read mcp_user_servers"),
+                }
+            }));
+        }
+
+        // The MCP view this user's sessions see: the access-filtered global runtime
+        // unioned with their per-user runtime (§7). `accessible_global` is a
+        // snapshot of `mcp_global_access`, captured at build time like fs membership.
+        let accessible_global: std::collections::HashSet<String> =
+            crate::db::mcp_global_access::server_names_for_user(&self.registry_pool, user_id)
+                .await
+                .unwrap_or_default()
+                .into_iter()
+                .collect();
+        let mcp_view: Arc<dyn McpProvider> = Arc::new(UserMcpView {
+            global: Arc::clone(&self.mcp),
+            user:   Arc::clone(&user_mcp),
+            accessible_global,
+        });
+
         let manager = Arc::new(ChatSessionManager::new(
             Arc::clone(&pool),
             Arc::clone(&self.registry_pool), // shared pool = system.db, for shared-memory injection
@@ -177,7 +242,7 @@ impl UserContextFactory {
             self.max_tool_result_chars,
             self.datetime_config.clone(),
             Arc::clone(&self.tools),
-            Arc::clone(&self.mcp),
+            mcp_view,
             Arc::clone(&approval),
             Arc::clone(&clarification),
             Arc::clone(&event_bus),
@@ -241,6 +306,7 @@ impl UserContextFactory {
             clarification,
             elicitation,
             inbox,
+            user_mcp,
             global_tx,
         }))
     }

crates/skald-core/src/tools/activate_tools.rs

@@ -5,7 +5,7 @@ use anyhow::Result;
 use serde_json::{Value, json};
 use sqlx::SqlitePool;
 
-use crate::mcp::McpManager;
+use crate::mcp::McpProvider;
 use crate::tools::tool_names::CONFIG_GROUP;
 use crate::tools::{Tool, ToolDescriptionLength, truncate_label, MAX_LABEL_SHORT};
 
@@ -37,7 +37,7 @@ pub struct ActivateTools {
     /// `None` for root agents (session-scoped grants).
     /// `Some(stack_id)` for sub-agents (stack-scoped grants, deleted on frame exit).
     pub stack_id:           Option<i64>,
-    pub mcp:                Arc<McpManager>,
+    pub mcp:                Arc<dyn McpProvider>,
     /// Shared in-memory grant set. Updated in-place on every call so subsequent
     /// rounds within the same turn see the new tools via `all_tool_defs()`.
     pub active_mcp_grants:  Arc<RwLock<HashSet<String>>>,

crates/skald-core/src/tools/list_items.rs

@@ -5,7 +5,6 @@ use serde_json::{Value, json};
 
 use crate::agents;
 use crate::cron::TaskManager;
-use crate::mcp::McpManager;
 use crate::plugin::PluginManager;
 use crate::tools::{Tool, ToolDescriptionLength};
 
@@ -19,14 +18,13 @@ use crate::tools::{Tool, ToolDescriptionLength};
 /// the ability to enumerate secret key names) and carries a `pattern` filter
 /// that would only apply to that one type.
 pub struct ListItems {
-    mcp:     Arc<McpManager>,
     plugins: Arc<PluginManager>,
     cron:    Arc<TaskManager>,
 }
 
 impl ListItems {
-    pub fn new(mcp: Arc<McpManager>, plugins: Arc<PluginManager>, cron: Arc<TaskManager>) -> Self {
-        Self { mcp, plugins, cron }
+    pub fn new(plugins: Arc<PluginManager>, cron: Arc<TaskManager>) -> Self {
+        Self { plugins, cron }
     }
 }
 
@@ -36,7 +34,6 @@ impl Tool for ListItems {
 
     fn description(&self) -> &str {
         "List configured items of a given type. Pass `type`:\n\
-         • `mcp` — MCP servers with status (running, error, disabled), description, friendly_name, and exposed tools.\n\
          • `plugins` — plugins with id, name, description, enabled flag (persisted), and running flag (live).\n\
          • `cron` — scheduled tasks/cron jobs with id, title, cron expression, agent_id, enabled, kind, last/next run.\n\
          • `agents` — sub-agents available to delegate to (id, name, description, optional `instructions` on how to call the agent well, optional client). Do NOT invoke the `main` agent.\n\
@@ -50,7 +47,7 @@ impl Tool for ListItems {
             "properties": {
                 "type": {
                     "type": "string",
-                    "enum": ["mcp", "plugins", "cron", "agents"],
+                    "enum": ["plugins", "cron", "agents"],
                     "description": "Which kind of item to list."
                 }
             }
@@ -67,12 +64,6 @@ impl Tool for ListItems {
             .ok_or_else(|| anyhow::anyhow!("list_items: missing required argument `type`"))?;
 
         match kind {
-            "mcp" => {
-                let infos = tokio::task::block_in_place(|| {
-                    tokio::runtime::Handle::current().block_on(self.mcp.list())
-                })?;
-                Ok(serde_json::to_string_pretty(&infos)?)
-            }
             "plugins" => {
                 let plugins = tokio::task::block_in_place(|| {
                     tokio::runtime::Handle::current().block_on(self.plugins.list())
@@ -124,7 +115,7 @@ impl Tool for ListItems {
                     .collect();
                 Ok(serde_json::to_string_pretty(&arr)?)
             }
-            other => anyhow::bail!("list_items: unknown type `{other}` (expected one of: mcp, plugins, cron, agents)"),
+            other => anyhow::bail!("list_items: unknown type `{other}` (expected one of: plugins, cron, agents)"),
         }
     }
 }

crates/skald-core/src/tools/mod.rs

@@ -43,7 +43,6 @@ pub mod list_secrets;
 pub mod notify;
 pub mod set_secret;
 pub mod read_notification;
-pub mod register_mcp;
 pub mod restart;
 pub mod show_file;
 pub mod toggle_item;

crates/skald-core/src/tools/register_mcp.rs

@@ -1,175 +0,0 @@
-use std::sync::Arc;
-
-use anyhow::Result;
-use serde_json::{Value, json};
-
-use crate::db::mcp_servers::UpsertParams;
-use crate::mcp::McpManager;
-use crate::tools::{Tool, ToolDescriptionLength};
-
-pub struct RegisterMcp {
-    mcp: Arc<McpManager>,
-}
-
-impl RegisterMcp {
-    pub fn new(mcp: Arc<McpManager>) -> Self { Self { mcp } }
-}
-
-impl Tool for RegisterMcp {
-    fn name(&self) -> &str { "register_mcp" }
-    fn category(&self) -> crate::tools::ToolCategory { crate::tools::ToolCategory::Config }
-
-    fn description(&self) -> &str {
-        "Register (or update) an MCP server and connect to it immediately. \
-         For stdio servers supply `command` and optionally `args` and `env`. \
-         For HTTP/SSE servers supply `url` and optionally `api_key`. \
-         Optionally provide `description` (what the server does) and `friendly_name` (display name for UI). \
-         Returns the list of tools exposed by the server once connected."
-    }
-
-    fn parameters_schema(&self) -> Value {
-        json!({
-            "type": "object",
-            "properties": {
-                "name": {
-                    "type": "string",
-                    "description": "Unique name for this MCP server (used to reference it in tool calls)."
-                },
-                "transport": {
-                    "type": "string",
-                    "enum": ["stdio", "http", "sse"],
-                    "description": "Connection transport. Use `stdio` for local processes, `http` for remote servers."
-                },
-                "command": {
-                    "type": "string",
-                    "description": "stdio only: executable to spawn (e.g. `npx`, `uvx`, path to binary)."
-                },
-                "args": {
-                    "type": "array",
-                    "items": { "type": "string" },
-                    "description": "stdio only: command-line arguments passed to the executable."
-                },
-                "env": {
-                    "type": "object",
-                    "additionalProperties": { "type": "string" },
-                    "description": "stdio only: extra environment variables. Values support `${VAR}` interpolation."
-                },
-                "url": {
-                    "type": "string",
-                    "description": "http/sse only: base URL of the remote MCP server."
-                },
-                "api_key": {
-                    "type": "string",
-                    "description": "http/sse only: API key sent as `Authorization: Bearer <key>`."
-                },
-                "description": {
-                    "type": "string",
-                    "description": "A short description of what this MCP server provides (shown in list_items type=mcp)."
-                },
-                "friendly_name": {
-                    "type": "string",
-                    "description": "A human-readable display name for this MCP server (e.g. 'Google Calendar')."
-                }
-            },
-            "required": ["name", "transport"]
-        })
-    }
-
-    fn describe(&self, args: &Value, _length: ToolDescriptionLength) -> String {
-        let name = args["name"].as_str().unwrap_or("?");
-        format!("register MCP `{name}`")
-    }
-
-    fn execute(&self, args: Value) -> Result<String> {
-        let name = args["name"].as_str()
-            .ok_or_else(|| anyhow::anyhow!("register_mcp: missing required argument `name`"))?;
-        let transport = args["transport"].as_str()
-            .ok_or_else(|| anyhow::anyhow!("register_mcp: missing required argument `transport`"))?;
-
-        let args_json = args["args"].as_array()
-            .map(|a| serde_json::to_string(a))
-            .transpose()?;
-        let env_json = args["env"].as_object()
-            .map(|o| serde_json::to_string(o))
-            .transpose()?;
-
-        let p = UpsertParams {
-            name,
-            transport,
-            command:       args["command"].as_str(),
-            args_json,
-            env_json,
-            url:           args["url"].as_str(),
-            api_key:       args["api_key"].as_str(),
-            description:   args["description"].as_str(),
-            friendly_name: args["friendly_name"].as_str(),
-        };
-
-        let tool_names = tokio::task::block_in_place(|| {
-            tokio::runtime::Handle::current().block_on(self.mcp.register(p))
-        })?;
-
-        Ok(format!(
-            "MCP server '{}' registered and connected. Tools: {}",
-            name,
-            if tool_names.is_empty() { "(none)".to_string() } else { tool_names.join(", ") },
-        ))
-    }
-}
-
-// ── delete_mcp ────────────────────────────────────────────────────────────────
-//
-// Destructive counterpart to `register_mcp`. Kept separate from `toggle_item`
-// (kind=mcp) for the same reason `delete_cron_job` is: toggling is reversible,
-// deletion is not, so the distinct tool can carry its own approval rule and the
-// LLM can't conflate "disable" with "remove". Both live here because both manage
-// the MCP-server lifecycle and hold only `Arc<McpManager>`.
-
-pub struct DeleteMcp {
-    mcp: Arc<McpManager>,
-}
-
-impl DeleteMcp {
-    pub fn new(mcp: Arc<McpManager>) -> Self { Self { mcp } }
-}
-
-impl Tool for DeleteMcp {
-    fn name(&self) -> &str { "delete_mcp" }
-    fn category(&self) -> crate::tools::ToolCategory { crate::tools::ToolCategory::Config }
-
-    fn description(&self) -> &str {
-        "Permanently delete (unregister) an MCP server by name: removes it from the \
-         database and disconnects it. This is irreversible — to temporarily turn a \
-         server off without losing its configuration, use \
-         `toggle_item(kind=\"mcp\", enabled=false)` instead."
-    }
-
-    fn parameters_schema(&self) -> Value {
-        json!({
-            "type": "object",
-            "required": ["name"],
-            "properties": {
-                "name": {
-                    "type": "string",
-                    "description": "Name of the MCP server to delete (from list_items type=mcp)."
-                }
-            }
-        })
-    }
-
-    fn describe(&self, args: &Value, _length: ToolDescriptionLength) -> String {
-        let name = args["name"].as_str().unwrap_or("?");
-        format!("delete MCP `{name}`")
-    }
-
-    fn execute(&self, args: Value) -> Result<String> {
-        let name = args["name"].as_str()
-            .ok_or_else(|| anyhow::anyhow!("delete_mcp: missing required argument `name`"))?;
-
-        tokio::task::block_in_place(|| {
-            tokio::runtime::Handle::current().block_on(self.mcp.unregister(name))
-        })?;
-
-        Ok(format!("MCP server '{name}' deleted and disconnected."))
-    }
-}

crates/skald-core/src/tools/toggle_item.rs

@@ -4,7 +4,6 @@ use anyhow::Result;
 use serde_json::{Value, json};
 
 use crate::cron::TaskManager;
-use crate::mcp::McpManager;
 use crate::plugin::PluginManager;
 use crate::tools::{Tool, ToolDescriptionLength};
 
@@ -16,14 +15,13 @@ use crate::tools::{Tool, ToolDescriptionLength};
 /// (irreversible) whereas toggling is reversible, and keeping it separate lets
 /// it carry a distinct approval rule.
 pub struct ToggleItem {
-    mcp:     Arc<McpManager>,
     plugins: Arc<PluginManager>,
     cron:    Arc<TaskManager>,
 }
 
 impl ToggleItem {
-    pub fn new(mcp: Arc<McpManager>, plugins: Arc<PluginManager>, cron: Arc<TaskManager>) -> Self {
-        Self { mcp, plugins, cron }
+    pub fn new(plugins: Arc<PluginManager>, cron: Arc<TaskManager>) -> Self {
+        Self { plugins, cron }
     }
 }
 
@@ -33,7 +31,6 @@ impl Tool for ToggleItem {
 
     fn description(&self) -> &str {
         "Enable or disable an item by kind. Pass `kind`, `id`, and `enabled`:\n\
-         • `mcp` — `id` is the server name. NOTE: a restart is required for the change to take full effect on running servers.\n\
          • `plugin` — `id` is the plugin id (e.g. \"telegram\"). Takes effect immediately (the plugin is started/stopped at once).\n\
          • `cron` — `id` is the numeric job id (from `list_items` type=cron). Re-enabling recalculates next_run_at.\n\
          Use `list_items` to find current names/ids and statuses."
@@ -46,12 +43,12 @@ impl Tool for ToggleItem {
             "properties": {
                 "kind": {
                     "type": "string",
-                    "enum": ["mcp", "plugin", "cron"],
+                    "enum": ["plugin", "cron"],
                     "description": "Which kind of item to toggle."
                 },
                 "id": {
                     "type": "string",
-                    "description": "MCP server name | plugin id | numeric cron job id (as a string)."
+                    "description": "plugin id | numeric cron job id (as a string)."
                 },
                 "enabled": {
                     "type": "boolean",
@@ -78,16 +75,6 @@ impl Tool for ToggleItem {
             .ok_or_else(|| anyhow::anyhow!("toggle_item: missing required argument `enabled`"))?;
 
         match kind {
-            "mcp" => {
-                tokio::task::block_in_place(|| {
-                    tokio::runtime::Handle::current().block_on(self.mcp.set_enabled(id, enabled))
-                })?;
-                Ok(format!(
-                    "MCP server '{}' is now {}. Note: a restart is required for the change to take effect on running servers.",
-                    id,
-                    if enabled { "enabled" } else { "disabled" }
-                ))
-            }
             "plugin" => {
                 tokio::task::block_in_place(|| {
                     tokio::runtime::Handle::current().block_on(self.plugins.toggle(id, enabled))
@@ -107,7 +94,7 @@ impl Tool for ToggleItem {
                     Ok(format!("No task with id {job_id}."))
                 }
             }
-            other => anyhow::bail!("toggle_item: unknown kind `{other}` (expected one of: mcp, plugin, cron)"),
+            other => anyhow::bail!("toggle_item: unknown kind `{other}` (expected one of: plugin, cron)"),
         }
     }
 }