forked from Rockachopa/Timmy-time-dashboard
904a7c564e
Replace the homebrew regex-based tool extraction and manual dispatch (tool_executor.py) with Agno's built-in Human-In-The-Loop confirmation: - Toolkit(requires_confirmation_tools=...) marks dangerous tools - agent.run() returns RunOutput with status=paused when confirmation needed - RunRequirement.confirm()/reject() + agent.continue_run() resumes execution Dashboard and Discord vendor both use the native flow. DuckDuckGo import isolated so its absence doesn't kill all tools. Test stubs cleaned up (agno is a real dependency, only truly optional packages stubbed). 1384 tests pass in parallel (~14s). Co-authored-by: Trip T <trip@local> Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
124 lines
3.7 KiB
Python
124 lines
3.7 KiB
Python
"""Tool safety classification and tool-call extraction helpers.
|
|
|
|
Classifies tools into tiers based on their potential impact:
|
|
- DANGEROUS: Can modify filesystem, execute code, or change system state.
|
|
Requires user confirmation before execution.
|
|
- SAFE: Read-only or purely computational. Executes without confirmation.
|
|
|
|
Also provides shared helpers for extracting hallucinated tool calls from
|
|
model output and formatting them for human review. Used by both the
|
|
Discord vendor and the dashboard chat route.
|
|
"""
|
|
|
|
import json
|
|
import re
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool classification
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Tools that require confirmation before execution.
|
|
DANGEROUS_TOOLS = frozenset(
|
|
{
|
|
"shell",
|
|
"python",
|
|
"write_file",
|
|
"aider",
|
|
"plan_and_execute",
|
|
}
|
|
)
|
|
|
|
# Tools that are safe to execute without confirmation.
|
|
SAFE_TOOLS = frozenset(
|
|
{
|
|
"web_search",
|
|
"calculator",
|
|
"memory_search",
|
|
"memory_read",
|
|
"memory_write",
|
|
"read_file",
|
|
"list_files",
|
|
"consult_grok",
|
|
"get_system_info",
|
|
"check_ollama_health",
|
|
"get_memory_status",
|
|
"list_swarm_agents",
|
|
}
|
|
)
|
|
|
|
|
|
def requires_confirmation(tool_name: str) -> bool:
|
|
"""Check if a tool requires user confirmation before execution.
|
|
|
|
Unknown tools default to requiring confirmation (safe-by-default).
|
|
"""
|
|
if tool_name in SAFE_TOOLS:
|
|
return False
|
|
return True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool call extraction from model output
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_TOOL_CALL_RE = re.compile(
|
|
r'\{\s*"name"\s*:\s*"([^"]+?)"\s*,\s*"(?:parameters|arguments)"\s*:\s*(\{.*?\})\s*\}',
|
|
re.DOTALL,
|
|
)
|
|
|
|
|
|
def extract_tool_calls(text: str) -> list[tuple[str, dict]]:
|
|
"""Extract hallucinated tool calls from model output.
|
|
|
|
Returns list of (tool_name, arguments_dict) tuples.
|
|
Handles both ``"arguments"`` and ``"parameters"`` JSON keys.
|
|
"""
|
|
if not text:
|
|
return []
|
|
|
|
results = []
|
|
for match in _TOOL_CALL_RE.finditer(text):
|
|
tool_name = match.group(1)
|
|
try:
|
|
args = json.loads(match.group(2))
|
|
except json.JSONDecodeError:
|
|
continue
|
|
results.append((tool_name, args))
|
|
|
|
return results
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Formatting helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def format_action_description(tool_name: str, tool_args: dict) -> str:
|
|
"""Format a human-readable description of a tool action."""
|
|
if tool_name == "shell":
|
|
cmd = tool_args.get("command") or tool_args.get("args", "")
|
|
if isinstance(cmd, list):
|
|
cmd = " ".join(cmd)
|
|
return f"Run shell command:\n`{cmd}`"
|
|
elif tool_name == "write_file":
|
|
path = tool_args.get("file_name", "unknown")
|
|
size = len(tool_args.get("contents", ""))
|
|
return f"Write file: `{path}` ({size} chars)"
|
|
elif tool_name == "python":
|
|
code = tool_args.get("code", "")[:200]
|
|
return f"Execute Python:\n```python\n{code}\n```"
|
|
else:
|
|
args_str = json.dumps(tool_args, indent=2)[:300]
|
|
return f"Execute `{tool_name}` with args:\n```json\n{args_str}\n```"
|
|
|
|
|
|
def get_impact_level(tool_name: str) -> str:
|
|
"""Return the impact level for a tool (high, medium, or low)."""
|
|
high_impact = {"shell", "python"}
|
|
medium_impact = {"write_file", "aider", "plan_and_execute"}
|
|
if tool_name in high_impact:
|
|
return "high"
|
|
if tool_name in medium_impact:
|
|
return "medium"
|
|
return "low"
|