perplexity/Timmy-time-dashboard
Archived
1
0
Fork 0
forked from Rockachopa/Timmy-time-dashboard
Code Issues Pull Requests 1 Activity

feat: add Aider AI tool to Forge's toolkit (#70)

Browse Source 
* test: remove hardcoded sleeps, add pytest-timeout

- Replace fixed time.sleep() calls with intelligent polling or WebDriverWait
- Add pytest-timeout dependency and --timeout=30 to prevent hangs
- Fixes test flakiness and improves test suite speed

* feat: add Aider AI tool to Forge's toolkit

- Add Aider tool that calls local Ollama (qwen2.5:14b) for AI coding assist
- Register tool in Forge's code toolkit
- Add functional tests for the Aider tool

---------

Co-authored-by: Alexander Payne <apayne@MM.local>
This commit is contained in:
Alexander Whitestone
2026-02-26 23:17:19 -05:00
committed by GitHub
parent 51140fb7f0
commit a5765c33b6
2 changed files with 184 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

161
src/timmy/tools.py
View File

@@ -42,6 +42,7 @@ try:
from agno.tools.file import FileTools
from agno.tools.python import PythonTools
from agno.tools.shell import ShellTools
_AGNO_TOOLS_AVAILABLE = True
except ImportError as e:
_AGNO_TOOLS_AVAILABLE = False
@@ -54,6 +55,7 @@ _TOOL_USAGE: dict[str, list[dict]] = {}
@dataclass
class ToolStats:
"""Statistics for a single tool."""
tool_name: str
call_count: int = 0
last_used: str | None = None
@@ -63,6 +65,7 @@ class ToolStats:
@dataclass
class PersonaTools:
"""Tools assigned to a persona/agent."""
agent_id: str
agent_name: str
toolkit: Toolkit
@@ -73,19 +76,21 @@ def _track_tool_usage(agent_id: str, tool_name: str, success: bool = True) -> No
"""Track tool usage for analytics."""
if agent_id not in _TOOL_USAGE:
_TOOL_USAGE[agent_id] = []
_TOOL_USAGE[agent_id].append({
"tool": tool_name,
"timestamp": datetime.now(timezone.utc).isoformat(),
"success": success,
})
_TOOL_USAGE[agent_id].append(
{
"tool": tool_name,
"timestamp": datetime.now(timezone.utc).isoformat(),
"success": success,
}
)
def get_tool_stats(agent_id: str | None = None) -> dict:
"""Get tool usage statistics.
Args:
agent_id: Optional agent ID to filter by. If None, returns stats for all agents.
Returns:
Dict with tool usage statistics.
"""
@@ -97,7 +102,7 @@ def get_tool_stats(agent_id: str | None = None) -> dict:
"tools_used": list(set(u["tool"] for u in usage)),
"recent_calls": usage[-10:] if usage else [],
}
# Return stats for all agents
all_stats = {}
for aid, usage in _TOOL_USAGE.items():
@@ -137,144 +142,203 @@ def calculator(expression: str) -> str:
def create_research_tools(base_dir: str | Path | None = None):
"""Create tools for research personas (Echo).
Includes: web search, file reading
"""
if not _AGNO_TOOLS_AVAILABLE:
raise ImportError(f"Agno tools not available: {_ImportError}")
toolkit = Toolkit(name="research")
# Web search via DuckDuckGo
search_tools = DuckDuckGoTools()
toolkit.register(search_tools.web_search, name="web_search")
# File reading
base_path = Path(base_dir) if base_dir else Path.cwd()
file_tools = FileTools(base_dir=base_path)
toolkit.register(file_tools.read_file, name="read_file")
toolkit.register(file_tools.list_files, name="list_files")
return toolkit
def create_code_tools(base_dir: str | Path | None = None):
"""Create tools for coding personas (Forge).
Includes: shell commands, python execution, file read/write
Includes: shell commands, python execution, file read/write, Aider AI assist
"""
if not _AGNO_TOOLS_AVAILABLE:
raise ImportError(f"Agno tools not available: {_ImportError}")
toolkit = Toolkit(name="code")
# Shell commands (sandboxed)
shell_tools = ShellTools()
toolkit.register(shell_tools.run_shell_command, name="shell")
# Python execution
python_tools = PythonTools()
toolkit.register(python_tools.run_python_code, name="python")
# File operations
base_path = Path(base_dir) if base_dir else Path.cwd()
file_tools = FileTools(base_dir=base_path)
toolkit.register(file_tools.read_file, name="read_file")
toolkit.register(file_tools.save_file, name="write_file")
toolkit.register(file_tools.list_files, name="list_files")
# Aider AI coding assistant (local with Ollama)
aider_tool = create_aider_tool(base_path)
toolkit.register(aider_tool.run_aider, name="aider")
return toolkit
def create_aider_tool(base_path: Path):
"""Create an Aider tool for AI-assisted coding."""
import subprocess
class AiderTool:
"""Tool that calls Aider (local AI coding assistant) for code generation."""
def __init__(self, base_dir: Path):
self.base_dir = base_dir
def run_aider(self, prompt: str, model: str = "qwen2.5:14b") -> str:
"""Run Aider to generate code changes.
Args:
prompt: What you want Aider to do (e.g., "add a fibonacci function")
model: Ollama model to use (default: qwen2.5:14b)
Returns:
Aider's response with the code changes made
"""
try:
# Run aider with the prompt
result = subprocess.run(
[
"aider",
"--no-git",
"--model",
f"ollama/{model}",
"--quiet",
prompt,
],
capture_output=True,
text=True,
timeout=120,
cwd=str(self.base_dir),
)
if result.returncode == 0:
return (
result.stdout
if result.stdout
else "Code changes applied successfully"
)
else:
return f"Aider error: {result.stderr}"
except FileNotFoundError:
return "Error: Aider not installed. Run: pip install aider"
except subprocess.TimeoutExpired:
return "Error: Aider timed out after 120 seconds"
except Exception as e:
return f"Error running Aider: {str(e)}"
return AiderTool(base_path)
def create_data_tools(base_dir: str | Path | None = None):
"""Create tools for data personas (Seer).
Includes: python execution, file reading, web search for data sources
"""
if not _AGNO_TOOLS_AVAILABLE:
raise ImportError(f"Agno tools not available: {_ImportError}")
toolkit = Toolkit(name="data")
# Python execution for analysis
python_tools = PythonTools()
toolkit.register(python_tools.run_python_code, name="python")
# File reading
base_path = Path(base_dir) if base_dir else Path.cwd()
file_tools = FileTools(base_dir=base_path)
toolkit.register(file_tools.read_file, name="read_file")
toolkit.register(file_tools.list_files, name="list_files")
# Web search for finding datasets
search_tools = DuckDuckGoTools()
toolkit.register(search_tools.web_search, name="web_search")
return toolkit
def create_writing_tools(base_dir: str | Path | None = None):
"""Create tools for writing personas (Quill).
Includes: file read/write
"""
if not _AGNO_TOOLS_AVAILABLE:
raise ImportError(f"Agno tools not available: {_ImportError}")
toolkit = Toolkit(name="writing")
# File operations
base_path = Path(base_dir) if base_dir else Path.cwd()
file_tools = FileTools(base_dir=base_path)
toolkit.register(file_tools.read_file, name="read_file")
toolkit.register(file_tools.save_file, name="write_file")
toolkit.register(file_tools.list_files, name="list_files")
return toolkit
def create_security_tools(base_dir: str | Path | None = None):
"""Create tools for security personas (Mace).
Includes: shell commands (for scanning), web search (for threat intel), file read
"""
if not _AGNO_TOOLS_AVAILABLE:
raise ImportError(f"Agno tools not available: {_ImportError}")
toolkit = Toolkit(name="security")
# Shell for running security scans
shell_tools = ShellTools()
toolkit.register(shell_tools.run_shell_command, name="shell")
# Web search for threat intelligence
search_tools = DuckDuckGoTools()
toolkit.register(search_tools.web_search, name="web_search")
# File reading for logs/configs
base_path = Path(base_dir) if base_dir else Path.cwd()
file_tools = FileTools(base_dir=base_path)
toolkit.register(file_tools.read_file, name="read_file")
toolkit.register(file_tools.list_files, name="list_files")
return toolkit
def create_devops_tools(base_dir: str | Path | None = None):
"""Create tools for DevOps personas (Helm).
Includes: shell commands, file read/write
"""
if not _AGNO_TOOLS_AVAILABLE:
raise ImportError(f"Agno tools not available: {_ImportError}")
toolkit = Toolkit(name="devops")
# Shell for deployment commands
shell_tools = ShellTools()
toolkit.register(shell_tools.run_shell_command, name="shell")
# File operations for config management
base_path = Path(base_dir) if base_dir else Path.cwd()
file_tools = FileTools(base_dir=base_path)
toolkit.register(file_tools.read_file, name="read_file")
toolkit.register(file_tools.save_file, name="write_file")
toolkit.register(file_tools.list_files, name="list_files")
return toolkit
@@ -305,6 +369,7 @@ def consult_grok(query: str) -> str:
# Log to Spark if available
try:
from spark.engine import spark_engine
spark_engine.on_tool_executed(
agent_id="timmy",
tool_name="consult_grok",
@@ -318,10 +383,13 @@ def consult_grok(query: str) -> str:
if not settings.grok_free:
try:
from lightning.factory import get_backend as get_ln_backend
ln = get_ln_backend()
sats = min(settings.grok_max_sats_per_query, 100)
inv = ln.create_invoice(sats, f"Grok query: {query[:50]}")
invoice_info = f"\n[Lightning invoice: {sats} sats — {inv.payment_request[:40]}...]"
invoice_info = (
f"\n[Lightning invoice: {sats} sats — {inv.payment_request[:40]}...]"
)
except Exception:
pass
@@ -370,6 +438,7 @@ def create_full_toolkit(base_dir: str | Path | None = None):
# Grok consultation — premium frontier reasoning (opt-in)
try:
from timmy.backends import grok_available
if grok_available():
toolkit.register(consult_grok, name="consult_grok")
logger.info("Grok consultation tool registered")
@@ -379,6 +448,7 @@ def create_full_toolkit(base_dir: str | Path | None = None):
# Memory search - semantic recall
try:
from timmy.semantic_memory import memory_search
toolkit.register(memory_search, name="memory_search")
except Exception:
logger.debug("Memory search not available")
@@ -414,13 +484,15 @@ def _create_stub_toolkit(name: str):
return toolkit
def get_tools_for_persona(persona_id: str, base_dir: str | Path | None = None) -> Toolkit | None:
def get_tools_for_persona(
persona_id: str, base_dir: str | Path | None = None
) -> Toolkit | None:
"""Get the appropriate toolkit for a persona.
Args:
persona_id: The persona ID (echo, mace, helm, seer, forge, quill)
base_dir: Optional base directory for file operations
Returns:
A Toolkit instance or None if persona_id is not recognized
"""
@@ -477,11 +549,17 @@ def get_all_available_tools() -> dict[str, dict]:
"description": "Premium frontier reasoning via xAI Grok (opt-in, Lightning-payable)",
"available_in": ["timmy"],
},
"aider": {
"name": "Aider AI Assistant",
"description": "Local AI coding assistant using Ollama (qwen2.5:14b or deepseek-coder)",
"available_in": ["forge", "timmy"],
},
}
# ── Git tools ─────────────────────────────────────────────────────────────
try:
from creative.tools.git_tools import GIT_TOOL_CATALOG
for tool_id, info in GIT_TOOL_CATALOG.items():
catalog[tool_id] = {
"name": info["name"],
@@ -494,6 +572,7 @@ def get_all_available_tools() -> dict[str, dict]:
# ── Image tools (Pixel) ───────────────────────────────────────────────────
try:
from creative.tools.image_tools import IMAGE_TOOL_CATALOG
for tool_id, info in IMAGE_TOOL_CATALOG.items():
catalog[tool_id] = {
"name": info["name"],
@@ -506,6 +585,7 @@ def get_all_available_tools() -> dict[str, dict]:
# ── Music tools (Lyra) ────────────────────────────────────────────────────
try:
from creative.tools.music_tools import MUSIC_TOOL_CATALOG
for tool_id, info in MUSIC_TOOL_CATALOG.items():
catalog[tool_id] = {
"name": info["name"],
@@ -518,6 +598,7 @@ def get_all_available_tools() -> dict[str, dict]:
# ── Video tools (Reel) ────────────────────────────────────────────────────
try:
from creative.tools.video_tools import VIDEO_TOOL_CATALOG
for tool_id, info in VIDEO_TOOL_CATALOG.items():
catalog[tool_id] = {
"name": info["name"],
@@ -530,6 +611,7 @@ def get_all_available_tools() -> dict[str, dict]:
# ── Creative pipeline (Director) ──────────────────────────────────────────
try:
from creative.director import DIRECTOR_TOOL_CATALOG
for tool_id, info in DIRECTOR_TOOL_CATALOG.items():
catalog[tool_id] = {
"name": info["name"],
@@ -542,6 +624,7 @@ def get_all_available_tools() -> dict[str, dict]:
# ── Assembler tools ───────────────────────────────────────────────────────
try:
from creative.assembler import ASSEMBLER_TOOL_CATALOG
for tool_id, info in ASSEMBLER_TOOL_CATALOG.items():
catalog[tool_id] = {
"name": info["name"],

65
tests/timmy/test_timmy_tools.py
View File

@@ -97,7 +97,17 @@ class TestGetToolStats:
class TestPersonaToolkits:
def test_all_expected_personas_present(self):
expected = {"echo", "mace", "helm", "seer", "forge", "quill", "pixel", "lyra", "reel"}
expected = {
"echo",
"mace",
"helm",
"seer",
"forge",
"quill",
"pixel",
"lyra",
"reel",
}
assert set(PERSONA_TOOLKITS.keys()) == expected
def test_get_tools_for_known_persona_raises_without_agno(self):
@@ -123,7 +133,14 @@ class TestPersonaToolkits:
class TestToolCatalog:
def test_catalog_contains_base_tools(self):
catalog = get_all_available_tools()
base_tools = {"web_search", "shell", "python", "read_file", "write_file", "list_files"}
base_tools = {
"web_search",
"shell",
"python",
"read_file",
"write_file",
"list_files",
}
for tool_id in base_tools:
assert tool_id in catalog, f"Missing base tool: {tool_id}"
@@ -137,7 +154,14 @@ class TestToolCatalog:
def test_catalog_timmy_has_all_base_tools(self):
catalog = get_all_available_tools()
base_tools = {"web_search", "shell", "python", "read_file", "write_file", "list_files"}
base_tools = {
"web_search",
"shell",
"python",
"read_file",
"write_file",
"list_files",
}
for tool_id in base_tools:
assert "timmy" in catalog[tool_id]["available_in"], (
f"Timmy missing tool: {tool_id}"
@@ -167,3 +191,38 @@ class TestToolCatalog:
# Should pick up image, music, video catalogs
all_keys = list(catalog.keys())
assert len(all_keys) > 6 # more than just base tools
def test_catalog_forge_has_aider(self):
"""Verify Aider AI tool is available in Forge's toolkit."""
catalog = get_all_available_tools()
assert "aider" in catalog
assert "forge" in catalog["aider"]["available_in"]
assert "timmy" in catalog["aider"]["available_in"]
class TestAiderTool:
"""Test the Aider AI coding assistant tool."""
def test_aider_tool_responds_to_simple_prompt(self):
"""Test Aider tool can respond to a simple prompt.
This is a smoke test - we just verify it returns something.
"""
from timmy.tools import create_aider_tool
from pathlib import Path
tool = create_aider_tool(Path.cwd())
# Call with a simple prompt - should return something (even if error)
result = tool.run_aider("what is 2+2", model="qwen2.5:14b")
# Should get a response (either success or error message)
assert result is not None
assert isinstance(result, str)
assert len(result) > 0
def test_aider_in_tool_catalog(self):
"""Verify Aider appears in the tool catalog."""
catalog = get_all_available_tools()
assert "aider" in catalog
assert "forge" in catalog["aider"]["available_in"]