CLI-41: Allow agents MCP management + HashPos Base64 + TUI reliability improvements

cecli/coders/agent_coder.py

@@ -32,7 +32,7 @@
 
 from .base_coder import Coder
 
-from cecli.helpers.coroutines import interruptible  # isort:skip
+from cecli.helpers import coroutines  # isort:skip
 
 logger = logging.getLogger(__name__)
 
@@ -110,7 +110,7 @@ def __init__(self, *args, **kwargs):
 
     def post_init(self):
         super().post_init()
-
+        self.coroutines = coroutines
         if not self._inherited_tools:
             # Populate per-instance tool and server filters from config
             self.registered_tools["included"] = set(
@@ -325,7 +325,9 @@ async def _exec_async():
                 self.io.tool_warning(f"Executing {tool_name} on {server.name} failed:\nError: {e}")
                 return f"Error executing tool call {tool_name}: {e}"
 
-        result, interrupted = await interruptible(_exec_async(), self.interrupt_event)
+        result, interrupted = await self.coroutines.interruptible(
+            _exec_async(), self.interrupt_event
+        )
 
         if interrupted:
             return "Tool execution interrupted by user."
@@ -625,7 +627,8 @@ def get_context_summary(self):
                 if percentage > 80:
                     result += "\n\n⚠ **Context is getting full!**\n"
                     result += "- Remove non-essential files via the `ContextManager` tool.\n"
-                    result += "- Keep only essential files in context for best performance"
+                    result += "- Remove unused MCP servers via the `RemoveMcp` tool to free context space.\n"
+                    result += "- Keep only essential files and MCP servers in context for best performance"
             result += "\n</context>"
             if not hasattr(self, "context_blocks_cache"):
                 self.context_blocks_cache = {}
@@ -659,7 +662,9 @@ def get_environment_info(self):
                     result += f"- Git repository: {rel_repo_dir} with {num_files:,} files\n"
                 except Exception:
                     result += "- Git repository: active but details unavailable\n"
-            else:
+                if self.mcp_manager and self.mcp_manager.connected_servers:
+                    num_mcp_servers = len(self.mcp_manager.connected_servers)
+                    result += f"- Connected MCP servers: {num_mcp_servers}\n"
                 result += "- Git repository: none\n"
             result += "</context>"
             return result
@@ -766,7 +771,7 @@ async def _execute_local_tools(self, tool_calls_list):
                     async def gather_and_await():
                         return await asyncio.gather(*tasks, return_exceptions=True)
 
-                    task_results, interrupted = await interruptible(
+                    task_results, interrupted = await self.coroutines.interruptible(
                         gather_and_await(), self.interrupt_event
                     )
 

cecli/coders/base_coder.py

@@ -423,9 +423,11 @@ def __init__(
         registered_servers=None,
         uuid: str = "",
         parent_uuid: str = "",
+        **kwargs,
     ):
         from cecli.helpers.agents.service import AgentService
 
+        self.original_kwargs = kwargs
         # initialize from args.map_cache_dir
         self.coroutines = coroutines
         # Per-instance tool and server filtering dictionaries

cecli/commands/__init__.py

@@ -35,6 +35,7 @@
 from .hot_reload import HotReloadCommand
 from .include_skill import IncludeSkillCommand
 from .lint import LintCommand
+from .list_mcp import ListMcpCommand
 from .list_sessions import ListSessionsCommand
 from .list_skills import ListSkillsCommand
 from .load import LoadCommand
@@ -123,6 +124,7 @@
 CommandRegistry.register(SwitchAgentCommand)
 CommandRegistry.register(IncludeSkillCommand)
 CommandRegistry.register(LintCommand)
+CommandRegistry.register(ListMcpCommand)
 CommandRegistry.register(ListSessionsCommand)
 CommandRegistry.register(ListSkillsCommand)
 CommandRegistry.register(LoadCommand)
@@ -210,6 +212,7 @@
     "LoadCommand",
     "LoadHookCommand",
     "LoadMcpCommand",
+    "ListMcpCommand",
     "LoadSessionCommand",
     "LoadSkillCommand",
     "LsCommand",

cecli/commands/list_mcp.py

@@ -0,0 +1,48 @@
+from cecli.commands.utils.base_command import BaseCommand
+from cecli.commands.utils.helpers import format_command_result
+
+
+class ListMcpCommand(BaseCommand):
+    NORM_NAME = "list-mcp"
+    DESCRIPTION = "List all loaded and configured MCP servers."
+
+    @classmethod
+    async def execute(cls, io, coder, args, **kwargs):
+        """Execute the list-mcp command."""
+        if not coder.mcp_manager:
+            return format_command_result(io, cls.NORM_NAME, "MCP manager is not configured.")
+
+        all_servers = coder.mcp_manager.servers
+        connected_servers = coder.mcp_manager.connected_servers
+
+        loaded_server_names = {server.name for server in connected_servers}
+        configured_servers = [
+            server for server in all_servers if server.name not in loaded_server_names
+        ]
+
+        result = []
+        if loaded_server_names:
+            result.append("Loaded MCP Servers:")
+            for name in sorted(list(loaded_server_names)):
+                result.append(f"- {name}")
+        else:
+            result.append("No MCP servers are currently loaded.")
+
+        result.append("")
+
+        if configured_servers:
+            result.append("Configured MCP Servers:")
+            for server in sorted(configured_servers, key=lambda s: s.name):
+                result.append(f"- {server.name}")
+        else:
+            result.append("No other MCP servers are configured.")
+
+        return format_command_result(io, cls.NORM_NAME, "\n".join(result))
+
+    @classmethod
+    def get_help(cls) -> str:
+        """Get help text for the list-mcp command."""
+        help_text = super().get_help()
+        help_text += "\nUsage:\n"
+        help_text += "  /list-mcp  # Lists all loaded and configured MCP servers\n"
+        return help_text

cecli/commands/remove_mcp.py

@@ -10,7 +10,7 @@
 
 class RemoveMcpCommand(BaseCommand):
     NORM_NAME = "remove-mcp"
-    DESCRIPTION = "Remove a MCP server by name, or use '*' to remove all"
+    DESCRIPTION = "Remove (unload) a MCP server by name, or use '*' to remove all"
 
     @classmethod
     async def execute(cls, io, coder, args, **kwargs):

cecli/tools/__init__.py

@@ -17,9 +17,12 @@
     git_show,
     git_status,
     grep,
+    list_mcp,
+    load_mcp,
     load_skill,
     ls,
     read_range,
+    remove_mcp,
     remove_skill,
     thinking,
     undo_change,
@@ -42,9 +45,12 @@
     git_show,
     git_status,
     grep,
+    list_mcp,
+    load_mcp,
     load_skill,
     ls,
     read_range,
+    remove_mcp,
     remove_skill,
     thinking,
     undo_change,

cecli/tools/list_mcp.py

@@ -0,0 +1,50 @@
+from cecli.tools.utils.base_tool import BaseTool
+
+
+class Tool(BaseTool):
+    NORM_NAME = "list-mcp"
+    SCHEMA = {
+        "type": "function",
+        "function": {
+            "name": "ListMcp",
+            "description": "List all loaded and configured MCP servers.",
+            "parameters": {
+                "type": "object",
+                "properties": {},
+                "required": [],
+            },
+        },
+    }
+
+    @classmethod
+    def execute(cls, coder, **kwargs):
+        """List all loaded and configured MCP servers."""
+        if not coder.mcp_manager:
+            return "MCP manager is not configured."
+
+        all_servers = coder.mcp_manager.servers
+        connected_servers = coder.mcp_manager.connected_servers
+
+        loaded_server_names = {server.name for server in connected_servers}
+        configured_servers = [
+            server for server in all_servers if server.name not in loaded_server_names
+        ]
+
+        result = []
+        if loaded_server_names:
+            result.append("Loaded MCP Servers:")
+            for name in sorted(list(loaded_server_names)):
+                result.append(f"- {name}")
+        else:
+            result.append("No MCP servers are currently loaded.")
+
+        result.append("")
+
+        if configured_servers:
+            result.append("Configured MCP Servers:")
+            for server in sorted(configured_servers, key=lambda s: s.name):
+                result.append(f"- {server.name}")
+        else:
+            result.append("No other MCP servers are configured.")
+
+        return "\n".join(result)

cecli/tools/load_mcp.py

@@ -0,0 +1,78 @@
+from typing import List
+
+from cecli.tools.utils.base_tool import BaseTool
+
+
+class Tool(BaseTool):
+    NORM_NAME = "load-mcp"
+    SCHEMA = {
+        "type": "function",
+        "function": {
+            "name": "LoadMCP",
+            "description": "Load MCP server(s) by name, or use '*' to load all enabled servers.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "servers": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": "A list of MCP server names to load. Use '*' to load all.",
+                    }
+                },
+                "required": ["servers"],
+            },
+        },
+    }
+
+    @classmethod
+    async def execute(cls, coder, servers: List[str]):
+        """Execute the load-mcp tool with given parameters."""
+        if not coder.mcp_manager or not coder.mcp_manager.servers:
+            return "No MCP servers found, nothing to load."
+
+        results = []
+        servers_to_load = []
+
+        if servers == ["*"]:
+            for server in coder.mcp_manager.servers:
+                if server.name in coder.mcp_manager.connected_servers:
+                    results.append(f"Server already loaded: {server.name}")
+                    continue
+                auto_connect = server.config.get("enabled", True)
+                if not auto_connect:
+                    results.append(f"Skipping server (not enabled by default): {server.name}")
+                    continue
+                servers_to_load.append(server)
+        else:
+            for server_name in servers:
+                server = coder.mcp_manager.get_server(server_name)
+                if server is None:
+                    results.append(f"MCP server {server_name} does not exist.")
+                else:
+                    servers_to_load.append(server)
+
+        if not servers_to_load and results:
+            return "\n".join(results)
+
+        # Process the loading
+        for server in servers_to_load:
+            server_name = server.name
+            if server_name in coder.mcp_manager.connected_servers:
+                results.append(f"Server already loaded: {server_name}")
+                continue
+
+            coder.interrupt_event.clear()
+            did_connect, interrupted = await coder.coroutines.interruptible(
+                coder.mcp_manager.connect_server(server_name),
+                coder.interrupt_event,
+            )
+
+            if interrupted:
+                results.append(f"Interrupted: {server_name}")
+                continue
+            if did_connect:
+                results.append(f"Loaded server: {server_name}")
+            else:
+                results.append(f"Unable to load server: {server_name}")
+
+        return "\n".join(results)

