Fix ultra simple server to be completely self-contained with zero external imports

src/dolibarr_mcp/ultra_simple_server.py

@@ -1,4 +1,4 @@
-"""Ultra-simple Dolibarr server - Zero compiled extensions, maximum Windows compatibility."""
+"""Ultra-simple Dolibarr server - COMPLETELY SELF-CONTAINED - Zero external dependencies issues."""
 
 import json
 import sys
@@ -6,21 +6,277 @@ import os
 import logging
 from typing import Any, Dict, List, Optional
 
-# Add src to path for imports
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
+# EVERYTHING is self-contained in this single file to avoid import issues
 
-# Import our ultra-simple components (no compiled extensions)
-from src.dolibarr_mcp.simple_client import SimpleDolibarrClient, SimpleDolibarrAPIError, SimpleConfig
+# ============================================================================
+# SELF-CONTAINED CONFIGURATION CLASS
+# ============================================================================
 
-# Configure logging
-logging.basicConfig(
-    level=logging.WARNING,
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-    handlers=[logging.StreamHandler(sys.stderr)]
-)
+class UltraSimpleConfig:
+    """Ultra-simple configuration - completely self-contained."""
+    
+    def __init__(self):
+        # Load .env manually
+        self.load_env()
+        
+        self.dolibarr_url = os.getenv("DOLIBARR_URL", "")
+        self.api_key = os.getenv("DOLIBARR_API_KEY", "")
+        self.log_level = os.getenv("LOG_LEVEL", "INFO")
+        
+        # Validate and fix URL
+        if not self.dolibarr_url or "your-dolibarr-instance" in self.dolibarr_url:
+            print("⚠️  DOLIBARR_URL not configured in .env file", file=sys.stderr)
+            self.dolibarr_url = "https://your-dolibarr-instance.com/api/index.php"
+        
+        if not self.api_key or "your_dolibarr_api_key" in self.api_key:
+            print("⚠️  DOLIBARR_API_KEY not configured in .env file", file=sys.stderr)
+            self.api_key = "placeholder_api_key"
+        
+        # Ensure URL format
+        if self.dolibarr_url and not self.dolibarr_url.endswith('/api/index.php'):
+            if '/api' not in self.dolibarr_url:
+                self.dolibarr_url = self.dolibarr_url.rstrip('/') + '/api/index.php'
+            elif not self.dolibarr_url.endswith('/index.php'):
+                self.dolibarr_url = self.dolibarr_url.rstrip('/') + '/index.php'
+    
+    def load_env(self):
+        """Load .env file manually - no python-dotenv needed."""
+        env_file = '.env'
+        if os.path.exists(env_file):
+            with open(env_file, 'r', encoding='utf-8') as f:
+                for line in f:
+                    line = line.strip()
+                    if line and not line.startswith('#') and '=' in line:
+                        key, value = line.split('=', 1)
+                        os.environ[key.strip()] = value.strip()
+
+# ============================================================================
+# SELF-CONTAINED API CLIENT
+# ============================================================================
+
+class UltraSimpleAPIError(Exception):
+    """Simple API error exception."""
+    def __init__(self, message: str, status_code: Optional[int] = None):
+        self.message = message
+        self.status_code = status_code
+        super().__init__(self.message)
+
+class UltraSimpleAPIClient:
+    """Ultra-simple Dolibarr client - completely self-contained."""
+    
+    def __init__(self, config: UltraSimpleConfig):
+        self.config = config
+        self.base_url = config.dolibarr_url.rstrip('/')
+        self.api_key = config.api_key
+        self.logger = logging.getLogger(__name__)
+        
+        # We'll use requests when needed, imported inside methods
+        
+    def _build_url(self, endpoint: str) -> str:
+        """Build full API URL."""
+        endpoint = endpoint.lstrip('/')
+        
+        # Special handling for status endpoint
+        if endpoint == "status":
+            base = self.base_url.replace('/index.php', '')
+            return f"{base}/status"
+        
+        return f"{self.base_url}/{endpoint}"
+    
+    def _make_request(
+        self, 
+        method: str, 
+        endpoint: str, 
+        params: Optional[Dict] = None,
+        data: Optional[Dict] = None
+    ) -> Dict[str, Any]:
+        """Make HTTP request to Dolibarr API."""
+        # Import requests here to avoid import issues
+        try:
+            import requests
+        except ImportError:
+            raise UltraSimpleAPIError("requests library not available. Please run setup_ultra.bat")
+        
+        url = self._build_url(endpoint)
+        
+        try:
+            self.logger.debug(f"Making {method} request to {url}")
+            
+            headers = {
+                "DOLAPIKEY": self.api_key,
+                "Content-Type": "application/json",
+                "Accept": "application/json",
+                "User-Agent": "Dolibarr-MCP-Ultra/1.0"
+            }
+            
+            kwargs = {
+                "params": params or {},
+                "timeout": 30,
+                "headers": headers
+            }
+            
+            if data and method.upper() in ["POST", "PUT"]:
+                kwargs["json"] = data
+            
+            response = requests.request(method, url, **kwargs)
+            
+            # Handle error responses
+            if response.status_code >= 400:
+                try:
+                    error_data = response.json()
+                    if isinstance(error_data, dict) and "error" in error_data:
+                        error_msg = str(error_data["error"])
+                    else:
+                        error_msg = f"HTTP {response.status_code}: {response.reason}"
+                except:
+                    error_msg = f"HTTP {response.status_code}: {response.reason}"
+                
+                raise UltraSimpleAPIError(error_msg, response.status_code)
+            
+            # Try to parse JSON response
+            try:
+                return response.json()
+            except:
+                return {"raw_response": response.text}
+        
+        except requests.RequestException as e:
+            # For status endpoint, try alternative
+            if endpoint == "status":
+                try:
+                    alt_url = f"{self.base_url.replace('/api/index.php', '')}/setup/modules"
+                    alt_response = requests.get(alt_url, headers=headers, timeout=10)
+                    if alt_response.status_code == 200:
+                        return {
+                            "success": 1,
+                            "dolibarr_version": "API Available",
+                            "api_version": "1.0"
+                        }
+                except:
+                    pass
+            
+            raise UltraSimpleAPIError(f"HTTP request failed: {str(e)}")
+        except Exception as e:
+            raise UltraSimpleAPIError(f"Unexpected error: {str(e)}")
+    
+    # API Methods
+    def get_status(self) -> Dict[str, Any]:
+        """Get API status."""
+        try:
+            return self._make_request("GET", "status")
+        except UltraSimpleAPIError:
+            try:
+                result = self._make_request("GET", "users?limit=1")
+                if result is not None:
+                    return {
+                        "success": 1,
+                        "dolibarr_version": "API Working",
+                        "api_version": "1.0"
+                    }
+            except:
+                pass
+            raise UltraSimpleAPIError("Cannot connect to Dolibarr API. Please check your configuration.")
+    
+    def get_users(self, limit: int = 100, page: int = 1) -> List[Dict[str, Any]]:
+        """Get list of users."""
+        params = {"limit": limit}
+        if page > 1:
+            params["page"] = page
+        result = self._make_request("GET", "users", params=params)
+        return result if isinstance(result, list) else []
+    
+    def get_user_by_id(self, user_id: int) -> Dict[str, Any]:
+        """Get specific user by ID."""
+        return self._make_request("GET", f"users/{user_id}")
+    
+    def create_user(self, **kwargs) -> Dict[str, Any]:
+        """Create a new user."""
+        return self._make_request("POST", "users", data=kwargs)
+    
+    def update_user(self, user_id: int, **kwargs) -> Dict[str, Any]:
+        """Update an existing user."""
+        return self._make_request("PUT", f"users/{user_id}", data=kwargs)
+    
+    def delete_user(self, user_id: int) -> Dict[str, Any]:
+        """Delete a user."""
+        return self._make_request("DELETE", f"users/{user_id}")
+    
+    def get_customers(self, limit: int = 100, page: int = 1) -> List[Dict[str, Any]]:
+        """Get list of customers."""
+        params = {"limit": limit}
+        if page > 1:
+            params["page"] = page
+        result = self._make_request("GET", "thirdparties", params=params)
+        return result if isinstance(result, list) else []
+    
+    def get_customer_by_id(self, customer_id: int) -> Dict[str, Any]:
+        """Get specific customer by ID."""
+        return self._make_request("GET", f"thirdparties/{customer_id}")
+    
+    def create_customer(self, name: str, **kwargs) -> Dict[str, Any]:
+        """Create a new customer."""
+        data = {
+            "name": name,
+            "status": kwargs.get("status", 1),
+            "client": 1 if kwargs.get("type", 1) in [1, 3] else 0,
+            "fournisseur": 1 if kwargs.get("type", 1) in [2, 3] else 0,
+            "country_id": kwargs.get("country_id", 1),
+        }
+        for field in ["email", "phone", "address", "town", "zip"]:
+            if field in kwargs:
+                data[field] = kwargs[field]
+        return self._make_request("POST", "thirdparties", data=data)
+    
+    def update_customer(self, customer_id: int, **kwargs) -> Dict[str, Any]:
+        """Update an existing customer."""
+        return self._make_request("PUT", f"thirdparties/{customer_id}", data=kwargs)
+    
+    def delete_customer(self, customer_id: int) -> Dict[str, Any]:
+        """Delete a customer."""
+        return self._make_request("DELETE", f"thirdparties/{customer_id}")
+    
+    def get_products(self, limit: int = 100) -> List[Dict[str, Any]]:
+        """Get list of products."""
+        params = {"limit": limit}
+        result = self._make_request("GET", "products", params=params)
+        return result if isinstance(result, list) else []
+    
+    def get_product_by_id(self, product_id: int) -> Dict[str, Any]:
+        """Get specific product by ID."""
+        return self._make_request("GET", f"products/{product_id}")
+    
+    def create_product(self, label: str, price: float, **kwargs) -> Dict[str, Any]:
+        """Create a new product."""
+        import time
+        ref = kwargs.get("ref", f"PROD-{int(time.time())}")
+        data = {
+            "ref": ref,
+            "label": label,
+            "price": price,
+            "price_ttc": price,
+        }
+        for field in ["description", "stock"]:
+            if field in kwargs:
+                data[field] = kwargs[field]
+        return self._make_request("POST", "products", data=data)
+    
+    def update_product(self, product_id: int, **kwargs) -> Dict[str, Any]:
+        """Update an existing product."""
+        return self._make_request("PUT", f"products/{product_id}", data=kwargs)
+    
+    def delete_product(self, product_id: int) -> Dict[str, Any]:
+        """Delete a product."""
+        return self._make_request("DELETE", f"products/{product_id}")
+    
+    def raw_api(self, method: str, endpoint: str, params: Optional[Dict] = None, data: Optional[Dict] = None) -> Dict[str, Any]:
+        """Make raw API call."""
+        return self._make_request(method, endpoint, params=params, data=data)
+
+# ============================================================================
+# ULTRA-SIMPLE SERVER
+# ============================================================================
 
 class UltraSimpleServer:
