Phase 3: Unified provider tools system for all LLM providers

tools/__init__.py

@@ -1,14 +1,33 @@
 """
 MaxClaw Agent Tools Package
-Provides custom tools for Proxmox and Ollama management
+Provides custom tools for Proxmox, Ollama, and unified provider management
 """
 
 from .proxmox_tools import get_proxmox_tools, ProxmoxTools
 from .ollama_tools import get_ollama_tools, OllamaTools
+from .provider_tools import (
+    get_provider_tools,
+    get_all_provider_tools,
+    BaseProviderTools,
+    OllamaProviderTools,
+    OpenAIProviderTools,
+    OpenRouterProviderTools
+)
 
 __all__ = [
+    # Proxmox tools
     'get_proxmox_tools',
     'ProxmoxTools',
+    
+    # Ollama tools (legacy)
     'get_ollama_tools',
-    'OllamaTools'
+    'OllamaTools',
+    
+    # Unified provider tools
+    'get_provider_tools',
+    'get_all_provider_tools',
+    'BaseProviderTools',
+    'OllamaProviderTools',
+    'OpenAIProviderTools',
+    'OpenRouterProviderTools'
 ]

tools/provider_tools.py

@@ -0,0 +1,630 @@
+"""
+Unified Provider Tools for MaxClaw Agent
+Provides a consistent interface for managing different LLM providers
+"""
+import os
+import json
+import requests
+from abc import ABC, abstractmethod
+from typing import Dict, List, Optional, Any, Tuple
+from langchain.tools import Tool
+
+
+class BaseProviderTools(ABC):
+    """Base class for provider-specific tools"""
+    
+    def __init__(self, provider_name: str, **kwargs):
+        """
+        Initialize provider tools
+        
+        Args:
+            provider_name: Name of the provider (e.g., "ollama", "openai", "openrouter")
+            **kwargs: Provider-specific configuration
+        """
+        self.provider_name = provider_name
+        self.config = kwargs
+    
+    @abstractmethod
+    def list_models(self) -> str:
+        """List available models for this provider"""
+        pass
+    
+    @abstractmethod
+    def get_model_info(self, model_name: str) -> str:
+        """Get detailed information about a specific model"""
+        pass
+    
+    @abstractmethod
+    def test_connection(self) -> str:
+        """Test connection to the provider"""
+        pass
+    
+    @abstractmethod
+    def get_default_model(self) -> str:
+        """Get the default model for this provider"""
+        pass
+    
+    @abstractmethod
+    def set_default_model(self, model_name: str) -> str:
+        """Set the default model for this provider"""
+        pass
+    
+    @abstractmethod
+    def get_provider_info(self) -> str:
+        """Get information about the provider configuration"""
+        pass
+    
+    def get_tools(self) -> List[Tool]:
+        """
+        Get a list of LangChain tools for this provider
+        
+        Returns:
+            List of tools
+        """
+        return [
+            Tool(
+                name=f"{self.provider_name}_list_models",
+                func=self.list_models,
+                description=f"List all available models for {self.provider_name}"
+            ),
+            Tool(
+                name=f"{self.provider_name}_model_info",
+                func=self.get_model_info,
+                description=f"Get detailed information about a specific {self.provider_name} model"
+            ),
+            Tool(
+                name=f"{self.provider_name}_test_connection",
+                func=self.test_connection,
+                description=f"Test connection to {self.provider_name} provider"
+            ),
+            Tool(
+                name=f"{self.provider_name}_get_default_model",
+                func=self.get_default_model,
+                description=f"Get the default model for {self.provider_name}"
+            ),
+            Tool(
+                name=f"{self.provider_name}_set_default_model",
+                func=self.set_default_model,
+                description=f"Set the default model for {self.provider_name}"
+            ),
+            Tool(
+                name=f"{self.provider_name}_info",
+                func=self.get_provider_info,
+                description=f"Get information about {self.provider_name} configuration"
+            )
+        ]
+
+
+class OllamaProviderTools(BaseProviderTools):
+    """Ollama provider tools implementation"""
+    
+    def __init__(self, base_url: str = "http://localhost:11434", **kwargs):
+        """
+        Initialize Ollama tools
+        
+        Args:
+            base_url: Base URL of the Ollama instance
+            **kwargs: Additional configuration
+        """
+        super().__init__("ollama", base_url=base_url, **kwargs)
+        self.base_url = base_url.rstrip('/')
+        self.api_url = f"{self.base_url}/api"
+    
+    def _make_request(self, endpoint: str, method: str = "GET", data: Optional[Dict] = None) -> Dict[str, Any]:
+        """Make a request to the Ollama API"""
+        try:
+            url = f"{self.api_url}/{endpoint}"
+            
+            if method.upper() == "GET":
+                response = requests.get(url, timeout=10)
+            elif method.upper() == "POST":
+                response = requests.post(url, json=data, timeout=30)
+            else:
+                return {"success": False, "error": f"Unsupported method: {method}"}
+            
+            if response.status_code == 200:
+                try:
+                    return {"success": True, "data": response.json()}
+                except json.JSONDecodeError:
+                    return {"success": True, "data": response.text}
+            else:
+                return {"success": False, "error": f"HTTP {response.status_code}: {response.text}"}
+                
+        except requests.exceptions.Timeout:
+            return {"success": False, "error": "Request timed out"}
+        except requests.exceptions.ConnectionError:
+            return {"success": False, "error": f"Connection failed to {self.base_url}"}
+        except Exception as e:
+            return {"success": False, "error": str(e)}
+    
+    def list_models(self) -> str:
+        """List all available models on the Ollama instance"""
+        result = self._make_request("tags")
+        
+        if not result["success"]:
+            return f"Error listing models: {result['error']}"
+        
+        models = result["data"].get("models", [])
+        if not models:
+            return "No models found on the Ollama instance."
+        
+        output = f"Available Ollama Models ({self.base_url}):\n"
+        output += "=" * 50 + "\n"
+        
+        for model in models:
+            name = model.get("name", "Unknown")
+            size = model.get("size", 0)
+            
+            # Convert size to human readable format
+            if size > 1024 * 1024 * 1024:
+                size_str = f"{size / (1024 * 1024 * 1024):.1f} GB"
+            elif size > 1024 * 1024:
+                size_str = f"{size / (1024 * 1024):.1f} MB"
+            else:
+                size_str = f"{size / 1024:.1f} KB"
+            
+            output += f"• {name} ({size_str})\n"
+        
+        return output
+    
+    def get_model_info(self, model_name: str) -> str:
+        """Get detailed information about a specific model"""
+        result = self._make_request("show", method="POST", data={"name": model_name})
+        
+        if not result["success"]:
+            return f"Error getting model info: {result['error']}"
+        
+        model_info = result["data"]
+        
+        output = f"Model Information: {model_name}\n"
+        output += "=" * 50 + "\n"
+        
+        if "license" in model_info:
+            output += f"License: {model_info['license']}\n"
+        
+        if "parameters" in model_info:
+            output += f"Parameters: {model_info['parameters']}\n"
+        
+        if "details" in model_info:
+            details = model_info["details"]
+            if "parameter_size" in details:
+                output += f"Parameter Size: {details['parameter_size']}\n"
+            if "quantization_level" in details:
+                output += f"Quantization Level: {details['quantization_level']}\n"
+        
+        return output
+    
+    def test_connection(self) -> str:
+        """Test connection to Ollama"""
+        try:
+            response = requests.get(f"{self.base_url}/", timeout=5)
+            if response.status_code == 200:
+                return f"✅ Ollama instance at {self.base_url} is running and responsive."
+            else:
+                return f"⚠️  Ollama instance at {self.base_url} returned status {response.status_code}."
+        except requests.exceptions.ConnectionError:
+            return f"❌ Cannot connect to Ollama instance at {self.base_url}."
+        except requests.exceptions.Timeout:
+            return f"❌ Connection to Ollama instance at {self.base_url} timed out."
+        except Exception as e:
+            return f"❌ Error checking Ollama status: {str(e)}"
+    
+    def get_default_model(self) -> str:
+        """Get the default model for Ollama"""
+        # Ollama doesn't have a built-in default model concept
+        # We'll return the first available model or a message
+        result = self._make_request("tags")
+        
+        if not result["success"]:
+            return f"Error getting models: {result['error']}"
+        
+        models = result["data"].get("models", [])
+        if not models:
+            return "No models available on Ollama instance."
+        
+        # Return the first model as default
+        first_model = models[0].get("name", "Unknown")
+        return f"Default model (first available): {first_model}"
+    
+    def set_default_model(self, model_name: str) -> str:
+        """Set the default model for Ollama"""
+        # Ollama doesn't have a built-in default model concept
+        # We'll just verify the model exists
+        result = self._make_request("show", method="POST", data={"name": model_name})
+        
+        if result["success"]:
+            return f"✅ Model '{model_name}' exists and can be used as default."
+        else:
+            return f"❌ Model '{model_name}' not found on Ollama instance."
+    
+    def get_provider_info(self) -> str:
+        """Get information about Ollama configuration"""
+        output = "Ollama Provider Configuration:\n"
+        output += "=" * 50 + "\n"
+        output += f"Base URL: {self.base_url}\n"
+        output += f"API URL: {self.api_url}\n"
+        
+        # Test connection
+        try:
+            response = requests.get(f"{self.base_url}/", timeout=5)
+            if response.status_code == 200:
+                output += "Status: ✅ Connected\n"
+            else:
+                output += f"Status: ⚠️  HTTP {response.status_code}\n"
+        except:
+            output += "Status: ❌ Not connected\n"
+        
+        # Get model count
+        result = self._make_request("tags")
+        if result["success"]:
+            models = result["data"].get("models", [])
+            output += f"Available Models: {len(models)}\n"
+        else:
+            output += "Available Models: Unknown\n"
+        
+        return output
+
+
+class OpenAIProviderTools(BaseProviderTools):
+    """OpenAI provider tools implementation"""
+    
+    def __init__(self, api_key: str, **kwargs):
+        """
+        Initialize OpenAI tools
+        
+        Args:
+            api_key: OpenAI API key
+            **kwargs: Additional configuration
+        """
+        super().__init__("openai", api_key=api_key, **kwargs)
+        self.api_key = api_key
+        self.api_url = "https://api.openai.com/v1"
+    
+    def _make_request(self, endpoint: str, method: str = "GET", data: Optional[Dict] = None) -> Dict[str, Any]:
+        """Make a request to the OpenAI API"""
+        try:
+            url = f"{self.api_url}/{endpoint}"
+            headers = {
+                "Authorization": f"Bearer {self.api_key}",
+                "Content-Type": "application/json"
+            }
+            
+            if method.upper() == "GET":
+                response = requests.get(url, headers=headers, timeout=10)
+            elif method.upper() == "POST":
+                response = requests.post(url, headers=headers, json=data, timeout=30)
+            else:
+                return {"success": False, "error": f"Unsupported method: {method}"}
+            
+            if response.status_code == 200:
+                try:
+                    return {"success": True, "data": response.json()}
+                except json.JSONDecodeError:
+                    return {"success": True, "data": response.text}
+            else:
+                return {"success": False, "error": f"HTTP {response.status_code}: {response.text}"}
+                
+        except requests.exceptions.Timeout:
+            return {"success": False, "error": "Request timed out"}
+        except requests.exceptions.ConnectionError:
+            return {"success": False, "error": "Connection failed to OpenAI API"}
+        except Exception as e:
+            return {"success": False, "error": str(e)}
+    
+    def list_models(self) -> str:
+        """List all available models on OpenAI"""
+        result = self._make_request("models")
+        
+        if not result["success"]:
+            return f"Error listing models: {result['error']}"
+        
+        models = result["data"].get("data", [])
+        if not models:
+            return "No models found on OpenAI."
+        
+        # Filter to show only common models
+        common_models = ["gpt-4o", "gpt-4o-mini", "gpt-4-turbo", "gpt-4", "gpt-3.5-turbo", "text-embedding-ada-002"]
+        
+        output = "Available OpenAI Models:\n"
+        output += "=" * 50 + "\n"
+        
+        for model in models:
+            model_id = model.get("id", "")
+            if model_id in common_models or "gpt" in model_id or "text-embedding" in model_id:
+                output += f"• {model_id}\n"
+        
+        return output
+    
+    def get_model_info(self, model_name: str) -> str:
+        """Get detailed information about a specific model"""
+        result = self._make_request(f"models/{model_name}")
+        
+        if not result["success"]:
+            return f"Error getting model info: {result['error']}"
+        
+        model_info = result["data"]
+        
+        output = f"Model Information: {model_name}\n"
+        output += "=" * 50 + "\n"
+        output += f"ID: {model_info.get('id', 'Unknown')}\n"
+        output += f"Object: {model_info.get('object', 'Unknown')}\n"
+        output += f"Created: {model_info.get('created', 'Unknown')}\n"
+        output += f"Owned by: {model_info.get('owned_by', 'Unknown')}\n"
+        
+        return output
+    
+    def test_connection(self) -> str:
+        """Test connection to OpenAI"""
+        try:
+            # Try to list models as a connection test
+            result = self._make_request("models")
+            if result["success"]:
+                return "✅ OpenAI API connection successful."
+            else:
+                return f"❌ OpenAI API connection failed: {result['error']}"
+        except Exception as e:
+            return f"❌ Error connecting to OpenAI: {str(e)}"
+    
+    def get_default_model(self) -> str:
+        """Get the default model for OpenAI"""
+        # OpenAI doesn't have a built-in default model concept
+        # Return a common default
+        return "Default model: gpt-4o (recommended for most tasks)"
+    
+    def set_default_model(self, model_name: str) -> str:
+        """Set the default model for OpenAI"""
+        # OpenAI doesn't have a built-in default model concept
+        # Just verify the model exists
+        result = self._make_request(f"models/{model_name}")
+        
+        if result["success"]:
+            return f"✅ Model '{model_name}' exists and can be used as default."
+        else:
+            return f"❌ Model '{model_name}' not found on OpenAI."
+    
+    def get_provider_info(self) -> str:
+        """Get information about OpenAI configuration"""
+        output = "OpenAI Provider Configuration:\n"
+        output += "=" * 50 + "\n"
+        output += f"API URL: {self.api_url}\n"
+        output += f"API Key: {'Set' if self.api_key else 'Not set'}\n"
+        
+        # Test connection
+        try:
+            result = self._make_request("models")
+            if result["success"]:
+                output += "Status: ✅ Connected\n"
+            else:
+                output += f"Status: ❌ Connection failed\n"
+        except:
+            output += "Status: ❌ Not connected\n"
+        
+        return output
+
+
+class OpenRouterProviderTools(BaseProviderTools):
+    """OpenRouter provider tools implementation"""
+    
+    def __init__(self, api_key: str, base_url: str = "https://openrouter.ai/api/v1", **kwargs):
+        """
+        Initialize OpenRouter tools
+        
+        Args:
+            api_key: OpenRouter API key
+            base_url: Base URL for OpenRouter API
+            **kwargs: Additional configuration
+        """
+        super().__init__("openrouter", api_key=api_key, base_url=base_url, **kwargs)
+        self.api_key = api_key
+        self.base_url = base_url.rstrip('/')
+        self.api_url = f"{self.base_url}"
+    
+    def _make_request(self, endpoint: str, method: str = "GET", data: Optional[Dict] = None) -> Dict[str, Any]:
+        """Make a request to the OpenRouter API"""
+        try:
+            url = f"{self.api_url}/{endpoint}"
+            headers = {
+                "Authorization": f"Bearer {self.api_key}",
+                "Content-Type": "application/json"
+            }
+            
+            if method.upper() == "GET":
+                response = requests.get(url, headers=headers, timeout=10)
+            elif method.upper() == "POST":
+                response = requests.post(url, headers=headers, json=data, timeout=30)
+            else:
+                return {"success": False, "error": f"Unsupported method: {method}"}
+            
+            if response.status_code == 200:
+                try:
+                    return {"success": True, "data": response.json()}
+                except json.JSONDecodeError:
+                    return {"success": True, "data": response.text}
+            else:
+                return {"success": False, "error": f"HTTP {response.status_code}: {response.text}"}
+                
+        except requests.exceptions.Timeout:
+            return {"success": False, "error": "Request timed out"}
+        except requests.exceptions.ConnectionError:
+            return {"success": False, "error": f"Connection failed to {self.base_url}"}
+        except Exception as e:
+            return {"success": False, "error": str(e)}
+    
+    def list_models(self) -> str:
+        """List all available models on OpenRouter"""
+        result = self._make_request("models")
+        
+        if not result["success"]:
+            return f"Error listing models: {result['error']}"
+        
+        models = result["data"].get("data", [])
+        if not models:
+            return "No models found on OpenRouter."
+        
+        output = "Available OpenRouter Models:\n"
+        output += "=" * 50 + "\n"
+        
+        for model in models[:20]:  # Limit to first 20 models
+            model_id = model.get("id", "")
+            name = model.get("name", "")
+            description = model.get("description", "")
+            
+            output += f"• {model_id}\n"
+            if name:
+                output += f"  Name: {name}\n"
+            if description:
+                output += f"  Description: {description[:100]}...\n"
+            output += "\n"
+        
+        return output
+    
+    def get_model_info(self, model_name: str) -> str:
+        """Get detailed information about a specific model"""
+        result = self._make_request(f"models/{model_name}")
+        
+        if not result["success"]:
+            return f"Error getting model info: {result['error']}"
+        
+        model_info = result["data"]
+        
+        output = f"Model Information: {model_name}\n"
+        output += "=" * 50 + "\n"
+        output += f"ID: {model_info.get('id', 'Unknown')}\n"
+        output += f"Name: {model_info.get('name', 'Unknown')}\n"
+        output += f"Description: {model_info.get('description', 'Unknown')}\n"
+        
+        if "pricing" in model_info:
+            pricing = model_info["pricing"]
+            output += f"Pricing: {pricing}\n"
+        
+        return output
+    
+    def test_connection(self) -> str:
+        """Test connection to OpenRouter"""
+        try:
+            # Try to list models as a connection test
+            result = self._make_request("models")
+            if result["success"]:
+                return "✅ OpenRouter API connection successful."
+            else:
+                return f"❌ OpenRouter API connection failed: {result['error']}"
+        except Exception as e:
+            return f"❌ Error connecting to OpenRouter: {str(e)}"
+    
+    def get_default_model(self) -> str:
+        """Get the default model for OpenRouter"""
+        # OpenRouter doesn't have a built-in default model concept
+        # Return a common default
+        return "Default model: anthropic/claude-3.5-sonnet (recommended for most tasks)"
+    
+    def set_default_model(self, model_name: str) -> str:
+        """Set the default model for OpenRouter"""
+        # OpenRouter doesn't have a built-in default model concept
+        # Just verify the model exists
+        result = self._make_request(f"models/{model_name}")
+        
+        if result["success"]:
+            return f"✅ Model '{model_name}' exists and can be used as default."
+        else:
+            return f"❌ Model '{model_name}' not found on OpenRouter."
+    
+    def get_provider_info(self) -> str:
+        """Get information about OpenRouter configuration"""
+        output = "OpenRouter Provider Configuration:\n"
+        output += "=" * 50 + "\n"
+        output += f"API URL: {self.api_url}\n"
+        output += f"API Key: {'Set' if self.api_key else 'Not set'}\n"
+        
+        # Test connection
+        try:
+            result = self._make_request("models")
+            if result["success"]:
+                output += "Status: ✅ Connected\n"
+            else:
+                output += f"Status: ❌ Connection failed\n"
+        except:
+            output += "Status: ❌ Not connected\n"
+        
+        return output
+
+
+def get_provider_tools(provider: str, **kwargs) -> List[Tool]:
+    """
+    Get tools for a specific provider
+    
+    Args:
+        provider: Provider name ("ollama", "openai", "openrouter")
+        **kwargs: Provider-specific configuration
+        
+    Returns:
+        List of tools for the provider
+    """
+    provider = provider.lower()
+    
+    if provider == "ollama":
+        base_url = kwargs.get("base_url", os.getenv("OLLAMA_BASE_URL", "http://localhost:11434"))
+        tools = OllamaProviderTools(base_url=base_url)
+        return tools.get_tools()
+    
+    elif provider == "openai":
+        api_key = kwargs.get("api_key", os.getenv("OPENAI_API_KEY"))
+        if not api_key:
+            print("Warning: OpenAI API key not provided")
+            return []
+        tools = OpenAIProviderTools(api_key=api_key)
+        return tools.get_tools()
+    
+    elif provider == "openrouter":
+        api_key = kwargs.get("api_key", os.getenv("OPENROUTER_API_KEY"))
+        base_url = kwargs.get("base_url", os.getenv("OPENROUTER_BASE_URL", "https://openrouter.ai/api/v1"))
+        if not api_key:
+            print("Warning: OpenRouter API key not provided")
+            return []
+        tools = OpenRouterProviderTools(api_key=api_key, base_url=base_url)
+        return tools.get_tools()
+    
+    else:
+        print(f"Warning: Unknown provider '{provider}'")
+        return []
+
+
+def get_all_provider_tools() -> List[Tool]:
+    """
+    Get tools for all configured providers
+    
+    Returns:
+        List of tools for all configured providers
+    """
+    all_tools = []
+    
+    # Check for Ollama configuration
+    ollama_url = os.getenv("OLLAMA_BASE_URL")
+    if ollama_url:
+        try:
+            ollama_tools = get_provider_tools("ollama", base_url=ollama_url)
+            all_tools.extend(ollama_tools)
+            print(f"✅ Added {len(ollama_tools)} Ollama tools")
+        except Exception as e:
+            print(f"⚠️  Could not load Ollama tools: {e}")
+    
+    # Check for OpenAI configuration
+    openai_key = os.getenv("OPENAI_API_KEY")
+    if openai_key:
+        try:
+            openai_tools = get_provider_tools("openai", api_key=openai_key)
+            all_tools.extend(openai_tools)
+            print(f"✅ Added {len(openai_tools)} OpenAI tools")
+        except Exception as e:
+            print(f"⚠️  Could not load OpenAI tools: {e}")
+    
+    # Check for OpenRouter configuration
+    openrouter_key = os.getenv("OPENROUTER_API_KEY")
+    if openrouter_key:
+        try:
+            openrouter_tools = get_provider_tools("openrouter", api_key=openrouter_key)
+            all_tools.extend(openrouter_tools)
+            print(f"✅ Added {len(openrouter_tools)} OpenRouter tools")
+        except Exception as e:
+            print(f"⚠️  Could not load OpenRouter tools: {e}")
+    
+    return all_tools