Add basic terminal command use

Adds a tool that enables Sodalis to use basic Linux commands in the terminal and include the results in responses to the user.

Author: Joshua92500

config/companions/default.yaml

@@ -2,3 +2,9 @@ name: Default
 model: qwen3:8b
 system_prompt: system_prompt.txt
 temperature: 0.7
+
+tools:
+  enabled: true
+  timeout: 10
+  max_output: 4096
+  max_rounds: 3

config/prompts/system_prompt.txt

@@ -1 +1,2 @@
-You are Sodalis, a local AI companion running on a Linux machine. Be practical, concise, and clear.
+You are a local AI companion running on a Linux machine. Be practical, concise, and clear.
+You can run read-only Linux commands when needed to answer questions. Use the tool format described below when you need to inspect the system.

src/sodalis/cli.py

@@ -2,15 +2,18 @@
 
 from sodalis.config import load_companion_config
 from sodalis.paths import ensure_runtime_dirs, get_prompts_dir
-from sodalis.llm import generate_response
+from sodalis.llm import generate_response, generate_with_tools
 from sodalis.db import init_db
 from sodalis.history import create_chat, delete_chat, list_chats, chat_exists, load_messages, save_message
 from sodalis.prompt_builder import build_prompt, load_prompt_layers
+from sodalis.tools import TOOL_SCHEMA_TEXT
 
 
-def run_chat(companion, system_prompt, companion_prompt, chat_id = None, resume_chat_id = None):
+def run_chat(companion, system_prompt, companion_prompt, chat_id = None, resume_chat_id = None, use_tools = True):
     print(f"Sodalis {companion['name']} ({companion['model']})")
 
+    tools_cfg = companion.get("tools", {})
+    
     history = []
     chat_id = None
 
