feat(codebase-tools): add codebase exploration and file search tools

agentic-framework/src/agentic_framework/cli.py

@@ -14,7 +14,7 @@
 
 load_dotenv()
 
-RUN_TIMEOUT_SECONDS = 90
+RUN_TIMEOUT_SECONDS = 600
 
 app = typer.Typer(
     name="agentic-framework",

agentic-framework/src/agentic_framework/core/developer_agent.py

@@ -0,0 +1,70 @@
+from typing import Any, Sequence
+
+from langchain_core.tools import StructuredTool
+
+from agentic_framework.constants import BASE_DIR
+from agentic_framework.core.langgraph_agent import LangGraphMCPAgent
+from agentic_framework.registry import AgentRegistry
+from agentic_framework.tools import (
+    CodeSearcher,
+    FileFragmentReaderTool,
+    FileOutlinerTool,
+    StructureExplorerTool,
+)
+
+
+@AgentRegistry.register("developer", mcp_servers=["webfetch"])
+class DeveloperAgent(LangGraphMCPAgent):
+    """
+    A specialized agent for codebase exploration and development.
+    Equipped with tools to search, structure, and read code, plus MCP capabilities.
+    """
+
+    @property
+    def system_prompt(self) -> str:
+        return """You are a Principal Software Engineer assistant.
+        Your goal is to help the user understand and maintain their codebase.
+        You have access to several specialized tools for:
+        1. Discovering the project structure.
+        2. Extracting class and function outlines from Python files.
+        3. Reading specific fragments of a file.
+        4. Searching the codebase for patterns using ripgrep.
+
+        When asked about the project structure, start with `discover_structure`.
+        When asked to explain a file, start with `get_file_outline` to get an overview.
+        Use `read_file_fragment` to read specific lines if you need more detail.
+        Use `code_search` for fast global pattern matching.
+
+        Always provide clear, concise explanations and suggest improvements when relevant.
+        You also have access to MCP tools like `webfetch` if you need to fetch information from the web.
+        """
+
+    def local_tools(self) -> Sequence[Any]:
+        # Initialize tool instances with project root
+        searcher = CodeSearcher(str(BASE_DIR))
+        explorer = StructureExplorerTool(str(BASE_DIR))
+        outliner = FileOutlinerTool(str(BASE_DIR))
+        reader = FileFragmentReaderTool(str(BASE_DIR))
+
+        return [
+            StructuredTool.from_function(
+                func=searcher.invoke,
+                name=searcher.name,
+                description=searcher.description,
+            ),
+            StructuredTool.from_function(
+                func=explorer.invoke,
+                name=explorer.name,
+                description=explorer.description,
+            ),
+            StructuredTool.from_function(
+                func=outliner.invoke,
+                name=outliner.name,
+                description=outliner.description,
+            ),
+            StructuredTool.from_function(
+                func=reader.invoke,
+                name=reader.name,
+                description=reader.description,
+            ),
+        ]

agentic-framework/src/agentic_framework/interfaces/base.py

@@ -25,6 +25,16 @@ def get_tools(self) -> List[Any]:
 class Tool(ABC):
     """Abstract Base Class for a Tool."""
 
+    @property
+    @abstractmethod
+    def name(self) -> str:
+        """The name of the tool."""
+
+    @property
+    @abstractmethod
+    def description(self) -> str:
+        """A description of what the tool does."""
+
     @abstractmethod
     def invoke(self, input_str: str) -> Any:
         """Execute the tool logic."""

agentic-framework/src/agentic_framework/mcp/config.py

@@ -29,6 +29,22 @@
 }
 
 
+def get_mcp_servers_config(
+    override: Dict[str, Dict[str, Any]] | None = None,
+) -> Dict[str, Dict[str, Any]]:
+    """Return MCP server config for MultiServerMCPClient.
+
+    Merges DEFAULT_MCP_SERVERS with optional override, then resolves
+    env-dependent values (e.g. TAVILY_API_KEY). Does not mutate any shared state.
+    """
+    base = {k: dict(v) for k, v in DEFAULT_MCP_SERVERS.items()}
+    if override:
+        for k, v in override.items():
+            base[k] = dict(base.get(k, {}))
+            base[k].update(v)
+    return {k: _resolve_server_config(k, v) for k, v in base.items()}
+
+
 def _resolve_server_config(server_name: str, raw: Dict[str, Any]) -> Dict[str, Any]:
     """Return a copy of server config with env-dependent values resolved."""
     import logging
