fix(mcp): same-named MCP servers with different credentials connect per profile; owner /reload-mcp keeps adopters' tools

Under gateway.multiplex_profiles every connection ledger in tools/mcp_tool.py
(_servers, _server_scope_keys/_server_tool_scopes, connecting/error/cooldown
maps, the circuit breaker, lazy schema-cache configs, trust metadata) was keyed
by the bare server NAME. The common per-tenant layout — each profile names its
server `github`/`notion` with its own token — gave only the first profile a
connection: the second profile's register_mcp_servers saw the name as "already
connected", refused to adopt it (different credentials, 4ddbcbd35e), and left
the profile silently tool-less with a healthy-looking `configured` status
(#106005 Bug 1/2, #91654). Siblings of the same bug: profile A's failing `x`
put profile B's healthy `x` into A's 10-minute connect cooldown and A's open
circuit breaker short-circuited B's calls; toolsets._resolve_toolset_memo was
not scope-keyed, so B resolved A's `mcp-<server>` tool names.

Keys are now the connection key from the new tools/mcp_tool_scope.py: the bare
name outside a multiplexer (single-profile processes are unchanged) and
(owner_scope, name) under one. Call-time lookups (_resolve_server_key) prefer
the calling scope's own connection, then a shared connection it adopted, so
identical-route profiles still share one subprocess. _select_new_servers,
the cooldown/breaker/trust maps, lazy registration and get_mcp_status all read
and write through the composite key; teardown resolves a task's key by
identity (the MCP loop has no profile context). The toolset memo key includes
registry.current_scope_key().

An owner's scoped /reload-mcp tore down its connection and, with it, every
adopting profile's tool overlay; nothing re-ran the adopters' discovery until
they reloaded. shutdown_mcp_servers(scope=) now records the orphaned adopters
and register_mcp_servers re-registers them under their own home + secret scope
once the owner's rediscovery pass completes.

Docs: multi-profile-gateways.md states the per-profile connection rule.

Fixes #106005
Fixes #91654
Co-authored-by: Bergmann89 <info@bergmann89.de>
Co-authored-by: Izzy-Gottz <srulynj@gmail.com>

tests/test_toolsets.py

@@ -337,7 +337,7 @@ class TestResolveToolsetMemo:
             f"got {get_toolset_calls['n']} calls"
         )
         assert (
-            "hermes-cli", True, registry_id, generation
+            "hermes-cli", True, registry_id, generation, registry.current_scope_key()
         ) in toolsets_mod._resolve_toolset_memo
 
     def test_generation_bump_invalidates_memo(self, monkeypatch):

tests/tools/test_mcp_multiplex_connection_keys.py

