Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions Dockerfile
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,22 @@ WORKDIR /app
# Copy from the official installer: https://github.com/astral-sh/uv
COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/

# Install Search Tools
# ripgrep: ultra-fast text searching
# fd-find: user-friendly alternative to 'find'
# fzf: general-purpose command-line fuzzy finder
RUN apt-get update && apt-get install -y --no-install-recommends \
ripgrep \
fd-find \
fzf \
&& rm -rf /var/lib/apt/lists/*

# Note: In Debian/Ubuntu, the 'fd' executable is renamed to 'fdfind'.
# We create a symbolic link so the agent can just call 'fd'.
RUN ln -s $(which fdfind) /usr/local/bin/fd

# Python Environment Setup

# Set environment variables
ENV PYTHONUNBUFFERED=1 \
PYTHONDONTWRITEBYTECODE=1 \
Expand Down
10 changes: 6 additions & 4 deletions agentic-framework/src/agentic_framework/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@
import typer
from dotenv import load_dotenv
from rich.console import Console
from rich.panel import Panel

from agentic_framework.constants import LOGS_DIR
from agentic_framework.mcp import MCPConnectionError, MCPProvider
Expand Down Expand Up @@ -57,7 +56,7 @@ def _handle_mcp_connection_error(error: MCPConnectionError) -> None:
elif error.__cause__:
console.print(f"[red] cause: {error.__cause__}[/red]")

console.print("\n[yellow]Suggestion:[/yellow] Ensure the MCP server URL is correct and you have network access.")
console.print("[yellow]Suggestion:[/yellow] Ensure the MCP server URL is correct and you have network access.")
if "web-fetch" in error.server_name:
console.print("[yellow]Note:[/yellow] web-fetch requires a valid remote URL. Check mcp/config.py")

Expand Down Expand Up @@ -93,7 +92,9 @@ def execute_agent(agent_name: str, input_text: str, timeout_sec: int) -> str:
def list_agents() -> None:
"""List all available agents."""
agents = AgentRegistry.list_agents()
console.print(Panel(f"[bold green]Available Agents:[/bold green] {', '.join(agents)}", title="Registry"))
console.print(
f"[bold magenta]Registry:[/bold magenta] [bold green]Available Agents:[/bold green] {', '.join(agents)}\n"
)


@app.callback(invoke_without_command=True)
Expand Down Expand Up @@ -127,7 +128,8 @@ def command(

try:
result = execute_agent(agent_name=agent_name, input_text=input_text, timeout_sec=timeout_sec)
console.print(Panel(result, title=f"[bold green]Result from {agent_name}[/bold green]"))
console.print(f"[bold green]Result from {agent_name}:[/bold green]")
console.print(result)
except typer.Exit:
raise
except TimeoutError as error:
Expand Down
16 changes: 12 additions & 4 deletions agentic-framework/src/agentic_framework/core/developer_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from agentic_framework.registry import AgentRegistry
from agentic_framework.tools import (
CodeSearcher,
FileFinderTool,
FileFragmentReaderTool,
FileOutlinerTool,
StructureExplorerTool,
Expand All @@ -25,11 +26,12 @@ 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.
1. Discovering the project structure and finding files by name (`find_files`).
2. Extracting class and function outlines from Python files (`get_file_outline`).
3. Reading specific fragments of a file (`read_file_fragment`).
4. Searching the codebase for patterns using ripgrep (`code_search`).

When you need to find a specific file by name, use `find_files`.
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.
Expand All @@ -42,6 +44,7 @@ def system_prompt(self) -> str:
def local_tools(self) -> Sequence[Any]:
# Initialize tool instances with project root
searcher = CodeSearcher(str(BASE_DIR))
finder = FileFinderTool(str(BASE_DIR))
explorer = StructureExplorerTool(str(BASE_DIR))
outliner = FileOutlinerTool(str(BASE_DIR))
reader = FileFragmentReaderTool(str(BASE_DIR))
Expand All @@ -52,6 +55,11 @@ def local_tools(self) -> Sequence[Any]:
name=searcher.name,
description=searcher.description,
),
StructuredTool.from_function(
func=finder.invoke,
name=finder.name,
description=finder.description,
),
StructuredTool.from_function(
func=explorer.invoke,
name=explorer.name,
Expand Down
2 changes: 2 additions & 0 deletions agentic-framework/src/agentic_framework/tools/__init__.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from .code_searcher import CodeSearcher
from .codebase_explorer import (
FileFinderTool,
FileFragmentReaderTool,
FileOutlinerTool,
StructureExplorerTool,
Expand All @@ -15,4 +16,5 @@
"StructureExplorerTool",
"FileOutlinerTool",
"FileFragmentReaderTool",
"FileFinderTool",
]
32 changes: 20 additions & 12 deletions agentic-framework/src/agentic_framework/tools/code_searcher.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import subprocess
from typing import Any, Dict, List, Union
from typing import Any, List, Union

from agentic_framework.interfaces.base import Tool

Expand All @@ -22,20 +22,28 @@ 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]]]:
def grep_search(self, query: str, glob: str = "*") -> Union[List[str], str]:
"""
Executes a search across the codebase.
Returns matches with line numbers and snippets.
Executes a search across the codebase using ripgrep.
Returns matches in vimgrep format (file:line:col:text).
"""
try:
# -n: line numbers, -C 1: context, -t: type
cmd = ["rg", "--json", "-t", file_type, query, self.root_dir]
# --vimgrep: file:line:col:text
# --smart-case: case-insensitive unless query has uppercase
# --iglob: filter by glob pattern
# --max-columns: limit long lines to avoid token blowup
cmd = ["rg", "--vimgrep", "--smart-case", "--max-columns", "500", "--iglob", glob, 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
if result.returncode == 0:
lines = result.stdout.splitlines()
if not lines:
return "No matches found."
return lines[:30] # Increased to 30 for more context

if result.stderr:
return f"Error executing rg: {result.stderr}"
return "No matches found."

except FileNotFoundError:
return ["Error: ripgrep (rg) is not installed on the host system."]
return "Error: ripgrep (rg) is not installed. Please install it to use this tool."
63 changes: 63 additions & 0 deletions agentic-framework/src/agentic_framework/tools/codebase_explorer.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import re
import subprocess
from pathlib import Path
from typing import Any, Dict, List, Optional, Union

Expand Down Expand Up @@ -134,3 +135,65 @@ def invoke(self, input_str: str) -> Any:
return "".join(lines[max(0, start_line - 1) : end_line])
except Exception as e:
return f"Error: {e}"


class FileFinderTool(CodebaseExplorer, Tool):
"""Tool to find files by name using 'fd'."""

@property
def name(self) -> str:
return "find_files"

@property
def description(self) -> str:
return (
"Fast file search by name using 'fd'. Returns paths relative to project root. "
"Input is a search pattern (regex or simple string)."
)

def invoke(self, pattern: str) -> Any:
try:
# First, get candidates using fd
# -H: hidden files, -I: ignore .gitignore for faster full list if needed,
# but usually we want to respect it, so let's stick to standard fd.
# We'll list all files and let fzf rank them.
fd_cmd = ["fd", "--color", "never", "-H", ".", str(self.root_dir)]
fd_result = subprocess.run(fd_cmd, capture_output=True, text=True, check=False)

if fd_result.returncode != 0:
if fd_result.stderr:
return f"Error executing fd: {fd_result.stderr}"
return "No files found."

all_files = fd_result.stdout
if not all_files:
return "No files found."

# Use fzf to rank the files based on the pattern
# fzf -f performs a non-interactive fuzzy search
try:
fzf_cmd = ["fzf", "-f", pattern]
fzf_result = subprocess.run(fzf_cmd, input=all_files, capture_output=True, text=True, check=False)
if fzf_result.returncode == 0:
ranked_files = fzf_result.stdout.splitlines()
else:
# Fallback to simple substring match if fzf returns nothing or fails
ranked_files = [f for f in all_files.splitlines() if pattern.lower() in f.lower()]
except FileNotFoundError:
# Fallback if fzf is missing
ranked_files = [f for f in all_files.splitlines() if pattern.lower() in f.lower()]

# Convert absolute paths back to relative and limit results
relative_files = []
for f in ranked_files[:30]: # Limit to top 30
try:
relative_files.append(str(Path(f).relative_to(self.root_dir)))
except ValueError:
relative_files.append(f)

if not relative_files:
return "No matches found for the given pattern."

return relative_files
except FileNotFoundError:
return "Error: Required search tools (fd) not found."