""" Tool registrations for the Explore agent. Registers our tool implementations with the PydanticAI agent. """ from pydantic_ai import Agent, RunContext from src.domains.agents.explore.agent import ExploreContext from src.domains.tools.file.edit import EditFileTool from src.domains.tools.file.glob import GlobFilesTool from src.domains.tools.file.read import ReadFileTool from src.domains.tools.file.write import WriteFileTool from src.domains.tools.search.grep import GrepContentTool from src.domains.tools.search.web import WebSearchTool from src.domains.tools.shell.bash import BashReadOnlyTool from src.domains.tools.shell.bash_full import BashTool def register_explore_tools(agent: Agent[ExploreContext, str]) -> None: """ Register all exploration tools with the agent. Each tool is wrapped to use context from RunContext. """ @agent.tool async def read_file( ctx: RunContext[ExploreContext], file_path: str, offset: int = 0, limit: int = 2000 ) -> str: """Read contents of a file with line numbers. Args: file_path: Absolute path to the file to read offset: Line number to start from (0-based, default: 0) limit: Maximum number of lines to read (default: 2000) Returns: File contents with line numbers, or error message. IMPORTANT: Always use absolute paths. Never guess file contents. """ tool = ReadFileTool(allowed_paths=ctx.deps.allowed_paths) result = await tool.execute( file_path=file_path, offset=offset, limit=limit ) return result.to_string() @agent.tool async def glob_files( ctx: RunContext[ExploreContext], pattern: str, path: str | None = None, limit: int = 100 ) -> str: """Find files matching a glob pattern. Args: pattern: Glob pattern (e.g., "**/*.py", "src/**/*.ts", "*.md") path: Directory to search in (default: working directory) limit: Maximum number of files to return (default: 100) Returns: List of absolute file paths, sorted by modification time (newest first). Examples: - "**/*.py" finds all Python files - "src/**/*.ts" finds TypeScript files in src/ - "**/test_*.py" finds all test files IMPORTANT: Use this to discover files before reading them. """ tool = GlobFilesTool(allowed_paths=ctx.deps.allowed_paths) search_path = path or ctx.deps.working_dir result = await tool.execute( pattern=pattern, path=search_path, limit=limit ) return result.to_string() @agent.tool async def grep_content( ctx: RunContext[ExploreContext], pattern: str, path: str | None = None, file_glob: str | None = None, context_lines: int = 0, case_sensitive: bool = True ) -> str: """Search file contents using regex pattern. Args: pattern: Regex pattern to search for (Python re syntax) path: Directory or file to search (default: working directory) file_glob: Filter files by glob (e.g., "*.py", "*.ts") context_lines: Lines of context before/after matches (default: 0) case_sensitive: Case-sensitive search (default: True) Returns: Matching lines with file paths and line numbers. Format: "filepath:line_num: content" Examples: - pattern="def.*__init__" finds init methods - pattern="class\\s+\\w+" finds class definitions - pattern="TODO|FIXME" finds todo comments IMPORTANT: Use this to search for code patterns. Escape regex special chars. """ tool = GrepContentTool(allowed_paths=ctx.deps.allowed_paths) search_path = path or ctx.deps.working_dir result = await tool.execute( pattern=pattern, path=search_path, file_glob=file_glob, context_lines=context_lines, case_sensitive=case_sensitive ) return result.to_string() @agent.tool async def bash_readonly( ctx: RunContext[ExploreContext], command: str, cwd: str | None = None, timeout: int = 30 ) -> str: """Execute a read-only bash command. ALLOWED commands: - File inspection: ls, find, cat, head, tail, wc, file, stat, tree, du - Git (read-only): git status, git log, git diff, git show, git branch - Text processing: grep, awk, sed (read-only), sort, uniq - System info: pwd, whoami, hostname, which FORBIDDEN: - File modification (rm, mv, cp, mkdir, touch) - Redirects (>, >>) - Command chaining (&&, ||, ;) - Network (curl, wget) Args: command: The bash command to execute cwd: Working directory (default: agent working directory) timeout: Timeout in seconds (default: 30) Returns: Command output or error message. Examples: - "ls -la" lists files with details - "git status" shows git status - "git log --oneline -10" shows recent commits """ tool = BashReadOnlyTool(allowed_paths=ctx.deps.allowed_paths) working_dir = cwd or ctx.deps.working_dir result = await tool.execute( command=command, cwd=working_dir, timeout=min(timeout, ctx.deps.timeout_seconds) ) return result.to_string() # === Write-capable tools === @agent.tool async def edit_file( ctx: RunContext[ExploreContext], file_path: str, old_string: str, new_string: str, replace_all: bool = False ) -> str: """Make targeted edits to a file using find-and-replace. Args: file_path: Absolute path to the file to edit old_string: The exact text to find and replace (must exist in file) new_string: The replacement text replace_all: If True, replace all occurrences. If False (default), old_string must be unique (appear exactly once). Returns: Success message with diff preview, or error. IMPORTANT: - old_string must exactly match file content (including whitespace) - By default, old_string must appear exactly once (for safety) - Always read the file first to verify exact content before editing """ tool = EditFileTool(allowed_paths=ctx.deps.allowed_paths) result = await tool.execute( file_path=file_path, old_string=old_string, new_string=new_string, replace_all=replace_all ) return result.to_string() @agent.tool async def write_file( ctx: RunContext[ExploreContext], file_path: str, content: str ) -> str: """Create a new file or overwrite an existing file. Args: file_path: Absolute path to the file to create/write content: The content to write to the file Returns: Success message with file path and size. IMPORTANT: - Parent directory must exist (use mkdir first if needed) - For editing existing files, prefer edit_file instead - Will overwrite existing files without confirmation """ tool = WriteFileTool(allowed_paths=ctx.deps.allowed_paths) result = await tool.execute( file_path=file_path, content=content ) return result.to_string() @agent.tool async def bash( ctx: RunContext[ExploreContext], command: str, cwd: str | None = None, timeout: int = 60 ) -> str: """Execute a bash command with write capabilities. ALLOWED: - File operations: ls, find, mkdir, touch, cp, mv, rm (single files) - Git (full): git add, git commit, git checkout, git merge, git pull - Python: python, pip install, pytest, mypy, ruff - Text processing: grep, awk, sed, sort - Command chaining: && and || are allowed FORBIDDEN: - sudo, su (privilege escalation) - Network: curl, wget, ssh, scp, rsync - Dangerous: rm -rf, chmod 777, dd, mkfs Args: command: The bash command to execute cwd: Working directory (default: agent working directory) timeout: Timeout in seconds (default: 60) Returns: Command output or error message. Examples: - "mkdir -p src/utils" creates directory - "git add . && git commit -m 'fix: bug'" commits changes - "pytest tests/ -v" runs tests - "rm old_file.py" removes single file """ tool = BashTool(allowed_paths=ctx.deps.allowed_paths) working_dir = cwd or ctx.deps.working_dir result = await tool.execute( command=command, cwd=working_dir, timeout=min(timeout, ctx.deps.timeout_seconds) ) return result.to_string() # === Web search === @agent.tool async def web_search( ctx: RunContext[ExploreContext], query: str, num_results: int = 5, categories: str | None = None ) -> str: """Search the web for current information. Args: query: Search query (e.g., "Python 3.12 new features") num_results: Number of results to return (1-10, default: 5) categories: Optional category filter ("general", "it", "news", "science") Returns: Search results with titles, URLs, and snippets. Use this for: - Current events or recent information - Documentation updates since your training - Facts you're uncertain about - Technical references with URLs IMPORTANT: Always include a "Sources:" section with URLs in your response. Examples: - query="FastAPI best practices 2024" - query="CVE-2024" categories="it" """ tool = WebSearchTool() result = await tool.execute( query=query, num_results=num_results, categories=categories ) return result.to_string()