from hopx_ai import Sandbox
import os
from typing import Dict, Any, List
class Plugin:
def __init__(self, plugin_id: str, code: str, permissions: List[str]):
self.plugin_id = plugin_id
self.code = code
self.permissions = permissions
class PluginSystem:
def __init__(self, api_key: str):
self.api_key = api_key
self.plugins = {}
self.plugin_sandboxes = {}
def register_plugin(self, plugin: Plugin):
"""Register a plugin"""
self.plugins[plugin.plugin_id] = plugin
def execute_plugin(self, plugin_id: str, context: Dict[str, Any] = None) -> Dict[str, Any]:
"""Execute plugin in isolated sandbox"""
if plugin_id not in self.plugins:
return {
"success": False,
"error": f"Plugin {plugin_id} not found"
}
plugin = self.plugins[plugin_id]
sandbox = None
try:
# Create fresh sandbox for each execution
sandbox = Sandbox.create(
template="code-interpreter",
api_key=self.api_key,
timeout_seconds=60
)
# Set up plugin environment with limited permissions
env_vars = {
"PLUGIN_ID": plugin_id,
"PERMISSIONS": ",".join(plugin.permissions)
}
sandbox.env.set_all(env_vars)
# Add context if provided
if context:
for key, value in context.items():
sandbox.env.set(f"CONTEXT_{key}", str(value))
# Execute plugin code
result = sandbox.run_code(plugin.code, timeout=30)
return {
"success": result.success,
"stdout": result.stdout,
"stderr": result.stderr,
"plugin_id": plugin_id
}
except Exception as e:
return {
"success": False,
"error": str(e),
"plugin_id": plugin_id
}
finally:
if sandbox:
sandbox.kill() # Always cleanup
def cleanup_all(self):
"""Clean up all plugin resources"""
for sandbox in self.plugin_sandboxes.values():
sandbox.kill()
self.plugin_sandboxes.clear()
# Usage
system = PluginSystem(api_key=os.getenv("HOPX_API_KEY"))
# Register plugin
system.register_plugin(Plugin(
plugin_id="calculator",
code="result = 2 + 2\nprint(f'Result: {result}')",
permissions=["compute"]
))
# Execute
result = system.execute_plugin("calculator")
print(result)
system.cleanup_all()