cecli/tools/remove_mcp.py

@@ -0,0 +1,77 @@
+from typing import List
+
+from cecli.tools.utils.base_tool import BaseTool
+
+
+class Tool(BaseTool):
+    NORM_NAME = "remove-mcp"
+    SCHEMA = {
+        "type": "function",
+        "function": {
+            "name": "RemoveMCP",
+            "description": (
+                "Remove (unload) MCP server(s) by name, or use '*' to remove all connected servers."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "servers": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": (
+                            "A list of MCP server names to remove. Use '*' to remove all."
+                        ),
+                    }
+                },
+                "required": ["servers"],
+            },
+        },
+    }
+
+    @classmethod
+    async def execute(cls, coder, servers: List[str]):
+        """Execute the remove-mcp tool with given parameters."""
+        if not coder.mcp_manager or not coder.mcp_manager.servers:
+            return "No MCP servers are configured."
+
+        results = []
+        servers_to_action = []
+
+        # Determine which servers to act on
+        if servers == ["*"]:
+            servers_to_action.extend(coder.mcp_manager.connected_servers.keys())
+        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:
+                    results.append(f"Server {server_name} is not currently connected.")
+                else:
+                    servers_to_action.append(server.name)
+
+        # If there are no servers to act on but we have preliminary results (like errors), return them
+        if not servers_to_action and results:
+            return "\n".join(results)
+
+        # If there are no servers to remove at all
+        if not servers_to_action:
+            return "No servers to remove."
+
+        # Process the removal
+        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}")
+                continue
+            if did_disconnect:
+                results.append(f"Removed server: {server_name}")
+            else:
+                results.append(f"Unable to remove server: {server_name}")
+
+        return "\n".join(results)

cecli/tui/app.py

@@ -750,7 +750,6 @@ def update_spinner(self, msg, agent_name: str | None = None):
     def show_error(self, message, agent_name: str | None = None):
         """Show an error message in the status bar."""
         status_bar = self.query_one("#status-bar", StatusBar)
-
         status_bar.show_notification(message, severity="error", timeout=5, agent_name=agent_name)
 
     def on_resize(self) -> None: