80 lines
2.3 KiB
Python
80 lines
2.3 KiB
Python
|
"""
|
||
|
|
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
|