-    """Ultra-simple server implementation - pure Python, zero compiled extensions."""
+    """Ultra-simple server - completely self-contained."""
     
     def __init__(self, name: str = "dolibarr-mcp-ultra"):
         self.name = name
@@ -30,132 +286,94 @@ class UltraSimpleServer:
     def init_client(self):
         """Initialize the Dolibarr client."""
         if not self.client:
-            config = SimpleConfig()
-            self.client = SimpleDolibarrClient(config)
+            config = UltraSimpleConfig()
+            self.client = UltraSimpleAPIClient(config)
     
     def get_available_tools(self) -> List[str]:
         """Get list of available tool names."""
         return [
-            "test_connection",
-            "get_status", 
-            "get_users",
-            "get_user_by_id",
-            "create_user",
-            "update_user", 
-            "delete_user",
-            "get_customers",
-            "get_customer_by_id",
-            "create_customer",
-            "update_customer",
-            "delete_customer",
-            "get_products",
-            "get_product_by_id", 
-            "create_product",
-            "update_product",
-            "delete_product",
-            "raw_api"
+            "test_connection", "get_status", "get_users", "get_user_by_id",
+            "create_user", "update_user", "delete_user", "get_customers",
+            "get_customer_by_id", "create_customer", "update_customer",
+            "delete_customer", "get_products", "get_product_by_id",
+            "create_product", "update_product", "delete_product", "raw_api"
         ]
     
     def handle_tool_call(self, tool_name: str, arguments: dict) -> Dict[str, Any]:
-        """Handle tool calls using the simple client."""
-        
+        """Handle tool calls."""
         try:
             self.init_client()
             
-            # System tools
             if tool_name == "test_connection":
                 result = self.client.get_status()
                 if 'success' not in result:
                     result = {"status": "success", "message": "API connection working", "data": result}
                 return {"success": True, "data": result}
-            
             elif tool_name == "get_status":
                 result = self.client.get_status()
                 return {"success": True, "data": result}
-            
-            # User management
             elif tool_name == "get_users":
                 result = self.client.get_users(
                     limit=arguments.get('limit', 100),
                     page=arguments.get('page', 1)
                 )
                 return {"success": True, "data": result}
-            
             elif tool_name == "get_user_by_id":
                 result = self.client.get_user_by_id(arguments['user_id'])
                 return {"success": True, "data": result}
-            
             elif tool_name == "create_user":
                 result = self.client.create_user(**arguments)
                 return {"success": True, "data": result}
-            
             elif tool_name == "update_user":
                 user_id = arguments.pop('user_id')
                 result = self.client.update_user(user_id, **arguments)
                 return {"success": True, "data": result}
-            
             elif tool_name == "delete_user":
                 result = self.client.delete_user(arguments['user_id'])
                 return {"success": True, "data": result}
-            
-            # Customer management
             elif tool_name == "get_customers":
                 result = self.client.get_customers(
                     limit=arguments.get('limit', 100),
                     page=arguments.get('page', 1)
                 )
                 return {"success": True, "data": result}
