sandbox tool base

backend/agent/tools/computer_use_tool.py

@@ -8,7 +8,7 @@ from typing import Optional, Dict
 import os
 
 from agentpress.tool import Tool, ToolResult, openapi_schema, xml_schema
-from sandbox.sandbox import SandboxToolsBase, Sandbox
+from sandbox.tool_base import SandboxToolsBase, Sandbox
 
 KEYBOARD_KEYS = [
     'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm',

backend/agent/tools/sb_browser_tool.py

@@ -3,7 +3,7 @@ import json
 
 from agentpress.tool import ToolResult, openapi_schema, xml_schema
 from agentpress.thread_manager import ThreadManager
-from sandbox.sandbox import SandboxToolsBase
+from sandbox.tool_base import SandboxToolsBase
 from utils.logger import logger
 
 

backend/agent/tools/sb_deploy_tool.py

@@ -1,7 +1,7 @@
 import os
 from dotenv import load_dotenv
 from agentpress.tool import ToolResult, openapi_schema, xml_schema
-from sandbox.sandbox import SandboxToolsBase
+from sandbox.tool_base import SandboxToolsBase
 from utils.files_utils import clean_path
 from agentpress.thread_manager import ThreadManager
 

backend/agent/tools/sb_expose_tool.py

@@ -1,5 +1,5 @@
 from agentpress.tool import ToolResult, openapi_schema, xml_schema
-from sandbox.sandbox import SandboxToolsBase
+from sandbox.tool_base import SandboxToolsBase
 from agentpress.thread_manager import ThreadManager
 
 class SandboxExposeTool(SandboxToolsBase):

backend/agent/tools/sb_files_tool.py

@@ -1,6 +1,6 @@
 
 from agentpress.tool import ToolResult, openapi_schema, xml_schema
-from sandbox.sandbox import SandboxToolsBase    
+from sandbox.tool_base import SandboxToolsBase    
 from utils.files_utils import should_exclude_file, clean_path
 from agentpress.thread_manager import ThreadManager
 from utils.logger import logger

backend/agent/tools/sb_shell_tool.py

@@ -1,7 +1,7 @@
 from typing import Optional, Dict
 from uuid import uuid4
 from agentpress.tool import ToolResult, openapi_schema, xml_schema
-from sandbox.sandbox import SandboxToolsBase
+from sandbox.tool_base import SandboxToolsBase
 from agentpress.thread_manager import ThreadManager
 
 class SandboxShellTool(SandboxToolsBase):

backend/agent/tools/sb_vision_tool.py

@@ -4,7 +4,7 @@ import mimetypes
 from typing import Optional
 
 from agentpress.tool import ToolResult, openapi_schema, xml_schema
-from sandbox.sandbox import SandboxToolsBase
+from sandbox.tool_base import SandboxToolsBase
 from agentpress.thread_manager import ThreadManager
 from utils.logger import logger
 import json

backend/sandbox/sandbox.py

@@ -1,14 +1,8 @@
-from typing import Optional
-
 from daytona_sdk import Daytona, DaytonaConfig, CreateSandboxParams, Sandbox, SessionExecuteRequest
 from daytona_api_client.models.workspace_state import WorkspaceState
 from dotenv import load_dotenv
-
-from agentpress.tool import Tool
 from utils.logger import logger
 from utils.config import config
-from utils.files_utils import clean_path
-from agentpress.thread_manager import ThreadManager
 
 load_dotenv()
 
@@ -130,83 +124,3 @@ def create_sandbox(password: str, project_id: str = None):
     logger.debug(f"Sandbox environment successfully initialized")
     return sandbox
 
-
-class SandboxToolsBase(Tool):
-    """Base class for all sandbox tools that provides project-based sandbox access."""
-    
-    # Class variable to track if sandbox URLs have been printed
-    _urls_printed = False
-    
-    def __init__(self, project_id: str, thread_manager: Optional[ThreadManager] = None):
-        super().__init__()
-        self.project_id = project_id
-        self.thread_manager = thread_manager
-        self.workspace_path = "/workspace"
-        self._sandbox = None
-        self._sandbox_id = None
-        self._sandbox_pass = None
-
-    async def _ensure_sandbox(self) -> Sandbox:
-        """Ensure we have a valid sandbox instance, retrieving it from the project if needed."""
-        if self._sandbox is None:
-            try:
-                # Get database client
-                client = await self.thread_manager.db.client
-                
-                # Get project data
-                project = await client.table('projects').select('*').eq('project_id', self.project_id).execute()
-                if not project.data or len(project.data) == 0:
-                    raise ValueError(f"Project {self.project_id} not found")
-                
-                project_data = project.data[0]
-                sandbox_info = project_data.get('sandbox', {})
-                
-                if not sandbox_info.get('id'):
-                    raise ValueError(f"No sandbox found for project {self.project_id}")
-                
-                # Store sandbox info
-                self._sandbox_id = sandbox_info['id']
-                self._sandbox_pass = sandbox_info.get('pass')
-                
-                # Get or start the sandbox
-                self._sandbox = await get_or_start_sandbox(self._sandbox_id)
-                
-                # # Log URLs if not already printed
-                # if not SandboxToolsBase._urls_printed:
-                #     vnc_link = self._sandbox.get_preview_link(6080)
-                #     website_link = self._sandbox.get_preview_link(8080)
-                    
-                #     vnc_url = vnc_link.url if hasattr(vnc_link, 'url') else str(vnc_link)
-                #     website_url = website_link.url if hasattr(website_link, 'url') else str(website_link)
-                    
-                #     print("\033[95m***")
-                #     print(f"VNC URL: {vnc_url}")
-                #     print(f"Website URL: {website_url}")
-                #     print("***\033[0m")
-                #     SandboxToolsBase._urls_printed = True
-                
-            except Exception as e:
-                logger.error(f"Error retrieving sandbox for project {self.project_id}: {str(e)}", exc_info=True)
-                raise e
-        
-        return self._sandbox
-
-    @property
-    def sandbox(self) -> Sandbox:
-        """Get the sandbox instance, ensuring it exists."""
-        if self._sandbox is None:
-            raise RuntimeError("Sandbox not initialized. Call _ensure_sandbox() first.")
-        return self._sandbox
-
-    @property
-    def sandbox_id(self) -> str:
-        """Get the sandbox ID, ensuring it exists."""
-        if self._sandbox_id is None:
-            raise RuntimeError("Sandbox ID not initialized. Call _ensure_sandbox() first.")
-        return self._sandbox_id
-
-    def clean_path(self, path: str) -> str:
-        """Clean and normalize a path to be relative to /workspace."""
-        cleaned_path = clean_path(path, self.workspace_path)
-        logger.debug(f"Cleaned path: {path} -> {cleaned_path}")
-        return cleaned_path

backend/sandbox/tool_base.py

@@ -0,0 +1,90 @@
+
+from typing import Optional
+
+from agentpress.thread_manager import ThreadManager
+from agentpress.tool import Tool
+from daytona_sdk import Sandbox
+from sandbox.sandbox import get_or_start_sandbox
+from utils import logger
+from utils.files_utils import clean_path
+
+
+class SandboxToolsBase(Tool):
+    """Base class for all sandbox tools that provides project-based sandbox access."""
+    
+    # Class variable to track if sandbox URLs have been printed
+    _urls_printed = False
+    
+    def __init__(self, project_id: str, thread_manager: Optional[ThreadManager] = None):
+        super().__init__()
+        self.project_id = project_id
+        self.thread_manager = thread_manager
+        self.workspace_path = "/workspace"
+        self._sandbox = None
+        self._sandbox_id = None
+        self._sandbox_pass = None
+
+    async def _ensure_sandbox(self) -> Sandbox:
+        """Ensure we have a valid sandbox instance, retrieving it from the project if needed."""
+        if self._sandbox is None:
+            try:
+                # Get database client
+                client = await self.thread_manager.db.client
+                
+                # Get project data
+                project = await client.table('projects').select('*').eq('project_id', self.project_id).execute()
+                if not project.data or len(project.data) == 0:
+                    raise ValueError(f"Project {self.project_id} not found")
+                
+                project_data = project.data[0]
+                sandbox_info = project_data.get('sandbox', {})
+                
+                if not sandbox_info.get('id'):
+                    raise ValueError(f"No sandbox found for project {self.project_id}")
+                
+                # Store sandbox info
+                self._sandbox_id = sandbox_info['id']
+                self._sandbox_pass = sandbox_info.get('pass')
+                
+                # Get or start the sandbox
+                self._sandbox = await get_or_start_sandbox(self._sandbox_id)
+                
+                # # Log URLs if not already printed
+                # if not SandboxToolsBase._urls_printed:
+                #     vnc_link = self._sandbox.get_preview_link(6080)
+                #     website_link = self._sandbox.get_preview_link(8080)
+                    
+                #     vnc_url = vnc_link.url if hasattr(vnc_link, 'url') else str(vnc_link)
+                #     website_url = website_link.url if hasattr(website_link, 'url') else str(website_link)
+                    
+                #     print("\033[95m***")
+                #     print(f"VNC URL: {vnc_url}")
+                #     print(f"Website URL: {website_url}")
+                #     print("***\033[0m")
+                #     SandboxToolsBase._urls_printed = True
+                
+            except Exception as e:
+                logger.error(f"Error retrieving sandbox for project {self.project_id}: {str(e)}", exc_info=True)
+                raise e
+        
+        return self._sandbox
+
+    @property
+    def sandbox(self) -> Sandbox:
+        """Get the sandbox instance, ensuring it exists."""
+        if self._sandbox is None:
+            raise RuntimeError("Sandbox not initialized. Call _ensure_sandbox() first.")
+        return self._sandbox
+
+    @property
+    def sandbox_id(self) -> str:
+        """Get the sandbox ID, ensuring it exists."""
+        if self._sandbox_id is None:
+            raise RuntimeError("Sandbox ID not initialized. Call _ensure_sandbox() first.")
+        return self._sandbox_id
+
+    def clean_path(self, path: str) -> str:
+        """Clean and normalize a path to be relative to /workspace."""
+        cleaned_path = clean_path(path, self.workspace_path)
+        logger.debug(f"Cleaned path: {path} -> {cleaned_path}")
+        return cleaned_path