feat: Add Morph codebase search integration

sample.config.toml

@@ -16,6 +16,7 @@ GEMINI = "<YOUR_GEMINI_API_KEY>"
 MISTRAL = "<YOUR_MISTRAL_API_KEY>"
 GROQ = "<YOUR_GROQ_API_KEY>"
 NETLIFY = "<YOUR_NETLIFY_API_KEY>"
+MORPH = "<YOUR_MORPH_API_KEY>"
 
 [API_ENDPOINTS]
 BING = "https://api.bing.microsoft.com/v7.0/search"

src/agents/agent.py

@@ -190,14 +190,22 @@ def subsequent_execute(self, prompt: str, project_name: str):
         self.agent_state.set_agent_active(project_name, True)
 
         conversation = self.project_manager.get_all_messages_formatted(project_name)
-        code_markdown = ReadCode(project_name).code_set_to_markdown()
 
         response, action = self.action.execute(conversation, project_name)
 
         self.project_manager.add_message_from_devika(project_name, response)
 
         print("\naction :: ", action, '\n')
 
+        read_code = ReadCode(project_name)
+        if action in ("answer", "feature", "bug"):
+            emit_agent("info", {"type": "info", "message": f"Searching codebase for: {prompt}"})
+            code_markdown = read_code.search_code(prompt)
+        elif action in ("run", "report"):
+            code_markdown = read_code.code_set_to_markdown()
+        else:
+            code_markdown = ""
+
         if action == "answer":
             response = self.answer.execute(
                 conversation=conversation,

src/config.py

@@ -84,6 +84,9 @@ def get_groq_api_key(self):
     def get_netlify_api_key(self):
         return self.config["API_KEYS"]["NETLIFY"]
 
+    def get_morph_api_key(self):
+        return self.config["API_KEYS"]["MORPH"]
+
     def get_sqlite_db(self):
         return self.config["STORAGE"]["SQLITE_DB"]
 
@@ -167,6 +170,10 @@ def set_netlify_api_key(self, key):
         self.config["API_KEYS"]["NETLIFY"] = key
         self.save_config()
 
+    def set_morph_api_key(self, key):
+        self.config["API_KEYS"]["MORPH"] = key
+        self.save_config()
+
     def set_logging_rest_api(self, value):
         self.config["LOGGING"]["LOG_REST_API"] = "true" if value else "false"
         self.save_config()

src/filesystem/read_code.py

@@ -1,6 +1,7 @@
 import os
 
 from src.config import Config
+from src.services import Morph
 
 """
 TODO: Replace this with `code2prompt` - https://github.com/mufeedvh/code2prompt
@@ -33,3 +34,18 @@ def code_set_to_markdown(self):
             markdown += f"```\n{code['code']}\n```\n\n"
             markdown += "---\n\n"
         return markdown
+
+    def search_code(self, query: str) -> str:
+        try:
+            results = Morph().search(self.directory_path, query)
+            if not results:
+                return self.code_set_to_markdown()
+        except Exception:
+            return self.code_set_to_markdown()
+
+        markdown = ""
+        for code in results:
+            markdown += f"### {code['filename']}:\n\n"
+            markdown += f"```\n{code['code']}\n```\n\n"
+            markdown += "---\n\n"
+        return markdown

src/services/__init__.py

@@ -1,3 +1,4 @@
 from .git import Git
 from .github import GitHub
-from .netlify import Netlify
+from .netlify import Netlify
+from .morph import Morph

src/services/morph.py

@@ -0,0 +1,171 @@
+import os
+import re
+import shutil
+import subprocess
+import xml.etree.ElementTree as ET
+from typing import List, Dict
+
+import requests
+
+from src.config import Config
+
+SKIP_DIRS = {".git", "node_modules", "__pycache__", ".venv"}
+SKIP_EXTS = {".pyc", ".so", ".dll", ".exe", ".png", ".jpg", ".gif", ".pdf", ".zip"}
+
+PROMPT = open("src/services/morph_prompt.txt", "r").read().strip()
+
+
+class Morph:
+    def __init__(self):
+        self.api_key = Config().get_morph_api_key()
+
+    def search(self, project_path: str, query: str) -> List[Dict]:
+        if not self.api_key:
+            return []
+        tree = Morph._file_tree(project_path)
+        if not tree:
+            return []
+        messages = [
+            {"role": "system", "content": PROMPT},
+            {"role": "user", "content": f"<repo_structure>\n{tree}\n</repo_structure>\n\n<search_string>\n{query}\n</search_string>"},
+        ]
+        for _ in range(4):
+            response = requests.post(
+                "https://api.morphllm.com/v1/chat/completions",
+                headers={"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"},
+                json={"model": "morph-warp-grep-v2", "messages": messages}, timeout=60,
+            )
+            response.raise_for_status()
+            text = response.json()["choices"][0]["message"]["content"]
+            messages.append({"role": "assistant", "content": text})
+            tool_calls = Morph._parse_tool_calls(text)
+            if not tool_calls:
+                break
+            tool_responses = []
+            for call in tool_calls:
+                name = call.get("tool_name", "")
+                if name == "finish":
+                    paths = [line.strip() for line in call.get("relevant_files", "").strip().splitlines() if line.strip()]
+                    return Morph._read_full_files(project_path, paths)
+                if name == "ripgrep":
+                    result = Morph._run_ripgrep(call.get("pattern", ""), project_path, call.get("glob", ""))
+                elif name == "read":
+                    valid_path = Morph._validate_path(call.get("path", ""), project_path)
+                    result = Morph._read_file(valid_path, call.get("line_ranges", "")) if valid_path else "Error: path outside project root"
+                elif name == "list_directory":
+                    valid_path = Morph._validate_path(call.get("path", ""), project_path)
+                    result = Morph._list_dir(valid_path) if valid_path else "Error: path outside project root"
+                else:
+                    continue
+                tool_responses.append(f"<tool_response>\n<tool_name>{name}</tool_name>\n<result>\n{result}\n</result>\n</tool_response>")
+            if tool_responses:
+                messages.append({"role": "user", "content": "\n".join(tool_responses)})
+        return []
+
+    @staticmethod
+    def _file_tree(project_path):
+        if not os.path.isdir(project_path):
+            return ""
+        lines = []
+        for root, dirs, files in os.walk(project_path):
+            dirs[:] = sorted(d for d in dirs if d not in SKIP_DIRS)
+            rel = os.path.relpath(root, project_path)
+            depth = 0 if rel == "." else rel.count(os.sep) + 1
+            if depth > 6:
+                dirs.clear()
+                continue
+            if rel != ".":
+                lines.append(f"{'  ' * depth}{os.path.basename(root)}/")
+            for file in sorted(files):
+                if not any(file.endswith(ext) for ext in SKIP_EXTS):
+                    lines.append(f"{'  ' * (depth + 1)}{file}")
+        return "\n".join(lines[:500])
+
+    @staticmethod
+    def _run_ripgrep(pattern, path, glob=""):
+        rg = shutil.which("rg")
+        if not rg:
+            return "(rg not found)"
+        cmd = [rg, "--max-count", "20", "--line-number", "--no-heading"]
+        if glob:
+            cmd.extend(["--glob", glob])
+        cmd.extend([pattern, path])
+        try:
+            out = subprocess.run(cmd, capture_output=True, text=True, timeout=15).stdout
+            return (out[:8000] + "\n... (truncated)") if len(out) > 8000 else (out or "(no matches)")
+        except Exception:
+            return "(rg failed)"
+
+    @staticmethod
+    def _read_file(path, line_ranges=""):
+        try:
+            with open(path, "r", errors="ignore") as f:
+                lines = f.readlines()
+        except Exception as exc:
+            return f"Error reading file: {exc}"
+        if not line_ranges:
+            content = "".join(lines[:500])
+            return content + f"\n... ({len(lines)} total lines, showing first 500)" if len(lines) > 500 else content
+        out = []
+        for spec in line_ranges.split(","):
+            spec = spec.strip()
+            try:
+                if "-" in spec:
+                    start, end = spec.split("-", 1)
+                    for i in range(max(1, int(start)) - 1, min(len(lines), int(end))):
+                        out.append(f"{i + 1}: {lines[i].rstrip()}")
+                else:
+                    line_num = int(spec)
+                    if 1 <= line_num <= len(lines):
+                        out.append(f"{line_num}: {lines[line_num - 1].rstrip()}")
+            except ValueError:
+                out.append(f"Error: invalid range {spec}")
+        return "\n".join(out)
+
+    @staticmethod
+    def _list_dir(path):
+        try:
+            entries = sorted(os.listdir(path))
+        except Exception as exc:
+            return f"Error listing directory: {exc}"
+        return "\n".join(f"{entry}/" if os.path.isdir(os.path.join(path, entry)) else entry for entry in entries if entry not in SKIP_DIRS)
+
+    @staticmethod
+    def _parse_tool_calls(text):
+        calls = []
+        for block in re.findall(r"<tool_call>(.*?)</tool_call>", text, re.DOTALL):
+            call = {}
+            try:
+                root = ET.fromstring(f"<root>{block}</root>")
+                for child in root:
+                    call[child.tag] = (child.text or "").strip()
+            except ET.ParseError:
+                for tag in ("tool_name", "pattern", "glob", "path", "line_ranges", "relevant_files"):
+                    m = re.search(rf"<{tag}>(.*?)</{tag}>", block, re.DOTALL)
+                    if m:
+                        call[tag] = m.group(1).strip()
+            if call.get("tool_name"):
+                calls.append(call)
+        return calls
+
+    @staticmethod
+    def _validate_path(path, project_root):
+        if not path:
+            return ""
+        full = os.path.join(project_root, path) if not os.path.isabs(path) else path
+        real, rroot = os.path.realpath(full), os.path.realpath(project_root)
+        return real if real == rroot or real.startswith(rroot + os.sep) else ""
+
+    @staticmethod
+    def _read_full_files(project_path, file_paths):
+        results = []
+        for file_path in file_paths:
+            valid_path = Morph._validate_path(file_path, project_path)
+            if not valid_path:
+                continue
+            try:
+                with open(valid_path, "r", errors="ignore") as f:
+                    results.append({"filename": valid_path, "code": f.read()})
+            except Exception:
+                continue
+        return results

src/services/morph_prompt.txt

@@ -0,0 +1,32 @@
+You are a code search agent. You have access to tools to search a codebase.
+Your goal is to find the most relevant code files for the user's query.
+
+You have these tools available (use XML tool_call blocks):
+
+<tool_call>
+<tool_name>ripgrep</tool_name>
+<pattern>REGEX_PATTERN</pattern>
+<glob>OPTIONAL_GLOB_PATTERN</glob>
+</tool_call>
+
+<tool_call>
+<tool_name>read</tool_name>
+<path>FILE_PATH</path>
+<line_ranges>START-END,START-END</line_ranges>
+</tool_call>
+
+<tool_call>
+<tool_name>list_directory</tool_name>
+<path>DIRECTORY_PATH</path>
+</tool_call>
+
+When you have found all relevant files, use the finish tool with the paths:
+<tool_call>
+<tool_name>finish</tool_name>
+<relevant_files>
+path/to/file1.py
+path/to/file2.js
+</relevant_files>
+</tool_call>
+
+Search strategically - use ripgrep to find patterns, read files to verify relevance, and finish when you have identified the key files.

ui/src/routes/settings/+page.svelte

@@ -52,7 +52,8 @@
       "GEMINI": settings["API_KEYS"]["GEMINI"],
       "MISTRAL": settings["API_KEYS"]["MISTRAL"],
       "GROQ": settings["API_KEYS"]["GROQ"],
-      "NETLIFY": settings["API_KEYS"]["NETLIFY"]
+      "NETLIFY": settings["API_KEYS"]["NETLIFY"],
+      "MORPH": settings["API_KEYS"]["MORPH"]
     };
     // make a copy of the original settings
     original = JSON.parse(JSON.stringify(settings));