File size: 2,102 Bytes
b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 a7c4301 b8e5043 | 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 | """Code executor - allows the AI to create new plugins at runtime."""
import yaml
from loguru import logger
from plugins.manager import PluginManager
# Global plugin manager (set during initialization)
_plugin_manager: PluginManager | None = None
def set_plugin_manager(manager: PluginManager):
"""Set the global plugin manager instance."""
global _plugin_manager
_plugin_manager = manager
async def run_create_plugin(code: str, name: str, description: str) -> str:
"""Create a new plugin from code.
The AI can call this to dynamically create new capabilities.
Args:
code: Python code for the plugin (must define a BasePlugin subclass).
name: Plugin name (used as directory name).
description: Human-readable description.
Returns:
Success or error message.
"""
if _plugin_manager is None:
return "Error: Plugin manager not initialized."
plugins_dir = _plugin_manager.plugins_dir
plugin_dir = plugins_dir / name
plugin_dir.mkdir(parents=True, exist_ok=True)
try:
# Write plugin.py
plugin_file = plugin_dir / "plugin.py"
plugin_file.write_text(code, encoding="utf-8")
# Write metadata.yaml
metadata = {
"name": name,
"description": description,
"author": "AI Generated",
"version": "1.0.0",
}
metadata_file = plugin_dir / "metadata.yaml"
with open(metadata_file, "w", encoding="utf-8") as f:
yaml.dump(metadata, f, allow_unicode=True)
# Write __init__.py
(plugin_dir / "__init__.py").write_text("", encoding="utf-8")
# Reload and activate
_plugin_manager.registry.discover_and_load(plugins_dir)
await _plugin_manager.activate_plugin(name)
logger.info(f"Plugin '{name}' created and activated")
return f"Plugin '{name}' created and activated successfully at {plugin_dir}"
except Exception as e:
logger.error(f"Failed to create plugin '{name}': {e}")
return f"Error creating plugin '{name}': {e}"
|