diff --git a/agentic-framework/src/agentic_framework/cli.py b/agentic-framework/src/agentic_framework/cli.py index 02926fe..48afae0 100644 --- a/agentic-framework/src/agentic_framework/cli.py +++ b/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", diff --git a/agentic-framework/src/agentic_framework/core/developer_agent.py b/agentic-framework/src/agentic_framework/core/developer_agent.py new file mode 100644 index 0000000..836dac3 --- /dev/null +++ b/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, + ), + ] diff --git a/agentic-framework/src/agentic_framework/interfaces/base.py b/agentic-framework/src/agentic_framework/interfaces/base.py index 7842944..e257bfa 100644 --- a/agentic-framework/src/agentic_framework/interfaces/base.py +++ b/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.""" diff --git a/agentic-framework/src/agentic_framework/mcp/config.py b/agentic-framework/src/agentic_framework/mcp/config.py index 8823c00..acb6bbc 100644 --- a/agentic-framework/src/agentic_framework/mcp/config.py +++ b/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()} diff --git a/agentic-framework/src/agentic_framework/mcp/provider.py b/agentic-framework/src/agentic_framework/mcp/provider.py index 0c65f4c..52d881d 100644 --- a/agentic-framework/src/agentic_framework/mcp/provider.py +++ b/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. """ diff --git a/agentic-framework/src/agentic_framework/tools/__init__.py b/agentic-framework/src/agentic_framework/tools/__init__.py index 4fe3810..f681878 100644 --- a/agentic-framework/src/agentic_framework/tools/__init__.py +++ b/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", +] diff --git a/agentic-framework/src/agentic_framework/tools/code_searcher.py b/agentic-framework/src/agentic_framework/tools/code_searcher.py new file mode 100644 index 0000000..d0b6ecf --- /dev/null +++ b/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."] diff --git a/agentic-framework/src/agentic_framework/tools/codebase_explorer.py b/agentic-framework/src/agentic_framework/tools/codebase_explorer.py new file mode 100644 index 0000000..bfaf749 --- /dev/null +++ b/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}" diff --git a/agentic-framework/src/agentic_framework/tools/example.py b/agentic-framework/src/agentic_framework/tools/example.py index 41b7d95..61ff80e 100644 --- a/agentic-framework/src/agentic_framework/tools/example.py +++ b/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." diff --git a/agentic-framework/src/agentic_framework/tools/web_search.py b/agentic-framework/src/agentic_framework/tools/web_search.py index 552e46b..1401dca 100644 --- a/agentic-framework/src/agentic_framework/tools/web_search.py +++ b/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)