feat: transform CLI into interactive AI coding agent

Author: hrabanazviking

mythic_vibe_cli/ai/tools.py

@@ -4,64 +4,160 @@
 import json
 import shlex
 import contextlib
+import os
+import subprocess
+from pathlib import Path
 from typing import Any
 
 from ..protocols.mcp_tools import build_tool_catalogue
 
 
 def get_agent_tools() -> list[dict[str, Any]]:
-    """Convert MCP tools into OpenAI-compatible tool schemas for the agent."""
+    """Convert MCP tools and OS tools into OpenAI-compatible tool schemas for the agent."""
     mcp_tools = build_tool_catalogue()
     tools = []
+    
+    # OS Level Tools
+    tools.extend([
+        {
+            "type": "function",
+            "function": {
+                "name": "read_file",
+                "description": "Read the contents of a file.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "path": {"type": "string", "description": "Path to the file to read."}
+                    },
+                    "required": ["path"],
+                    "additionalProperties": False,
+                }
+            }
+        },
+        {
+            "type": "function",
+            "function": {
+                "name": "write_file",
+                "description": "Write or overwrite the contents of a file.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "path": {"type": "string", "description": "Path to the file."},
+                        "content": {"type": "string", "description": "Content to write."}
+                    },
+                    "required": ["path", "content"],
+                    "additionalProperties": False,
+                }
+            }
+        },
+        {
+            "type": "function",
+            "function": {
+                "name": "run_command",
+                "description": "Run a shell command in the project directory.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "command": {"type": "string", "description": "The shell command to execute."}
+                    },
+                    "required": ["command"],
+                    "additionalProperties": False,
+                }
+            }
+        },
+        {
+            "type": "function",
+            "function": {
+                "name": "list_dir",
+                "description": "List the contents of a directory.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "path": {"type": "string", "description": "Path to the directory."}
+                    },
+                    "required": ["path"],
+                    "additionalProperties": False,
+                }
+            }
+        }
+    ])
+
     for mcp_tool in mcp_tools:
-        # Some providers prefer parameters to be complete JSON schema
         parameters = dict(mcp_tool.input_schema)
-        # Ensure additionalProperties is False if required for strict structured output
-        # But we leave it as is from mcp_tools.py which already sets it.
         tools.append({
             "type": "function",
             "function": {
-                "name": mcp_tool.name.replace(".", "_"),  # e.g. mythic_vibe.memory_search -> mythic_vibe_memory_search
+                "name": mcp_tool.name.replace(".", "_"),  
                 "description": mcp_tool.description,
                 "parameters": parameters,
             }
         })
     return tools
 
 
-def execute_tool(name: str, arguments: dict[str, Any]) -> str:
+def execute_tool(name: str, arguments: dict[str, Any], project_root: Path | None = None) -> str:
     """Execute a tool request and return its stdout/stderr as a string."""
+    
+    root_path = project_root if project_root is not None else Path.cwd()
+    
+    if name == "read_file":
+        path = root_path / arguments.get("path", "")
+        try:
+            return path.read_text(encoding="utf-8")
+        except Exception as exc:
+            return f"Failed to read file: {exc}"
+            
+    if name == "write_file":
+        path = root_path / arguments.get("path", "")
+        try:
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(arguments.get("content", ""), encoding="utf-8")
+            return f"Successfully wrote to {path}"
+        except Exception as exc:
+            return f"Failed to write file: {exc}"
+            
+    if name == "run_command":
+        command = arguments.get("command", "")
+        try:
+            result = subprocess.run(
+                command, shell=True, cwd=str(root_path), 
+                stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True
+            )
+            return result.stdout or f"Command executed with exit code {result.returncode}"
+        except Exception as exc:
+            return f"Command failed: {exc}"
+            
+    if name == "list_dir":
+        path = root_path / arguments.get("path", ".")
+        try:
+            items = os.listdir(path)
+            return "\\n".join(sorted(items))
+        except Exception as exc:
+            return f"Failed to list directory: {exc}"
+
     from ..app import main
 
-    # Tools are prefixed with mythic_vibe_
     if not name.startswith("mythic_vibe_"):
         return f"Error: Unknown tool prefix {name}"
 
-    # Extract the actual command name
     command_name = name[len("mythic_vibe_"):]
-    
-    # argv array from the tool arguments
     tool_argv = arguments.get("argv", [])
     if not isinstance(tool_argv, list):
         return f"Error: argv must be a list, got {type(tool_argv)}"
 
-    # Ensure they are strings
     tool_argv = [str(arg) for arg in tool_argv]
