Spaces:
Sleeping
Sleeping
File size: 3,288 Bytes
44e7e06 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 |
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 |