Update tool_executor.py

tool_executor.py

@@ -1,193 +1,7 @@
-import os
-import requests
-import json
-from typing import Dict, Optional
-import subprocess
-import tempfile
-
 class ToolExecutor:
-    """Execute various tools like web search, code execution, and N8N workflows."""
-    
     def __init__(self):
-        """Initialize tool executor with API keys."""
-        self.tavily_key = os.getenv("TAVILY_API_KEY")
-        self.n8n_webhook = os.getenv("N8N_WEBHOOK_URL")
-    
-    def execute_tools(self, query: str) -> str:
-        """Determine and execute appropriate tools based on query."""
-        query_lower = query.lower()
-        results = []
-        
-        # Web search
-        if any(keyword in query_lower for keyword in ['search', 'find', 'look up', 'what is', 'who is', 'google']):
-            if self.tavily_key:
-                search_result = self.web_search(query)
-                if search_result:
-                    results.append(f"**Web Search Results:**\n{search_result}")
-            else:
-                results.append("Web search not configured (missing TAVILY_API_KEY)")
-        
-        # Code execution
-        if any(keyword in query_lower for keyword in ['calculate', 'compute', 'run code', 'execute', 'python']):
-            # Try to extract code from query
-            code = self._extract_code(query)
-            if code:
-                exec_result = self.execute_code(code)
-                if exec_result:
-                    results.append(f"**Code Execution Result:**\n{exec_result}")
-        
-        # N8N workflow
-        if any(keyword in query_lower for keyword in ['add task', 'create task', 'workflow', 'automate', 'trigger']):
-            if self.n8n_webhook:
-                n8n_result = self.trigger_n8n_workflow({'query': query})
-                if n8n_result:
-                    results.append(f"**Workflow Triggered:**\n{n8n_result}")
-            else:
-                results.append("N8N workflow not configured (missing N8N_WEBHOOK_URL)")
-        
-        return '\n\n'.join(results) if results else ""
-    
-    def web_search(self, query: str, max_results: int = 3) -> str:
-        """Search the web using Tavily API."""
-        if not self.tavily_key:
-            return "Web search is not configured. Please add TAVILY_API_KEY."
-        
-        try:
-            url = "https://api.tavily.com/search"
-            payload = {
-                "api_key": self.tavily_key,
-                "query": query,
-                "max_results": max_results,
-                "search_depth": "basic",
-                "include_answer": True
-            }
-            
-            response = requests.post(url, json=payload, timeout=10)
-            response.raise_for_status()
-            data = response.json()
-            
-            # Format results
-            results = []
-            
-            # Include direct answer if available
-            if data.get('answer'):
-                results.append(f"**Answer:** {data['answer']}")
-            
-            # Include search results
-            if data.get('results'):
-                results.append("\n**Sources:**")
-                for i, result in enumerate(data['results'][:max_results], 1):
-                    title = result.get('title', 'No title')
-                    url = result.get('url', '')
-                    content = result.get('content', '')[:200]  # Truncate
-                    results.append(f"{i}. **{title}**\n   {content}...\n   {url}")
-            
-            return '\n'.join(results) if results else "No results found."
-        
-        except Exception as e:
-            return f"Error performing web search: {str(e)}"
-    
-    def execute_code(self, code: str, language: str = "python") -> str:
-        """Execute code safely in a sandboxed environment."""
-        if language != "python":
-            return "Only Python code execution is currently supported."
-        
-        try:
-            # Create a temporary file for the code
-            with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
-                f.write(code)
-                temp_file = f.name
-            
-            # Execute with timeout
-            result = subprocess.run(
-                ['python3', temp_file],
-                capture_output=True,
-                text=True,
-                timeout=5
-            )
-            
-            # Clean up
-            os.unlink(temp_file)
-            
-            # Combine stdout and stderr
-            output = []
-            if result.stdout:
-                output.append(f"**Output:**\n{result.stdout}")
-            if result.stderr:
-                output.append(f"**Errors:**\n{result.stderr}")
-            
-            return '\n'.join(output) if output else "Code executed successfully with no output."
-        
-        except subprocess.TimeoutExpired:
-            return "Code execution timed out (max 5 seconds)."
-        except Exception as e:
-            return f"Error executing code: {str(e)}"
-    
-    def trigger_n8n_workflow(self, data: Dict) -> str:
-        """Trigger an N8N workflow via webhook."""
-        if not self.n8n_webhook:
-            return "N8N workflow is not configured. Please add N8N_WEBHOOK_URL."
-        
-        try:
-            response = requests.post(self.n8n_webhook, json=data, timeout=10)
-            response.raise_for_status()
-            
-            # Try to parse response
-            try:
-                result = response.json()
-                return f"Workflow executed successfully: {json.dumps(result, indent=2)}"
-            except:
-                return f"Workflow executed successfully. Status: {response.status_code}"
-        
-        except Exception as e:
-            return f"Error triggering N8N workflow: {str(e)}"
-    
-    def _extract_code(self, query: str) -> Optional[str]:
-        """Extract code from query if present."""
-        # Look for code blocks
-        import re
-        
-        # Check for markdown code blocks
-        code_block_pattern = r'```(?:python)?\n(.*?)```'
-        matches = re.findall(code_block_pattern, query, re.DOTALL)
-        if matches:
-            return matches[0].strip()
-        
-        # Check for inline code
-        if 'print(' in query or 'def ' in query or 'for ' in query:
-            # Try to extract code-like content
-            lines = query.split('\n')
-            code_lines = [line for line in lines if any(kw in line for kw in ['print', 'def', 'for', 'if', 'while', '='])]
-            if code_lines:
-                return '\n'.join(code_lines)
-        
-        return None
-    
-    def calculate(self, expression: str) -> str:
-        """Safely evaluate mathematical expressions."""
-        try:
-            # Use Python's eval but only for safe math operations
-            allowed_names = {
-                'abs': abs, 'round': round, 'min': min, 'max': max,
-                'sum': sum, 'pow': pow, 'len': len
-            }
-            
-            # Import math functions
-            import math
-            for name in dir(math):
-                if not name.startswith('_'):
-                    allowed_names[name] = getattr(math, name)
-            
-            result = eval(expression, {"__builtins__": {}}, allowed_names)
-            return f"Result: {result}"
-        except Exception as e:
-            return f"Error calculating: {str(e)}"
-    
-    def get_available_tools(self) -> Dict[str, bool]:
-        """Get status of available tools."""
-        return {
-            "web_search": bool(self.tavily_key),
-            "code_execution": True,  # Always available
-            "n8n_workflows": bool(self.n8n_webhook),
-            "calculator": True  # Always available
-        }
+        print("ToolExecutor initialized.")
+
+    def execute_tools(self, message: str) -> str:
+        print(f"Placeholder: Executing tools for message: '{message}'")
+        return "No tools were executed."