@@ -0,0 +1,134 @@
+"""Two multiplexed profiles that name the same MCP server with different credentials are two
+connections (#106005, #91654): the ledgers in ``tools.mcp_tool`` are keyed per owning profile
+scope, and an owner's scoped reload re-registers the profiles that had adopted its connection."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+from hermes_constants import hermes_home_key, reset_hermes_home_override, set_hermes_home_override
+
+
+def _tool():
+    return SimpleNamespace(name="t", description="d", inputSchema={"type": "object", "properties": {}},
+                           annotations=None)
+
+
+def _server(name, cfg):
+    return SimpleNamespace(name=name, session=object(), _config=cfg, _tools=[_tool()], tool_timeout=30,
+                           initialize_result=None, _registered_tool_names=[], _sampling=None)
+
+
+@pytest.fixture
+def two_profiles(tmp_path, monkeypatch):
+    """Multiplex on, clean MCP ledgers, a scope switcher for homes A and B; restores everything."""
+    import tools.mcp_tool as core
+    from tools import mcp_tool_config as _config
+    from tools.registry import registry
+
+    homes = {k: tmp_path / "profiles" / k for k in ("a", "b")}
+    for home in homes.values():
+        home.mkdir(parents=True)
+    monkeypatch.setattr("agent.secret_scope.is_multiplex_active", lambda: True)
+    monkeypatch.setattr(core, "_ensure_mcp_sdk", lambda: True)
+    monkeypatch.setattr(_config, "_filter_suspicious_mcp_servers", lambda servers: servers)
+    ledgers = ("_servers", "_server_scope_keys", "_server_tool_scopes", "_server_connecting",
+               "_server_connect_errors", "_server_connect_retry_after", "_server_connect_failures",
+               "_server_error_counts", "_server_breaker_opened_at", "_lazy_server_configs",
+               "_mcp_tool_server_names", "_orphaned_adopters")
+    saved = {n: type(getattr(core, n))(getattr(core, n)) for n in ledgers}
+    for n in ledgers:
+        getattr(core, n).clear()
+    tokens = []
+
+    def enter(which):
+        tokens.append(set_hermes_home_override(homes[which]))
+        return hermes_home_key(homes[which])
+
+    yield enter
+    for tool_name in list(registry.get_tool_names_for_toolset("mcp-x")):
+        for home in homes.values():
+            registry.deregister(tool_name, scope=hermes_home_key(home))
+    for token in reversed(tokens):
+        reset_hermes_home_override(token)
+    for n in ledgers:
+        getattr(core, n).clear()
+        getattr(core, n).update(saved[n])
+
+
+def test_same_named_server_with_other_credentials_is_a_separate_connection(two_profiles):
+    import tools.mcp_tool as core
+    from tools import mcp_tool_discovery as disc, mcp_tool_handlers as handlers
+    from tools import mcp_tool_registration as reg
+    from tools.registry import registry
+    import toolsets
+
+    cfg_a = {"url": "https://mcp.example/x", "headers": {"Authorization": "Bearer A"}}
+    cfg_b = {"url": "https://mcp.example/x", "headers": {"Authorization": "Bearer B"}}
+
+    two_profiles("a")
+    srv_a = _server("x", cfg_a)
+    disc._adopt_server("x", srv_a)
+    srv_a._registered_tool_names = reg._register_server_tools("x", srv_a, cfg_a)
+    assert toolsets.resolve_toolset("mcp-x") == ["mcp__x__t"]
+    for _ in range(core._CIRCUIT_BREAKER_THRESHOLD):
+        core._bump_server_error("x")
+    disc._note_connect_failure("y", RuntimeError("boom"))
+
+    two_profiles("b")
+    # B's own view: no tools yet, its memo is not A's, and A's connection is not "connected" for B.
+    assert registry.get_tool_names_for_toolset("mcp-x") == []
+    assert toolsets.resolve_toolset("mcp-x") == []
+    assert disc.get_mcp_status({"x": cfg_b})[0]["status"] == "configured"
+    # B's differently-authenticated 'x' is a connect candidate, not shadowed by A's ledger entries.
+    assert "x" in disc._select_new_servers({"x": cfg_b})
+    assert not disc._connect_cooldown_active("y")
+    assert handlers._check_circuit_breaker("x") is None
+
+
+def test_owner_reload_reregisters_profiles_that_adopted_its_connection(two_profiles):
+    import tools.mcp_tool as core
+    from tools import mcp_tool_discovery as disc, mcp_tool_lifecycle as lifecycle
+    from tools import mcp_tool_registration as reg
+    from tools.registry import registry
+
+    cfg = {"url": "https://mcp.example/x", "headers": {"Authorization": "Bearer shared"}}
+    scope_a = two_profiles("a")
+    srv_a = _server("x", cfg)
+    disc._adopt_server("x", srv_a)
+    srv_a._registered_tool_names = reg._register_server_tools("x", srv_a, cfg)
+
+    two_profiles("b")
+    assert reg.register_connected_into_current_scope({"x": cfg}) == 1
+    assert registry.get_tool_names_for_toolset("mcp-x") == ["mcp__x__t"]
+
+    # Owner A: scoped shutdown (no MCP loop here, so emulate the task teardown), then rediscovery.
+    two_profiles("a")
+    with patch.object(lifecycle._loop, "_stop_mcp_loop", lambda **_kw: False):
+        lifecycle.shutdown_mcp_servers(scope=scope_a)
+    for tool_name in list(srv_a._registered_tool_names):
+        reg._deregister_mcp_tool_all_scopes(srv_a, tool_name)
+    with core._lock:
+        for key in [k for k, v in core._servers.items() if v is srv_a]:
+            core._servers.pop(key)
+            core._server_scope_keys.pop(key, None)
+            core._server_tool_scopes.pop(key, None)
+
+    def fake_pass(new_servers):
+        for name, config in new_servers.items():
+            srv = _server(name, config)
+            disc._adopt_server(name, srv)
+            srv._registered_tool_names = reg._register_server_tools(name, srv, config)
+
+    with patch.object(disc, "_run_discovery_pass", fake_pass), \
+            patch.object(disc._loop, "_ensure_mcp_loop", lambda: None), \
+            patch("tools.mcp_tool_config._load_mcp_config", lambda: {"x": cfg}):
+        disc.register_mcp_servers({"x": cfg})
+
+    # B never reloaded, yet has its tools back on the owner's new identical connection.
+    two_profiles("b")
+    assert registry.get_tool_names_for_toolset("mcp-x") == ["mcp__x__t"]
+    assert disc.get_mcp_status({"x": cfg})[0]["status"] == "connected"

tools/mcp_tool.py

@@ -396,22 +396,30 @@ class MCPServerTask(MCPServerRunMixin, MCPServerTransportMixin, MCPServerHealthM
 
 
 # ---- Module-level state (every mutation under ``_lock``) ----
+#
+# Every ledger below is keyed by the CONNECTION KEY from ``tools.mcp_tool_scope``: the bare
+# server name outside a multiplexer, ``(owner_scope, name)`` under one. Two profiles naming the
+# same server with their own credentials are two connections; a name-keyed ledger let the first
+# profile's connection shadow the second's (never connected, silently tool-less — #106005).
 
-_servers: Dict[str, MCPServerTask] = {}
+_servers: Dict[Any, MCPServerTask] = {}
 # Profile registry scope per live connection (None outside multiplex) so a multiplexed
 # /reload-mcp tears down only its own profile's servers.
-_server_scope_keys: Dict[str, Optional[str]] = {}
+_server_scope_keys: Dict[Any, Optional[str]] = {}
 # Registry scopes that have adopted a live server connection. The owning scope above remains
 # authoritative for connection teardown; this set preserves visibility for shared connections.
-_server_tool_scopes: Dict[str, set] = {}
-_server_connecting: set[str] = set()
-_server_connect_errors: Dict[str, str] = {}
+_server_tool_scopes: Dict[Any, set] = {}
+_server_connecting: set = set()
+_server_connect_errors: Dict[Any, str] = {}
+# adopter scope -> server names whose shared connection an owner's scoped shutdown tore down;
+# drained by the next discovery pass so the adopter is re-registered (see mcp_tool_lifecycle).
+_orphaned_adopters: Dict[str, set] = {}
 # Lazy startup: servers registered from the schema cache without connecting; popped on
 # first real connection.
-# Keyed by server name; entries are popped once a real connection is established on first use. See #56832.
-_lazy_server_configs: Dict[str, dict] = {}
-_lazy_server_fingerprints: Dict[str, str] = {}
-_lazy_server_tool_names: Dict[str, List[str]] = {}
+# Keyed by connection key; entries are popped once a real connection is established on first use. See #56832.
+_lazy_server_configs: Dict[Any, dict] = {}
+_lazy_server_fingerprints: Dict[Any, str] = {}
+_lazy_server_tool_names: Dict[Any, List[str]] = {}
 # Task-local claim around ``_connect_server``: discovery retains a recoverable parked task
 # while standalone probes never publish failed servers into module-global ownership.
 _connect_server_claim: contextvars.ContextVar[Optional[Callable[[MCPServerTask], None]]] = (
@@ -432,8 +440,8 @@ _connect_server_claim: contextvars.ContextVar[Optional[Callable[[MCPServerTask],
 # ``retry_after`` deadline with exponential backoff. ``register_mcp_servers`` skips a server whose cooldown
 # has not elapsed, so a chronically failing server is retried on a backoff schedule instead of on every
 # worker session -- isolating it from the rest of the bridge. A successful connection clears the state.
-_server_connect_retry_after: Dict[str, float] = {}   # name -> monotonic deadline
-_server_connect_failures: Dict[str, int] = {}        # name -> consecutive failures
+_server_connect_retry_after: Dict[Any, float] = {}   # connection key -> monotonic deadline
+_server_connect_failures: Dict[Any, int] = {}        # connection key -> consecutive failures
 _CONNECT_RETRY_BASE_BACKOFF_SEC, _CONNECT_RETRY_MAX_BACKOFF_SEC = 30.0, 600.0
 
 # Per-server circuit breaker: closed -> open (calls short-circuit until the cooldown) ->
@@ -446,8 +454,8 @@ _CONNECT_RETRY_BASE_BACKOFF_SEC, _CONNECT_RETRY_MAX_BACKOFF_SEC = 30.0, 600.0
 # ``_server_breaker_opened_at`` records the monotonic timestamp when the breaker most recently transitioned
 # into the open state. Use the ``_bump_server_error`` / ``_reset_server_error`` helpers to mutate this state
 # — they keep the count and timestamp in sync.
-_server_error_counts: Dict[str, int] = {}
-_server_breaker_opened_at: Dict[str, float] = {}
+_server_error_counts: Dict[Any, int] = {}
+_server_breaker_opened_at: Dict[Any, float] = {}
 _CIRCUIT_BREAKER_THRESHOLD, _CIRCUIT_BREAKER_COOLDOWN_SEC = 3, 60.0
 
 # Trust-tier gating (``trust: full | untrusted``): on an untrusted server every write-capable
@@ -456,24 +464,29 @@ _CIRCUIT_BREAKER_THRESHOLD, _CIRCUIT_BREAKER_COOLDOWN_SEC = 3, 60.0
 # already warned about, never widen access. Missing trust = full; unrecognized = untrusted (a
 # typo must never disable the gate). Classified at CALL time from DISCOVERY data: no schema
 # mutation, prompt cache intact.
-_server_trust_levels: Dict[str, str] = {}
-_tool_read_only_hints: Dict[str, Dict[str, bool]] = {}
+_server_trust_levels: Dict[Any, str] = {}
+_tool_read_only_hints: Dict[Any, Dict[str, bool]] = {}
 
 _TRUST_FULL, _TRUST_UNTRUSTED = "full", "untrusted"
 
 
 def _bump_server_error(server_name: str) -> None:
-    """Count a failure; at the threshold (re)stamp the breaker-open time."""
-    n = _server_error_counts.get(server_name, 0) + 1
-    _server_error_counts[server_name] = n
+    """Count a failure; at the threshold (re)stamp the breaker-open time. Keyed by the calling
+    scope's connection so one profile's failing server never opens another profile's breaker."""
+    from tools.mcp_tool_scope import _resolve_server_key
+    key = _resolve_server_key(server_name)
+    n = _server_error_counts.get(key, 0) + 1
+    _server_error_counts[key] = n
     if n >= _CIRCUIT_BREAKER_THRESHOLD:
