feat: Python tool system, Command Palette, @file/@folder mentions #21
Conversation
- Add Python tool system via bridge server with 11 tools (glob, grep, read_file, write_file, edit_file, delete_file, delete_dir, move, mkdir, stat, web_fetch) - Implement Ctrl+P Command Palette for quick command access - Add @file and @folder mentions with CWD-aware autocomplete - Update mention handlers to include file contents and directory listings - Add file/folder icons in suggestion UI 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
|
Caution Review failedFailed to post review comments 📝 WalkthroughSummary by CodeRabbit
✏️ Tip: You can customize this high-level summary in your review settings. WalkthroughIntroduces a Python tool framework (base models, registry, file operations, network utilities) with public API re-exports, integrates tool schemas into workflow agents, and adds UI features for command palette navigation, folder mention support, and file/folder path autocompletion. Changes
Sequence DiagramssequenceDiagram
participant Agent as Agent<br/>(Tool Executor)
participant ToolRegistry as Tool Registry<br/>(Lookup)
participant Tool as Tool Instance<br/>(Metadata & Function)
participant FileSystem as File System<br/>(Execution Target)
Agent->>ToolRegistry: get_tool_schemas()<br/>→ [Tool schema, ...]
ToolRegistry->>Agent: Return OpenAI<br/>compatible schemas
Note over Agent: Agent selects tool<br/>and calls with params
Agent->>Tool: execute(**kwargs)
activate Tool
Tool->>Tool: Validate params
Tool->>FileSystem: Call func() with args
activate FileSystem
alt Success
FileSystem->>Tool: Return result
Tool->>Agent: ToolResult.ok(output)
else Exception
FileSystem->>Tool: Raise error
Tool->>Agent: ToolResult.fail(error msg)
end
deactivate FileSystem
deactivate Tool
sequenceDiagram
participant User as User<br/>(Keyboard Input)
participant InputHook as Input Hook<br/>(useInputMode)
participant FileAPI as File System API<br/>(getFileSuggestions)
participant UI as Suggestion List<br/>(UI Display)
User->>InputHook: Type `@file` /src<br/>(mention + path)
InputHook->>InputHook: Detect path-based<br/>mention pattern
InputHook->>FileAPI: fetchFileSuggestions(cwd, query)
activate FileAPI
FileAPI->>FileAPI: List directory entries<br/>Filter by query
FileAPI->>InputHook: Return FileSuggestion[]<br/>(files/folders with paths)
deactivate FileAPI
InputHook->>InputHook: Merge with mention<br/>into suggestions
InputHook->>UI: Update suggestions list
activate UI
UI->>UI: Map file suggestions<br/>to UI items with icons
UI->>User: Display path-aware<br/>autocomplete results
deactivate UI
User->>UI: Select file/folder
UI->>User: Populate `@mention`<br/>with selected path
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
Pre-merge checks and finishing touches❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
Comment |
Summary of ChangesHello @Zochory, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request significantly enhances the interactive capabilities of the application by integrating a robust Python tool system, a user-friendly Command Palette, and intelligent file/folder mentions. These features aim to streamline agent interactions, provide powerful file system and network manipulation capabilities, and improve the overall user experience with intuitive command and context input. Highlights
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
![github-code-quality[bot]](https://avatars.githubusercontent.com/u/9919?s=60&v=4){kind=small}
|
|
||
| try { | ||
| // Determine the directory to list | ||
| let dirPath = cwd; |
| */ | ||
|
|
||
| import { readdir, lstat } from "fs/promises"; | ||
| import { resolve, extname, relative, dirname, basename } from "path"; |
| # Get permissions (Unix only) | ||
| permissions = None | ||
| try: | ||
| mode = stat_result.st_mode |
| """Network tools for the coding agent.""" | ||
|
|
||
| import json | ||
| from typing import Optional, Dict, Any |
| @@ -0,0 +1,141 @@ | |||
| """Tool registry and decorator for registering tools.""" | |||
|
|
|||
| from typing import Any, Awaitable, Callable, Dict, List, Optional | |||
| """Tool registry and decorator for registering tools.""" | ||
|
|
||
| from typing import Any, Awaitable, Callable, Dict, List, Optional | ||
| from functools import wraps |
| "writable": os.access(file_path, os.W_OK), | ||
| "executable": os.access(file_path, os.X_OK), | ||
| } | ||
| except Exception: |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a powerful set of new features, including a Python-based tool system, a command palette for quick access to commands, and @file/@folder mentions with autocomplete. The implementation is comprehensive and significantly enhances the CLI's capabilities. The Python tool system is particularly well-designed, with a clear registration process and a solid set of base classes. My review focuses on opportunities to improve maintainability, refine some API designs, and ensure the new features are as robust and easy to use as possible. I've provided suggestions to reduce code duplication, clarify a tool's API, and improve the tool registration mechanism. Overall, this is an excellent and impactful contribution.
| async def web_fetch( | ||
| url: str, | ||
| method: str = "GET", | ||
| headers: Optional[Dict[str, str]] = None, | ||
| timeout: int = DEFAULT_TIMEOUT, |
There was a problem hiding this comment.
The web_fetch tool's design for handling POST requests is unconventional. It expects the request body to be passed within the headers dictionary under the key "body", which is then removed. A request body is not a header, and this design is confusing and error-prone.
I recommend refactoring the function to accept a dedicated body parameter. This would make the tool's API much clearer and more aligned with standard practices.
| async def web_fetch( | |
| url: str, | |
| method: str = "GET", | |
| headers: Optional[Dict[str, str]] = None, | |
| timeout: int = DEFAULT_TIMEOUT, | |
| async def web_fetch( | |
| url: str, | |
| method: str = "GET", | |
| headers: Optional[Dict[str, str]] = None, | |
| body: Optional[Any] = None, | |
| timeout: int = DEFAULT_TIMEOUT, | |
| ) -> ToolResult: |
|
|
||
|
|
||
| @tool( | ||
| description="Read the contents of a file. Returns up to 120KB by default.", |
There was a problem hiding this comment.
The description for read_file states that it "Returns up to 120KB by default," but the READ_MAX_SIZE constant used for the limit is set to 10MB. This discrepancy could be misleading for the LLM using this tool. Please update the description to accurately reflect the 10MB limit.
| ) -> Callable[[Callable[..., Awaitable[ToolResult]]], Tool]: | ||
| """ | ||
| Decorator to register a function as a tool. | ||
| Args: | ||
| name: Optional custom name for the tool (defaults to function name) | ||
| description: Description of what the tool does | ||
| risk_level: Risk level for permission system | ||
| Example: | ||
| @tool(description="Read file contents", risk_level=ToolRiskLevel.LOW) | ||
| async def read_file(path: str) -> ToolResult: | ||
| ... | ||
| """ | ||
| def decorator(func: Callable[..., Awaitable[ToolResult]]) -> Tool: |
There was a problem hiding this comment.
The return type hint for the decorator function is -> Tool, but the function actually returns func, which is a Callable. This mismatch is incorrect and would be flagged by type checkers. The decorator's purpose is to register the function and return the original callable, so the type hints should reflect this behavior.
| ) -> Callable[[Callable[..., Awaitable[ToolResult]]], Tool]: | |
| """ | |
| Decorator to register a function as a tool. | |
| Args: | |
| name: Optional custom name for the tool (defaults to function name) | |
| description: Description of what the tool does | |
| risk_level: Risk level for permission system | |
| Example: | |
| @tool(description="Read file contents", risk_level=ToolRiskLevel.LOW) | |
| async def read_file(path: str) -> ToolResult: | |
| ... | |
| """ | |
| def decorator(func: Callable[..., Awaitable[ToolResult]]) -> Tool: | |
| ) -> Callable[[Callable[..., Awaitable[ToolResult]]], Callable[..., Awaitable[ToolResult]]]: | |
| """ | |
| Decorator to register a function as a tool. | |
| Args: | |
| name: Optional custom name for the tool (defaults to function name) | |
| description: Description of what the tool does | |
| risk_level: Risk level for permission system | |
| Example: | |
| @tool(description="Read file contents", risk_level=ToolRiskLevel.LOW) | |
| async def read_file(path: str) -> ToolResult: | |
| ... | |
| """ | |
| def decorator(func: Callable[..., Awaitable[ToolResult]]) -> Callable[..., Awaitable[ToolResult]]: |
| parameters[param_name] = ToolParam( | ||
| type=param_type, | ||
| description=param.description if hasattr(param, "description") else "", | ||
| default=default, | ||
| ) |
There was a problem hiding this comment.
The tool decorator does not currently extract descriptions for tool parameters. The expression param.description if hasattr(param, "description") else "" will always yield an empty string, as inspect.Parameter objects lack a description attribute. To enhance the generated tool schemas for the LLM, consider parsing the function's docstring (e.g., using a standard format like Google's or reStructuredText) to extract these parameter descriptions. This would make the tools more self-documenting and easier for the agent to use correctly.
| const filteredCount = context.commandPaletteCommands.filter((cmd) => { | ||
| if (!context.commandPaletteSearch) return true; | ||
| const query = context.commandPaletteSearch.toLowerCase(); | ||
| return ( | ||
| cmd.name.toLowerCase().includes(query) || | ||
| cmd.description.toLowerCase().includes(query) || | ||
| cmd.keywords?.some((k) => k.toLowerCase().includes(query)) | ||
| ); | ||
| }).length; |
There was a problem hiding this comment.
The logic for filtering commands in the command palette is duplicated here and in the CommandPalette.tsx component. This violates the DRY (Don't Repeat Yourself) principle and makes the code harder to maintain, as any changes to the filtering logic will need to be made in two separate places. This logic should be extracted into a shared utility function or a memoized selector that both the hook and the component can consume.
| const lines = await Promise.all( | ||
| limited.map(async (entry) => { | ||
| const isDir = entry.isDirectory() || (entry.isSymbolicLink() && (await lstat(resolve(fullPath, entry.name))).isDirectory()); | ||
| const icon = isDir ? "[DIR]" : "[FILE]"; | ||
| return `${icon} ${entry.name}`; | ||
| }) | ||
| ); |
There was a problem hiding this comment.
The logic to determine if a symbolic link points to a directory is duplicated here and in src/utils/fileSuggestions.ts (in the listDirectoryEntries function). This creates redundancy and increases the maintenance burden.
To improve code reuse, this logic should be extracted into a shared utility function. A good place for it would be in src/utils/fileSuggestions.ts, from where it could be exported and used in both files. For example, you could create a function like export async function isDirectory(entry: Dirent, fullPath: string): Promise<boolean> { ... }.
There was a problem hiding this comment.
Pull request overview
This PR introduces a comprehensive tool system for AI agents, a command palette for quick navigation, and enhanced @file/@folder mentions with autocomplete. The changes enable agents to perform file system operations, fetch web content, and provide users with a more efficient interface for file references.
Key Changes:
- Implemented 11 Python tools (glob, grep, read_file, write_file, edit_file, delete_file, delete_dir, move, mkdir, stat, web_fetch) with a decorator-based registry system
- Added Ctrl+P Command Palette with fuzzy search and keyboard navigation
- Extended @mention system to support @file and @folder with CWD-aware path autocomplete
Reviewed changes
Copilot reviewed 17 out of 17 changed files in this pull request and generated 26 comments.
Show a summary per file
| File | Description |
|---|---|
| bridge/tools/base.py | Defines base classes for tool system: Tool, ToolResult, ToolParam, ToolSchema, and risk levels |
| bridge/tools/registry.py | Implements decorator-based tool registration with automatic parameter detection from function signatures |
| bridge/tools/file_tools.py | Provides 10 file system tools for reading, writing, editing, deleting, and inspecting files/directories |
| bridge/tools/net_tools.py | Implements web_fetch tool for HTTP GET/POST requests with content type handling |
| bridge/tools/init.py | Exports all tool functions and registry utilities for use by the bridge server |
| bridge/workflow_factory.py | Integrates tools into agent workflows, providing all agents with file system capabilities |
| src/components/CommandPalette.tsx | New modal component displaying searchable command list grouped alphabetically |
| src/components/SuggestionList.tsx | Enhanced to display file/folder suggestions with icons and metadata |
| src/hooks/useKeyboardShortcuts.ts | Adds Ctrl+P handler and Command Palette navigation (arrow keys, tab, enter, escape) |
| src/hooks/useInputMode.ts | Extends mention detection to fetch file/folder suggestions dynamically based on path input |
| src/utils/fileSuggestions.ts | New utility for listing directory contents, fuzzy file matching, and path safety validation |
| src/mentionHandlers.ts | Adds @folder support for directory listings alongside existing @file file content inclusion |
| src/commands.ts | Adds @folder to mention types and marks @file/@folder as supporting path autocomplete |
| src/types.ts | Extends UISuggestion to include "file" and "folder" kinds with optional path field |
| src/tools/prompt.ts | Updates system prompt with comprehensive tool documentation and usage guidelines |
| src/index.tsx | Integrates CommandPalette component with state management and keyboard shortcut handlers |
| conductor.json | New configuration file for external tooling (setup and run scripts) |
| """File system tools for the coding agent.""" | ||
|
|
||
| import os | ||
| import re | ||
| from pathlib import Path | ||
| from typing import List, Optional, Dict, Any | ||
|
|
||
| from .base import ToolResult, ToolRiskLevel | ||
| from .registry import tool | ||
|
|
||
|
|
||
| # Maximum number of files to return from glob | ||
| GLOB_MAX_FILES = 1000 | ||
|
|
||
| # Maximum file size to read (10MB) | ||
| READ_MAX_SIZE = 10 * 1024 * 1024 | ||
|
|
||
| # Maximum lines to return from grep | ||
| GREP_MAX_RESULTS = 100 | ||
|
|
||
|
|
||
| @tool( | ||
| description="Find files matching a glob pattern. Use **/*.ts for recursive search.", | ||
| risk_level=ToolRiskLevel.LOW, | ||
| ) | ||
| async def glob( | ||
| pattern: str, | ||
| cwd: str = ".", | ||
| recursive: bool = True, | ||
| ) -> ToolResult: | ||
| """Find files matching a glob pattern.""" | ||
| try: | ||
| base = Path(cwd) | ||
|
|
||
| # Limit depth for safety | ||
| if "**" in pattern and not recursive: | ||
| pattern = pattern.replace("**", "*", 1) | ||
|
|
||
| if recursive: | ||
| matches = list(base.glob(pattern)) | ||
| else: | ||
| matches = list(base.glob(pattern)) | ||
|
|
||
| # Limit results for safety | ||
| matches = matches[:GLOB_MAX_FILES] | ||
|
|
||
| return ToolResult.ok([str(m) for m in matches]) | ||
| except Exception as e: | ||
| return ToolResult.fail(f"glob error: {e}") | ||
|
|
||
|
|
||
| @tool( | ||
| description="Search for text pattern in files. Returns matching lines with line numbers.", | ||
| risk_level=ToolRiskLevel.LOW, | ||
| ) | ||
| async def grep( | ||
| pattern: str, | ||
| path: str = ".", | ||
| glob: Optional[str] = None, | ||
| recursive: bool = True, | ||
| ) -> ToolResult: | ||
| """Search for text in files using regex.""" | ||
| try: | ||
| base_path = Path(path) | ||
| results: List[Dict[str, Any]] = [] | ||
| count = 0 | ||
|
|
||
| # Build glob patterns to search | ||
| if glob: | ||
| pattern_path = Path(glob) | ||
| if pattern_path.is_absolute(): | ||
| base_path = pattern_path.parent | ||
| glob_pattern = pattern_path.name | ||
| else: | ||
| glob_pattern = glob | ||
| else: | ||
| glob_pattern = "**/*" if recursive else "*" | ||
|
|
||
| # Compile regex for efficiency | ||
| try: | ||
| regex = re.compile(pattern) | ||
| except re.error as e: | ||
| return ToolResult.fail(f"Invalid regex pattern: {e}") | ||
|
|
||
| # Search files | ||
| for file_path in base_path.glob(glob_pattern): | ||
| if not file_path.is_file(): | ||
| continue | ||
|
|
||
| # Skip binary files | ||
| try: | ||
| content = file_path.read_text(encoding="utf-8") | ||
| except (UnicodeDecodeError, PermissionError): | ||
| continue | ||
|
|
||
| # Search line by line | ||
| for line_num, line in enumerate(content.splitlines(), 1): | ||
| if regex.search(line): | ||
| results.append({ | ||
| "file": str(file_path), | ||
| "line": line_num, | ||
| "content": line.strip(), | ||
| }) | ||
| count += 1 | ||
| if count >= GREP_MAX_RESULTS: | ||
| break | ||
|
|
||
| if count >= GREP_MAX_RESULTS: | ||
| break | ||
|
|
||
| return ToolResult.ok(results) | ||
| except Exception as e: | ||
| return ToolResult.fail(f"grep error: {e}") | ||
|
|
||
|
|
||
| @tool( | ||
| description="Read the contents of a file. Returns up to 120KB by default.", | ||
| risk_level=ToolRiskLevel.LOW, | ||
| ) | ||
| async def read_file( | ||
| path: str, | ||
| offset: int = 0, | ||
| limit: int = -1, | ||
| ) -> ToolResult: | ||
| """Read file contents.""" | ||
| try: | ||
| file_path = Path(path) | ||
|
|
||
| if not file_path.exists(): | ||
| return ToolResult.fail(f"File not found: {path}") | ||
|
|
||
| if not file_path.is_file(): | ||
| return ToolResult.fail(f"Not a file: {path}") | ||
|
|
||
| # Check file size | ||
| size = file_path.stat().st_size | ||
| if size > READ_MAX_SIZE: | ||
| return ToolResult.fail(f"File too large ({size} bytes, max {READ_MAX_SIZE})") | ||
|
|
||
| content = file_path.read_text(encoding="utf-8") | ||
|
|
||
| # Handle offset and limit | ||
| lines = content.splitlines() | ||
|
|
||
| if offset > 0: | ||
| lines = lines[offset:] | ||
|
|
||
| if limit > 0: | ||
| lines = lines[:limit] | ||
|
|
||
| result = "\n".join(lines) | ||
|
|
||
| # Truncate if still too large | ||
| if len(result) > READ_MAX_SIZE: | ||
| result = result[:READ_MAX_SIZE] + "\n... (truncated)" | ||
|
|
||
| return ToolResult.ok(result) | ||
| except Exception as e: | ||
| return ToolResult.fail(f"read_file error: {e}") | ||
|
|
||
|
|
||
| @tool( | ||
| description="Create or overwrite a file with the given content. Creates parent directories if needed.", | ||
| risk_level=ToolRiskLevel.MEDIUM, | ||
| ) | ||
| async def write_file( | ||
| path: str, | ||
| content: str, | ||
| create_dirs: bool = True, | ||
| ) -> ToolResult: | ||
| """Write content to a file.""" | ||
| try: | ||
| file_path = Path(path) | ||
|
|
||
| # Create parent directories if needed | ||
| if create_dirs: | ||
| file_path.parent.mkdir(parents=True, exist_ok=True) | ||
|
|
||
| file_path.write_text(content, encoding="utf-8") | ||
|
|
||
| return ToolResult.ok({ | ||
| "path": str(file_path), | ||
| "size": len(content), | ||
| "lines": len(content.splitlines()), | ||
| }) | ||
| except Exception as e: | ||
| return ToolResult.fail(f"write_file error: {e}") | ||
|
|
||
|
|
||
| @tool( | ||
| description="Edit a file by replacing old_text with new_text. Uses exact text matching.", | ||
| risk_level=ToolRiskLevel.HIGH, | ||
| ) | ||
| async def edit_file( | ||
| path: str, | ||
| old_text: str, | ||
| new_text: str, | ||
| ) -> ToolResult: | ||
| """Replace specific text in a file (diff-based edit).""" | ||
| try: | ||
| file_path = Path(path) | ||
|
|
||
| if not file_path.exists(): | ||
| return ToolResult.fail(f"File not found: {path}") | ||
|
|
||
| content = file_path.read_text(encoding="utf-8") | ||
|
|
||
| if old_text not in content: | ||
| return ToolResult.fail(f"Text not found in file: {old_text[:100]}...") | ||
|
|
||
| new_content = content.replace(old_text, new_text, 1) | ||
| file_path.write_text(new_content, encoding="utf-8") | ||
|
|
||
| return ToolResult.ok({ | ||
| "path": str(file_path), | ||
| "changes": 1, | ||
| }) | ||
| except Exception as e: | ||
| return ToolResult.fail(f"edit_file error: {e}") | ||
|
|
||
|
|
||
| @tool( | ||
| description="Delete a single file. This action cannot be undone.", | ||
| risk_level=ToolRiskLevel.HIGH, | ||
| ) | ||
| async def delete_file(path: str) -> ToolResult: | ||
| """Delete a file.""" | ||
| try: | ||
| file_path = Path(path) | ||
|
|
||
| if not file_path.exists(): | ||
| return ToolResult.fail(f"File not found: {path}") | ||
|
|
||
| if not file_path.is_file(): | ||
| return ToolResult.fail(f"Not a file: {path}") | ||
|
|
||
| file_path.unlink() | ||
|
|
||
| return ToolResult.ok({ | ||
| "path": str(file_path), | ||
| "deleted": True, | ||
| }) | ||
| except Exception as e: | ||
| return ToolResult.fail(f"delete_file error: {e}") | ||
|
|
||
|
|
||
| @tool( | ||
| description="Delete an empty directory or recursively delete a directory and all contents.", | ||
| risk_level=ToolRiskLevel.HIGH, | ||
| ) | ||
| async def delete_dir( | ||
| path: str, | ||
| recursive: bool = False, | ||
| ) -> ToolResult: | ||
| """Delete a directory.""" | ||
| try: | ||
| dir_path = Path(path) | ||
|
|
||
| if not dir_path.exists(): | ||
| return ToolResult.fail(f"Directory not found: {path}") | ||
|
|
||
| if not dir_path.is_dir(): | ||
| return ToolResult.fail(f"Not a directory: {path}") | ||
|
|
||
| import shutil | ||
|
|
||
| if recursive: | ||
| shutil.rmtree(dir_path) | ||
| else: | ||
| dir_path.rmdir() | ||
|
|
||
| return ToolResult.ok({ | ||
| "path": str(dir_path), | ||
| "deleted": True, | ||
| "recursive": recursive, | ||
| }) | ||
| except Exception as e: | ||
| return ToolResult.fail(f"delete_dir error: {e}") | ||
|
|
||
|
|
||
| @tool( | ||
| description="Move or rename a file or directory from source to destination.", | ||
| risk_level=ToolRiskLevel.HIGH, | ||
| ) | ||
| async def move( | ||
| source: str, | ||
| destination: str, | ||
| ) -> ToolResult: | ||
| """Move or rename a file/directory.""" | ||
| try: | ||
| src_path = Path(source) | ||
| dst_path = Path(destination) | ||
|
|
||
| if not src_path.exists(): | ||
| return ToolResult.fail(f"Source not found: {source}") | ||
|
|
||
| # Create parent dirs for destination | ||
| dst_path.parent.mkdir(parents=True, exist_ok=True) | ||
|
|
||
| # Use shutil.move which handles both move and rename | ||
| import shutil | ||
| shutil.move(str(src_path), str(dst_path)) | ||
|
|
||
| return ToolResult.ok({ | ||
| "source": str(src_path), | ||
| "destination": str(dst_path), | ||
| "moved": True, | ||
| }) | ||
| except Exception as e: | ||
| return ToolResult.fail(f"move error: {e}") | ||
|
|
||
|
|
||
| @tool( | ||
| description="Create a directory. Use recursive=true to create nested directories.", | ||
| risk_level=ToolRiskLevel.MEDIUM, | ||
| ) | ||
| async def mkdir( | ||
| path: str, | ||
| recursive: bool = True, | ||
| ) -> ToolResult: | ||
| """Create a directory.""" | ||
| try: | ||
| dir_path = Path(path) | ||
|
|
||
| if recursive: | ||
| dir_path.mkdir(parents=True, exist_ok=True) | ||
| else: | ||
| dir_path.mkdir(exist_ok=True) | ||
|
|
||
| return ToolResult.ok({ | ||
| "path": str(dir_path), | ||
| "created": True, | ||
| }) | ||
| except FileExistsError: | ||
| return ToolResult.fail(f"Path exists and is not a directory: {path}") | ||
| except Exception as e: | ||
| return ToolResult.fail(f"mkdir error: {e}") | ||
|
|
||
|
|
||
| @tool( | ||
| description="Get file metadata including size, modification time, and file type.", | ||
| risk_level=ToolRiskLevel.LOW, | ||
| ) | ||
| async def stat(path: str) -> ToolResult: | ||
| """Get file metadata.""" | ||
| try: | ||
| file_path = Path(path) | ||
|
|
||
| if not file_path.exists(): | ||
| return ToolResult.fail(f"Path not found: {path}") | ||
|
|
||
| stat_result = file_path.stat() | ||
|
|
||
| # Get file type | ||
| if file_path.is_file(): | ||
| file_type = "file" | ||
| elif file_path.is_dir(): | ||
| file_type = "directory" | ||
| elif file_path.is_symlink(): | ||
| file_type = "symlink" | ||
| else: | ||
| file_type = "unknown" | ||
|
|
||
| # Get permissions (Unix only) | ||
| permissions = None | ||
| try: | ||
| mode = stat_result.st_mode | ||
| permissions = { | ||
| "readable": os.access(file_path, os.R_OK), | ||
| "writable": os.access(file_path, os.W_OK), | ||
| "executable": os.access(file_path, os.X_OK), | ||
| } | ||
| except Exception: | ||
| pass | ||
|
|
||
| return ToolResult.ok({ | ||
| "path": str(file_path), | ||
| "size": stat_result.st_size, | ||
| "mtime": stat_result.st_mtime, | ||
| "ctime": stat_result.st_ctime, | ||
| "type": file_type, | ||
| "permissions": permissions, | ||
| }) | ||
| except Exception as e: | ||
| return ToolResult.fail(f"stat error: {e}") |
There was a problem hiding this comment.
The Python file tools lack path traversal protection. All file operations accept arbitrary paths without validating that they are within the intended working directory. An agent or attacker could use patterns like "../../../etc/passwd" to access files outside the project directory. Consider adding a path validation function that checks resolved paths are within an allowed directory boundary before performing file operations.
| def decorator(func: Callable[..., Awaitable[ToolResult]]) -> Tool: | ||
| tool_name = name or func.__name__ | ||
|
|
||
| # Build parameters from function signature annotations | ||
| import inspect | ||
| sig = inspect.signature(func) | ||
| parameters: Dict[str, ToolParam] = {} | ||
| required: List[str] = [] | ||
|
|
||
| for param_name, param in sig.parameters.items(): | ||
| # Skip 'self' and 'cls' | ||
| if param_name in ("self", "cls"): | ||
| continue | ||
|
|
||
| # Get annotation or default to string | ||
| annotation = param.annotation | ||
| param_type = "string" | ||
|
|
||
| if annotation is not None: | ||
| # Handle Optional types | ||
| origin = getattr(annotation, "__origin__", None) | ||
| if annotation is str: | ||
| param_type = "string" | ||
| elif annotation is int: | ||
| param_type = "integer" | ||
| elif annotation is float: | ||
| param_type = "number" | ||
| elif annotation is bool: | ||
| param_type = "boolean" | ||
| elif annotation is list or annotation is List: | ||
| param_type = "array" | ||
| elif origin is list: | ||
| param_type = "array" | ||
| elif annotation is dict or annotation is Dict: | ||
| param_type = "object" | ||
| elif origin is dict: | ||
| param_type = "object" | ||
| elif getattr(annotation, "__name__", None) == "Path": | ||
| param_type = "string" | ||
| # Handle Optional types like Optional[str] | ||
| if origin is list: | ||
| # Get the inner type | ||
| args = getattr(annotation, "__args__", []) | ||
| if args: | ||
| inner = args[0] | ||
| if inner is str: | ||
| param_type = "array" | ||
| elif inner is int: | ||
| param_type = "array" | ||
|
|
||
| # Get default value | ||
| default = None | ||
| if param.default != inspect.Parameter.empty: | ||
| default = param.default | ||
|
|
||
| # Check if required | ||
| if default is None and param.default == inspect.Parameter.empty: | ||
| required.append(param_name) | ||
|
|
||
| parameters[param_name] = ToolParam( | ||
| type=param_type, | ||
| description=param.description if hasattr(param, "description") else "", | ||
| default=default, | ||
| ) | ||
|
|
||
| # Create tool instance | ||
| tool_instance = Tool( | ||
| name=tool_name, | ||
| description=description or func.__doc__ or "", | ||
| parameters=parameters, | ||
| required=required, | ||
| risk_level=risk_level, | ||
| func=func, | ||
| ) | ||
|
|
||
| # Register it | ||
| _TOOL_REGISTRY[tool_name] = tool_instance | ||
|
|
||
| return func # Return the original function | ||
|
|
||
| return decorator |
There was a problem hiding this comment.
The decorator's return type annotation indicates it returns a Tool, but line 109 actually returns the original function. This creates a type mismatch - the decorator is typed as Callable[[Callable[..., Awaitable[ToolResult]]], Tool] but the inner decorator function returns 'func' not 'tool_instance'. This will cause type checking errors when the decorated functions are used. The decorator should either return the Tool instance or update the type annotation to reflect that it returns the original function.
| # Check if required | ||
| if default is None and param.default == inspect.Parameter.empty: |
There was a problem hiding this comment.
The parameter detection logic on line 87 incorrectly marks parameters with default=None as required. The condition checks 'if default is None and param.default == inspect.Parameter.empty' which will be false when a parameter has an explicit None default (e.g., 'path: Optional[str] = None'). This means optional parameters with None defaults will be treated as required. Consider checking only 'param.default == inspect.Parameter.empty' to determine if a parameter is required.
| # Check if required | |
| if default is None and param.default == inspect.Parameter.empty: | |
| # Check if required: a parameter is required only if it has no default | |
| if param.default == inspect.Parameter.empty: |
| # Handle Optional types like Optional[str] | ||
| if origin is list: | ||
| # Get the inner type | ||
| args = getattr(annotation, "__args__", []) | ||
| if args: | ||
| inner = args[0] | ||
| if inner is str: | ||
| param_type = "array" | ||
| elif inner is int: | ||
| param_type = "array" |
There was a problem hiding this comment.
The parameter type detection logic handles the 'origin is list' case twice (lines 62-63 and 71-79), with the second block overwriting param_type to "array" regardless of what was already detected. The second block at lines 71-79 checks 'if origin is list' again but doesn't consider that this was already handled. This creates redundant and potentially conflicting logic. Consider consolidating the list type handling into a single block.
| # Handle Optional types like Optional[str] | |
| if origin is list: | |
| # Get the inner type | |
| args = getattr(annotation, "__args__", []) | |
| if args: | |
| inner = args[0] | |
| if inner is str: | |
| param_type = "array" | |
| elif inner is int: | |
| param_type = "array" |
| async def grep( | ||
| pattern: str, | ||
| path: str = ".", | ||
| glob: Optional[str] = None, |
There was a problem hiding this comment.
The grep function has a parameter named 'glob' which shadows the glob function defined earlier in the same file. While Python allows this, it makes the code confusing and prevents calling the glob function from within grep if needed. On line 59, the parameter is named 'glob: Optional[str]' but there's already a 'glob' function at line 26. Consider renaming the parameter to something like 'glob_pattern' or 'file_pattern' for clarity.
| # Get permissions (Unix only) | ||
| permissions = None | ||
| try: | ||
| mode = stat_result.st_mode |
There was a problem hiding this comment.
Variable mode is not used.
| """Network tools for the coding agent.""" | ||
|
|
||
| import json | ||
| from typing import Optional, Dict, Any |
There was a problem hiding this comment.
Import of 'Any' is not used.
| from typing import Optional, Dict, Any | |
| from typing import Optional, Dict |
| @@ -0,0 +1,141 @@ | |||
| """Tool registry and decorator for registering tools.""" | |||
|
|
|||
| from typing import Any, Awaitable, Callable, Dict, List, Optional | |||
There was a problem hiding this comment.
Import of 'Any' is not used.
| from typing import Any, Awaitable, Callable, Dict, List, Optional | |
| from typing import Awaitable, Callable, Dict, List, Optional |
| """Tool registry and decorator for registering tools.""" | ||
|
|
||
| from typing import Any, Awaitable, Callable, Dict, List, Optional | ||
| from functools import wraps |
There was a problem hiding this comment.
Import of 'wraps' is not used.
| from functools import wraps |
| except Exception: | ||
| pass |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| except Exception: | |
| pass | |
| except OSError: | |
| # Best-effort permission detection; leave permissions as None if it fails. | |
| permissions = None |
Description
Adds three major features: Python tool system via bridge server, Ctrl+P Command Palette, and @file/@folder mentions with CWD-aware autocomplete.
Type of Change
Changes Made
Testing
bun run typecheck)