"""Hot-reload all tools without restarting the bot."""
from __future__ import annotations
import json
import logging
from tools.alter_privileges import PRIVILEGES, has_privilege
logger = logging.getLogger(__name__)
TOOL_NAME = "reload_tools"
TOOL_DESCRIPTION = (
"Reload all tools registered with the bot's tool system. "
"Useful when tools have been added or modified at runtime."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {},
}
[docs]
async def run(ctx=None, **_kwargs) -> str:
"""Execute this tool and return the result.
Args:
ctx: Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
try:
registry = getattr(ctx, "tool_registry", None) if ctx else None
config = getattr(ctx, "config", None) if ctx else None
if registry is None:
return json.dumps({"success": False, "error": "Tool registry not available via ctx"})
if config is None:
return json.dumps({"success": False, "error": "Config not available via ctx"})
user_id = getattr(ctx, "user_id", "") or ""
redis = getattr(ctx, "redis", None)
if not await has_privilege(redis, user_id, PRIVILEGES["UNSANDBOXED_EXEC"], config):
return json.dumps({
"success": False,
"error": (
"reload_tools requires UNSANDBOXED_EXEC. "
"Ask an admin to grant it with alter_privileges."
),
})
from tool_loader import load_tools
old_count = len(registry)
old_permissions = dict(registry._permissions)
registry._tools.clear()
registry.invalidate_cache()
load_tools(config.tools_dir, registry)
registry._permissions = old_permissions
new_count = len(registry)
logger.info("Tool reload completed: %d -> %d tools", old_count, new_count)
return json.dumps({
"success": True,
"old_count": old_count,
"new_count": new_count,
"message": f"Tools reloaded successfully: {old_count} -> {new_count} tools",
})
except Exception as e:
logger.error("Error reloading tools: %s", e, exc_info=True)
return json.dumps({"success": False, "error": f"Error reloading tools: {e}"})