diff --git a/src/timmy/tools.py b/src/timmy/tools.py index 3eb7f6e..1749c77 100644 --- a/src/timmy/tools.py +++ b/src/timmy/tools.py @@ -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"], diff --git a/tests/timmy/test_timmy_tools.py b/tests/timmy/test_timmy_tools.py index 816ad57..df4ea25 100644 --- a/tests/timmy/test_timmy_tools.py +++ b/tests/timmy/test_timmy_tools.py @@ -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"]