@@ -45,7 +48,17 @@ def run_chat(companion, system_prompt, companion_prompt, chat_id = None, resume_
 
             prompt = build_prompt(system_prompt, companion_prompt, history, user_input)
 
-            response = generate_response(
+            if use_tools and tools_cfg.get("enabled", True):
+                response = generate_with_tools(
+                    prompt,
+                    companion["model"],
+                    companion["temperature"],
+                    max_rounds=tools_cfg.get("max_rounds", 3),
+                    timeout=tools_cfg.get("timeout", 10),
+                    max_output=tools_cfg.get("max_output", 4096),
+                )
+            else:
+                response = generate_response(
                     prompt, 
                     companion["model"], 
                     companion["temperature"]
@@ -63,15 +76,26 @@ def run_chat(companion, system_prompt, companion_prompt, chat_id = None, resume_
             break
 
 
-def run_once(companion, system_prompt, message):
+def run_once(companion, system_prompt, message, use_tools=True):
 
+    tools_cfg = companion.get("tools", {})
     full_prompt = f"{system_prompt}\n\nUSER: {message}\nASSISTANT:"
 
-    response = generate_response(
-        prompt=full_prompt,
-        model=companion["model"],
-        temperature=companion["temperature"],
-        )   
+    if use_tools and tools_cfg.get("enabled", True):
+        response = generate_with_tools(
+            prompt=full_prompt,
+            model=companion["model"],
+            temperature=companion["temperature"],
+            max_rounds=tools_cfg.get("max_rounds", 3),
+            timeout=tools_cfg.get("timeout", 10),
+            max_output=tools_cfg.get("max_output", 4096),
+        )
+    else:
+        response = generate_response(
+            prompt=full_prompt,
+            model=companion["model"],
+            temperature=companion["temperature"],
+        )
 
     print(response)
 
@@ -142,13 +166,23 @@ def main() -> None:
             metavar="CHAT_ID",
             help="Delete a saved chat by ID",
             )
+    parser.add_argument(
+            "--no-tools",
+            action="store_true",
+            help="Disable tool use for this invocation",
+            )
 
     args = parser.parse_args()
     ensure_runtime_dirs()
 
     companion = load_companion_config(args.companion)
     companion_name = companion.get("name", args.companion)
     system_prompt, companion_prompt = load_prompt_layers(companion)
+
+    use_tools = not args.no_tools
+    tools_cfg = companion.get("tools", {})
+    if use_tools and tools_cfg.get("enabled", True):
+        system_prompt = system_prompt + "\n\n" + TOOL_SCHEMA_TEXT
 
     if args.history:
         chats = list_chats()
@@ -171,9 +205,9 @@ def main() -> None:
 
     if args.chat is not None:
         if args.chat == "new":
-            run_chat(companion, system_prompt, companion_prompt)
+            run_chat(companion, system_prompt, companion_prompt, use_tools=use_tools)
         else:
-            run_chat(companion, system_prompt, companion_prompt, resume_chat_id=int(args.chat))
+            run_chat(companion, system_prompt, companion_prompt, resume_chat_id=int(args.chat), use_tools=use_tools)
 
 
 if __name__ == "__main__":

src/sodalis/llm.py

@@ -1,5 +1,7 @@
 import requests
 
+from sodalis.tools import parse_tool_call, run_command, ToolError
+
 
 def generate_response(
         prompt: str,
@@ -23,3 +25,34 @@ def generate_response(
         data = response.json()
 
         return data.get("response", "").strip()
+
+
+def generate_with_tools(
+        prompt: str,
+        model: str,
+        temperature: float,
+        max_rounds: int = 3,
+        timeout: int = 10,
+        max_output: int = 4096,
+        ) -> str:
+    response_text = generate_response(prompt, model, temperature)
+
+    for _ in range(max_rounds):
+        tool_call = parse_tool_call(response_text)
+        if tool_call is None:
+            break
+
+        try:
+            output = run_command(
+                tool_call["command"],
+                tool_call["args"],
+                timeout=timeout,
+                max_output=max_output,
+            )
+        except ToolError as e:
+            output = f"Error: {e}"
+
+        followup = prompt + f"\n\n[ASSISTANT]\n{response_text}\n\n[TOOL OUTPUT]\n{output}\n\n[ASSISTANT]"
+        response_text = generate_response(followup, model, temperature)
+
+    return response_text

src/sodalis/tools.py

@@ -0,0 +1,130 @@
+import json
+import re
+import shutil
+import subprocess
+from pathlib import Path
+
+
+class ToolError(Exception):
+    pass
+
+
+# Resolve allowed commands at import time to prevent PATH hijacking
+_COMMAND_NAMES = [
+    "cat", "grep", "echo", "ls", "pwd", "head", "tail", "wc",
+    "find", "stat", "file", "du", "df", "uname", "whoami", "id",
+    "date", "printenv", "env",
+]
+
+ALLOWED_COMMANDS: dict[str, Path | None] = {}
+for _cmd in _COMMAND_NAMES:
+    _resolved = shutil.which(_cmd)
+    if _resolved:
+        ALLOWED_COMMANDS[_cmd] = Path(_resolved)
+
+# Commands that must never appear, even as arguments
+_BLOCKED_TOKENS = {"sudo", "su", "doas"}
+
+_MAX_ARG_LEN = 1024
+
+_TOOL_CALL_RE = re.compile(
+    r"```tool\s*\n(\{.*?\})\s*\n```",
+    re.DOTALL,
+)
+
+
+TOOL_SCHEMA_TEXT = """\
+You have access to read-only Linux commands. To run a command, output a fenced \
+JSON block tagged "tool" with the keys "command" (string) and "args" (list of strings).
+
+Example:
+```tool
+{"command": "ls", "args": ["-la", "/tmp"]}
+```
+
+Available commands: """ + ", ".join(sorted(ALLOWED_COMMANDS.keys())) + """
+
+Rules:
+- Only one tool call per response.
+- Do NOT wrap the tool block inside another code block.
+- If you don't need a command, just reply normally with no tool block.
+- The tool output will be provided back to you so you can give a final answer.
+"""
+
+
+def _validate_args(args: list[str]) -> None:
+    for arg in args:
+        if "\x00" in arg:
+            raise ToolError("Null bytes are not allowed in arguments")
+        if len(arg) > _MAX_ARG_LEN:
+            raise ToolError(f"Argument exceeds maximum length of {_MAX_ARG_LEN}")
+        lower = arg.lower()
+        for blocked in _BLOCKED_TOKENS:
+            if lower == blocked:
+                raise ToolError(f"Blocked token in arguments: {blocked}")
+
+
+def run_command(
+    name: str,
+    args: list[str],
+    timeout: int = 10,
+    max_output: int = 4096,
+) -> str:
+    if name not in ALLOWED_COMMANDS:
+        raise ToolError(f"Command not allowed: {name}")
+
+    resolved = ALLOWED_COMMANDS[name]
+    if resolved is None:
+        raise ToolError(f"Command not found on system: {name}")
+
+    _validate_args(args)
+
+    cmd = [str(resolved)] + args
+
+    try:
+        result = subprocess.run(
+            cmd,
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+            shell=False,
+        )
+    except subprocess.TimeoutExpired:
+        raise ToolError(f"Command timed out after {timeout}s")
+
+    output = result.stdout + result.stderr
+
+    if result.returncode != 0:
+        raise ToolError(
+            f"Command exited with code {result.returncode}:\n"
+            + output[:max_output]
+        )
+
+    if len(output) > max_output:
+        output = output[:max_output] + "\n... (output truncated)"
+
+    return output
+
+
+def parse_tool_call(response: str) -> dict | None:
+    match = _TOOL_CALL_RE.search(response)
+    if not match:
+        return None
+
+    try:
+        data = json.loads(match.group(1))
+    except json.JSONDecodeError:
+        return None
+
+    if not isinstance(data, dict):
+        return None
+    if "command" not in data or not isinstance(data["command"], str):
+        return None
+    if "args" not in data:
+        data["args"] = []
+    if not isinstance(data["args"], list):
+        return None
+    if not all(isinstance(a, str) for a in data["args"]):
+        return None
+
+    return data