Make load and remove MCP tools match their command counterparts

Author: Your Name

cecli/tools/load_mcp.py

@@ -1,5 +1,9 @@
 from typing import List
 
+from cecli.commands.utils.helpers import (
+    iter_all_coders,
+    update_server_registration,
+)
 from cecli.tools.utils.base_tool import BaseTool
 
 
@@ -35,7 +39,7 @@ async def execute(cls, coder, servers: List[str]):
 
         if servers == ["*"]:
             for server in coder.mcp_manager.servers:
-                if server.name in coder.mcp_manager.connected_servers:
+                if server.name in {s.name for s in coder.mcp_manager.connected_servers}:
                     results.append(f"Server already loaded: {server.name}")
                     continue
                 auto_connect = server.config.get("enabled", True)
@@ -54,10 +58,24 @@ async def execute(cls, coder, servers: List[str]):
         if not servers_to_load and results:
             return "\n".join(results)
 
+        # Before connecting any new server, convert coders with empty included sets
+        # to explicit include lists of all currently connected MCP servers.
+        # This moves them from "implicitly include all" to explicit state-machine
+        # management, preventing the new server from being implicitly available
+        # to all coders.
+        connected_names = {s.name for s in coder.mcp_manager.connected_servers}
+        if connected_names:
+            for c in iter_all_coders(coder):
+                if not c.registered_servers["included"]:
+                    included = set(connected_names) - c.registered_servers["excluded"]
+                    if c.edit_format in ("agent", "subagent"):
+                        included.add("Local")  # "local" is always available
+                    c.registered_servers["included"] = included
+
         # Process the loading
         for server in servers_to_load:
             server_name = server.name
-            if server_name in coder.mcp_manager.connected_servers:
+            if server_name in {s.name for s in coder.mcp_manager.connected_servers}:
                 results.append(f"Server already loaded: {server_name}")
                 continue
 
@@ -71,6 +89,15 @@ async def execute(cls, coder, servers: List[str]):
                 results.append(f"Interrupted: {server_name}")
                 continue
             if did_connect:
+                # Force-include on the primary (active) coder
+                update_server_registration(coder, server_name, "include", force=True)
+
+                # Safe-exclude on all other coders (respects existing inclusions)
+                for other_coder in iter_all_coders(coder):
+                    if other_coder is coder:
+                        continue
+                    update_server_registration(other_coder, server_name, "exclude", force=False)
+
                 results.append(f"Loaded server: {server_name}")
             else:
                 results.append(f"Unable to load server: {server_name}")

cecli/tools/remove_mcp.py

@@ -1,5 +1,9 @@
 from typing import List
 
+from cecli.commands.utils.helpers import (
+    is_server_globally_excluded,
+    update_server_registration,
+)
 from cecli.tools.utils.base_tool import BaseTool
 
 
@@ -39,13 +43,13 @@ async def execute(cls, coder, servers: List[str]):
 
         # Determine which servers to act on
         if servers == ["*"]:
-            servers_to_action.extend(coder.mcp_manager.connected_servers.keys())
+            servers_to_action.extend(s.name for s in coder.mcp_manager.connected_servers)
         else:
             for server_name in servers:
                 server = coder.mcp_manager.get_server(server_name)
                 if not server:
                     results.append(f"MCP server {server_name} does not exist.")
-                elif server.name not in coder.mcp_manager.connected_servers:
+                elif server.name not in {s.name for s in coder.mcp_manager.connected_servers}:
                     results.append(f"Server {server_name} is not currently connected.")
                 else:
                     servers_to_action.append(server.name)
@@ -58,20 +62,34 @@ async def execute(cls, coder, servers: List[str]):
         if not servers_to_action:
             return "No servers to remove."
 
-        # Process the removal
+        # Process the removal with server registration awareness
         for server_name in servers_to_action:
-            coder.interrupt_event.clear()
-            did_disconnect, interrupted = await coder.coroutines.interruptible(
-                coder.mcp_manager.disconnect_server(server_name),
-                coder.interrupt_event,
-            )
-
-            if interrupted:
-                results.append(f"Interrupted: {server_name}")
+            # Never remove the "local" server
+            if server_name == "Local":
+                results.append("Cannot remove 'Local' server")
                 continue
-            if did_disconnect:
-                results.append(f"Removed server: {server_name}")
+
+            # Force-exclude on the primary (active) coder
+            update_server_registration(coder, server_name, "exclude", force=True)
+
+            # Check if all coders in the hierarchy have this server excluded
+            all_excluded = is_server_globally_excluded(coder, server_name)
+
+            if all_excluded:
+                coder.interrupt_event.clear()
+                did_disconnect, interrupted = await coder.coroutines.interruptible(
+                    coder.mcp_manager.disconnect_server(server_name),
+                    coder.interrupt_event,
+                )
+
+                if interrupted:
+                    results.append(f"Interrupted: {server_name}")
+                    continue
+                if did_disconnect:
+                    results.append(f"Removed server: {server_name}")
+                else:
+                    results.append(f"Unable to remove server: {server_name}")
             else:
-                results.append(f"Unable to remove server: {server_name}")
+                results.append(f"Removed from active coder, still active for others: {server_name}")
 
         return "\n".join(results)