| """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"), |
| } |
|
|
