File size: 4,131 Bytes
4f96544 | 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 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | """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"),
}
|