-
     full_argv = [command_name] + tool_argv
 
-    # Capture output
     stdout = io.StringIO()
     stderr = io.StringIO()
 
     try:
         with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
             exit_code = main(full_argv)
     except SystemExit as exc:
-        # Argparse might call sys.exit, we catch it so the shell doesn't die.
         exit_code = exc.code if exc.code is not None else 0
     except Exception as exc:
-        return f"Tool execution crashed: {exc}\n{stderr.getvalue()}"
+        return f"Tool execution crashed: {exc}\\n{stderr.getvalue()}"
 
     output = stdout.getvalue()
     err_output = stderr.getvalue()
@@ -70,7 +166,7 @@ def execute_tool(name: str, arguments: dict[str, Any]) -> str:
     if output:
         result.append(output.strip())
     if err_output:
-        result.append(f"STDERR:\n{err_output.strip()}")
+        result.append(f"STDERR:\\n{err_output.strip()}")
     result.append(f"Exit code: {exit_code}")
 
-    return "\n".join(result)
+    return "\\n".join(result)

mythic_vibe_cli/repl.py

@@ -330,11 +330,7 @@ def _handle_test_command(
 def _looks_like_command(stripped: str, main_commands: set[str]) -> bool:
     if stripped.startswith("/"):
         return True
-    try:
-        argv = shlex.split(stripped)
-    except ValueError:
-        return True
-    return bool(argv and argv[0] in main_commands)
+    return False
 
 
 def _known_command_names(_main: Callable[[list[str]], int]) -> set[str]:
@@ -522,7 +518,7 @@ def _answer_with_selected_model(prompt: str, stdout: IO[str], context: ShellCont
 
                 print(f"     [Executing {tool_name}...]", file=stdout)
                 try:
-                    tool_output = execute_tool(tool_name, args)
+                    tool_output = execute_tool(tool_name, args, context.project_root)
                 except Exception as exc:
                     tool_output = f"Tool execution error: {exc}"
 
@@ -626,10 +622,27 @@ def run_shell(
     last_test_result: dict[str, str] = {}
     _print_banner(out_stream, shell_context)
 
+    use_prompt_toolkit = False
+    session = None
+    if hasattr(in_stream, "isatty") and in_stream.isatty():
+        try:
+            from prompt_toolkit import PromptSession
+            from prompt_toolkit.history import InMemoryHistory
+            session = PromptSession(history=InMemoryHistory())
+            use_prompt_toolkit = True
+        except ImportError:
+            pass
+
     while True:
-        print(PROMPT, end="", file=out_stream, flush=True)
         try:
-            line = in_stream.readline()
+            if use_prompt_toolkit and session is not None:
+                try:
+                    line = session.prompt(PROMPT) + "\n"
+                except EOFError:
+                    line = ""
+            else:
+                print(PROMPT, end="", file=out_stream, flush=True)
+                line = in_stream.readline()
         except KeyboardInterrupt:
             print("(interrupted; type /quit to exit)", file=out_stream)
             continue

pyproject.toml

@@ -70,7 +70,8 @@ docs = [
   "pymdown-extensions>=10.0"
 ]
 ux = [
-  "rich>=13.0"
+  "rich>=13.0",
+  "prompt_toolkit>=3.0"
 ]
 tui = [
   "textual>=0.80"
@@ -105,6 +106,7 @@ dev = [
   "mkdocs-material>=9.5",
   "mypy>=1.10",
   "openai>=1.40",
+  "prompt_toolkit>=3.0",
   "pymdown-extensions>=10.0",
   "pytest>=8.0",
   "pytest-cov>=5.0",

tests/test_repl.py

@@ -126,12 +126,12 @@ def test_real_command_dispatches_through_main(self) -> None:
         self.assertEqual(len(fake.calls), 1)
         self.assertEqual(fake.calls[0], ["scan", "--path", "."])
 
-    def test_bare_command_dispatches_through_main(self) -> None:
-        """A line without a leading slash is also dispatched."""
+    def test_bare_command_does_not_dispatch_through_main(self) -> None:
+        """A line without a leading slash is NOT dispatched to main."""
         fake = _FakeMain()
         code, _out, _err = self._drive(["status --json\n", "/quit\n"], main=fake)
         self.assertEqual(code, SUCCESS)
-        self.assertEqual(fake.calls, [["status", "--json"]])
+        self.assertEqual(fake.calls, [])
 
     def test_non_zero_exit_code_is_surfaced_but_loop_continues(self) -> None:
         fake = _FakeMain(return_codes=[USER_INPUT_ERROR])