Spaces:
Sleeping
Sleeping
| """Tool selector for remote execution steps.""" | |
| import sqlite3 | |
| from typing import Any | |
| from api.deps import get_logger, load_config | |
| logger = get_logger("kapo.agent.tool_selector") | |
| class ToolSelectorAgent: | |
| def __init__(self): | |
| self.cfg = load_config() | |
| def _load_catalog(self) -> list[dict[str, Any]]: | |
| tools_db = self.cfg.get("TOOLS_DB_PATH") | |
| if not tools_db or not str(tools_db).endswith(".db"): | |
| return [] | |
| try: | |
| conn = sqlite3.connect(str(tools_db)) | |
| cur = conn.cursor() | |
| cur.execute( | |
| """ | |
| SELECT tool_name, install_command, path, description, installed | |
| FROM tools | |
| """ | |
| ) | |
| rows = cur.fetchall() | |
| conn.close() | |
| except sqlite3.OperationalError: | |
| return [] | |
| except Exception: | |
| logger.exception("Tool catalog load failed") | |
| return [] | |
| return [ | |
| { | |
| "tool_name": row[0], | |
| "install_command": row[1], | |
| "path": row[2], | |
| "description": row[3] or "", | |
| "installed": bool(row[4]), | |
| } | |
| for row in rows | |
| ] | |
| def _step_text(self, step: dict[str, Any]) -> str: | |
| for key in ("input", "description", "title", "summary", "task", "prompt"): | |
| value = str(step.get(key, "")).strip() | |
| if value: | |
| return value | |
| return "" | |
| def _fallback_tool(self, tool_hint: str) -> dict[str, Any]: | |
| name = f"fallback_{tool_hint or 'command'}" | |
| return { | |
| "tool_name": name, | |
| "description": "Synthetic fallback tool selected by ToolSelectorAgent", | |
| "path": tool_hint or "", | |
| "installed": True, | |
| } | |
| def _fallback_command(self, step: dict[str, Any], tool_hint: str = "") -> str: | |
| action = str(step.get("action", "")).strip().lower() | |
| step_input = self._step_text(step) | |
| explicit_command = str(step.get("command", "")).strip() | |
| if explicit_command: | |
| return explicit_command | |
| if action == "research": | |
| return f"echo Research task queued: {step_input}".strip() | |
| if action in {"verify", "summarize", "respond", "synthesize"}: | |
| if step_input: | |
| return f"python -c \"print({step_input!r})\"" | |
| return "echo Verification requested" | |
| if action in {"execute", "collect_requirements", "analyze", "decompose", "collect_context"}: | |
| return f"python -c \"print({step_input!r})\"" if step_input else "python -c \"print('step received')\"" | |
| if tool_hint in {"python", "py"} or not tool_hint: | |
| return f"python -c \"print({step_input!r})\"" if step_input else "python -c \"print('step received')\"" | |
| if step_input: | |
| return f"echo {step_input}" | |
| return "echo Step received" | |
| def select_tool(self, step: dict[str, Any]) -> dict[str, Any]: | |
| action = str(step.get("action", "")).strip().lower() | |
| tool_hint = str(step.get("tool_hint", "")).strip().lower() | |
| step_input = self._step_text(step).lower() | |
| catalog = self._load_catalog() | |
| for tool in catalog: | |
| haystack = " ".join([tool["tool_name"], tool["description"], tool["path"] or ""]).lower() | |
| if tool_hint and tool_hint in haystack: | |
| return {"command": tool["path"] or tool["tool_name"], "files": {}, "env": {}, "tool": tool} | |
| if action and action in haystack: | |
| return {"command": tool["path"] or tool["tool_name"], "files": {}, "env": {}, "tool": tool} | |
| if step_input and any(token in haystack for token in step_input.split()[:4]): | |
| return {"command": tool["path"] or tool["tool_name"], "files": {}, "env": {}, "tool": tool} | |
| return { | |
| "command": self._fallback_command(step, tool_hint=tool_hint), | |
| "files": step.get("files", {}), | |
| "env": step.get("env", {}), | |
| "tool": self._fallback_tool(tool_hint or "python"), | |
| } | |