-            
             elif tool_name == "get_customer_by_id":
                 result = self.client.get_customer_by_id(arguments['customer_id'])
                 return {"success": True, "data": result}
-            
             elif tool_name == "create_customer":
                 result = self.client.create_customer(**arguments)
                 return {"success": True, "data": result}
-            
             elif tool_name == "update_customer":
                 customer_id = arguments.pop('customer_id')
                 result = self.client.update_customer(customer_id, **arguments)
                 return {"success": True, "data": result}
-            
             elif tool_name == "delete_customer":
                 result = self.client.delete_customer(arguments['customer_id'])
                 return {"success": True, "data": result}
-            
-            # Product management
             elif tool_name == "get_products":
                 result = self.client.get_products(limit=arguments.get('limit', 100))
                 return {"success": True, "data": result}
-            
             elif tool_name == "get_product_by_id":
                 result = self.client.get_product_by_id(arguments['product_id'])
                 return {"success": True, "data": result}
-            
             elif tool_name == "create_product":
                 result = self.client.create_product(**arguments)
                 return {"success": True, "data": result}
-            
             elif tool_name == "update_product":
                 product_id = arguments.pop('product_id')
                 result = self.client.update_product(product_id, **arguments)
                 return {"success": True, "data": result}
-            
             elif tool_name == "delete_product":
                 result = self.client.delete_product(arguments['product_id'])
                 return {"success": True, "data": result}
-            
-            # Raw API access
             elif tool_name == "raw_api":
                 result = self.client.raw_api(**arguments)
                 return {"success": True, "data": result}
-            
             else:
                 return {"error": f"Unknown tool: {tool_name}", "type": "unknown_tool"}
         
-        except SimpleDolibarrAPIError as e:
+        except UltraSimpleAPIError as e:
             return {"error": f"Dolibarr API Error: {str(e)}", "type": "api_error"}
-        
         except Exception as e:
             self.logger.error(f"Tool execution error: {e}")
             return {"error": f"Tool execution failed: {str(e)}", "type": "internal_error"}
@@ -165,15 +383,15 @@ class UltraSimpleServer:
         return json.dumps(content, indent=2, ensure_ascii=False)
     
     def run_interactive(self):
-        """Run server in interactive mode for testing."""
+        """Run server in interactive mode."""
         print("🚀 Ultra-Simple Dolibarr MCP Server (Maximum Windows Compatibility)", file=sys.stderr)
         print("✅ ZERO compiled extensions - NO .pyd files!", file=sys.stderr)
-        print("✅ Only pure Python + requests library", file=sys.stderr)
+        print("✅ Completely self-contained - no import issues!", file=sys.stderr)
         print("", file=sys.stderr)
         
         # Test configuration
         try:
-            config = SimpleConfig()
+            config = UltraSimpleConfig()
             
             if "your-dolibarr-instance" in config.dolibarr_url:
                 print("⚠️  DOLIBARR_URL not configured in .env file", file=sys.stderr)
@@ -212,12 +430,10 @@ class UltraSimpleServer:
                 
                 if command == "exit":
                     break
-                    
                 elif command == "list":
                     print("Available tools:")
                     for i, tool in enumerate(tools, 1):
                         print(f"  {i:2}. {tool}")
-                
                 elif command == "help":
                     print("Commands:")
                     print("  list                    - Show all available tools")
@@ -231,18 +447,15 @@ class UltraSimpleServer:
                     print("  test get_users          - Get first 5 users")
                     print("  test get_customers      - Get first 5 customers")
                     print("  test get_products       - Get first 5 products")
-                
                 elif command == "config":
-                    config = SimpleConfig()
+                    config = UltraSimpleConfig()
                     print(f"Configuration:")
                     print(f"  URL: {config.dolibarr_url}")
                     print(f"  API Key: {'*' * min(len(config.api_key), 10)}...")
                     print(f"  Log Level: {config.log_level}")
-                    
                 elif command.startswith("test "):
                     tool_name = command[5:].strip()
                     
-                    # Quick test implementations
                     if tool_name == "test_connection":
                         result = self.handle_tool_call("test_connection", {})
                     elif tool_name == "get_status":
@@ -262,7 +475,6 @@ class UltraSimpleServer:
                         continue
                     
                     print(self.format_response(result))
-                
                 elif command:
                     print("Unknown command. Type 'help' for available commands.")
                     
@@ -271,19 +483,13 @@ class UltraSimpleServer:
             except Exception as e:
                 print(f"Error: {e}")
         
-        # Cleanup
-        if self.client:
-            self.client.close()
-        
         print("\n👋 Goodbye!")
 
-
 def main():
     """Main entry point."""
     server = UltraSimpleServer("dolibarr-mcp-ultra")
     server.run_interactive()
 
-
 if __name__ == "__main__":
     try:
         main()