|
|
import asyncio |
|
|
import logging |
|
|
import abc |
|
|
from typing import Dict, List, Any, Optional |
|
|
|
|
|
from aworld.sandbox.base import Sandbox |
|
|
from aworld.sandbox.models import SandboxStatus, SandboxEnvType, SandboxInfo |
|
|
from aworld.sandbox.run.mcp_servers import McpServers |
|
|
|
|
|
|
|
|
class BaseSandbox(Sandbox): |
|
|
""" |
|
|
Base sandbox implementation with common functionality for all sandbox types. |
|
|
This class implements common methods and provides a foundation for specific sandbox implementations. |
|
|
""" |
|
|
|
|
|
def __init__( |
|
|
self, |
|
|
sandbox_id: Optional[str] = None, |
|
|
env_type: Optional[int] = None, |
|
|
metadata: Optional[Dict[str, str]] = None, |
|
|
timeout: Optional[int] = None, |
|
|
mcp_servers: Optional[List[str]] = None, |
|
|
mcp_config: Optional[Any] = None, |
|
|
): |
|
|
""" |
|
|
Initialize a new BaseSandbox instance. |
|
|
|
|
|
Args: |
|
|
sandbox_id: Unique identifier for the sandbox. If None, one will be generated. |
|
|
env_type: The environment type (LOCAL, K8S, SUPERCOMPUTER). |
|
|
metadata: Additional metadata for the sandbox. |
|
|
timeout: Timeout for sandbox operations. |
|
|
mcp_servers: List of MCP servers to use. |
|
|
mcp_config: Configuration for MCP servers. |
|
|
""" |
|
|
super().__init__( |
|
|
sandbox_id=sandbox_id, |
|
|
env_type=env_type, |
|
|
metadata=metadata, |
|
|
timeout=timeout, |
|
|
mcp_servers=mcp_servers, |
|
|
mcp_config=mcp_config |
|
|
) |
|
|
self._logger = self._setup_logger() |
|
|
|
|
|
def _setup_logger(self): |
|
|
""" |
|
|
Set up a logger for the sandbox instance. |
|
|
|
|
|
Returns: |
|
|
logging.Logger: Configured logger instance. |
|
|
""" |
|
|
logger = logging.getLogger(f"sandbox.{self.__class__.__name__}.{self.sandbox_id[:8]}") |
|
|
return logger |
|
|
|
|
|
def get_info(self) -> SandboxInfo: |
|
|
""" |
|
|
Get information about the sandbox. |
|
|
|
|
|
Returns: |
|
|
SandboxInfo: Information about the sandbox. |
|
|
""" |
|
|
return { |
|
|
"sandbox_id": self.sandbox_id, |
|
|
"status": self.status, |
|
|
"metadata": self.metadata, |
|
|
"env_type": self.env_type |
|
|
} |
|
|
|
|
|
@property |
|
|
def mcpservers(self) -> McpServers: |
|
|
""" |
|
|
Module for running MCP servers in the sandbox. |
|
|
This property provides access to the MCP servers instance. |
|
|
|
|
|
Returns: |
|
|
McpServers: The MCP servers instance. |
|
|
""" |
|
|
if hasattr(self, '_mcpservers'): |
|
|
return self._mcpservers |
|
|
return None |
|
|
|
|
|
@abc.abstractmethod |
|
|
async def cleanup(self) -> bool: |
|
|
""" |
|
|
Clean up sandbox resources. |
|
|
This method must be implemented by subclasses to provide environment-specific cleanup. |
|
|
|
|
|
Returns: |
|
|
bool: True if cleanup was successful, False otherwise. |
|
|
""" |
|
|
pass |
|
|
|
|
|
@abc.abstractmethod |
|
|
async def remove(self) -> bool: |
|
|
""" |
|
|
Remove the sandbox. |
|
|
This method must be implemented by subclasses to provide environment-specific removal. |
|
|
|
|
|
Returns: |
|
|
bool: True if removal was successful, False otherwise. |
|
|
""" |
|
|
pass |
