""" Runtime Manager Manages multiple runtimes and routes tool execution to the appropriate runtime. """ from typing import Any, Dict, Optional from tools.runtime.base import BaseRuntime from tools.runtime.python_runtime import PythonRuntime from tools.runtime.js_runtime import JavaScriptRuntime from tools.runtime.native_runtime import NativeRuntime class RuntimeManager: """Runtime manager for multi-runtime tool execution""" def __init__(self): self._runtimes: Dict[str, BaseRuntime] = { "python": PythonRuntime(), "javascript": JavaScriptRuntime(), "native": NativeRuntime(), } def get_runtime(self, name: str) -> Optional[BaseRuntime]: """Get runtime by name""" return self._runtimes.get(name) def register_runtime(self, name: str, runtime: BaseRuntime) -> None: """Register a custom runtime""" self._runtimes[name] = runtime async def execute( self, runtime_name: str, entry: str, command: str, parameters: Dict[str, Any], timeout: int = 30000, ) -> Dict[str, Any]: """Execute tool through the specified runtime""" runtime = self.get_runtime(runtime_name) if not runtime: return { "status": "error", "error": f"Unknown runtime: {runtime_name}", } # Validate first is_valid = await runtime.validate(entry) if not is_valid: return { "status": "error", "error": f"Validation failed for runtime {runtime_name}, entry: {entry}", } return await runtime.execute(entry, command, parameters, timeout) def list_runtimes(self) -> list: """List all registered runtimes""" return [ { "name": name, "available": runtime.get_name() == name, } for name, runtime in self._runtimes.items() ] # Global runtime manager instance _runtime_manager: Optional[RuntimeManager] = None def get_runtime_manager() -> RuntimeManager: """Get the global runtime manager instance""" global _runtime_manager if _runtime_manager is None: _runtime_manager = RuntimeManager() return _runtime_manager