@@ -50,19 +66,3 @@ def _resolve_server_config(server_name: str, raw: Dict[str, Any]) -> Dict[str, A
             out["headers"] = out.get("headers", {})
             out["headers"]["X-API-Key"] = key
     return out
-
-
-def get_mcp_servers_config(
-    override: Dict[str, Dict[str, Any]] | None = None,
-) -> Dict[str, Dict[str, Any]]:
-    """Return MCP server config for MultiServerMCPClient.
-
-    Merges DEFAULT_MCP_SERVERS with optional override, then resolves
-    env-dependent values (e.g. TAVILY_API_KEY). Does not mutate any shared state.
-    """
-    base = {k: dict(v) for k, v in DEFAULT_MCP_SERVERS.items()}
-    if override:
-        for k, v in override.items():
-            base[k] = dict(base.get(k, {}))
-            base[k].update(v)
-    return {k: _resolve_server_config(k, v) for k, v in base.items()}

agentic-framework/src/agentic_framework/mcp/provider.py

@@ -69,7 +69,7 @@ async def tool_session(self, fail_fast: bool = True):
         """
         Async context manager: load MCP tools with sessions that close on exit.
 
-        IMPORTANT: Connections are now opened sequentially. While slightly slower than
+        IMPORTANT: Connections are opened sequentially. While slightly slower than
         parallel, this avoids "Attempted to exit cancel scope in a different task"
         errors from anyio (used by mcp) which requires task identity for cleanup.
         """

agentic-framework/src/agentic_framework/tools/__init__.py

@@ -1,4 +1,18 @@
+from .code_searcher import CodeSearcher
+from .codebase_explorer import (
+    FileFragmentReaderTool,
+    FileOutlinerTool,
+    StructureExplorerTool,
+)
 from .example import CalculatorTool, WeatherTool
 from .web_search import WebSearchTool
 
-__all__ = ["CalculatorTool", "WeatherTool", "WebSearchTool"]
+__all__ = [
+    "CalculatorTool",
+    "WeatherTool",
+    "WebSearchTool",
+    "CodeSearcher",
+    "StructureExplorerTool",
+    "FileOutlinerTool",
+    "FileFragmentReaderTool",
+]

agentic-framework/src/agentic_framework/tools/code_searcher.py

@@ -0,0 +1,41 @@
+import subprocess
+from typing import Any, Dict, List, Union
+
+from agentic_framework.interfaces.base import Tool
+
+
+class CodeSearcher(Tool):
+    """Wraps ripgrep (rg) for ultra-fast codebase querying."""
+
+    def __init__(self, root_dir: str):
+        self.root_dir = root_dir
+
+    @property
+    def name(self) -> str:
+        return "code_search"
+
+    @property
+    def description(self) -> str:
+        return "Searches the codebase for a given pattern using ripgrep. Returns top 20 matches."
+
+    def invoke(self, input_str: str) -> Any:
+        """Executes a search across the codebase. input_str is the search query."""
+        return self.grep_search(input_str)
+
+    def grep_search(self, query: str, file_type: str = "py") -> Union[List[str], List[Dict[str, Any]]]:
+        """
+        Executes a search across the codebase.
+        Returns matches with line numbers and snippets.
+        """
+        try:
+            # -n: line numbers, -C 1: context, -t: type
+            cmd = ["rg", "--json", "-t", file_type, query, self.root_dir]
+            result = subprocess.run(cmd, capture_output=True, text=True, check=False)
+
+            # Note: In a production framework, you'd parse the JSON stream from rg
+            # For this MVP, we return a simplified string representation
+            if result.returncode != 0:
+                return []
+            return result.stdout.splitlines()[:20]  # Limit to top 20 for token safety
+        except FileNotFoundError:
+            return ["Error: ripgrep (rg) is not installed on the host system."]

agentic-framework/src/agentic_framework/tools/codebase_explorer.py

@@ -0,0 +1,136 @@
+import re
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Union
+
+from agentic_framework.interfaces.base import Tool
+
+
+class CodebaseExplorer:
+    """
+    A utility for agents to discover and navigate codebases.
+    Base class for specific codebase tools.
+    """
+
+    def __init__(self, root_dir: Union[str, Path], ignore_patterns: Optional[List[str]] = None):
+        self.root_dir = Path(root_dir).resolve()
+        # Standard noise reduction for agentic context
+        self.ignore_patterns = ignore_patterns or [
+            r"\.git",
+            r"__pycache__",
+            r"node_modules",
+            r"\.venv",
+            r"dist",
+            r"build",
+            r"\.mypy_cache",
+            r"\.pytest_cache",
+        ]
+
+    def _is_ignored(self, path: Path) -> bool:
+        try:
+            rel_path = str(path.relative_to(self.root_dir))
+        except ValueError:
+            return True
+        return any(re.search(pattern, rel_path) for pattern in self.ignore_patterns)
+
+
+class StructureExplorerTool(CodebaseExplorer, Tool):
+    """Tool to discover the directory structure of the project."""
+
+    @property
+    def name(self) -> str:
+        return "discover_structure"
+
+    @property
+    def description(self) -> str:
+        return "Lists files and directories recursively up to a certain depth. Helps understand project layout."
+
+    def invoke(self, input_str: str) -> Any:
+        # Input can be max_depth as string, defaults to 3
+        try:
+            max_depth = int(input_str) if input_str.isdigit() else 3
+        except Exception:
+            max_depth = 3
+        return self._build_tree(self.root_dir, depth=0, max_depth=max_depth)
+
+    def _build_tree(self, current_dir: Path, depth: int, max_depth: int) -> Dict[str, Any]:
+        tree: Dict[str, Any] = {"name": current_dir.name or str(current_dir), "type": "directory", "children": []}
+
+        if depth >= max_depth:
+            return tree
+
+        try:
+            for item in sorted(current_dir.iterdir()):
+                if self._is_ignored(item):
+                    continue
+
+                if item.is_dir():
+                    tree["children"].append(self._build_tree(item, depth + 1, max_depth))
+                else:
+                    tree["children"].append(
+                        {"name": item.name, "type": "file", "size_kb": round(item.stat().st_size / 1024, 2)}
+                    )
+        except PermissionError:
+            tree["error"] = "Permission denied"
+
+        return tree
+
+
+class FileOutlinerTool(CodebaseExplorer, Tool):
+    """Tool to extract high-level signatures from a file."""
+
+    @property
+    def name(self) -> str:
+        return "get_file_outline"
+
+    @property
+    def description(self) -> str:
+        return "Extracts classes and functions signatures from a Python file to skim its content."
+
+    def invoke(self, file_path: str) -> Any:
+        full_path = self.root_dir / file_path
+        if not full_path.exists() or not full_path.is_file():
+            return f"Error: File {file_path} not found."
+
+        outline = []
+        pattern = re.compile(r"^\s*(class\s+\w+|async\s+def\s+\w+|def\s+\w+)")
+
+        try:
+            with open(full_path, "r", encoding="utf-8") as f:
+                for line in f:
+                    match = pattern.match(line)
+                    if match:
+                        outline.append(line.strip())
+            return outline
+        except Exception as e:
+            return f"Error reading file: {e}"
+
+
+class FileFragmentReaderTool(CodebaseExplorer, Tool):
+    """Tool to read a specific range of a file."""
+
+    @property
+    def name(self) -> str:
+        return "read_file_fragment"
+
+    @property
+    def description(self) -> str:
+        return "Reads a specific line range from a file. Input format: 'path:start:end' (e.g. 'src/cli.py:1:20')."
+
+    def invoke(self, input_str: str) -> Any:
+        try:
+            parts = input_str.split(":")
+            if len(parts) < 3:
+                return "Error: Invalid input format. Use 'path:start:end'."
+            file_path = ":".join(parts[:-2])
+            start_line = int(parts[-2])
+            end_line = int(parts[-1])
+
+            full_path = self.root_dir / file_path
+            if not full_path.exists():
+                return f"Error: File {file_path} not found."
+
+            with open(full_path, "r", encoding="utf-8") as f:
+                lines = f.readlines()
+                return "".join(lines[max(0, start_line - 1) : end_line])
+        except Exception as e:
+            return f"Error: {e}"

agentic-framework/src/agentic_framework/tools/example.py

@@ -4,6 +4,14 @@
 class CalculatorTool(Tool):
     """A simple calculator tool."""
 
+    @property
+    def name(self) -> str:
+        return "calculator"
+
+    @property
+    def description(self) -> str:
+        return "Evaluate a mathematical expression. Input should be a valid Python expression string."
+
     def invoke(self, input_str: str) -> str:
         """Evaluate a mathematical expression."""
         try:
@@ -16,5 +24,13 @@ def invoke(self, input_str: str) -> str:
 class WeatherTool(Tool):
     """A mock weather tool."""
 
+    @property
+    def name(self) -> str:
+        return "weather"
+
+    @property
+    def description(self) -> str:
+        return "Get the current weather for a location."
+
     def invoke(self, input_str: str) -> str:
         return f"The weather in {input_str} is currently sunny."

agentic-framework/src/agentic_framework/tools/web_search.py

@@ -11,6 +11,14 @@ class WebSearchTool(Tool):
     def __init__(self):
         self.tavily_client = TavilyClient()
 
+    @property
+    def name(self) -> str:
+        return "web_search"
+
+    @property
+    def description(self) -> str:
+        return "Search the web for information using Tavily API."
+
     def invoke(self, query: str) -> Dict[str, Any]:
         """Search the web for information"""
         return self.tavily_client.search(query)