feat(gateway): per-profile MCP server lifecycle RPCs (mcp.servers.*) (#86473)

Adds the full MCP setup surface as profile-scoped gateway RPCs so a
desktop client (Bot Mode's bot editor, the core Capabilities tab) can
add/configure/test/authenticate/remove MCP servers for ANY profile, not
just the launch profile:

- mcp.servers.list (profile) -> configured servers (transport, auth,
  oauth_tokens_present, enabled, tool names; no secret values)
- mcp.servers.add (profile, name, config|preset, bearer_token?) -> reuses
  mcp_config._apply_mcp_preset / _save_mcp_server / _save_bearer_auth_token
- mcp.servers.set_api_key (profile, name, value, env_var?) -> http auth
  header template or stdio env ref, via save_env_value
- mcp.servers.test (profile, name) -> _probe_single_server + oauth state
- mcp.servers.remove (profile, name)
- mcp.servers.oauth.start/poll (profile, name[, session_id]) -> mirrors the
  PROVIDER oauth session/poll model (not the FastAPI dashboard flow): a
  background worker drives the same interactive machinery 'hermes mcp login'
  uses, capturing the browser redirect on a local loopback listener. Client
  opens auth_url via openExternal and polls until status=='approved'.

All handlers are profile-scoped via set_hermes_home_override in try/finally
(mirrors skills.manage). Shared helpers live in tui_gateway/mcp_rpc_helpers.py
and are aliased onto server.py's namespace so the rebound handler bodies
(HandlerRegistry.install) can resolve them — a plain def in methods_tools is
unreachable post-rebind. Reuses hermes_cli/mcp_config.py throughout; no config
logic duplicated; no raw yaml near config.yaml (config-read-guard safe).

Tests: tests/tui_gateway/test_mcp_profile_rpcs.py, 8 E2E against real temp
HERMES_HOME profiles asserting add/list/set_api_key/remove land in the RIGHT
profile's config.yaml and not the launch profile's. 8/8. Registration +
live mcp.servers.list verified in an imported gateway.

Co-authored-by: Teknium <teknium1@users.noreply.github.com>

tests/tui_gateway/test_mcp_profile_rpcs.py

@@ -0,0 +1,248 @@
+"""E2E tests for the per-profile MCP lifecycle RPCs (mcp.servers.*).
+
+These drive the real registered gateway handlers against a real temp
+``HERMES_HOME`` with named profile dirs — no mocks of the config/mcp layer — and
+assert that every write lands in the RIGHT profile's ``config.yaml`` / ``.env``
+and NEVER leaks into the launch (default) profile.
+
+Covered: add + list + set_api_key + remove, profile isolation, and the
+duplicate/not-found error envelopes.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import pytest
+
+import tui_gateway.server as server
+
+
+@pytest.fixture
+def hermes_root(tmp_path, monkeypatch):
+    """A temp HERMES_HOME root with two named profiles: 'work' and 'other'.
+
+    Pointing HERMES_HOME at a dir outside ~/.hermes makes it the profile ROOT
+    (get_default_hermes_root's Docker/custom branch), so named profiles live at
+    ``<root>/profiles/<name>/`` and the launch/default profile is ``<root>``.
+    """
+    root = tmp_path / "hermes_home"
+    (root / "profiles" / "work").mkdir(parents=True)
+    (root / "profiles" / "other").mkdir(parents=True)
+    monkeypatch.setenv("HERMES_HOME", str(root))
+    # Make sure no stale process-wide home override leaks in from another test.
+    from hermes_constants import get_hermes_home_override
+
+    assert get_hermes_home_override() is None
+    return root
+
+
+def _call(method, params=None):
+    handler = server._methods[method]
+    return handler(1, params or {})
+
+
+def _result(resp):
+    assert "error" not in resp, resp.get("error")
+    return resp["result"]
+
+
+def _read_yaml(path: Path) -> dict:
+    """Read a config.yaml directly for assertions (test-side, not the guarded loader)."""
+    import yaml
+
+    if not path.is_file():
+        return {}
+    with open(path, "r", encoding="utf-8") as f:
+        return yaml.safe_load(f) or {}
+
+
+def test_add_lands_in_named_profile_only(hermes_root):
+    root = hermes_root
+    resp = _call(
+        "mcp.servers.add",
+        {
+            "profile": "work",
+            "name": "weather",
+            "config": {"url": "https://mcp.example.com/weather"},
+        },
+    )
+    result = _result(resp)
+    assert result["ok"] is True
+    assert result["server"]["transport"] == "http"
+    assert result["server"]["url"] == "https://mcp.example.com/weather"
+
+    work_cfg = _read_yaml(root / "profiles" / "work" / "config.yaml")
+    assert "weather" in work_cfg.get("mcp_servers", {})
+    assert work_cfg["mcp_servers"]["weather"]["url"] == "https://mcp.example.com/weather"
+
+    # The launch/default profile and the sibling profile stay untouched.
+    default_cfg = _read_yaml(root / "config.yaml")
+    assert "weather" not in default_cfg.get("mcp_servers", {})
+    other_cfg = _read_yaml(root / "profiles" / "other" / "config.yaml")
+    assert "weather" not in other_cfg.get("mcp_servers", {})
+
+
+def test_list_reflects_the_scoped_profile(hermes_root):
+    _result(
+        _call(
+            "mcp.servers.add",
+            {"profile": "work", "name": "svc-a", "config": {"command": "svc-a-bin"}},
+        )
+    )
+    _result(
+        _call(
+            "mcp.servers.add",
+            {"profile": "other", "name": "svc-b", "config": {"command": "svc-b-bin"}},
+        )
+    )
+
+    work_names = [s["name"] for s in _result(_call("mcp.servers.list", {"profile": "work"}))["servers"]]
+    other_names = [s["name"] for s in _result(_call("mcp.servers.list", {"profile": "other"}))["servers"]]
+
+    assert work_names == ["svc-a"]
+    assert other_names == ["svc-b"]
+
+    # stdio transport surfaced correctly.
+    work_server = _result(_call("mcp.servers.list", {"profile": "work"}))["servers"][0]
+    assert work_server["transport"] == "stdio"
+    assert work_server["command"] == "svc-a-bin"
+
+
+def test_set_api_key_writes_env_and_header_to_right_profile(hermes_root):
+    root = hermes_root
+    _result(
+        _call(
+            "mcp.servers.add",
+            {
+                "profile": "work",
+                "name": "gizmo",
+                "config": {"url": "https://mcp.example.com/gizmo"},
+            },
+        )
+    )
+
+    resp = _result(
+        _call(
+            "mcp.servers.set_api_key",
+            {"profile": "work", "name": "gizmo", "value": "sk-secret-123"},
+        )
+    )
+    assert resp["ok"] is True
+    env_var = resp["env_var"]
+    assert env_var == "MCP_GIZMO_API_KEY"
+
+    # The secret is in the work profile's .env — and NOT the default profile's.
+    work_env = (root / "profiles" / "work" / ".env").read_text(encoding="utf-8")
+    assert "MCP_GIZMO_API_KEY=sk-secret-123" in work_env
+    assert not (root / ".env").exists() or "sk-secret-123" not in (root / ".env").read_text(
+        encoding="utf-8"
+    )
+
+    # config.yaml stores only the interpolation template, never the raw secret.
+    work_cfg = _read_yaml(root / "profiles" / "work" / "config.yaml")
+    headers = work_cfg["mcp_servers"]["gizmo"]["headers"]
+    assert headers["Authorization"] == "Bearer ${MCP_GIZMO_API_KEY}"
+    assert "sk-secret-123" not in str(work_cfg)
+
+
+def test_set_api_key_stdio_references_env_block(hermes_root):
+    root = hermes_root
+    _result(
+        _call(
+            "mcp.servers.add",
+            {"profile": "work", "name": "localtool", "config": {"command": "localtool-bin"}},
+        )
+    )
+    resp = _result(
+        _call(
+            "mcp.servers.set_api_key",
+            {
+                "profile": "work",
+                "name": "localtool",
+                "env_var": "LOCALTOOL_TOKEN",
+                "value": "tok-xyz",
+            },
+        )
+    )
+    assert resp["env_var"] == "LOCALTOOL_TOKEN"
+
+    work_cfg = _read_yaml(root / "profiles" / "work" / "config.yaml")
+    env_block = work_cfg["mcp_servers"]["localtool"]["env"]
+    assert env_block["LOCALTOOL_TOKEN"] == "${LOCALTOOL_TOKEN}"
+    work_env = (root / "profiles" / "work" / ".env").read_text(encoding="utf-8")
+    assert "LOCALTOOL_TOKEN=tok-xyz" in work_env
+
+
+def test_remove_scoped_to_profile(hermes_root):
+    root = hermes_root
+    _result(
+        _call(
+            "mcp.servers.add",
+            {"profile": "work", "name": "temp", "config": {"command": "temp-bin"}},
+        )
+    )
+    # Same-named server in a different profile must be unaffected by the remove.
+    _result(
+        _call(
+            "mcp.servers.add",
+            {"profile": "other", "name": "temp", "config": {"command": "temp-bin"}},
+        )
+    )
+
+    resp = _result(_call("mcp.servers.remove", {"profile": "work", "name": "temp"}))
+    assert resp["removed"] is True
+
+    assert "temp" not in _read_yaml(root / "profiles" / "work" / "config.yaml").get("mcp_servers", {})
+    # The 'other' profile still has its server.
+    assert "temp" in _read_yaml(root / "profiles" / "other" / "config.yaml").get("mcp_servers", {})
+
+
+def test_add_duplicate_and_missing_errors(hermes_root):
+    _result(
+        _call(
+            "mcp.servers.add",
+            {"profile": "work", "name": "dup", "config": {"command": "dup-bin"}},
+        )
+    )
+    dup = _call(
+        "mcp.servers.add",
+        {"profile": "work", "name": "dup", "config": {"command": "dup-bin"}},
+    )
+    assert "error" in dup
+    assert dup["error"]["code"] == 4090
+
+    missing = _call("mcp.servers.remove", {"profile": "work", "name": "nope"})
+    assert "error" in missing
+    assert missing["error"]["code"] == 4064
+
+    bad_profile = _call(
+        "mcp.servers.add",
+        {"profile": "ghost", "name": "x", "config": {"command": "x"}},
+    )
+    assert "error" in bad_profile
+    assert bad_profile["error"]["code"] == 4064
+
+
+def test_add_requires_transport(hermes_root):
+    resp = _call("mcp.servers.add", {"profile": "work", "name": "empty", "config": {}})
+    assert "error" in resp
+    assert resp["error"]["code"] == 4063
+
+
+def test_default_profile_add_when_profile_omitted(hermes_root):
+    root = hermes_root
+    _result(
+        _call(
+            "mcp.servers.add",
+            {"name": "rootsvc", "config": {"command": "rootsvc-bin"}},
+        )
+    )
+    # Omitted profile → launch/default profile == HERMES_HOME root config.yaml.
+    default_cfg = _read_yaml(root / "config.yaml")
+    assert "rootsvc" in default_cfg.get("mcp_servers", {})
+    # ...and NOT in a named profile.
+    assert "rootsvc" not in _read_yaml(root / "profiles" / "work" / "config.yaml").get(
+        "mcp_servers", {}
+    )

tui_gateway/mcp_oauth_sessions.py

@@ -0,0 +1,339 @@
+"""Session-backed MCP OAuth flows for the gateway (mcp.servers.oauth.*).
+
+This mirrors the *provider* OAuth model used by the dashboard
+(``/api/providers/oauth/{id}/start`` + ``/poll/{session_id}``) rather than the
+FastAPI-request-coupled MCP dashboard flow: a ``start`` primitive kicks off a
+background worker and returns ``{session_id, auth_url, flow}``; a ``poll``
+primitive reports ``{status: pending|approved|error}`` until the tokens land on
+disk for that server in that profile.
+
+The underlying token machinery is the *same* one the CLI ``hermes mcp login``
+uses — ``hermes_cli.mcp_config._probe_single_server`` under
+``tools.mcp_oauth.force_interactive_oauth`` — so no OAuth logic is reimplemented
+here. The only new piece is decoupling the two browser callbacks (authorization
+URL out, ``code``/``state`` back in) from a FastAPI ``Request``:
+
+* ``tools.mcp_dashboard_oauth.DashboardOAuthFlow`` already provides the two
+  thread-safe rendezvous points (``publish_authorization_url`` /
+  ``deliver_callback``). We reuse it verbatim as the bridge object.
+* Instead of routing the browser redirect through a FastAPI callback route, we
+  run a tiny loopback HTTP listener on ``127.0.0.1:<port>/callback`` and set the
+  flow's ``redirect_uri`` to it. When the provider redirects the user's browser
+  there, the listener calls ``flow.deliver_callback(...)``. This is the same
+  loopback strategy the CLI uses by default, just wired to the shared bridge.
+
+Client contract (what the desktop plugin does):
+  1. call ``mcp.servers.oauth.start(profile, name)`` → ``{session_id, auth_url}``
+  2. open ``auth_url`` in the native browser (``openExternal``)
+  3. poll ``mcp.servers.oauth.poll(profile, name, session_id)`` until
+     ``status == "approved"`` (tokens persisted) or ``"error"``.
+"""
+
+from __future__ import annotations
+
+import http.server
+import secrets
+import threading
+import time
+from pathlib import Path
+from typing import Any, Dict, Optional
+from urllib.parse import parse_qs, urlparse
+
+# Session registry: session_id -> record. A record wraps the shared
+# DashboardOAuthFlow bridge plus a bit of gateway bookkeeping.
+_sessions: Dict[str, Dict[str, Any]] = {}
+_sessions_lock = threading.Lock()
+
+# How long a completed/abandoned session lingers before GC (seconds).
+_SESSION_TTL_SECONDS = 900
+# Cap concurrent in-flight flows so a runaway client can't exhaust ports/threads.
+_MAX_PENDING = 12
+
+
+def _gc_sessions() -> None:
+    """Drop expired sessions. Called opportunistically on start."""
+    cutoff = time.time() - _SESSION_TTL_SECONDS
+    with _sessions_lock:
+        stale = [sid for sid, rec in _sessions.items() if rec["created_at"] < cutoff]
+        for sid in stale:
+            rec = _sessions.pop(sid, None)
+            if rec is not None:
+                _shutdown_listener(rec)
+
+
+def _shutdown_listener(rec: Dict[str, Any]) -> None:
+    server = rec.get("httpd")
+    if server is not None:
+        try:
+            server.shutdown()
+        except Exception:
+            pass
+        try:
+            server.server_close()
+        except Exception:
+            pass
+        rec["httpd"] = None
+
+
+def _start_loopback_listener(flow) -> "http.server.HTTPServer":
+    """Bind a loopback callback listener that feeds the flow's deliver_callback.
+
+    Returns the running HTTPServer (already serving on a daemon thread). The
+    bound port is read back off ``server.server_address`` so the caller can set
+    ``flow.redirect_uri`` to the matching ``/callback`` URL BEFORE the worker
+    starts the OAuth flow (the redirect URI must be pinned at authorization).
+    """
+
+    class _Handler(http.server.BaseHTTPRequestHandler):
+        def do_GET(self):  # noqa: N802 — stdlib naming
+            parsed = urlparse(self.path)
+            if parsed.path.rstrip("/") not in ("/callback", ""):
+                self.send_response(404)
+                self.end_headers()
+                return
+            qs = parse_qs(parsed.query)
+            code = (qs.get("code") or [None])[0]
+            state = (qs.get("state") or [None])[0]
+            error = (qs.get("error") or [None])[0]
+            body = b"<h1>Authorization received</h1><p>You can close this tab and return to Hermes.</p>"
+            status = 200
+            try:
+                flow.deliver_callback(code=code, state=state, error=error)
+            except Exception:
+                body = b"<h1>OAuth callback rejected</h1><p>The callback was invalid or already used.</p>"
+                status = 400
+            self.send_response(status)
+            self.send_header("Content-Type", "text/html; charset=utf-8")
+            self.end_headers()
+            try:
+                self.wfile.write(body)
+            except Exception:
+                pass
+
+        def log_message(self, *_a):  # silence stdlib request logging
+            return
+
+    httpd = http.server.HTTPServer(("127.0.0.1", 0), _Handler)
+    threading.Thread(
+        target=httpd.serve_forever,
+        kwargs={"poll_interval": 0.5},
+        daemon=True,
+        name=f"mcp-oauth-cb-{flow.server_name}",
+    ).start()
+    return httpd
+
+
+def _worker(session_id: str, hermes_home: str, server_name: str, cfg: dict, reconnect_live: bool) -> None:
+    """Drive the interactive MCP OAuth probe under the shared dashboard bridge.
+
+    Structurally identical to ``web_server._run_dashboard_mcp_oauth`` — the same
+    HERMES_HOME override + secret-scope + force_interactive_oauth +
+    dashboard_oauth_flow wrapping around ``_probe_single_server`` — but keyed to
+    our session record instead of a FastAPI request. On success the token file
+    exists on disk (verified via ``_oauth_tokens_present``) and the server config
+    is (re)saved into the profile's config.yaml.
+    """
+    from hermes_cli.mcp_config import (
+        _oauth_tokens_present,
+        _probe_single_server,
+        _save_mcp_server,
+    )
+    from hermes_constants import reset_hermes_home_override, set_hermes_home_override
+
+    rec = _sessions.get(session_id)
+    flow = rec["flow"] if rec else None
+    try:
+        from agent.secret_scope import (
+            build_profile_secret_scope,
+            reset_secret_scope,
+            set_secret_scope,
+        )
+        from tools.mcp_dashboard_oauth import dashboard_oauth_flow
+        from tools.mcp_oauth import force_interactive_oauth
+        from tools.mcp_oauth_manager import get_manager
+
+        home_token = set_hermes_home_override(hermes_home)
+        secret_token = set_secret_scope(build_profile_secret_scope(Path(hermes_home)))
+        try:
+            with force_interactive_oauth(), dashboard_oauth_flow(flow):
+                from tools.mcp_oauth import HermesTokenStorage
+
+                manager = get_manager()
+                storage = HermesTokenStorage(server_name)
+                backup = storage.snapshot()
+                previous_entry = None
+                try:
+                    previous_entry = manager.remove(server_name, hermes_home=hermes_home)
+                    tools = _probe_single_server(
+                        server_name,
+                        cfg,
+                        connect_timeout=max(float(cfg.get("connect_timeout", 0) or 0), 315),
+                    )
+                    if not _oauth_tokens_present(server_name):
+                        raise RuntimeError(
+                            "The server responded, but no OAuth token was obtained — "
+                            "this provider may require a manually-registered OAuth client."
+                        )
+                    _save_mcp_server(server_name, cfg)
+                    if flow is not None:
+                        flow.tools = [{"name": t, "description": d} for t, d in tools]
+                        flow.mark_approved()
+                    if reconnect_live:
+                        from tools.mcp_tool import reconnect_mcp_server
+
+                        reconnect_mcp_server(server_name)
+                except Exception:
+                    storage.restore(backup, only_if_absent=True)
+                    manager.restore_entry(server_name, previous_entry, hermes_home=hermes_home)
+                    raise
+        finally:
+            reset_secret_scope(secret_token)
+            reset_hermes_home_override(home_token)
+    except Exception as exc:
+        msg = str(exc)
+        try:
+            from tools.mcp_oauth import humanize_oauth_registration_error
+
+            humanized = humanize_oauth_registration_error(
+                server_name, exc, server_url=cfg.get("url") if isinstance(cfg, dict) else None
+            )
+            if humanized:
+                msg = humanized
+        except Exception:
+            pass
+        if flow is not None:
+            flow.mark_error(msg)
+    finally:
+        if flow is not None:
+            flow.mark_worker_done()
+        if rec is not None:
+            _shutdown_listener(rec)
+
+
+def start_flow(
+    hermes_home: str,
+    server_name: str,
+    cfg: dict,
+    *,
+    reconnect_live: bool = False,
+    url_timeout: float = 30.0,
+) -> Dict[str, Any]:
+    """Begin an MCP OAuth flow and return ``{session_id, auth_url, flow}``.
+
+    ``cfg`` is the server's resolved config dict (must have ``url`` and be
+    OAuth-capable). ``hermes_home`` is the already-resolved profile home dir
+    string. Blocks up to ``url_timeout`` for the worker to publish the browser
+    authorization URL, then returns it.
+    """
+    from tools.mcp_dashboard_oauth import DashboardOAuthFlow
+
+    _gc_sessions()
+
+    with _sessions_lock:
+        pending = sum(
+            1
+            for r in _sessions.values()
+            if not r["flow"].worker_done
+        )
+        if pending >= _MAX_PENDING:
+            raise RuntimeError("Too many MCP OAuth flows are already in progress")
+        if any(
+            r["server_name"] == server_name
+            and r["hermes_home"] == hermes_home
+            and not r["flow"].worker_done
+            for r in _sessions.values()
+        ):
+            raise RuntimeError(f"MCP OAuth for '{server_name}' is already in progress")
+
+    session_id = secrets.token_urlsafe(24)
+    flow = DashboardOAuthFlow(
+        flow_id=session_id,
+        server_name=server_name,
+        profile=None,
+        hermes_home=hermes_home,
+        redirect_uri="",  # set below once the loopback port is known
+        reconnect_live=reconnect_live,
+    )
+    httpd = _start_loopback_listener(flow)
+    port = httpd.server_address[1]
+    flow.redirect_uri = f"http://127.0.0.1:{port}/callback"
+
+    rec = {
+        "session_id": session_id,
+        "server_name": server_name,
+        "hermes_home": hermes_home,
+        "flow": flow,
+        "httpd": httpd,
+        "created_at": time.time(),
+    }
+    with _sessions_lock:
+        _sessions[session_id] = rec
+
+    threading.Thread(
+        target=_worker,
+        args=(session_id, hermes_home, server_name, dict(cfg), reconnect_live),
+        daemon=True,
+        name=f"mcp-oauth-{server_name}",
+    ).start()
+
+    try:
+        auth_url = None
+        # wait_for_authorization_url is async; run its wait synchronously.
+        deadline = time.time() + url_timeout
+        while time.time() < deadline:
+            snap = flow.snapshot()
+            if snap.get("authorization_url"):
+                auth_url = snap["authorization_url"]
+                break
+            if snap.get("status") == "error":
+                raise RuntimeError(snap.get("error") or "MCP OAuth flow failed before authorization")
+            time.sleep(0.1)
+        if not auth_url:
+            raise TimeoutError("Timed out waiting for MCP authorization URL")
+    except Exception:
+        flow.mark_error("Timed out waiting for MCP authorization URL")
+        _shutdown_listener(rec)
+        raise
+
+    return {
+        "session_id": session_id,
+        "auth_url": auth_url,
+        # "pkce" mirrors the provider-OAuth ``flow`` discriminator: the client
+        # opens a URL then polls (no user_code to type, unlike device_code).
+        "flow": "pkce",
+    }
+
+
+def poll_flow(session_id: str, server_name: str) -> Dict[str, Any]:
+    """Poll a session's status → ``{status, error_message?, auth_url?, tools?}``.
+
+    ``status`` is one of ``pending`` | ``approved`` | ``error`` — the same
+    vocabulary as the provider poll endpoint (``authorization_required`` from
+    the underlying bridge maps to ``pending`` since the client only needs to
+    know whether to keep waiting).
+    """
+    with _sessions_lock:
+        rec = _sessions.get(session_id)
+    if rec is None:
+        return {"status": "error", "error_message": "OAuth session not found or expired"}
+    if rec["server_name"] != server_name:
+        return {"status": "error", "error_message": "server name mismatch for session"}
+
+    flow = rec["flow"]
+    snap = flow.snapshot()
+    raw = snap.get("status")
+    if raw == "approved":
+        status = "approved"
+    elif raw == "error":
+        status = "error"
+    else:
+        status = "pending"
+    out: Dict[str, Any] = {
+        "session_id": session_id,
+        "status": status,
+        "error_message": snap.get("error"),
+        "auth_url": snap.get("authorization_url"),
+    }
+    if status == "approved":
+        out["tools"] = list(getattr(flow, "tools", []) or [])
+    return out

tui_gateway/mcp_rpc_helpers.py

@@ -0,0 +1,74 @@
+"""Shared helpers for the per-profile MCP lifecycle RPCs (mcp.servers.*).
+
+These live in their own module (not methods_tools) because methods_tools
+handlers are rebound onto ``tui_gateway.server``'s globals at install time
+(see method_ctx.HandlerRegistry.install); a plain module-level def in
+methods_tools would not be reachable from a rebound handler body. Handlers
+import these at call time instead.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Optional, Tuple
+
+
+def resolve_profile(rid, params, err_fn) -> Tuple[Optional[Any], Optional[dict]]:
+    """Resolve the optional ``profile`` param to a HERMES_HOME override token.
+
+    Returns ``(token, error)``: ``token`` is None for the launch profile (no
+    override) or an opaque reset token; ``error`` is a JSON-RPC error dict
+    (built via ``err_fn``) when the named profile doesn't exist. Callers reset
+    ``token`` in a finally via :func:`reset_profile`.
+    """
+    profile = str(params.get("profile") or "").strip()
+    if not profile:
+        return None, None
+    from hermes_cli.profiles import get_profile_dir
+    from hermes_constants import set_hermes_home_override
+
+    profile_dir = get_profile_dir(profile)
+    if not profile_dir or not profile_dir.is_dir():
+        return None, err_fn(rid, 4064, f"profile '{profile}' not found")
+    return set_hermes_home_override(str(profile_dir)), None
+
+
+def reset_profile(token) -> None:
+    if token is not None:
+        try:
+            from hermes_constants import reset_hermes_home_override
+
+            reset_hermes_home_override(token)
+        except Exception:
+            pass
+
+
+def summarize_server(name: str, cfg: dict) -> Dict[str, Any]:
+    """Serialize one server's config for a UI (no secret values).
+
+    Mirrors web_server._mcp_server_summary plus an ``oauth_tokens_present``
+    flag so a UI can tell an OAuth server that still needs authentication from
+    one already authenticated.
+    """
+    from hermes_cli.mcp_config import _oauth_tokens_present
+
+    cfg = cfg if isinstance(cfg, dict) else {}
+    transport = "http" if cfg.get("url") else ("stdio" if cfg.get("command") else "unknown")
+    auth = cfg.get("auth")
+    headers = cfg.get("headers") or {}
+    if not auth and isinstance(headers, dict) and any(
+        str(key).lower() == "authorization" for key in headers
+    ):
+        auth = "header"
+    tokens_present = _oauth_tokens_present(name) if auth == "oauth" else None
+    return {
+        "name": name,
+        "transport": transport,
+        "url": cfg.get("url"),
+        "command": cfg.get("command"),
+        "args": list(cfg.get("args") or []),
+        "env": sorted(str(k) for k in (cfg.get("env") or {})),
+        "auth": auth,
+        "oauth_tokens_present": tokens_present,
+        "enabled": cfg.get("enabled", True) is not False,
+        "tools": cfg.get("tools"),
+    }

tui_gateway/methods_tools.py

@@ -1901,6 +1901,402 @@ def _(rid, params: dict) -> dict:
                 pass
 
 
+# ─── Per-profile MCP server lifecycle (mcp.servers.*) ────────────────────────
+#
+# Gateway RPCs mirroring the dashboard's REST surface
+# (hermes_cli/web_routers/mcp.py) so a desktop plugin can manage MCP servers for
+# ANY profile, not just the launch profile. Each accepts an optional ``profile``
+# param that scopes HERMES_HOME via set_hermes_home_override (omitted/None = the
+# launch profile) in a try/finally, exactly like ``skills.manage`` / ``mcp.catalog``.
+# All persistence reuses hermes_cli/mcp_config.py helpers — no logic is duplicated.
+# Shared helpers (resolve_profile / reset_profile / summarize_server) live in
+# tui_gateway.mcp_rpc_helpers and are imported at call time: these handlers are
+# rebound onto server.py's globals at install time, so a plain module-level def
+# here would not be reachable from the rebound handler body.
+
+
+@method("mcp.servers.list")
+def _(rid, params: dict) -> dict:
+    """List a profile's configured MCP servers.
+
+    Params: optional ``profile``. Result: ``{servers: [{name, transport, url,
+    command, args, env (key names only), auth, oauth_tokens_present, enabled,
+    tools}]}``. Reuses ``mcp_config._get_mcp_servers`` under the home override.
+    """
+    token, err = _mcp_resolve_profile(rid, params)
+    if err:
+        return err
+    try:
+        from hermes_cli.mcp_config import _get_mcp_servers
+
+        servers = _get_mcp_servers()
+        return _ok(
+            rid,
+            {
+                "servers": [
+                    _mcp_summarize_server(name, cfg)
+                    for name, cfg in sorted(servers.items())
+                ]
+            },
+        )
+    except Exception as e:
+        return _err(rid, 5024, str(e))
+    finally:
+        _mcp_reset_profile(token)
+
+
+@method("mcp.servers.add")
+def _(rid, params: dict) -> dict:
+    """Add/save an MCP server to a profile's config.yaml.
+
+    Params: optional ``profile``, ``name`` (required), and EITHER:
+      - ``preset`` (a catalog preset id) → applied via ``_apply_mcp_preset``, or
+      - ``config`` (an mcp_servers entry dict: url/command/args/env/headers/
+        auth/tools) → saved via ``_save_mcp_server``.
+    If ``bearer_token`` is given (header auth), it is written to the profile's
+    .env via ``_save_bearer_auth_token`` and only the safe ``Authorization``
+    header template is persisted in config.yaml.
+
+    Result: ``{ok: true, name, server: <summary>}``. Duplicate names error.
+    """
+    name = str(params.get("name") or "").strip()
+    if not name:
+        return _err(rid, 4063, "name required")
+    token, err = _mcp_resolve_profile(rid, params)
+    if err:
+        return err
+    try:
+        from hermes_cli.mcp_config import (
+            _apply_mcp_preset,
+            _get_mcp_servers,
+            _save_bearer_auth_token,
+            _save_mcp_server,
+        )
+
+        if name in _get_mcp_servers():
+            return _err(rid, 4090, f"server '{name}' already exists")
+
+        preset = str(params.get("preset") or "").strip()
+        raw_cfg = params.get("config")
+        server_config: dict = dict(raw_cfg) if isinstance(raw_cfg, dict) else {}
+
+        if preset:
+            # _apply_mcp_preset fills url/command/args from a known preset when
+            # transport details were omitted; it mutates server_config in place.
+            _apply_mcp_preset(
+                name,
+                preset_name=preset,
+                url=server_config.get("url"),
+                command=server_config.get("command"),
+                cmd_args=list(server_config.get("args") or []),
+                server_config=server_config,
+            )
+
+        if not server_config.get("url") and not server_config.get("command"):
+            return _err(
+                rid,
+                4063,
+                "config must specify a 'url' (http) or 'command' (stdio), or a valid 'preset'",
+            )
+
+        bearer_token = params.get("bearer_token")
+        if bearer_token:
+            # Persist the secret in .env; store only the interpolation template.
+            server_config["headers"] = _save_bearer_auth_token(name, str(bearer_token))
+
+        if not _save_mcp_server(name, server_config):
+            return _err(
+                rid,
+                4001,
+                f"server '{name}' rejected: suspicious command/args configuration",
+            )
+        saved = _get_mcp_servers().get(name, server_config)
+        return _ok(rid, {"ok": True, "name": name, "server": _mcp_summarize_server(name, saved)})
+    except Exception as e:
+        return _err(rid, 5024, str(e))
+    finally:
+        _mcp_reset_profile(token)
+
+
+@method("mcp.servers.set_api_key")
+def _(rid, params: dict) -> dict:
+    """Store a required API key / credential for a server in a profile.
+
+    Params: optional ``profile``, ``name`` (required), ``value`` (required,
+    the secret), and optional ``env_var`` (defaults to the server's canonical
+    ``MCP_<NAME>_API_KEY`` key). The secret is written to that profile's .env
+    via ``save_env_value``; the config.yaml entry is updated to reference it —
+    a header template ``Authorization: Bearer ${ENV}`` for http servers, or an
+    ``env: {VAR: "${ENV}"}`` reference for stdio servers — matching how
+    ``cmd_mcp_configure`` / ``_save_bearer_auth_token`` wire secrets.
+
+    Result: ``{ok: true, name, env_var, server: <summary>}``.
+    """
+    name = str(params.get("name") or "").strip()
+    if not name:
+        return _err(rid, 4063, "name required")
+    value = params.get("value")
+    if value is None or str(value) == "":
+        return _err(rid, 4063, "value required")
+    token, err = _mcp_resolve_profile(rid, params)
+    if err:
+        return err
+    try:
+        from hermes_cli.config import load_config, save_config, save_env_value
+        from hermes_cli.mcp_config import (
+            _bearer_auth_headers,
+            _env_key_for_server,
+            _get_mcp_servers,
+            _strip_bearer_prefix,
+        )
+
+        servers = _get_mcp_servers()
+        if name not in servers:
+            return _err(rid, 4064, f"server '{name}' not found")
+
+        env_var = str(params.get("env_var") or "").strip() or _env_key_for_server(name)
+
+        entry = servers[name]
+        if not isinstance(entry, dict):
+            return _err(rid, 4001, "malformed server config")
+
+        if entry.get("url"):
+            # http/sse server: store a bearer token + Authorization template.
+            normalized = _strip_bearer_prefix(str(value))
+            if not normalized or normalized.lower() == "bearer":
+                return _err(rid, 4063, "value is not a valid credential")
+            save_env_value(env_var, normalized)
+            if env_var == _env_key_for_server(name):
+                headers = _bearer_auth_headers(name)
+            else:
+                headers = {"Authorization": f"Bearer ${{{env_var}}}"}
+            entry["headers"] = headers
+        else:
+            # stdio server: reference the secret from the process env block.
+            save_env_value(env_var, str(value))
+            env_block = entry.get("env")
+            if not isinstance(env_block, dict):
+                env_block = {}
+            env_block[env_var] = f"${{{env_var}}}"
+            entry["env"] = env_block
+
+        cfg = load_config()
+        cfg.setdefault("mcp_servers", {})[name] = entry
+        save_config(cfg)
+        return _ok(
+            rid,
+            {
+                "ok": True,
+                "name": name,
+                "env_var": env_var,
+                "server": _mcp_summarize_server(name, entry),
+            },
+        )
+    except Exception as e:
+        return _err(rid, 5024, str(e))
+    finally:
+        _mcp_reset_profile(token)
+
+
+@method("mcp.servers.test")
+def _(rid, params: dict) -> dict:
+    """Probe a profile's MCP server: connect, list tools, disconnect.
+
+    Params: optional ``profile``, ``name`` (required). Result on success:
+    ``{ok: true, tools: [{name, description}], prompts, resources,
+    oauth_tokens_present}``. On failure: ``{ok: false, error, tools: [],
+    oauth_needed}``. Reuses ``mcp_config._probe_single_server`` +
+    ``_oauth_tokens_present`` — same logic as the /test dashboard route.
+
+    Runs on the RPC thread pool (see _LONG_HANDLERS): a cold stdio `npx`
+    spawn can block for many seconds.
+    """
+    name = str(params.get("name") or "").strip()
+    if not name:
+        return _err(rid, 4063, "name required")
+    token, err = _mcp_resolve_profile(rid, params)
+    if err:
+        return err
+    try:
+        from hermes_cli.mcp_config import (
+            _get_mcp_servers,
+            _oauth_tokens_present,
+            _probe_single_server,
+        )
+
+        servers = _get_mcp_servers()
+        if name not in servers:
+            return _err(rid, 4064, f"server '{name}' not found")
+
+        cfg = servers[name]
+        # An `auth: oauth` server that serves tools/list anonymously would probe
+        # OK with no token — a false green. Require a token on disk for it.
+        needs_oauth_token = cfg.get("auth") == "oauth"
+        details: dict = {}
+        try:
+            tools = _probe_single_server(name, cfg, details=details)
+            token_present = _oauth_tokens_present(name) if needs_oauth_token else True
+        except Exception as exc:
+            return _ok(
+                rid,
+                {
+                    "ok": False,
+                    "error": str(exc),
+                    "tools": [],
+                    "oauth_needed": needs_oauth_token,
+                    "oauth_tokens_present": _oauth_tokens_present(name)
+                    if needs_oauth_token
+                    else None,
+                },
+            )
+        if not token_present:
+            return _ok(
+                rid,
+                {
+                    "ok": False,
+                    "error": "OAuth authentication required — no token found.",
+                    "tools": [],
+                    "oauth_needed": True,
+                    "oauth_tokens_present": False,
+                },
+            )
+        return _ok(
+            rid,
+            {
+                "ok": True,
+                "tools": [{"name": t, "description": d} for t, d in tools],
+                "prompts": details.get("prompts", 0),
+                "resources": details.get("resources", 0),
+                "oauth_needed": needs_oauth_token,
+                "oauth_tokens_present": True if needs_oauth_token else None,
+            },
+        )
+    except Exception as e:
+        return _err(rid, 5024, str(e))
+    finally:
+        _mcp_reset_profile(token)
+
+
+@method("mcp.servers.remove")
+def _(rid, params: dict) -> dict:
+    """Remove a server from a profile's config.yaml.
+
+    Params: optional ``profile``, ``name`` (required). Result:
+    ``{ok: true, removed: bool}``. Reuses ``mcp_config._remove_mcp_server``.
+    """
+    name = str(params.get("name") or "").strip()
+    if not name:
+        return _err(rid, 4063, "name required")
+    token, err = _mcp_resolve_profile(rid, params)
+    if err:
+        return err
+    try:
+        from hermes_cli.mcp_config import _remove_mcp_server
+
+        removed = _remove_mcp_server(name)
+        if not removed:
+            return _err(rid, 4064, f"server '{name}' not found")
+        return _ok(rid, {"ok": True, "removed": True})
+    except Exception as e:
+        return _err(rid, 5024, str(e))
+    finally:
+        _mcp_reset_profile(token)
+
+
+@method("mcp.servers.oauth.start")
+def _(rid, params: dict) -> dict:
+    """Begin a session-backed OAuth flow for an MCP server in a profile.
+
+    Params: optional ``profile``, ``name`` (required). Result:
+    ``{ok: true, session_id, auth_url, flow: "pkce"}``.
+
+    The client (desktop) opens ``auth_url`` in the native browser
+    (``window.hermesDesktop.openExternal``) and then polls
+    ``mcp.servers.oauth.poll`` with the returned ``session_id`` until
+    ``status == "approved"``. This mirrors the provider-OAuth start/poll model
+    (``/api/providers/oauth/{id}/start`` + ``/poll``): a background worker drives
+    the SAME interactive MCP OAuth machinery ``hermes mcp login`` uses
+    (``_probe_single_server`` under ``force_interactive_oauth``), and a loopback
+    listener captures the browser redirect — no FastAPI request object needed.
+
+    Runs on the RPC thread pool (see _LONG_HANDLERS): start blocks briefly for
+    the authorization URL to be published.
+    """
+    name = str(params.get("name") or "").strip()
+    if not name:
+        return _err(rid, 4063, "name required")
+    token, err = _mcp_resolve_profile(rid, params)
+    if err:
+        return err
+    try:
+        from hermes_cli.mcp_config import _get_mcp_servers
+        from hermes_constants import get_hermes_home
+        from tui_gateway import mcp_oauth_sessions
+
+        servers = _get_mcp_servers()
+        if name not in servers:
+            return _err(rid, 4064, f"server '{name}' not found")
+        cfg = dict(servers[name])
+        if not cfg.get("url"):
+            return _err(
+                rid, 4001, "stdio servers authenticate via env keys, not OAuth"
+            )
+        if cfg.get("headers") and cfg.get("auth") != "oauth":
+            return _err(
+                rid, 4001, "this server uses header/API-key auth, not OAuth"
+            )
+        cfg["auth"] = "oauth"
+
+        hermes_home = str(get_hermes_home().expanduser().resolve(strict=False))
+        result = mcp_oauth_sessions.start_flow(hermes_home, name, cfg)
+        return _ok(
+            rid,
+            {
+                "ok": True,
+                "session_id": result["session_id"],
+                "auth_url": result["auth_url"],
+                "flow": result["flow"],
+            },
+        )
+    except Exception as e:
+        return _err(rid, 5024, str(e))
+    finally:
+        _mcp_reset_profile(token)
+
+
+@method("mcp.servers.oauth.poll")
+def _(rid, params: dict) -> dict:
+    """Poll a session-backed MCP OAuth flow.
+
+    Params: optional ``profile``, ``name`` (required), ``session_id`` (required,
+    from ``mcp.servers.oauth.start``). Result: ``{ok: true, status:
+    "pending"|"approved"|"error", error_message?, auth_url?, tools?}``.
+
+    On ``approved`` the OAuth tokens have been persisted for that server in that
+    profile (verified via ``_oauth_tokens_present`` inside the worker). The
+    profile scope is applied here too so a same-profile reconnect / token read
+    resolves correctly.
+    """
+    name = str(params.get("name") or "").strip()
+    if not name:
+        return _err(rid, 4063, "name required")
+    session_id = str(params.get("session_id") or "").strip()
+    if not session_id:
+        return _err(rid, 4063, "session_id required")
+    token, err = _mcp_resolve_profile(rid, params)
+    if err:
+        return err
+    try:
+        from tui_gateway import mcp_oauth_sessions
+
+        result = mcp_oauth_sessions.poll_flow(session_id, name)
+        return _ok(rid, {"ok": True, **result})
+    except Exception as e:
+        return _err(rid, 5024, str(e))
+    finally:
+        _mcp_reset_profile(token)
+
+
 @method("skills.reload")
 def _(rid, params: dict) -> dict:
     try:

tui_gateway/server.py

@@ -248,6 +248,12 @@ _LONG_HANDLERS = frozenset(
         # dead after a few skin switches. The handler serializes concurrent
         # reloads via _mcp_reload_lock.
         "reload.mcp",
+        # MCP server test/OAuth RPCs block on network: a probe spawns a stdio
+        # server (cold `npx` cold start = many seconds) or connects to a remote
+        # endpoint; oauth.start blocks up to ~30s waiting for the provider to
+        # publish an authorization URL. Keep them off the reader thread.
+        "mcp.servers.test",
+        "mcp.servers.oauth.start",
         "process.list",
         # profiles.list runs list_profiles() (recursive skill-tree walk per
         # profile) and opens each profile's state.db for the last-session
@@ -14668,6 +14674,27 @@ def _browser_disconnect(rid) -> dict:
 
 
 
+# Per-profile MCP lifecycle helpers (mcp.servers.* handlers). Defined on THIS
+# namespace so the rebound handler bodies (register() below) can resolve them,
+# same as _ok/_err — a plain def in methods_tools would be unreachable.
+from .mcp_rpc_helpers import (  # noqa: E402
+    reset_profile as _mcp_reset_profile,
+    summarize_server as _mcp_summarize_server_impl,
+)
+
+
+def _mcp_resolve_profile(rid, params):  # noqa: E402
+    # Bind this namespace's _err so the helper's error envelopes match every
+    # other handler's shape; handlers call this with just (rid, params).
+    from .mcp_rpc_helpers import resolve_profile as _rp
+
+    return _rp(rid, params, _err)
+
+
+def _mcp_summarize_server(name, cfg):  # noqa: E402
+    return _mcp_summarize_server_impl(name, cfg)
+
+
 # ── Split @method handler modules (see method_ctx.py) ────────────────
 # Imported at the end of this module so every global the handlers close
 # over already exists; register() rebinds them onto this namespace.