diff --git a/pentestagent/interface/tui.py b/pentestagent/interface/tui.py index 8d595b8..0a9da76 100644 --- a/pentestagent/interface/tui.py +++ b/pentestagent/interface/tui.py @@ -533,6 +533,8 @@ class MCPScreen(ModalScreen): connected_icon = "🟢" if mcp.connected else "🔴" text.append(f"Connected: {connected_icon}\n", style="#9a9a9a") + if mcp.last_error: + text.append(f"Last error message: {mcp.last_error}\n", style="#e91b1b") elif tool is not None: text.append(f"{tool['description']}\n", style="#d4d4d4") else: diff --git a/pentestagent/mcp/manager.py b/pentestagent/mcp/manager.py index 210863a..7dd3674 100644 --- a/pentestagent/mcp/manager.py +++ b/pentestagent/mcp/manager.py @@ -41,19 +41,31 @@ class MCPServer: name: str config: MCPServerConfig - transport: MCPTransport + transport: Optional[MCPTransport] tools: List[dict] = field(default_factory=list) connected: bool = False # Lock for serializing all communication with this server # Prevents message ID collisions and transport interleaving _lock: asyncio.Lock = field(default_factory=asyncio.Lock) + last_error: str = "" #If something went wrong, get the last error message. + async def disconnect(self): """Disconnect from the server.""" if self.connected: - await self.transport.disconnect() + if self.transport: + await self.transport.disconnect() self.connected = False + def is_enabled(self) -> bool: + return self.config.enabled + + def enable(self): + self.config.enabled = True + + def disable(self): + self.config.enabled = False + class MCPManager: """Manages MCP server connections and exposes tools to agents.""" @@ -236,7 +248,14 @@ class MCPManager: except Exception: pass print(f"[MCP] Failed to connect to {config.name}: {e}") - return None + return MCPServer( + name=config.name, + config=config, + transport=None, + tools=[], + connected=False, + last_error=str(e) + ) async def call_tool(self, server_name: str, tool_name: str, arguments: dict) -> Any: server = self.servers.get(server_name)