-        _server_breaker_opened_at[server_name] = time.monotonic()
+        _server_breaker_opened_at[key] = time.monotonic()
 
 
 def _reset_server_error(server_name: str) -> None:
     """Close the breaker on any unambiguous success signal."""
-    _server_error_counts[server_name] = 0
-    _server_breaker_opened_at.pop(server_name, None)
+    from tools.mcp_tool_scope import _resolve_server_key
+    key = _resolve_server_key(server_name)
+    _server_error_counts[key] = 0
+    _server_breaker_opened_at.pop(key, None)
 
 
 # Raw server names opted into parallel tool calls (``foo-bar``/``foo_bar`` sanitize alike but
@@ -622,20 +635,22 @@ def _mcp_registry_scope() -> Optional[str]:
     return registry.current_scope_key()
 
 
-def _server_registry_scope(name: str) -> Optional[str]:
-    """Scope owning *name*'s tools: the one captured at adoption (teardown runs on the MCP
-    loop without the discovering profile's context), else the current one."""
-    if name in _server_scope_keys:
-        return _server_scope_keys[name]
-    return _mcp_registry_scope()
+def _server_registry_scope(key) -> Optional[str]:
+    """Scope owning the connection under *key*'s tools: the one captured at adoption (teardown
+    runs on the MCP loop without the discovering profile's context), else the current one."""
+    if key in _server_scope_keys:
+        return _server_scope_keys[key]
+    from tools.mcp_tool_scope import _key_scope
+    return _key_scope(key) or _mcp_registry_scope()
 
 
-def _server_visible_in_scope(name: str, scope: Optional[str]) -> bool:
-    """Whether a live server is visible from ``scope`` without changing its teardown owner."""
+def _server_visible_in_scope(key, scope: Optional[str]) -> bool:
+    """Whether the live connection under *key* is visible from ``scope`` without changing its
+    teardown owner."""
     if scope is None:
         return True
-    return (_server_scope_keys.get(name) == scope
-            or scope in _server_tool_scopes.get(name, ()))
+    return (_server_scope_keys.get(key) == scope
+            or scope in _server_tool_scopes.get(key, ()))
 
 
 # Cross-process discovery guard: advisory file lock so gateway + CLI + TUI don't all discover.

tools/mcp_tool_discovery.py

@@ -16,27 +16,31 @@ from tools import mcp_tool_lifecycle as _lifecycle
 from tools import mcp_tool_loop as _loop
 from tools import mcp_tool_registration as _registration
 from tools.mcp_tool_schema import MCP_TOOL_NAME_PREFIX
+from tools.mcp_tool_scope import _key_name, _resolve_server_key, _server_key
 
 logger = logging.getLogger("tools.mcp_tool")
 
 
 def _record_connect_failure(server_name: str) -> None:
     """Stamp a geometric, capped retry cooldown after a failed connect (under ``_lock``)."""
-    n = _core._server_connect_failures.get(server_name, 0) + 1
-    _core._server_connect_failures[server_name] = n
+    key = _server_key(server_name)
+    n = _core._server_connect_failures.get(key, 0) + 1
+    _core._server_connect_failures[key] = n
     backoff = min(_core._CONNECT_RETRY_BASE_BACKOFF_SEC * (2 ** (n - 1)), _core._CONNECT_RETRY_MAX_BACKOFF_SEC)
-    _core._server_connect_retry_after[server_name] = time.monotonic() + backoff
+    _core._server_connect_retry_after[key] = time.monotonic() + backoff
 
 
 def _clear_connect_failure(server_name: str) -> None:
     """Clear the connect-cooldown state after a successful connection."""
-    _core._server_connect_failures.pop(server_name, None)
-    _core._server_connect_retry_after.pop(server_name, None)
+    key = _server_key(server_name)
+    _core._server_connect_failures.pop(key, None)
+    _core._server_connect_retry_after.pop(key, None)
 
 
 def _connect_cooldown_active(server_name: str) -> bool:
-    """True if ``server_name`` is still within its retry cooldown."""
-    deadline = _core._server_connect_retry_after.get(server_name)
+    """True if ``server_name`` is still within its retry cooldown (this scope's connection: one
+    profile's failing ``x`` must not shadow another profile's healthy ``x``)."""
+    deadline = _core._server_connect_retry_after.get(_server_key(server_name))
     return deadline is not None and time.monotonic() < deadline
 
 
@@ -111,8 +115,9 @@ def _note_connect_failure(name: str, exc: BaseException) -> str:
     """Record a failed connect (under ``_lock``): error text for status, cooldown stamp."""
     message = _errors._format_connect_error(exc)
     with _core._lock:
-        _core._server_connecting.discard(name)
-        _core._server_connect_errors[name] = message
+        key = _server_key(name)
+        _core._server_connecting.discard(key)
+        _core._server_connect_errors[key] = message
         _record_connect_failure(name)
     return message
 
@@ -120,16 +125,18 @@ def _note_connect_failure(name: str, exc: BaseException) -> str:
 def _note_connect_success(name: str) -> None:
     """Clear connecting/error/cooldown state after a successful connect (under ``_lock``)."""
     with _core._lock:
-        _core._server_connecting.discard(name)
-        _core._server_connect_errors.pop(name, None)
+        key = _server_key(name)
+        _core._server_connecting.discard(key)
+        _core._server_connect_errors.pop(key, None)
         _clear_connect_failure(name)
 
 
 def _adopt_server(name: str, server: _core.MCPServerTask) -> None:
-    """Publish *server* into ``_servers`` with its owning registry scope (under ``_lock``)."""
+    """Publish *server* into ``_servers`` under the connecting scope's key (under ``_lock``)."""
     with _core._lock:
-        _core._servers[name] = server
-        _core._server_scope_keys[name] = _core._mcp_registry_scope()
+        key = _server_key(name)
+        _core._servers[key] = server
+        _core._server_scope_keys[key] = _core._mcp_registry_scope()
 
 
 def _ensure_lazy_server_connected(server_name: str) -> bool:
@@ -140,15 +147,16 @@ def _ensure_lazy_server_connected(server_name: str) -> bool:
     See #50394.
     """
     with _core._lock:
-        server = _core._servers.get(server_name)
+        key = _resolve_server_key(server_name)
+        server = _core._servers.get(key)
         if server is not None and server.session is not None:
             return True
-        config = _core._lazy_server_configs.get(server_name)
+        config = _core._lazy_server_configs.get(key)
         if (not config or _connect_cooldown_active(server_name)
-                or server_name in _core._server_connecting):
+                or key in _core._server_connecting):
             return False
-        _core._server_connecting.add(server_name)
-        _core._server_connect_errors.pop(server_name, None)
+        _core._server_connecting.add(key)
+        _core._server_connect_errors.pop(key, None)
     logger.info("MCP server '%s': lazy start on first use", server_name)
     _loop._ensure_mcp_loop()
     connect_timeout = config.get("connect_timeout", _core._DEFAULT_CONNECT_TIMEOUT)
@@ -160,16 +168,16 @@ def _ensure_lazy_server_connected(server_name: str) -> bool:
         return False
     _note_connect_success(server_name)
     with _core._lock:
-        _core._lazy_server_configs.pop(server_name, None)
-        stale_fingerprint = _core._lazy_server_fingerprints.pop(server_name, None)
-        cached_names = _core._lazy_server_tool_names.pop(server_name, None) or []
-        server = _core._servers.get(server_name)
+        _core._lazy_server_configs.pop(key, None)
+        stale_fingerprint = _core._lazy_server_fingerprints.pop(key, None)
+        cached_names = _core._lazy_server_tool_names.pop(key, None) or []
+        server = _core._servers.get(key)
         live_names = set(getattr(server, "_registered_tool_names", []) or [])
     # The cached manifest may advertise tools the live server no longer serves.
     phantom_names = [n for n in cached_names if n not in live_names]
     if phantom_names:
         for tool_name in phantom_names:
-            _registration._deregister_mcp_tool_all_scopes(server_name, tool_name)
+            _registration._deregister_mcp_tool_all_scopes(key, tool_name)
         logger.info("MCP server '%s': deregistered %d phantom cached tool(s) not served live (stale schema-cache "
                     "fingerprint %s): %s", server_name, len(phantom_names), stale_fingerprint, ", ".join(phantom_names))
     return server is not None and server.session is not None
@@ -183,8 +191,9 @@ def _get_connected_server_for_call(server_name: str) -> Optional[_core.MCPServer
     AND the resource/prompt utility handlers all trigger the deferred spawn (#56832).
     """
     with _core._lock:
-        server = _core._servers.get(server_name)
-        is_lazy = server_name in _core._lazy_server_configs
+        key = _resolve_server_key(server_name)
+        server = _core._servers.get(key)
+        is_lazy = key in _core._lazy_server_configs
     if is_lazy and (server is None or server.session is None):
         _ensure_lazy_server_connected(server_name)
     elif server is not None and server.session is None and server._is_recycled_stdio():
@@ -192,7 +201,7 @@ def _get_connected_server_for_call(server_name: str) -> Optional[_core.MCPServer
     else:
         return server
     with _core._lock:
-        return _core._servers.get(server_name)
+        return _core._servers.get(key)
 
 
 async def _discover_and_register_server(name: str, config: dict) -> List[str]:
@@ -217,8 +226,9 @@ async def _discover_and_register_server(name: str, config: dict) -> List[str]:
     finally:
         _core._connect_server_claim.reset(claim_token)
     with _core._lock:
-        _core._server_connecting.discard(name)
-        _core._server_connect_errors.pop(name, None)
+        key = _server_key(name)
+        _core._server_connecting.discard(key)
+        _core._server_connect_errors.pop(key, None)
     _adopt_server(name, server)
     registered_names = _registration._register_server_tools(name, server, config)
     server._registered_tool_names = list(registered_names)
@@ -232,21 +242,25 @@ def _select_new_servers(servers: Dict[str, dict]) -> Dict[str, dict]:
     refresh per-server bookkeeping. Known servers without a live session are parked or
     mid-reconnect with tools deregistered, so nothing else can nudge them: signal a reconnect."""
     with _core._lock:
-        connecting = set(_core._server_connecting)
+        current_scope = _core._mcp_registry_scope()
+        # This scope's own connections OR shared ones it adopted (``register_connected_into_current_scope``
+        # ran first): a same-named server owned by ANOTHER profile with other credentials is not
+        # "connected" for us and must be a candidate, or this profile ends up silently tool-less.
+        keys = {k: _resolve_server_key(k, current_scope, current=False) for k in servers}
         # Only attempt servers that aren't already connected (or currently connecting) and are enabled.
         # Checking ``_server_connecting`` prevents duplicate subprocess spawns when ``discover_mcp_tools()``
         # is called from multiple entry-points before the first batch finishes (#58862).
         new_servers = {
             k: v for k, v in servers.items()
-            if k not in _core._servers and k not in connecting and k not in _core._lazy_server_configs
+            if keys[k] not in _core._servers and keys[k] not in _core._server_connecting
+            and keys[k] not in _core._lazy_server_configs
             and _enabled(v) and not _connect_cooldown_active(k)}
-        stale_cached = [_core._servers[k] for k in servers
-                        if k in _core._servers and getattr(_core._servers[k], "session", None) is None]
-        _core._server_connecting.update(new_servers)
-        current_scope = _core._mcp_registry_scope()
+        stale_cached = [_core._servers[keys[k]] for k in servers
+                        if keys[k] in _core._servers and getattr(_core._servers[keys[k]], "session", None) is None]
         for srv_name in new_servers:
-            _core._server_scope_keys[srv_name] = current_scope
-            _core._server_connect_errors.pop(srv_name, None)
+            _core._server_connecting.add(keys[srv_name])
+            _core._server_scope_keys[keys[srv_name]] = current_scope
+            _core._server_connect_errors.pop(keys[srv_name], None)
         # Track which servers opt-in to parallel tool calls (idempotent).
         for srv_name, srv_cfg in servers.items():
             if _parse_boolish(srv_cfg.get("supports_parallel_tool_calls", False), default=False):
@@ -278,13 +292,13 @@ def _register_lazy_from_cache(new_servers: Dict[str, dict]) -> Tuple[Dict[str, d
         if not entry:
             continue
         with _core._lock:
-            _core._server_connecting.discard(name)
+            _core._server_connecting.discard(_server_key(name))
         try:
             names = _registration._register_from_cache_sync(name, cfg, entry)
         except Exception as exc:
             logger.warning("Failed lazy MCP registration for '%s': %s", name, exc)
             with _core._lock:
-                _core._server_connecting.add(name)
+                _core._server_connecting.add(_server_key(name))
             continue
         eager_servers.pop(name, None)
         lazy_registered += len(names)
@@ -321,13 +335,14 @@ def _run_discovery_pass(new_servers: Dict[str, dict]) -> None:
         # Stranded _server_connecting entries would block future reconnects.
         how = "timed out" if isinstance(_e, TimeoutError) else "interrupted"
         with _core._lock:
-            stale = [n for n in new_servers if n in _core._server_connecting]
+            stale = [n for n in new_servers if _server_key(n) in _core._server_connecting]
             if stale:
                 logger.warning("MCP discovery %s while %d server(s) were still connecting; clearing stale "
                                "connecting set: %s", how, len(stale), ", ".join(stale))
-                _core._server_connecting.difference_update(stale)
                 for _sn in stale:
-                    _core._server_connect_errors.setdefault(_sn, f"Connection attempt {how} during discovery")
+                    _core._server_connecting.discard(_server_key(_sn))
+                    _core._server_connect_errors.setdefault(
+                        _server_key(_sn), f"Connection attempt {how} during discovery")
         raise
     finally:
         if _was_interrupted:
@@ -337,8 +352,10 @@ def _run_discovery_pass(new_servers: Dict[str, dict]) -> None:
 def _connected_summary(names, *, lazy_tools: int = 0, lazy_servers: int = 0) -> Tuple[int, int, int]:
     """(tool count, connected count, failed count) for candidate names, plus lazy servers."""
     with _core._lock:
-        connected = [n for n in names if n in _core._servers and n not in _core._server_connect_errors]
-        tool_count = sum(len(getattr(_core._servers[n], "_registered_tool_names", [])) for n in connected)
+        keys = {n: _server_key(n) for n in names}
+        connected = [n for n in names
+                     if keys[n] in _core._servers and keys[n] not in _core._server_connect_errors]
+        tool_count = sum(len(getattr(_core._servers[keys[n]], "_registered_tool_names", [])) for n in connected)
     failed = len(names) - len(connected)
     return tool_count + lazy_tools, len(connected) + lazy_servers, failed
 
@@ -360,6 +377,15 @@ def register_mcp_servers(servers: Dict[str, dict]) -> List[str]:
         logger.debug("MCP SDK not available -- skipping explicit MCP registration")
         return []
     servers = _config._filter_suspicious_mcp_servers(servers)
+    try:
+        return _register_mcp_servers(servers)
+    finally:
+        # An owner's scoped reload orphaned adopters of its shared connections: now that this
+        # pass (its rediscovery) is done, give them their tools back under their own scope.
+        _lifecycle._reregister_orphaned_adopters()
+
+
+def _register_mcp_servers(servers: Dict[str, dict]) -> List[str]:
     scoped_healed = _registration.register_connected_into_current_scope(servers)
     if not servers:
         logger.debug("No explicit MCP servers provided")
@@ -434,9 +460,10 @@ def discover_mcp_tools(allowed_mcp_names: Optional[List[str]] = None) -> List[st
     cookie = _acquire_discovery_lock_with_retry()
     try:
         with _core._lock:
-            connecting = set(_core._server_connecting)
+            keys = {name: _resolve_server_key(name) for name in servers}
             new_server_names = [name for name, cfg in servers.items()
-                                if name not in _core._servers and name not in connecting and _enabled(cfg)]
+                                if keys[name] not in _core._servers and keys[name] not in _core._server_connecting
+                                and _enabled(cfg)]
         tool_names = register_mcp_servers(servers)
         if new_server_names:
             _log_summary("  MCP:", new_server_names)
@@ -465,15 +492,15 @@ def get_mcp_status(configured: Optional[Dict[str, dict]] = None, *, include_runt
         return []
     current_scope = _core._mcp_registry_scope()
     with _core._lock:
-        def visible(name: str) -> bool:
+        def visible(key) -> bool:
             # Runtime state belongs to the profile that adopted it; under a multiplexer only that
             # profile's view may show it, and ``include_runtime=False`` hides the launch profile's
             # servers from a status read scoped to a different profile.
-            return include_runtime and _core._server_visible_in_scope(name, current_scope)
+            return include_runtime and _core._server_visible_in_scope(key, current_scope)
 
-        active_servers = {n: s for n, s in _core._servers.items() if visible(n)}
-        connecting = {n for n in _core._server_connecting if visible(n)}
-        connect_errors = {n: e for n, e in _core._server_connect_errors.items() if visible(n)}
+        active_servers = {_key_name(k): s for k, s in _core._servers.items() if visible(k)}
+        connecting = {_key_name(k) for k in _core._server_connecting if visible(k)}
+        connect_errors = {_key_name(k): e for k, e in _core._server_connect_errors.items() if visible(k)}
 
     result: List[dict] = []
     for name, cfg in configured.items():

tools/mcp_tool_handlers.py

@@ -41,8 +41,10 @@ _STDIO_OUTCOME_UNCERTAIN_MSG = (
 def _trust_gate_check(server_name: str, tool_name: str) -> Optional[str]:
     """Approval gate for write-capable tools on ``trust: untrusted`` servers. None to proceed,
     else a ``tool_error``. Fail-closed: approval-system errors block."""
-    if (_core._server_trust_levels.get(server_name, _core._TRUST_FULL) != _core._TRUST_UNTRUSTED
-            or _core._tool_read_only_hints.get(server_name, {}).get(tool_name) is True):
+    from tools.mcp_tool_scope import _resolve_server_key
+    key = _resolve_server_key(server_name)
+    if (_core._server_trust_levels.get(key, _core._TRUST_FULL) != _core._TRUST_UNTRUSTED
+            or _core._tool_read_only_hints.get(key, {}).get(tool_name) is True):
         return None
     try:  # lazy: tools.approval routes the prompt to whichever surface owns the session
         from tools.approval_prompt import request_elicitation_consent
@@ -67,8 +69,10 @@ def _trust_gate_check(server_name: str, tool_name: str) -> Optional[str]:
 def _check_circuit_breaker(server_name: str) -> Optional[str]:
     """Open-breaker error, or None when calls may proceed. After the cooldown the breaker is
     half-open: the next call probes; success resets, failure re-bumps and re-arms the cooldown."""
-    failures = _core._server_error_counts.get(server_name, 0)
-    age = time.monotonic() - _core._server_breaker_opened_at.get(server_name, 0.0)
+    from tools.mcp_tool_scope import _resolve_server_key
+    key = _resolve_server_key(server_name)
+    failures = _core._server_error_counts.get(key, 0)
+    age = time.monotonic() - _core._server_breaker_opened_at.get(key, 0.0)
     if failures < _core._CIRCUIT_BREAKER_THRESHOLD or age >= _core._CIRCUIT_BREAKER_COOLDOWN_SEC:
         return None
     return tool_error(f"MCP server '{server_name}' is unreachable after {failures} consecutive failures. "
@@ -120,8 +124,9 @@ def _mcp_loop_running() -> bool:
 def _lookup_reconnectable_server(server_name: str, require_loop: bool = False):
     """The registered server object when it can be signalled to reconnect, else None.
     With *require_loop*, also None unless the MCP loop is running (nothing to wait on)."""
+    from tools.mcp_tool_scope import _resolve_server_key
     with _core._lock:
-        srv = _core._servers.get(server_name)
+        srv = _core._servers.get(_resolve_server_key(server_name))
     ok = srv is not None and hasattr(srv, "_reconnect_event") and (_mcp_loop_running() or not require_loop)
     return srv if ok else None
 
@@ -591,9 +596,12 @@ _make_get_prompt_handler = _make_utility_handler(
 
 def _make_check_fn(server_name: str):
     """Connection-alive check; lazy (schema-cache registered) servers count as available."""
+    from tools.mcp_tool_scope import _resolve_server_key
+
     def _check() -> bool:
         with _core._lock:
-            server = _core._servers.get(server_name)
+            key = _resolve_server_key(server_name)
+            server = _core._servers.get(key)
             return ((server is not None and (server.session is not None or server._is_recycled_stdio()))
-                    or server_name in _core._lazy_server_configs)
+                    or key in _core._lazy_server_configs)
     return _check

tools/mcp_tool_health.py

@@ -127,7 +127,7 @@ class MCPServerHealthMixin:
         from tools.registry import registry
         for tool_name in tool_names:
             if registry.get_toolset_for_tool(tool_name) == f"mcp-{self.name}":
-                _registration._deregister_mcp_tool_all_scopes(self.name, tool_name)
+                _registration._deregister_mcp_tool_all_scopes(self, tool_name)
 
     async def _refresh_tools(self):
         """Re-fetch tools on ``tools/list_changed`` and update the registry. The lock serializes rapid-fire

tools/mcp_tool_lifecycle.py

@@ -84,16 +84,47 @@ def _filter_mcp_children(pids: set) -> set:
     return kept
 
 
-def _clear_connect_cooldowns(names=None) -> None:
+def _clear_connect_cooldowns(keys=None) -> None:
     """Drop connect-retry cooldowns: a restart must re-attempt every server immediately, not
     honour a stale per-server backoff. Caller holds ``_core._lock``."""
-    if names is None:
+    if keys is None:
         _core._server_connect_retry_after.clear()
         _core._server_connect_failures.clear()
     else:
-        for name in names:
-            _core._server_connect_retry_after.pop(name, None)
-            _core._server_connect_failures.pop(name, None)
+        for key in keys:
+            _core._server_connect_retry_after.pop(key, None)
+            _core._server_connect_failures.pop(key, None)
+
+
+def _reregister_orphaned_adopters() -> None:
+    """Re-run MCP registration for profiles whose ADOPTED shared connection an owner's
+    ``/reload-mcp`` just tore down. Their tools vanished with the owner's teardown and nothing
+    re-runs their discovery until THEY reload, so they sat tool-less behind a healthy-looking
+    status (#106005). Runs after the owner's rediscovery, under each adopter's own home + secret
+    scope (its ``${VAR}`` refs must resolve to ITS credentials): the adopter re-adopts the owner's
+    new identical connection or connects its own."""
+    with _core._lock:
+        pending = dict(_core._orphaned_adopters)
+        _core._orphaned_adopters.clear()
+    if not pending:
+        return
+    from pathlib import Path
+    from agent.secret_scope import build_profile_secret_scope, reset_secret_scope, set_secret_scope
+    from hermes_constants import reset_hermes_home_override, set_hermes_home_override
+    from tools import mcp_tool_discovery as _discovery
+    from tools.mcp_tool_config import _load_mcp_config
+    for adopter, names in pending.items():
+        home_token = set_hermes_home_override(adopter)
+        secret_token = set_secret_scope(build_profile_secret_scope(Path(adopter)))
+        try:
+            servers = {n: c for n, c in (_load_mcp_config() or {}).items() if n in names}
+            if servers:
+                _discovery.register_mcp_servers(servers)
+        except Exception:
+            logger.debug("MCP: re-registration for profile scope %s failed", adopter, exc_info=True)
+        finally:
+            reset_secret_scope(secret_token)
+            reset_hermes_home_override(home_token)
 
 
 def shutdown_mcp_servers(*, scope: Optional[str] = None):
@@ -103,23 +134,32 @@ def shutdown_mcp_servers(*, scope: Optional[str] = None):
     (its ``/reload-mcp`` must not kill other profiles') and leaves the shared loop running if
     anything else is still connected."""
     with _core._lock:
-        selected = [name for name in _core._servers if scope is None or _core._server_scope_keys.get(name) == scope]
-        servers_snapshot = [_core._servers[name] for name in selected]
+        selected = [key for key in _core._servers if scope is None or _core._server_scope_keys.get(key) == scope]
+        servers_snapshot = [_core._servers[key] for key in selected]
         selected_status = (
             set(_core._servers) | set(_core._server_scope_keys)
             | set(_core._server_tool_scopes)
             | set(_core._server_connecting) | set(_core._server_connect_errors)
             if scope is None else {
-                name for name, owner in _core._server_scope_keys.items() if owner == scope
+                key for key, owner in _core._server_scope_keys.items() if owner == scope
             }
         )
+        # Adopters of the connections being torn down lose their overlays with the tasks' own
+        # ``_deregister_tools``; remember them so the next discovery pass re-registers them
+        # (``_reregister_orphaned_adopters``).
+        if scope is not None:
+            from tools.mcp_tool_scope import _key_name
+            for key in selected:
+                for adopter in _core._server_tool_scopes.get(key, ()):
+                    if adopter != scope:
+                        _core._orphaned_adopters.setdefault(adopter, set()).add(_key_name(key))
 
     def clear_selected_status():
         _core._server_connecting.difference_update(selected_status)
-        for name in selected_status:
-            _core._server_connect_errors.pop(name, None)
-            _core._server_scope_keys.pop(name, None)
-            _core._server_tool_scopes.pop(name, None)
+        for key in selected_status:
+            _core._server_connect_errors.pop(key, None)
+            _core._server_scope_keys.pop(key, None)
+            _core._server_tool_scopes.pop(key, None)
 
     # Fast path: nothing to shut down. The connect-cooldown maps can still be populated here — a server that
     # failed to connect is never recorded in ``_servers`` (that is the very premise of the #50394 cooldown),
@@ -132,9 +172,9 @@ def shutdown_mcp_servers(*, scope: Optional[str] = None):
                 if isinstance(result, Exception):
                     logger.debug("Error closing MCP server '%s': %s", server.name, result)
             with _core._lock:
-                for name in selected:
-                    _core._servers.pop(name, None)
-                    _core._server_scope_keys.pop(name, None)
+                for key in selected:
+                    _core._servers.pop(key, None)
+                    _core._server_scope_keys.pop(key, None)
                 clear_selected_status()
                 _clear_connect_cooldowns(None if scope is None else selected_status)
 

tools/mcp_tool_loop.py

@@ -194,8 +194,9 @@ def _signal_reconnect(server: Any) -> bool:
 
 def reconnect_mcp_server(server_name: str) -> bool:
     """Ask a currently-live MCP server to rebuild after external re-auth."""
+    from tools.mcp_tool_scope import _resolve_server_key
     with _core._lock:
-        server = _core._servers.get(server_name)
+        server = _core._servers.get(_resolve_server_key(server_name))
     return server is not None and _signal_reconnect(server)
 
 

tools/mcp_tool_registration.py

@@ -17,6 +17,7 @@ from tools.mcp_tool_handlers import (
     _make_list_resources_handler, _make_read_resource_handler)
 from tools.mcp_tool_schema import (
     _UTILITY_CAPABILITY_ATTRS, _build_utility_schemas, _normalize_name_filter, matches_name_filter)
+from tools.mcp_tool_scope import _key_name, _resolve_server_key, _server_key
 
 if TYPE_CHECKING:  # pragma: no cover
     from tools.mcp_tool import MCPServerTask
@@ -54,8 +55,9 @@ def _record_tool_trust_metadata(server_name: str, config: dict, tools: List[Any]
     """Capture per-server trust and per-tool readOnlyHint at discovery — the security boundary: the call-time gate
     classifies from data we control, never re-read server-supplied state."""
     with _core._lock:
-        _core._server_trust_levels[server_name] = _normalize_server_trust((config or {}).get("trust"))
-        hints = _core._tool_read_only_hints.setdefault(server_name, {})
+        key = _resolve_server_key(server_name)
+        _core._server_trust_levels[key] = _normalize_server_trust((config or {}).get("trust"))
+        hints = _core._tool_read_only_hints.setdefault(key, {})
         hints.update({t.name: _annotation_read_only_hint(t) for t in tools if getattr(t, "name", None)})
 
 
@@ -71,49 +73,65 @@ def _forget_mcp_tool_server(tool_name: str) -> None:
         _core._mcp_tool_server_names.pop(tool_name, None)
 
 
-def _deregister_mcp_tool_all_scopes(server_name: str, tool_name: str) -> None:
-    """Deregister one server tool from every profile overlay that owns it."""
+def _server_key_for_task(server) -> object:
+    """Connection key of a live ``MCPServerTask`` (teardown runs on the MCP loop without the
+    discovering profile's context, so the key is found by identity, never re-derived)."""
+    with _core._lock:
+        for key, live in _core._servers.items():
+            if live is server:
+                return key
+    return _server_key(server.name)
+
+
+def _deregister_mcp_tool_all_scopes(server, tool_name: str) -> None:
+    """Deregister one server tool from every profile overlay that owns it. *server* is the
+    live task or a connection key."""
     from tools.registry import registry
 
+    key = server if isinstance(server, (str, tuple)) else _server_key_for_task(server)
     with _core._lock:
-        scopes = set(_core._server_tool_scopes.get(server_name, ()))
+        scopes = set(_core._server_tool_scopes.get(key, ()))
         if not scopes:
-            scopes = {_core._server_registry_scope(server_name)}
+            scopes = {_core._server_registry_scope(key)}
     for scope in scopes:
         registry.deregister(tool_name, scope=scope)
     _forget_mcp_tool_server(tool_name)
-    _restore_server_toolset_alias(server_name)
+    _restore_server_toolset_alias(key)
 
 
-def _restore_server_toolset_alias(server_name: str) -> None:
-    """Keep the process-global alias while any profile still owns this server's tools."""
+def _restore_server_toolset_alias(key) -> None:
+    """Keep the process-global alias while any profile still owns tools of a server with this
+    name — including another profile's same-named connection (the alias is per NAME; the
+    registry deregister dropped it after checking only one scope)."""
     from tools.registry import registry
 
+    server_name = _key_name(key)
     with _core._lock:
-        server = _core._servers.get(server_name)
-        scopes = set(_core._server_tool_scopes.get(server_name, ()))
-        tool_names = list(getattr(server, "_registered_tool_names", ()) if server is not None else ())
+        owned = [(server, set(_core._server_tool_scopes.get(k, ())))
+                 for k, server in _core._servers.items() if _key_name(k) == server_name]
     if any(
         registry.snapshot_registration(tool_name, scope=scope) is not None
-        for scope in scopes for tool_name in tool_names
+        for server, scopes in owned for scope in scopes
+        for tool_name in getattr(server, "_registered_tool_names", ())
     ):
         registry.register_toolset_alias(server_name, f"mcp-{server_name}")
 
 
-def _remove_server_scope(server_name: str, scope: str) -> None:
+def _remove_server_scope(key, scope: str) -> None:
     """Remove one profile's MCP overlay for a shared live connection."""
     from tools.registry import registry
 
+    server_name = _key_name(key)
     for tool_name in registry.get_tool_names_for_toolset(f"mcp-{server_name}"):
         registry.deregister(tool_name, scope=scope)
     with _core._lock:
-        scopes = set(_core._server_tool_scopes.get(server_name, ()))
+        scopes = set(_core._server_tool_scopes.get(key, ()))
         scopes.discard(scope)
         if scopes:
-            _core._server_tool_scopes[server_name] = scopes
+            _core._server_tool_scopes[key] = scopes
         else:
-            _core._server_tool_scopes.pop(server_name, None)
-    _restore_server_toolset_alias(server_name)
+            _core._server_tool_scopes.pop(key, None)
+    _restore_server_toolset_alias(key)
 
 
 def _select_utility_schemas(server_name: str, server: "MCPServerTask", config: dict) -> List[dict]:
@@ -153,12 +171,12 @@ def _existing_tool_names() -> List[str]:
 
         with _core._lock:
             server_names = [
-                name for name in _core._servers
-                if _core._server_visible_in_scope(name, scope)
+                _key_name(key) for key in _core._servers
+                if _core._server_visible_in_scope(key, scope)
             ]
             server_names.extend(
-                name for name in _core._lazy_server_tool_names
-                if name not in _core._servers
+                _key_name(key) for key in _core._lazy_server_tool_names
+                if key not in _core._servers and _core._server_visible_in_scope(key, scope)
             )
         return sorted({
             tool_name
@@ -171,8 +189,8 @@ def _existing_tool_names() -> List[str]:
         names.extend(server._registered_tool_names if hasattr(server, "_registered_tool_names")
                      else (_schema._convert_mcp_schema(server.name, t)["name"] for t in server._tools))
     with _core._lock:
-        names.extend(n for sname, tool_names in _core._lazy_server_tool_names.items()
-                     if sname not in _core._servers for n in tool_names)
+        names.extend(n for key, tool_names in _core._lazy_server_tool_names.items()
+                     if key not in _core._servers for n in tool_names)
     return names
 
 
@@ -288,14 +306,17 @@ def _resolve_name_collisions(name: str, candidates: List[_Candidate]) -> List[_C
 
 
 def _register_candidates(name: str, candidates: List[_Candidate], *, check_fn: Callable,
-                         scope: Callable[[], Optional[str]], lazy: bool) -> List[str]:
+                         scope: Callable[[], Optional[str]], lazy: bool, key=None) -> List[str]:
     """Register candidates under toolset ``mcp-{name}``; returns the names that landed. The
     ownership pre-check is advisory (servers connect in parallel): ``registry.register()`` is
-    the atomic gate and its verdict is re-read after every call."""
+    the atomic gate and its verdict is re-read after every call. *key* is the connection whose
+    ``_server_tool_scopes`` records the registering scope (default: this scope's own)."""
     from tools.registry import registry
     toolset_name = f"mcp-{name}"
     registered: List[str] = []
     scope_value = scope()
+    if key is None:
+        key = _server_key(name, scope_value, current=False)
     for c in candidates:
         existing_toolset = registry.get_toolset_for_tool(c.registry_name)
         if existing_toolset and existing_toolset != toolset_name:  # foreign owner: skip, preserve it
@@ -317,7 +338,7 @@ def _register_candidates(name: str, candidates: List[_Candidate], *, check_fn: C
             _track_mcp_tool_server(c.registry_name, name)
             if scope_value is not None:
                 with _core._lock:
-                    _core._server_tool_scopes.setdefault(name, set()).add(scope_value)
+                    _core._server_tool_scopes.setdefault(key, set()).add(scope_value)
             registered.append(c.registry_name)
         elif not lazy:
             logger.error("MCP server '%s': registration of %s as '%s' was rejected by the registry; "
@@ -362,12 +383,13 @@ def _register_server_tools(name: str, server: "MCPServerTask", config: dict) ->
     refresh); returns the names. Toolset aliases derive from the live registry, not
     ``toolsets.TOOLSETS``; lossy normalization collisions (``read-file``/``read_file``) fail closed."""
     should_register = _make_tool_filter(name, config)
+    key = _server_key_for_task(server)
     _record_tool_trust_metadata(name, config, server._tools)
     candidates = _tool_candidates(name, server._tools, should_register, server.tool_timeout)
     candidates += _utility_candidates(name, _select_utility_schemas(name, server, config), server.tool_timeout)
     registered = _register_candidates(
         name, _resolve_name_collisions(name, candidates),
-        check_fn=_make_check_fn(name), scope=lambda: _core._server_registry_scope(name), lazy=False)
+        check_fn=_make_check_fn(name), scope=lambda: _core._server_registry_scope(key), lazy=False, key=key)
     if registered:
         _write_schema_cache(name, server, config, should_register)
     return registered
@@ -420,28 +442,34 @@ def _register_connected_into_current_scope(servers: dict) -> int:
 
     with _core._lock:
         stale = []
-        for name, scopes in _core._server_tool_scopes.items():
+        for key, scopes in _core._server_tool_scopes.items():
             if scope not in scopes:
                 continue
-            server = _core._servers.get(name)
-            config = servers.get(name)
+            server = _core._servers.get(key)
+            config = servers.get(_key_name(key))
             if (config is None or not _server_enabled(config) or server is None
                     or getattr(server, "session", None) is None or not _same_server_route(server, config)):
-                stale.append(name)
-    for name in stale:
-        _remove_server_scope(name, scope)
+                stale.append(key)
+    for key in stale:
+        _remove_server_scope(key, scope)
 
     registered_servers = 0
     for name, config in servers.items():
         if not _server_enabled(config):
             continue
         with _core._lock:
-            server = _core._servers.get(name)
-        if server is None or getattr(server, "session", None) is None or not _same_server_route(server, config):
+            if _server_key(name, scope, current=False) in _core._servers:
+                continue  # this profile has its own connection for the name
+            # Any other profile's live connection with the same route AND credentials is shareable.
+            shared = [(key, live) for key, live in _core._servers.items()
+                      if _key_name(key) == name and getattr(live, "session", None) is not None
+                      and _same_server_route(live, config)]
+        if not shared:
             continue
+        key, server = shared[0]
         # Visibility for this profile: the owner keeps teardown, this scope sees the connection.
         with _core._lock:
-            _core._server_tool_scopes.setdefault(name, set()).add(scope)
+            _core._server_tool_scopes.setdefault(key, set()).add(scope)
         if registry.get_tool_names_for_toolset(f"mcp-{name}"):
             continue
         candidates = _tool_candidates(name, server._tools, _make_tool_filter(name, config), server.tool_timeout)
@@ -449,7 +477,7 @@ def _register_connected_into_current_scope(servers: dict) -> int:
             name, _select_utility_schemas(name, server, config), server.tool_timeout)
         names = _register_candidates(
             name, _resolve_name_collisions(name, candidates),
-            check_fn=_make_check_fn(name), scope=lambda: scope, lazy=False)
+            check_fn=_make_check_fn(name), scope=lambda: scope, lazy=False, key=key)
         if names:
             registered_servers += 1
             with _core._lock:
@@ -476,8 +504,9 @@ def _register_from_cache_sync(name: str, config: dict, entry: dict) -> List[str]
         name, candidates, check_fn=_make_check_fn(name), scope=_core._mcp_registry_scope, lazy=True)
     if registered:
         with _core._lock:
-            _core._lazy_server_configs[name] = dict(config)
-            _core._lazy_server_fingerprints[name] = config_fingerprint(config)
-            _core._lazy_server_tool_names[name] = list(registered)
+            key = _server_key(name)
+            _core._lazy_server_configs[key] = dict(config)
+            _core._lazy_server_fingerprints[key] = config_fingerprint(config)
+            _core._lazy_server_tool_names[key] = list(registered)
         logger.info("MCP server '%s' (lazy): registered %d tool(s) from schema cache", name, len(registered))
     return registered

tools/mcp_tool_scope.py

@@ -0,0 +1,63 @@
+"""Connection-ledger keys for tools.mcp_tool under a profile multiplexer.
+
+Every ledger in ``tools.mcp_tool`` (``_servers``, connecting/error/cooldown maps, circuit
+breaker, lazy configs, trust metadata) is keyed by a *connection key*: the bare server name
+outside a multiplexer (single-profile processes are unchanged, byte for byte), and
+``(owner_scope, name)`` under one. Two profiles that both configure ``github`` with their own
+token are two connections; keying by name alone let the first profile's connection shadow the
+second's forever — its ``register_mcp_servers`` saw the name as "already connected", adopted
+nothing (different credentials) and left the profile silently tool-less (#106005, #91654).
+
+A profile may still *adopt* another profile's live connection when the route and credentials
+match (``mcp_tool_registration._same_server_route``); ``_server_tool_scopes[key]`` records every
+scope that has done so, and ``_resolve_server_key`` finds that shared connection for a caller
+whose own scope has none.
+"""
+
+from __future__ import annotations
+
+from typing import Optional, Tuple, Union
+
+from tools.mcp_tool_common import _core
+
+ServerKey = Union[str, Tuple[str, str]]
+
+
+def _server_key(name: str, scope: Optional[str] = None, *, current: bool = True) -> ServerKey:
+    """Connection key for *name* owned by *scope* (the current registry scope when *current*).
+    ``None`` scope (no multiplexer) keeps the bare name."""
+    if scope is None and current:
+        scope = _core._mcp_registry_scope()
+    return name if scope is None else (scope, name)
+
+
+def _key_name(key: ServerKey) -> str:
+    return key[1] if isinstance(key, tuple) else key
+
+
+def _key_scope(key: ServerKey) -> Optional[str]:
+    """Owning registry scope encoded in *key* (None for a bare, unscoped key)."""
+    return key[0] if isinstance(key, tuple) else None
+
+
+def _key_visible_in_scope(key: ServerKey, scope: Optional[str]) -> bool:
+    """Whether the connection under *key* serves *scope*: owned by it or adopted into it.
+    Caller holds ``_core._lock`` or tolerates a racy read (status surfaces)."""
+    if scope is None:
+        return True
+    return _key_scope(key) == scope or scope in _core._server_tool_scopes.get(key, ())
+
+
+def _resolve_server_key(name: str, scope: Optional[str] = None, *, current: bool = True) -> ServerKey:
+    """The connection key a call to *name* from *scope* must use: the scope's own connection
+    (live, connecting or lazily registered) when it has one, else a shared connection it
+    adopted, else its own (not yet existing) key so bookkeeping lands under this scope."""
+    if scope is None and current:
+        scope = _core._mcp_registry_scope()
+    own = _server_key(name, scope, current=False)
+    if scope is None or own in _core._servers or own in _core._lazy_server_configs:
+        return own
+    for key, scopes in _core._server_tool_scopes.items():
+        if scope in scopes and _key_name(key) == name and key in _core._servers:
+            return key
+    return own

tools/mcp_tool_server_run.py

@@ -414,5 +414,5 @@ class MCPServerRunMixin:
         """Drop this server's tools from the registry (idempotent); on shutdown AND budget
         exhaustion, so a dead server never leaves phantom tools in the prompt."""
         for tool_name in list(getattr(self, "_registered_tool_names", [])):
-            _registration._deregister_mcp_tool_all_scopes(self.name, tool_name)
+            _registration._deregister_mcp_tool_all_scopes(self, tool_name)
         self._registered_tool_names = []

tools/mcp_tool_transport.py

@@ -449,10 +449,11 @@ class MCPServerTransportMixin:
         if self._registered_tool_names:
             return
         with _core._lock:
-            owned = _core._servers.get(self.name) is self
+            owned = [key for key, live in _core._servers.items() if live is self]
         if not owned and not self._ready.is_set():
             return
         self._registered_tool_names = _registration._register_server_tools(self.name, self, self._config)
         with _core._lock:  # a retained initial-failure server that just published tools has recovered
-            if _core._servers.get(self.name) is self:
-                _core._server_connect_errors.pop(self.name, None)
+            for key in owned:
+                if _core._servers.get(key) is self:
+                    _core._server_connect_errors.pop(key, None)

toolsets.py

@@ -325,9 +325,11 @@ def bundle_non_core_tools(toolset_name: str) -> Set[str]:
     return to_remove - core
 
 
-# Memo keyed on (name, include_registry, id(registry), registry generation);
-# engages only at the public entry (visited is None).
-_resolve_toolset_memo: Dict[Tuple[str, bool, int, int], List[str]] = {}
+# Memo keyed on (name, include_registry, id(registry), registry generation, profile scope);
+# engages only at the public entry (visited is None). The scope is part of the key because a
+# multiplexed process resolves ``mcp-<server>`` per profile overlay: without it profile B got
+# profile A's tool names for a server B never connected (#106005).
+_resolve_toolset_memo: Dict[Tuple[str, bool, int, int, str], List[str]] = {}
 
 
 def _plugin_platform_bundle(name: str) -> List[str]:
@@ -361,7 +363,7 @@ def resolve_toolset(name: str, visited: Set[str] = None, *, include_registry: bo
     """
     external_call = visited is None
     if external_call:
-        memo_key = (name, include_registry, *_registry_generation())
+        memo_key = (name, include_registry, *_registry_generation(), _registry_call("current_scope_key", ""))
         cached = _resolve_toolset_memo.get(memo_key)
         if cached is not None:
             return list(cached)

website/docs/user-guide/multi-profile-gateways.md

@@ -237,7 +237,11 @@ into a shared environment. Subprocesses like MCP servers and Kanban workers only
 ever see their own profile's secrets — including credentials injected by an
 external secret source (1Password, Bitwarden, …): a stdio MCP server started for
 profile B receives B's value for such a name, or nothing if B has none, never the
-default profile's. Terminal settings
+default profile's. MCP servers are connected **per profile**: two profiles that
+both name a server `github` with their own token get two connections and each
+sees only its own tools; profiles whose `mcp_servers` entry is identical (same
+route *and* credentials) share one connection, and an owner's `/reload-mcp`
+re-registers the sharing profiles' tools without them reloading. Terminal settings
 (`terminal.backend`, `terminal.cwd`, `terminal.docker_volumes`,
 `terminal.docker_shared_container_key`, SSH targets, …) are likewise resolved
 per profile on every routed turn: a profile that omits a terminal key gets the