mirror of
https://github.com/tiennm99/serena.git
synced 2026-10-15 00:18:04 +00:00
Merge pull request #282 from oraios/refactoring/tools
Refactor tools.py into a package with several modules
This commit is contained in:
9 files changed
+1392
-1341
No files matched your search
-1341
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,8 @@
|
||||
# ruff: noqa
|
||||
from .tools_base import *
|
||||
from .file_tools import *
|
||||
from .ls_tools import *
|
||||
from .memory_tools import *
|
||||
from .cmd_tools import *
|
||||
from .config_tools import *
|
||||
from .workflow_tools import *
|
||||
@@ -0,0 +1,43 @@
|
||||
"""
|
||||
Tools supporting the execution of (external) commands
|
||||
"""
|
||||
|
||||
from serena.tools import TOOL_DEFAULT_MAX_ANSWER_LENGTH, Tool, ToolMarkerCanEdit
|
||||
from serena.util.shell import execute_shell_command
|
||||
|
||||
|
||||
class ExecuteShellCommandTool(Tool, ToolMarkerCanEdit):
|
||||
"""
|
||||
Executes a shell command.
|
||||
"""
|
||||
|
||||
def apply(
|
||||
self,
|
||||
command: str,
|
||||
cwd: str | None = None,
|
||||
capture_stderr: bool = True,
|
||||
max_answer_chars: int = TOOL_DEFAULT_MAX_ANSWER_LENGTH,
|
||||
) -> str:
|
||||
"""
|
||||
Execute a shell command and return its output.
|
||||
|
||||
IMPORTANT: you should always consider the memory about suggested shell commands before using this tool.
|
||||
If this memory was not loaded in the current conversation, you should load it using the `read_memory` tool
|
||||
before using this tool.
|
||||
|
||||
You should have at least once looked at the suggested shell commands from the corresponding memory
|
||||
created during the onboarding process before using this tool.
|
||||
Never execute unsafe shell commands like `rm -rf /` or similar! Generally be very careful with deletions.
|
||||
|
||||
:param command: the shell command to execute
|
||||
:param cwd: the working directory to execute the command in. If None, the project root will be used.
|
||||
:param capture_stderr: whether to capture and return stderr output
|
||||
:param max_answer_chars: if the output is longer than this number of characters,
|
||||
no content will be returned. Don't adjust unless there is really no other way to get the content
|
||||
required for the task.
|
||||
:return: a JSON object containing the command's stdout and optionally stderr output
|
||||
"""
|
||||
_cwd = cwd or self.get_project_root()
|
||||
result = execute_shell_command(command, cwd=_cwd, capture_stderr=capture_stderr)
|
||||
result = result.json()
|
||||
return self._limit_length(result, max_answer_chars)
|
||||
@@ -0,0 +1,89 @@
|
||||
import json
|
||||
|
||||
from serena.config.context_mode import SerenaAgentMode
|
||||
from serena.tools import Tool, ToolMarkerDoesNotRequireActiveProject
|
||||
|
||||
|
||||
class ActivateProjectTool(Tool, ToolMarkerDoesNotRequireActiveProject):
|
||||
"""
|
||||
Activates a project by name.
|
||||
"""
|
||||
|
||||
def apply(self, project: str) -> str:
|
||||
"""
|
||||
Activates the project with the given name.
|
||||
|
||||
:param project: the name of a registered project to activate or a path to a project directory
|
||||
"""
|
||||
from ..config.serena_config import ProjectConfig
|
||||
|
||||
active_project, new_project_generated, new_project_config_generated = self.agent.activate_project_from_path_or_name(project)
|
||||
if new_project_generated:
|
||||
result_str = (
|
||||
f"Created and activated a new project with name {active_project.project_name} at {active_project.project_root}, language: {active_project.project_config.language.value}. "
|
||||
+ "You can activate this project later by name."
|
||||
)
|
||||
else:
|
||||
result_str = f"Activated existing project with name {active_project.project_name} at {active_project.project_root}, language: {active_project.project_config.language.value}"
|
||||
if new_project_config_generated:
|
||||
result_str += (
|
||||
f"\nNote: A new project configuration was autogenerated because the given path did not contain a {ProjectConfig.SERENA_DEFAULT_PROJECT_FILE} file."
|
||||
+ f"You can now edit the project configuration in the file {active_project.path_to_project_yml()}. In particular, you may want to edit the project name and the initial prompt."
|
||||
)
|
||||
|
||||
if active_project.project_config.initial_prompt:
|
||||
result_str += f"\nAdditional project information:\n {active_project.project_config.initial_prompt}"
|
||||
result_str += (
|
||||
f"\nAvailable memories:\n {json.dumps(list(self.memories_manager.list_memories()))}"
|
||||
+ "You should not read these memories directly, but rather use the `read_memory` tool to read them later if needed for the task."
|
||||
)
|
||||
result_str += f"\nAvailable tools:\n {json.dumps(self.agent.get_active_tool_names())}"
|
||||
return result_str
|
||||
|
||||
|
||||
class RemoveProjectTool(Tool, ToolMarkerDoesNotRequireActiveProject):
|
||||
"""
|
||||
Removes a project from the Serena configuration.
|
||||
"""
|
||||
|
||||
def apply(self, project_name: str) -> str:
|
||||
"""
|
||||
Removes a project from the Serena configuration.
|
||||
|
||||
:param project_name: Name of the project to remove
|
||||
"""
|
||||
self.agent.serena_config.remove_project(project_name)
|
||||
return f"Successfully removed project '{project_name}' from configuration."
|
||||
|
||||
|
||||
class SwitchModesTool(Tool):
|
||||
"""
|
||||
Activates modes by providing a list of their names
|
||||
"""
|
||||
|
||||
def apply(self, modes: list[str]) -> str:
|
||||
"""
|
||||
Activates the desired modes, like ["editing", "interactive"] or ["planning", "one-shot"]
|
||||
|
||||
:param modes: the names of the modes to activate
|
||||
"""
|
||||
mode_instances = [SerenaAgentMode.load(mode) for mode in modes]
|
||||
self.agent.set_modes(mode_instances)
|
||||
|
||||
# Inform the Agent about the activated modes and the currently active tools
|
||||
result_str = f"Successfully activated modes: {', '.join([mode.name for mode in mode_instances])}" + "\n"
|
||||
result_str += "\n".join([mode_instance.prompt for mode_instance in mode_instances]) + "\n"
|
||||
result_str += f"Currently active tools: {', '.join(self.agent.get_active_tool_names())}"
|
||||
return result_str
|
||||
|
||||
|
||||
class GetCurrentConfigTool(Tool):
|
||||
"""
|
||||
Prints the current configuration of the agent, including the active and available projects, tools, contexts, and modes.
|
||||
"""
|
||||
|
||||
def apply(self) -> str:
|
||||
"""
|
||||
Print the current configuration of the agent, including the active and available projects, tools, contexts, and modes.
|
||||
"""
|
||||
return self.agent.get_current_config_overview()
|
||||
@@ -0,0 +1,393 @@
|
||||
"""
|
||||
File and file system-related tools, specifically for
|
||||
* listing directory contents
|
||||
* reading files
|
||||
* creating files
|
||||
* editing at the file level
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
from collections import defaultdict
|
||||
from fnmatch import fnmatch
|
||||
from pathlib import Path
|
||||
|
||||
from serena.text_utils import search_files
|
||||
from serena.tools import SUCCESS_RESULT, TOOL_DEFAULT_MAX_ANSWER_LENGTH, EditedFileContext, Tool, ToolMarkerCanEdit
|
||||
from serena.util.file_system import scan_directory
|
||||
|
||||
|
||||
class ReadFileTool(Tool):
|
||||
"""
|
||||
Reads a file within the project directory.
|
||||
"""
|
||||
|
||||
def apply(
|
||||
self, relative_path: str, start_line: int = 0, end_line: int | None = None, max_answer_chars: int = TOOL_DEFAULT_MAX_ANSWER_LENGTH
|
||||
) -> str:
|
||||
"""
|
||||
Reads the given file or a chunk of it. Generally, symbolic operations
|
||||
like find_symbol or find_referencing_symbols should be preferred if you know which symbols you are looking for.
|
||||
Reading the entire file is only recommended if there is no other way to get the content required for the task.
|
||||
|
||||
:param relative_path: the relative path to the file to read
|
||||
:param start_line: the 0-based index of the first line to be retrieved.
|
||||
:param end_line: the 0-based index of the last line to be retrieved (inclusive). If None, read until the end of the file.
|
||||
:param max_answer_chars: if the file (chunk) is longer than this number of characters,
|
||||
no content will be returned. Don't adjust unless there is really no other way to get the content
|
||||
required for the task.
|
||||
:return: the full text of the file at the given relative path
|
||||
"""
|
||||
self.agent.validate_relative_path(relative_path)
|
||||
|
||||
result = self.language_server.retrieve_full_file_content(relative_path)
|
||||
result_lines = result.splitlines()
|
||||
if end_line is None:
|
||||
result_lines = result_lines[start_line:]
|
||||
else:
|
||||
self.lines_read.add_lines_read(relative_path, (start_line, end_line))
|
||||
result_lines = result_lines[start_line : end_line + 1]
|
||||
result = "\n".join(result_lines)
|
||||
|
||||
return self._limit_length(result, max_answer_chars)
|
||||
|
||||
|
||||
class CreateTextFileTool(Tool, ToolMarkerCanEdit):
|
||||
"""
|
||||
Creates/overwrites a file in the project directory.
|
||||
"""
|
||||
|
||||
def apply(self, relative_path: str, content: str) -> str:
|
||||
"""
|
||||
Write a new file (or overwrite an existing file). For existing files, it is strongly recommended
|
||||
to use symbolic operations like replace_symbol_body or insert_after_symbol/insert_before_symbol, if possible.
|
||||
You can also use insert_at_line to insert content at a specific line for existing files if the symbolic operations
|
||||
are not the right choice for what you want to do.
|
||||
|
||||
If ever used on an existing file, the content has to be the complete content of that file (so it
|
||||
may never end with something like "The remaining content of the file is left unchanged.").
|
||||
For operations that just replace a part of a file, use the replace_lines or the symbolic editing tools instead.
|
||||
|
||||
:param relative_path: the relative path to the file to create
|
||||
:param content: the (utf-8-encoded) content to write to the file
|
||||
:return: a message indicating success or failure
|
||||
"""
|
||||
self.agent.validate_relative_path(relative_path)
|
||||
|
||||
abs_path = (Path(self.get_project_root()) / relative_path).resolve()
|
||||
will_overwrite_existing = abs_path.exists()
|
||||
|
||||
abs_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
abs_path.write_text(content, encoding="utf-8")
|
||||
answer = f"File created: {relative_path}."
|
||||
if will_overwrite_existing:
|
||||
answer += " Overwrote existing file."
|
||||
return answer
|
||||
|
||||
|
||||
class ListDirTool(Tool):
|
||||
"""
|
||||
Lists files and directories in the given directory (optionally with recursion).
|
||||
"""
|
||||
|
||||
def apply(self, relative_path: str, recursive: bool, max_answer_chars: int = TOOL_DEFAULT_MAX_ANSWER_LENGTH) -> str:
|
||||
"""
|
||||
Lists all non-gitignored files and directories in the given directory (optionally with recursion).
|
||||
|
||||
:param relative_path: the relative path to the directory to list; pass "." to scan the project root
|
||||
:param recursive: whether to scan subdirectories recursively
|
||||
:param max_answer_chars: if the output is longer than this number of characters,
|
||||
no content will be returned. Don't adjust unless there is really no other way to get the content
|
||||
required for the task.
|
||||
:return: a JSON object with the names of directories and files within the given directory
|
||||
"""
|
||||
self.agent.validate_relative_path(relative_path)
|
||||
|
||||
dirs, files = scan_directory(
|
||||
os.path.join(self.get_project_root(), relative_path),
|
||||
relative_to=self.get_project_root(),
|
||||
recursive=recursive,
|
||||
is_ignored_dir=self.agent.path_is_gitignored,
|
||||
is_ignored_file=self.agent.path_is_gitignored,
|
||||
)
|
||||
|
||||
result = json.dumps({"dirs": dirs, "files": files})
|
||||
return self._limit_length(result, max_answer_chars)
|
||||
|
||||
|
||||
class FindFileTool(Tool):
|
||||
"""
|
||||
Finds files in the given relative paths
|
||||
"""
|
||||
|
||||
def apply(self, file_mask: str, relative_path: str) -> str:
|
||||
"""
|
||||
Finds non-gitignored files matching the given file mask within the given relative path
|
||||
|
||||
:param file_mask: the filename or file mask (using the wildcards * or ?) to search for
|
||||
:param relative_path: the relative path to the directory to search in; pass "." to scan the project root
|
||||
:return: a JSON object with the list of matching files
|
||||
"""
|
||||
self.agent.validate_relative_path(relative_path)
|
||||
|
||||
dir_to_scan = os.path.join(self.get_project_root(), relative_path)
|
||||
|
||||
# find the files by ignoring everything that doesn't match
|
||||
def is_ignored_file(abs_path: str) -> bool:
|
||||
if self.agent.path_is_gitignored(abs_path):
|
||||
return True
|
||||
filename = os.path.basename(abs_path)
|
||||
return not fnmatch(filename, file_mask)
|
||||
|
||||
dirs, files = scan_directory(
|
||||
path=dir_to_scan,
|
||||
recursive=True,
|
||||
is_ignored_dir=self.agent.path_is_gitignored,
|
||||
is_ignored_file=is_ignored_file,
|
||||
relative_to=self.get_project_root(),
|
||||
)
|
||||
|
||||
result = json.dumps({"files": files})
|
||||
return result
|
||||
|
||||
|
||||
class ReplaceRegexTool(Tool, ToolMarkerCanEdit):
|
||||
"""
|
||||
Replaces content in a file by using regular expressions.
|
||||
"""
|
||||
|
||||
def apply(
|
||||
self,
|
||||
relative_path: str,
|
||||
regex: str,
|
||||
repl: str,
|
||||
allow_multiple_occurrences: bool = False,
|
||||
) -> str:
|
||||
r"""
|
||||
Replaces one or more occurrences of the given regular expression.
|
||||
This is the preferred way to replace content in a file whenever the symbol-level
|
||||
tools are not appropriate.
|
||||
Even large sections of code can be replaced by providing a concise regular expression of
|
||||
the form "beginning.*?end-of-text-to-be-replaced".
|
||||
Always try to use wildcards to avoid specifying the exact content of the code to be replaced,
|
||||
especially if it spans several lines.
|
||||
|
||||
IMPORTANT: REMEMBER TO USE WILDCARDS WHEN APPROPRIATE! I WILL BE VERY UNHAPPY IF YOU WRITE LONG REGEXES WITHOUT USING WILDCARDS INSTEAD!
|
||||
|
||||
:param relative_path: the relative path to the file
|
||||
:param regex: a Python-style regular expression, matches of which will be replaced.
|
||||
Dot matches all characters, multi-line matching is enabled.
|
||||
:param repl: the string to replace the matched content with, which may contain
|
||||
backreferences like \1, \2, etc.
|
||||
:param allow_multiple_occurrences: if True, the regex may match multiple occurrences in the file
|
||||
and all of them will be replaced.
|
||||
If this is set to False and the regex matches multiple occurrences, an error will be returned
|
||||
(and you may retry with a revised, more specific regex).
|
||||
"""
|
||||
self.agent.validate_relative_path(relative_path)
|
||||
with EditedFileContext(relative_path, self.agent) as context:
|
||||
original_content = context.get_original_content()
|
||||
updated_content, n = re.subn(regex, repl, original_content, flags=re.DOTALL | re.MULTILINE)
|
||||
if n == 0:
|
||||
return f"Error: No matches found for regex '{regex}' in file '{relative_path}'."
|
||||
if not allow_multiple_occurrences and n > 1:
|
||||
return (
|
||||
f"Error: Regex '{regex}' matches {n} occurrences in file '{relative_path}'. "
|
||||
"Please revise the regex to be more specific or enable allow_multiple_occurrences if this is expected."
|
||||
)
|
||||
context.set_updated_content(updated_content)
|
||||
return SUCCESS_RESULT
|
||||
|
||||
|
||||
class DeleteLinesTool(Tool, ToolMarkerCanEdit):
|
||||
"""
|
||||
Deletes a range of lines within a file.
|
||||
"""
|
||||
|
||||
def apply(
|
||||
self,
|
||||
relative_path: str,
|
||||
start_line: int,
|
||||
end_line: int,
|
||||
) -> str:
|
||||
"""
|
||||
Deletes the given lines in the file.
|
||||
Requires that the same range of lines was previously read using the `read_file` tool to verify correctness
|
||||
of the operation.
|
||||
|
||||
:param relative_path: the relative path to the file
|
||||
:param start_line: the 0-based index of the first line to be deleted
|
||||
:param end_line: the 0-based index of the last line to be deleted
|
||||
"""
|
||||
if not self.lines_read.were_lines_read(relative_path, (start_line, end_line)):
|
||||
read_lines_tool = self.agent.get_tool(ReadFileTool)
|
||||
return f"Error: Must call `{read_lines_tool.get_name_from_cls()}` first to read exactly the affected lines."
|
||||
self.symbol_manager.delete_lines(relative_path, start_line, end_line)
|
||||
return SUCCESS_RESULT
|
||||
|
||||
|
||||
class ReplaceLinesTool(Tool, ToolMarkerCanEdit):
|
||||
"""
|
||||
Replaces a range of lines within a file with new content.
|
||||
"""
|
||||
|
||||
def apply(
|
||||
self,
|
||||
relative_path: str,
|
||||
start_line: int,
|
||||
end_line: int,
|
||||
content: str,
|
||||
) -> str:
|
||||
"""
|
||||
Replaces the given range of lines in the given file.
|
||||
Requires that the same range of lines was previously read using the `read_file` tool to verify correctness
|
||||
of the operation.
|
||||
|
||||
:param relative_path: the relative path to the file
|
||||
:param start_line: the 0-based index of the first line to be deleted
|
||||
:param end_line: the 0-based index of the last line to be deleted
|
||||
:param content: the content to insert
|
||||
"""
|
||||
if not content.endswith("\n"):
|
||||
content += "\n"
|
||||
result = self.agent.get_tool(DeleteLinesTool).apply(relative_path, start_line, end_line)
|
||||
if result != SUCCESS_RESULT:
|
||||
return result
|
||||
self.agent.get_tool(InsertAtLineTool).apply(relative_path, start_line, content)
|
||||
return SUCCESS_RESULT
|
||||
|
||||
|
||||
class InsertAtLineTool(Tool, ToolMarkerCanEdit):
|
||||
"""
|
||||
Inserts content at a given line in a file.
|
||||
"""
|
||||
|
||||
def apply(
|
||||
self,
|
||||
relative_path: str,
|
||||
line: int,
|
||||
content: str,
|
||||
) -> str:
|
||||
"""
|
||||
Inserts the given content at the given line in the file, pushing existing content of the line down.
|
||||
In general, symbolic insert operations like insert_after_symbol or insert_before_symbol should be preferred if you know which
|
||||
symbol you are looking for.
|
||||
However, this can also be useful for small targeted edits of the body of a longer symbol (without replacing the entire body).
|
||||
|
||||
:param relative_path: the relative path to the file
|
||||
:param line: the 0-based index of the line to insert content at
|
||||
:param content: the content to be inserted
|
||||
"""
|
||||
if not content.endswith("\n"):
|
||||
content += "\n"
|
||||
self.symbol_manager.insert_at_line(relative_path, line, content)
|
||||
return SUCCESS_RESULT
|
||||
|
||||
|
||||
class SearchForPatternTool(Tool):
|
||||
"""
|
||||
Performs a search for a pattern in the project.
|
||||
"""
|
||||
|
||||
def apply(
|
||||
self,
|
||||
substring_pattern: str,
|
||||
context_lines_before: int = 0,
|
||||
context_lines_after: int = 0,
|
||||
paths_include_glob: str | None = None,
|
||||
paths_exclude_glob: str | None = None,
|
||||
relative_path: str = "",
|
||||
restrict_search_to_code_files: bool = False,
|
||||
max_answer_chars: int = TOOL_DEFAULT_MAX_ANSWER_LENGTH,
|
||||
) -> str:
|
||||
"""
|
||||
Offers a flexible search for arbitrary patterns in the codebase, including the
|
||||
possibility to search in non-code files.
|
||||
Generally, symbolic operations like find_symbol or find_referencing_symbols
|
||||
should be preferred if you know which symbols you are looking for.
|
||||
|
||||
Pattern Matching Logic:
|
||||
For each match, the returned result will contain the full lines where the
|
||||
substring pattern is found, as well as optionally some lines before and after it. The pattern will be compiled with
|
||||
DOTALL, meaning that the dot will match all characters including newlines.
|
||||
This also means that it never makes sense to have .* at the beginning or end of the pattern,
|
||||
but it may make sense to have it in the middle for complex patterns.
|
||||
If a pattern matches multiple lines, all those lines will be part of the match.
|
||||
Be careful to not use greedy quantifiers unnecessarily, it is usually better to use non-greedy quantifiers like .*? to avoid
|
||||
matching too much content.
|
||||
|
||||
File Selection Logic:
|
||||
The files in which the search is performed can be restricted very flexibly.
|
||||
Using `restrict_search_to_code_files` is useful if you are only interested in code symbols (i.e., those
|
||||
symbols that can be manipulated with symbolic tools like find_symbol).
|
||||
You can also restrict the search to a specific file or directory,
|
||||
and provide glob patterns to include or exclude certain files on top of that.
|
||||
The globs are matched against relative file paths from the project root (not to the `relative_path` parameter that
|
||||
is used to further restrict the search).
|
||||
Smartly combining the various restrictions allows you to perform very targeted searches.
|
||||
|
||||
|
||||
:param substring_pattern: Regular expression for a substring pattern to search for
|
||||
:param context_lines_before: Number of lines of context to include before each match
|
||||
:param context_lines_after: Number of lines of context to include after each match
|
||||
:param paths_include_glob: optional glob pattern specifying files to include in the search.
|
||||
Matches against relative file paths from the project root (e.g., "*.py", "src/**/*.ts").
|
||||
Only matches files, not directories.
|
||||
:param paths_exclude_glob: optional glob pattern specifying files to exclude from the search.
|
||||
Matches against relative file paths from the project root (e.g., "*test*", "**/*_generated.py").
|
||||
Takes precedence over paths_include_glob. Only matches files, not directories.
|
||||
:param relative_path: only subpaths of this path (relative to the repo root) will be analyzed. If a path to a single
|
||||
file is passed, only that will be searched. The path must exist, otherwise a `FileNotFoundError` is raised.
|
||||
:param max_answer_chars: if the output is longer than this number of characters,
|
||||
no content will be returned. Don't adjust unless there is really no other way to get the content
|
||||
required for the task. Instead, if the output is too long, you should
|
||||
make a stricter query.
|
||||
:param restrict_search_to_code_files: whether to restrict the search to only those files where
|
||||
analyzed code symbols can be found. Otherwise, will search all non-ignored files.
|
||||
Set this to True if your search is only meant to discover code that can be manipulated with symbolic tools.
|
||||
For example, for finding classes or methods from a name pattern.
|
||||
Setting to False is a better choice if you also want to search in non-code files, like in html or yaml files,
|
||||
which is why it is the default.
|
||||
:return: A JSON object mapping file paths to lists of matched consecutive lines (with context, if requested).
|
||||
"""
|
||||
abs_path = os.path.join(self.get_project_root(), relative_path)
|
||||
if not os.path.exists(abs_path):
|
||||
raise FileNotFoundError(f"Relative path {relative_path} does not exist.")
|
||||
|
||||
if restrict_search_to_code_files:
|
||||
matches = self.language_server.search_files_for_pattern(
|
||||
pattern=substring_pattern,
|
||||
relative_path=relative_path,
|
||||
context_lines_before=context_lines_before,
|
||||
context_lines_after=context_lines_after,
|
||||
paths_include_glob=paths_include_glob,
|
||||
paths_exclude_glob=paths_exclude_glob,
|
||||
)
|
||||
else:
|
||||
if os.path.isfile(abs_path):
|
||||
rel_paths_to_search = [relative_path]
|
||||
else:
|
||||
dirs, rel_paths_to_search = scan_directory(
|
||||
path=abs_path,
|
||||
recursive=True,
|
||||
is_ignored_dir=self.agent.path_is_gitignored,
|
||||
is_ignored_file=self.agent.path_is_gitignored,
|
||||
relative_to=self.get_project_root(),
|
||||
)
|
||||
# TODO (maybe): not super efficient to walk through the files again and filter if glob patterns are provided
|
||||
# but it probably never matters and this version required no further refactoring
|
||||
matches = search_files(
|
||||
rel_paths_to_search,
|
||||
substring_pattern,
|
||||
root_path=self.get_project_root(),
|
||||
paths_include_glob=paths_include_glob,
|
||||
paths_exclude_glob=paths_exclude_glob,
|
||||
)
|
||||
# group matches by file
|
||||
file_to_matches: dict[str, list[str]] = defaultdict(list)
|
||||
for match in matches:
|
||||
assert match.source_file_path is not None
|
||||
file_to_matches[match.source_file_path].append(match.to_display_string())
|
||||
result = json.dumps(file_to_matches)
|
||||
return self._limit_length(result, max_answer_chars)
|
||||
@@ -0,0 +1,279 @@
|
||||
"""
|
||||
Language server-related tools
|
||||
"""
|
||||
|
||||
import json
|
||||
from collections.abc import Sequence
|
||||
from copy import copy
|
||||
from typing import Any
|
||||
|
||||
from serena.tools import SUCCESS_RESULT, TOOL_DEFAULT_MAX_ANSWER_LENGTH, Tool, ToolMarkerCanEdit
|
||||
from solidlsp.ls_types import SymbolKind
|
||||
|
||||
|
||||
def _sanitize_symbol_dict(symbol_dict: dict[str, Any]) -> dict[str, Any]:
|
||||
"""
|
||||
Sanitize a symbol dictionary inplace by removing unnecessary information.
|
||||
"""
|
||||
# We replace the location entry, which repeats line information already included in body_location
|
||||
# and has unnecessary information on column, by just the relative path.
|
||||
symbol_dict = copy(symbol_dict)
|
||||
s_relative_path = symbol_dict.get("location", {}).get("relative_path")
|
||||
if s_relative_path is not None:
|
||||
symbol_dict["relative_path"] = s_relative_path
|
||||
symbol_dict.pop("location", None)
|
||||
# also remove name, name_path should be enough
|
||||
symbol_dict.pop("name")
|
||||
return symbol_dict
|
||||
|
||||
|
||||
class RestartLanguageServerTool(Tool):
|
||||
"""Restarts the language server, may be necessary when edits not through Serena happen."""
|
||||
|
||||
def apply(self) -> str:
|
||||
"""Use this tool only on explicit user request or after confirmation.
|
||||
It may be necessary to restart the language server if the user performs edits
|
||||
not through Serena, so the language server state becomes outdated and further editing attempts lead to errors.
|
||||
|
||||
If such editing errors happen, you should suggest using this tool.
|
||||
"""
|
||||
self.agent.reset_language_server()
|
||||
return SUCCESS_RESULT
|
||||
|
||||
|
||||
class GetSymbolsOverviewTool(Tool):
|
||||
"""
|
||||
Gets an overview of the top-level symbols defined in a given file or directory.
|
||||
"""
|
||||
|
||||
def apply(self, relative_path: str, max_answer_chars: int = TOOL_DEFAULT_MAX_ANSWER_LENGTH) -> str:
|
||||
"""
|
||||
Gets an overview of the given file or directory.
|
||||
For each analyzed file, we list the top-level symbols in the file (name_path, kind).
|
||||
Use this tool to get a high-level understanding of the code symbols.
|
||||
Calling this is often a good idea before more targeted reading, searching or editing operations on the code symbols.
|
||||
|
||||
:param relative_path: the relative path to the file or directory to get the overview of
|
||||
:param max_answer_chars: if the overview is longer than this number of characters,
|
||||
no content will be returned. Don't adjust unless there is really no other way to get the content
|
||||
required for the task. If the overview is too long, you should use a smaller directory instead,
|
||||
(e.g. a subdirectory).
|
||||
:return: a JSON object mapping relative paths of all contained files to info about top-level symbols in the file (name_path, kind).
|
||||
"""
|
||||
path_to_symbol_infos = self.language_server.request_overview(relative_path)
|
||||
result = {}
|
||||
for file_path, symbols in path_to_symbol_infos.items():
|
||||
# TODO: maybe include not just top-level symbols? We could filter by kind to exclude variables
|
||||
# The language server methods would need to be adjusted for this.
|
||||
result[file_path] = [{"name_path": symbol[0], "kind": int(symbol[1])} for symbol in symbols]
|
||||
|
||||
result_json_str = json.dumps(result)
|
||||
return self._limit_length(result_json_str, max_answer_chars)
|
||||
|
||||
|
||||
class FindSymbolTool(Tool):
|
||||
"""
|
||||
Performs a global (or local) search for symbols with/containing a given name/substring (optionally filtered by type).
|
||||
"""
|
||||
|
||||
def apply(
|
||||
self,
|
||||
name_path: str,
|
||||
depth: int = 0,
|
||||
relative_path: str | None = None,
|
||||
include_body: bool = False,
|
||||
include_kinds: list[int] | None = None,
|
||||
exclude_kinds: list[int] | None = None,
|
||||
substring_matching: bool = False,
|
||||
max_answer_chars: int = TOOL_DEFAULT_MAX_ANSWER_LENGTH,
|
||||
) -> str:
|
||||
"""
|
||||
Retrieves information on all symbols/code entities (classes, methods, etc.) based on the given `name_path`,
|
||||
which represents a pattern for the symbol's path within the symbol tree of a single file.
|
||||
The returned symbol location can be used for edits or further queries.
|
||||
Specify `depth > 0` to retrieve children (e.g., methods of a class).
|
||||
|
||||
The matching behavior is determined by the structure of `name_path`, which can
|
||||
either be a simple name (e.g. "method") or a name path like "class/method" (relative name path)
|
||||
or "/class/method" (absolute name path). Note that the name path is not a path in the file system
|
||||
but rather a path in the symbol tree **within a single file**. Thus, file or directory names should never
|
||||
be included in the `name_path`. For restricting the search to a single file or directory,
|
||||
the `within_relative_path` parameter should be used instead. The retrieved symbols' `name_path` attribute
|
||||
will always be composed of symbol names, never file or directory names.
|
||||
|
||||
Key aspects of the name path matching behavior:
|
||||
- Trailing slashes in `name_path` play no role and are ignored.
|
||||
- The name of the retrieved symbols will match (either exactly or as a substring)
|
||||
the last segment of `name_path`, while other segments will restrict the search to symbols that
|
||||
have a desired sequence of ancestors.
|
||||
- If there is no starting or intermediate slash in `name_path`, there is no
|
||||
restriction on the ancestor symbols. For example, passing `method` will match
|
||||
against symbols with name paths like `method`, `class/method`, `class/nested_class/method`, etc.
|
||||
- If `name_path` contains a `/` but doesn't start with a `/`, the matching is restricted to symbols
|
||||
with the same ancestors as the last segment of `name_path`. For example, passing `class/method` will match against
|
||||
`class/method` as well as `nested_class/class/method` but not `method`.
|
||||
- If `name_path` starts with a `/`, it will be treated as an absolute name path pattern, meaning
|
||||
that the first segment of it must match the first segment of the symbol's name path.
|
||||
For example, passing `/class` will match only against top-level symbols like `class` but not against `nested_class/class`.
|
||||
Passing `/class/method` will match against `class/method` but not `nested_class/class/method` or `method`.
|
||||
|
||||
|
||||
:param name_path: The name path pattern to search for, see above for details.
|
||||
:param depth: Depth to retrieve descendants (e.g., 1 for class methods/attributes).
|
||||
:param relative_path: Optional. Restrict search to this file or directory. If None, searches entire codebase.
|
||||
If a directory is passed, the search will be restricted to the files in that directory.
|
||||
If a file is passed, the search will be restricted to that file.
|
||||
If you have some knowledge about the codebase, you should use this parameter, as it will significantly
|
||||
speed up the search as well as reduce the number of results.
|
||||
:param include_body: If True, include the symbol's source code. Use judiciously.
|
||||
:param include_kinds: Optional. List of LSP symbol kind integers to include. (e.g., 5 for Class, 12 for Function).
|
||||
Valid kinds: 1=file, 2=module, 3=namespace, 4=package, 5=class, 6=method, 7=property, 8=field, 9=constructor, 10=enum,
|
||||
11=interface, 12=function, 13=variable, 14=constant, 15=string, 16=number, 17=boolean, 18=array, 19=object,
|
||||
20=key, 21=null, 22=enum member, 23=struct, 24=event, 25=operator, 26=type parameter
|
||||
:param exclude_kinds: Optional. List of LSP symbol kind integers to exclude. Takes precedence over `include_kinds`.
|
||||
:param substring_matching: If True, use substring matching for the last segment of `name`.
|
||||
:param max_answer_chars: Max characters for the JSON result. If exceeded, no content is returned.
|
||||
:return: JSON string: a list of symbols (with locations) matching the name.
|
||||
"""
|
||||
parsed_include_kinds: Sequence[SymbolKind] | None = [SymbolKind(k) for k in include_kinds] if include_kinds else None
|
||||
parsed_exclude_kinds: Sequence[SymbolKind] | None = [SymbolKind(k) for k in exclude_kinds] if exclude_kinds else None
|
||||
symbols = self.symbol_manager.find_by_name(
|
||||
name_path,
|
||||
include_body=include_body,
|
||||
include_kinds=parsed_include_kinds,
|
||||
exclude_kinds=parsed_exclude_kinds,
|
||||
substring_matching=substring_matching,
|
||||
within_relative_path=relative_path,
|
||||
)
|
||||
symbol_dicts = [_sanitize_symbol_dict(s.to_dict(kind=True, location=True, depth=depth, include_body=include_body)) for s in symbols]
|
||||
result = json.dumps(symbol_dicts)
|
||||
return self._limit_length(result, max_answer_chars)
|
||||
|
||||
|
||||
class FindReferencingSymbolsTool(Tool):
|
||||
"""
|
||||
Finds symbols that reference the symbol at the given location (optionally filtered by type).
|
||||
"""
|
||||
|
||||
def apply(
|
||||
self,
|
||||
name_path: str,
|
||||
relative_path: str,
|
||||
include_kinds: list[int] | None = None,
|
||||
exclude_kinds: list[int] | None = None,
|
||||
max_answer_chars: int = TOOL_DEFAULT_MAX_ANSWER_LENGTH,
|
||||
) -> str:
|
||||
"""
|
||||
Finds symbols that reference the symbol at the given `name_path`. The result will contain metadata about the referencing symbols
|
||||
as well as a short code snippet around the reference (unless `include_body` is True, then the short snippet will be omitted).
|
||||
Note that among other kinds of references, this function can be used to find (direct) subclasses of a class,
|
||||
as subclasses are referencing symbols that have the kind class.
|
||||
|
||||
:param name_path: for finding the symbol to find references for, same logic as in the `find_symbol` tool.
|
||||
:param relative_path: the relative path to the file containing the symbol for which to find references.
|
||||
Note that here you can't pass a directory but must pass a file.
|
||||
:param include_kinds: same as in the `find_symbol` tool.
|
||||
:param exclude_kinds: same as in the `find_symbol` tool.
|
||||
:param max_answer_chars: same as in the `find_symbol` tool.
|
||||
:return: a list of JSON objects with the symbols referencing the requested symbol
|
||||
"""
|
||||
include_body = False # It is probably never a good idea to include the body of the referencing symbols
|
||||
parsed_include_kinds: Sequence[SymbolKind] | None = [SymbolKind(k) for k in include_kinds] if include_kinds else None
|
||||
parsed_exclude_kinds: Sequence[SymbolKind] | None = [SymbolKind(k) for k in exclude_kinds] if exclude_kinds else None
|
||||
references_in_symbols = self.symbol_manager.find_referencing_symbols(
|
||||
name_path,
|
||||
relative_file_path=relative_path,
|
||||
include_body=include_body,
|
||||
include_kinds=parsed_include_kinds,
|
||||
exclude_kinds=parsed_exclude_kinds,
|
||||
)
|
||||
reference_dicts = []
|
||||
for ref in references_in_symbols:
|
||||
ref_dict = ref.symbol.to_dict(kind=True, location=True, depth=0, include_body=include_body)
|
||||
ref_dict = _sanitize_symbol_dict(ref_dict)
|
||||
if not include_body:
|
||||
ref_relative_path = ref.symbol.location.relative_path
|
||||
assert ref_relative_path is not None, f"Referencing symbol {ref.symbol.name} has no relative path, this is likely a bug."
|
||||
content_around_ref = self.language_server.retrieve_content_around_line(
|
||||
relative_file_path=ref_relative_path, line=ref.line, context_lines_before=1, context_lines_after=1
|
||||
)
|
||||
ref_dict["content_around_reference"] = content_around_ref.to_display_string()
|
||||
reference_dicts.append(ref_dict)
|
||||
result = json.dumps(reference_dicts)
|
||||
return self._limit_length(result, max_answer_chars)
|
||||
|
||||
|
||||
class ReplaceSymbolBodyTool(Tool, ToolMarkerCanEdit):
|
||||
"""
|
||||
Replaces the full definition of a symbol.
|
||||
"""
|
||||
|
||||
def apply(
|
||||
self,
|
||||
name_path: str,
|
||||
relative_path: str,
|
||||
body: str,
|
||||
) -> str:
|
||||
r"""
|
||||
Replaces the body of the symbol with the given `name_path`.
|
||||
|
||||
:param name_path: for finding the symbol to replace, same logic as in the `find_symbol` tool.
|
||||
:param relative_path: the relative path to the file containing the symbol
|
||||
:param body: the new symbol body. Important: Begin directly with the symbol definition and provide no
|
||||
leading indentation for the first line (but do indent the rest of the body according to the context).
|
||||
"""
|
||||
self.symbol_manager.replace_body(
|
||||
name_path,
|
||||
relative_file_path=relative_path,
|
||||
body=body,
|
||||
use_same_indentation=False,
|
||||
)
|
||||
return SUCCESS_RESULT
|
||||
|
||||
|
||||
class InsertAfterSymbolTool(Tool, ToolMarkerCanEdit):
|
||||
"""
|
||||
Inserts content after the end of the definition of a given symbol.
|
||||
"""
|
||||
|
||||
def apply(
|
||||
self,
|
||||
name_path: str,
|
||||
relative_path: str,
|
||||
body: str,
|
||||
) -> str:
|
||||
"""
|
||||
Inserts the given body/content after the end of the definition of the given symbol (via the symbol's location).
|
||||
A typical use case is to insert a new class, function, method, field or variable assignment.
|
||||
|
||||
:param name_path: name path of the symbol after which to insert content (definitions in the `find_symbol` tool apply)
|
||||
:param relative_path: the relative path to the file containing the symbol
|
||||
:param body: the body/content to be inserted. The inserted code shall begin with the next line after
|
||||
the symbol.
|
||||
"""
|
||||
self.symbol_manager.insert_after_symbol(name_path, relative_file_path=relative_path, body=body, use_same_indentation=False)
|
||||
return SUCCESS_RESULT
|
||||
|
||||
|
||||
class InsertBeforeSymbolTool(Tool, ToolMarkerCanEdit):
|
||||
"""
|
||||
Inserts content before the beginning of the definition of a given symbol.
|
||||
"""
|
||||
|
||||
def apply(
|
||||
self,
|
||||
name_path: str,
|
||||
relative_path: str,
|
||||
body: str,
|
||||
) -> str:
|
||||
"""
|
||||
Inserts the given body/content before the beginning of the definition of the given symbol (via the symbol's location).
|
||||
A typical use case is to insert a new class, function, method, field or variable assignment.
|
||||
It also can be used to insert a new import statement before the first symbol in the file.
|
||||
|
||||
:param name_path: name path of the symbol before which to insert content (definitions in the `find_symbol` tool apply)
|
||||
:param relative_path: the relative path to the file containing the symbol
|
||||
:param body: the body/content to be inserted before the line in which the referenced symbol is defined
|
||||
"""
|
||||
self.symbol_manager.insert_before_symbol(name_path, relative_file_path=relative_path, body=body, use_same_indentation=False)
|
||||
return SUCCESS_RESULT
|
||||
@@ -0,0 +1,69 @@
|
||||
import json
|
||||
|
||||
from serena.tools import TOOL_DEFAULT_MAX_ANSWER_LENGTH, Tool
|
||||
|
||||
|
||||
class WriteMemoryTool(Tool):
|
||||
"""
|
||||
Writes a named memory (for future reference) to Serena's project-specific memory store.
|
||||
"""
|
||||
|
||||
def apply(self, memory_name: str, content: str, max_answer_chars: int = TOOL_DEFAULT_MAX_ANSWER_LENGTH) -> str:
|
||||
"""
|
||||
Write some information about this project that can be useful for future tasks to a memory.
|
||||
Use markdown formatting for the content.
|
||||
The information should be short and to the point.
|
||||
The memory name should be meaningful, such that from the name you can infer what the information is about.
|
||||
It is better to have multiple small memories than to have a single large one because
|
||||
memories will be read one by one and we only ever want to read relevant memories.
|
||||
|
||||
This tool is either called during the onboarding process or when you have identified
|
||||
something worth remembering about the project from the past conversation.
|
||||
"""
|
||||
if len(content) > max_answer_chars:
|
||||
raise ValueError(
|
||||
f"Content for {memory_name} is too long. Max length is {max_answer_chars} characters. " + "Please make the content shorter."
|
||||
)
|
||||
|
||||
return self.memories_manager.save_memory(memory_name, content)
|
||||
|
||||
|
||||
class ReadMemoryTool(Tool):
|
||||
"""
|
||||
Reads the memory with the given name from Serena's project-specific memory store.
|
||||
"""
|
||||
|
||||
def apply(self, memory_file_name: str, max_answer_chars: int = TOOL_DEFAULT_MAX_ANSWER_LENGTH) -> str:
|
||||
"""
|
||||
Read the content of a memory file. This tool should only be used if the information
|
||||
is relevant to the current task. You can infer whether the information
|
||||
is relevant from the memory file name.
|
||||
You should not read the same memory file multiple times in the same conversation.
|
||||
"""
|
||||
return self.memories_manager.load_memory(memory_file_name)
|
||||
|
||||
|
||||
class ListMemoriesTool(Tool):
|
||||
"""
|
||||
Lists memories in Serena's project-specific memory store.
|
||||
"""
|
||||
|
||||
def apply(self) -> str:
|
||||
"""
|
||||
List available memories. Any memory can be read using the `read_memory` tool.
|
||||
"""
|
||||
return json.dumps(self.memories_manager.list_memories())
|
||||
|
||||
|
||||
class DeleteMemoryTool(Tool):
|
||||
"""
|
||||
Deletes a memory from Serena's project-specific memory store.
|
||||
"""
|
||||
|
||||
def apply(self, memory_file_name: str) -> str:
|
||||
"""
|
||||
Delete a memory file. Should only happen if a user asks for it explicitly,
|
||||
for example by saying that the information retrieved from a memory file is no longer correct
|
||||
or no longer relevant for the project.
|
||||
"""
|
||||
return self.memories_manager.delete_memory(memory_file_name)
|
||||
@@ -0,0 +1,372 @@
|
||||
import inspect
|
||||
import os
|
||||
import traceback
|
||||
from abc import ABC, abstractmethod
|
||||
from collections.abc import Callable, Generator, Iterable
|
||||
from copy import copy
|
||||
from types import TracebackType
|
||||
from typing import TYPE_CHECKING, Any, Self, TypeVar
|
||||
|
||||
from mcp.server.fastmcp.utilities.func_metadata import FuncMetadata, func_metadata
|
||||
from sensai.util import logging
|
||||
from sensai.util.string import dict_string
|
||||
|
||||
from serena.prompt_factory import PromptFactory
|
||||
from serena.symbol import SymbolManager
|
||||
from serena.util.inspection import iter_subclasses
|
||||
from solidlsp import SolidLanguageServer
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from serena.agent import LinesRead, MemoriesManager, SerenaAgent
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
T = TypeVar("T")
|
||||
SUCCESS_RESULT = "OK"
|
||||
|
||||
|
||||
class Component(ABC):
|
||||
def __init__(self, agent: "SerenaAgent"):
|
||||
self.agent = agent
|
||||
|
||||
@property
|
||||
def language_server(self) -> SolidLanguageServer:
|
||||
assert self.agent.language_server is not None
|
||||
return self.agent.language_server
|
||||
|
||||
def get_project_root(self) -> str:
|
||||
"""
|
||||
:return: the root directory of the active project, raises a ValueError if no active project configuration is set
|
||||
"""
|
||||
return self.agent.get_project_root()
|
||||
|
||||
@property
|
||||
def prompt_factory(self) -> PromptFactory:
|
||||
return self.agent.prompt_factory
|
||||
|
||||
@property
|
||||
def memories_manager(self) -> "MemoriesManager":
|
||||
assert self.agent.memories_manager is not None
|
||||
return self.agent.memories_manager
|
||||
|
||||
@property
|
||||
def symbol_manager(self) -> SymbolManager:
|
||||
assert self.agent.symbol_manager is not None
|
||||
return self.agent.symbol_manager
|
||||
|
||||
@property
|
||||
def lines_read(self) -> "LinesRead":
|
||||
assert self.agent.lines_read is not None
|
||||
return self.agent.lines_read
|
||||
|
||||
|
||||
TOOL_DEFAULT_MAX_ANSWER_LENGTH = int(2e5)
|
||||
|
||||
|
||||
class ToolMarkerCanEdit:
|
||||
"""
|
||||
Marker class for all tools that can perform editing operations on files.
|
||||
"""
|
||||
|
||||
|
||||
class ToolMarkerDoesNotRequireActiveProject:
|
||||
pass
|
||||
|
||||
|
||||
class ToolInterface(ABC):
|
||||
"""Protocol defining the complete interface that make_tool() expects from a tool."""
|
||||
|
||||
@abstractmethod
|
||||
def get_name(self) -> str:
|
||||
"""Get the tool name."""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def get_apply_docstring(self) -> str:
|
||||
"""Get the docstring for the tool application, used by the MCP server."""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def get_apply_fn_metadata(self) -> FuncMetadata:
|
||||
"""Get the metadata for the tool application function, used by the MCP server."""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def apply_ex(self, log_call: bool = True, catch_exceptions: bool = True, **kwargs: Any) -> str:
|
||||
"""Apply the tool with logging and exception handling."""
|
||||
...
|
||||
|
||||
|
||||
class Tool(Component, ToolInterface):
|
||||
# NOTE: each tool should implement the apply method, which is then used in
|
||||
# the central method of the Tool class `apply_ex`.
|
||||
# Failure to do so will result in a RuntimeError at tool execution time.
|
||||
# The apply method is not declared as part of the base Tool interface since we cannot
|
||||
# know the signature of the (input parameters of the) method in advance.
|
||||
#
|
||||
# The docstring and types of the apply method are used to generate the tool description
|
||||
# (which is use by the LLM, so a good description is important)
|
||||
# and to validate the tool call arguments.
|
||||
|
||||
@classmethod
|
||||
def get_name_from_cls(cls) -> str:
|
||||
name = cls.__name__
|
||||
if name.endswith("Tool"):
|
||||
name = name[:-4]
|
||||
# convert to snake_case
|
||||
name = "".join(["_" + c.lower() if c.isupper() else c for c in name]).lstrip("_")
|
||||
return name
|
||||
|
||||
def get_name(self) -> str:
|
||||
return self.get_name_from_cls()
|
||||
|
||||
def get_apply_fn(self) -> Callable:
|
||||
apply_fn = getattr(self, "apply")
|
||||
if apply_fn is None:
|
||||
raise RuntimeError(f"apply not defined in {self}. Did you forget to implement it?")
|
||||
return apply_fn
|
||||
|
||||
@classmethod
|
||||
def can_edit(cls) -> bool:
|
||||
"""
|
||||
Returns whether this tool can perform editing operations on code.
|
||||
|
||||
:return: True if the tool can edit code, False otherwise
|
||||
"""
|
||||
return issubclass(cls, ToolMarkerCanEdit)
|
||||
|
||||
@classmethod
|
||||
def get_tool_description(cls) -> str:
|
||||
docstring = cls.__doc__
|
||||
if docstring is None:
|
||||
return ""
|
||||
return docstring.strip()
|
||||
|
||||
@classmethod
|
||||
def get_apply_docstring_from_cls(cls) -> str:
|
||||
"""Get the docstring for the apply method from the class (static metadata).
|
||||
Needed for creating MCP tools in a separate process without running into serialization issues.
|
||||
"""
|
||||
# First try to get from __dict__ to handle dynamic docstring changes
|
||||
if "apply" in cls.__dict__:
|
||||
apply_fn = cls.__dict__["apply"]
|
||||
else:
|
||||
# Fall back to getattr for inherited methods
|
||||
apply_fn = getattr(cls, "apply", None)
|
||||
if apply_fn is None:
|
||||
raise AttributeError(f"apply method not defined in {cls}. Did you forget to implement it?")
|
||||
|
||||
docstring = apply_fn.__doc__
|
||||
if not docstring:
|
||||
raise AttributeError(f"apply method has no (or empty) docstring in {cls}. Did you forget to implement it?")
|
||||
return docstring.strip()
|
||||
|
||||
def get_apply_docstring(self) -> str:
|
||||
"""Get the docstring for the apply method (instance method implementing ToolProtocol)."""
|
||||
return self.get_apply_docstring_from_cls()
|
||||
|
||||
def get_apply_fn_metadata(self) -> FuncMetadata:
|
||||
"""Get the metadata for the apply method (instance method implementing ToolProtocol)."""
|
||||
return self.get_apply_fn_metadata_from_cls()
|
||||
|
||||
@classmethod
|
||||
def get_apply_fn_metadata_from_cls(cls) -> FuncMetadata:
|
||||
"""Get the metadata for the apply method from the class (static metadata).
|
||||
Needed for creating MCP tools in a separate process without running into serialization issues.
|
||||
"""
|
||||
# First try to get from __dict__ to handle dynamic docstring changes
|
||||
if "apply" in cls.__dict__:
|
||||
apply_fn = cls.__dict__["apply"]
|
||||
else:
|
||||
# Fall back to getattr for inherited methods
|
||||
apply_fn = getattr(cls, "apply", None)
|
||||
if apply_fn is None:
|
||||
raise AttributeError(f"apply method not defined in {cls}. Did you forget to implement it?")
|
||||
|
||||
return func_metadata(apply_fn, skip_names=["self", "cls"])
|
||||
|
||||
def _log_tool_application(self, frame: Any) -> None:
|
||||
params = {}
|
||||
ignored_params = {"self", "log_call", "catch_exceptions", "args", "apply_fn"}
|
||||
for param, value in frame.f_locals.items():
|
||||
if param in ignored_params:
|
||||
continue
|
||||
if param == "kwargs":
|
||||
params.update(value)
|
||||
else:
|
||||
params[param] = value
|
||||
log.info(f"{self.get_name_from_cls()}: {dict_string(params)}")
|
||||
|
||||
@staticmethod
|
||||
def _limit_length(result: str, max_answer_chars: int) -> str:
|
||||
if (n_chars := len(result)) > max_answer_chars:
|
||||
result = (
|
||||
f"The answer is too long ({n_chars} characters). "
|
||||
+ "Please try a more specific tool query or raise the max_answer_chars parameter."
|
||||
)
|
||||
return result
|
||||
|
||||
def is_active(self) -> bool:
|
||||
return self.agent.tool_is_active(self.__class__)
|
||||
|
||||
def apply_ex(self, log_call: bool = True, catch_exceptions: bool = True, **kwargs) -> str: # type: ignore
|
||||
"""
|
||||
Applies the tool with the given arguments
|
||||
"""
|
||||
|
||||
def task() -> str:
|
||||
apply_fn = self.get_apply_fn()
|
||||
|
||||
try:
|
||||
if not self.is_active():
|
||||
return f"Error: Tool '{self.get_name_from_cls()}' is not active. Active tools: {self.agent.get_active_tool_names()}"
|
||||
except Exception as e:
|
||||
return f"RuntimeError while checking if tool {self.get_name_from_cls()} is active: {e}"
|
||||
|
||||
if log_call:
|
||||
self._log_tool_application(inspect.currentframe())
|
||||
try:
|
||||
# check whether the tool requires an active project and language server
|
||||
if not isinstance(self, ToolMarkerDoesNotRequireActiveProject):
|
||||
if self.agent._active_project is None:
|
||||
return (
|
||||
"Error: No active project. Ask to user to select a project from this list: "
|
||||
+ f"{self.agent.serena_config.project_names}"
|
||||
)
|
||||
if not self.agent.is_language_server_running():
|
||||
log.info("Language server is not running. Starting it ...")
|
||||
self.agent.reset_language_server()
|
||||
|
||||
# apply the actual tool
|
||||
result = apply_fn(**kwargs)
|
||||
|
||||
except Exception as e:
|
||||
if not catch_exceptions:
|
||||
raise
|
||||
msg = f"Error executing tool: {e}\n{traceback.format_exc()}"
|
||||
log.error(
|
||||
f"Error executing tool: {e}. "
|
||||
f"Consider restarting the language server to solve this (especially, if it's a timeout of a symbolic operation)",
|
||||
exc_info=e,
|
||||
)
|
||||
result = msg
|
||||
|
||||
if log_call:
|
||||
log.info(f"Result: {result}")
|
||||
|
||||
try:
|
||||
self.language_server.save_cache()
|
||||
except Exception as e:
|
||||
log.error(f"Error saving language server cache: {e}")
|
||||
|
||||
return result
|
||||
|
||||
future = self.agent.issue_task(task, name=self.__class__.__name__)
|
||||
return future.result(timeout=self.agent.serena_config.tool_timeout)
|
||||
|
||||
|
||||
class EditedFileContext:
|
||||
"""
|
||||
Context manager for file editing.
|
||||
|
||||
Create the context, then use `set_updated_content` to set the new content, the original content
|
||||
being provided in `original_content`.
|
||||
When exiting the context without an exception, the updated content will be written back to the file.
|
||||
"""
|
||||
|
||||
def __init__(self, relative_path: str, agent: "SerenaAgent"):
|
||||
self._project = agent.get_active_project()
|
||||
assert self._project is not None
|
||||
self._abs_path = os.path.join(self._project.project_root, relative_path)
|
||||
if not os.path.isfile(self._abs_path):
|
||||
raise FileNotFoundError(f"File {self._abs_path} does not exist.")
|
||||
with open(self._abs_path, encoding=self._project.project_config.encoding) as f:
|
||||
self._original_content = f.read()
|
||||
self._updated_content: str | None = None
|
||||
|
||||
def __enter__(self) -> Self:
|
||||
return self
|
||||
|
||||
def get_original_content(self) -> str:
|
||||
"""
|
||||
:return: the original content of the file before any modifications.
|
||||
"""
|
||||
return self._original_content
|
||||
|
||||
def set_updated_content(self, content: str) -> None:
|
||||
"""
|
||||
Sets the updated content of the file, which will be written back to the file
|
||||
when the context is exited without an exception.
|
||||
|
||||
:param content: the updated content of the file
|
||||
"""
|
||||
self._updated_content = content
|
||||
|
||||
def __exit__(self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None) -> None:
|
||||
if self._updated_content is not None and exc_type is None:
|
||||
assert self._project is not None
|
||||
with open(self._abs_path, "w", encoding=self._project.project_config.encoding) as f:
|
||||
f.write(self._updated_content)
|
||||
log.info(f"Updated content written to {self._abs_path}")
|
||||
# Language servers should automatically detect the change and update its state accordingly.
|
||||
# If they do not, we may have to add a call to notify it.
|
||||
|
||||
|
||||
class ToolRegistry:
|
||||
_tool_dict: dict[str, type[Tool]] | None = None
|
||||
"""maps tool name to the corresponding tool class"""
|
||||
|
||||
@staticmethod
|
||||
def _iter_tool_classes() -> Generator[type[Tool], None, None]:
|
||||
"""
|
||||
Iterate over Tool subclasses.
|
||||
"""
|
||||
for cls in iter_subclasses(Tool):
|
||||
if cls.__module__.startswith("serena.tools"):
|
||||
yield cls
|
||||
|
||||
@classmethod
|
||||
def _get_tool_dict(cls) -> dict[str, type[Tool]]:
|
||||
if cls._tool_dict is None:
|
||||
cls._tool_dict = {}
|
||||
for tool_class in cls._iter_tool_classes():
|
||||
name = tool_class.get_name_from_cls()
|
||||
if name in cls._tool_dict:
|
||||
raise ValueError(f"Duplicate tool name found: {name}. Tool classes must have unique names.")
|
||||
cls._tool_dict[name] = tool_class
|
||||
return cls._tool_dict
|
||||
|
||||
@classmethod
|
||||
def get_tool_class_by_name(cls, tool_name: str) -> type[Tool]:
|
||||
try:
|
||||
return cls._get_tool_dict()[tool_name]
|
||||
except KeyError as e:
|
||||
available_tools = "\n".join(ToolRegistry.get_tool_names())
|
||||
raise ValueError(f"Tool with name {tool_name} not found. Available tools:\n{available_tools}") from e
|
||||
|
||||
@classmethod
|
||||
def get_all_tool_classes(cls) -> list[type[Tool]]:
|
||||
return list(cls._get_tool_dict().values())
|
||||
|
||||
@classmethod
|
||||
def get_tool_names(cls) -> list[str]:
|
||||
return list(cls._get_tool_dict().keys())
|
||||
|
||||
@classmethod
|
||||
def tool_dict(cls) -> dict[str, type[Tool]]:
|
||||
"""Maps tool name to the corresponding tool class"""
|
||||
return copy(cls._get_tool_dict())
|
||||
|
||||
@classmethod
|
||||
def print_tool_overview(cls, tools: Iterable[type[Tool] | Tool] | None = None) -> None:
|
||||
"""
|
||||
Print a summary of the tools. If no tools are passed, a summary of all tools is printed.
|
||||
"""
|
||||
if tools is None:
|
||||
tools = cls._get_tool_dict().values()
|
||||
|
||||
tool_dict: dict[str, type[Tool] | Tool] = {}
|
||||
for tool_class in tools:
|
||||
tool_dict[tool_class.get_name_from_cls()] = tool_class
|
||||
for tool_name in sorted(tool_dict.keys()):
|
||||
tool_class = tool_dict[tool_name]
|
||||
print(f" * `{tool_name}`: {tool_class.get_tool_description().strip()}")
|
||||
@@ -0,0 +1,139 @@
|
||||
"""
|
||||
Tools supporting the general workflow of the agent
|
||||
"""
|
||||
|
||||
import json
|
||||
import platform
|
||||
|
||||
from serena.tools import Tool, ToolMarkerDoesNotRequireActiveProject
|
||||
|
||||
|
||||
class CheckOnboardingPerformedTool(Tool):
|
||||
"""
|
||||
Checks whether project onboarding was already performed.
|
||||
"""
|
||||
|
||||
def apply(self) -> str:
|
||||
"""
|
||||
Checks whether project onboarding was already performed.
|
||||
You should always call this tool before beginning to actually work on the project/after activating a project,
|
||||
but after calling the initial instructions tool.
|
||||
"""
|
||||
from .memory_tools import ListMemoriesTool
|
||||
|
||||
list_memories_tool = self.agent.get_tool(ListMemoriesTool)
|
||||
memories = json.loads(list_memories_tool.apply())
|
||||
if len(memories) == 0:
|
||||
return (
|
||||
"Onboarding not performed yet (no memories available). "
|
||||
+ "You should perform onboarding by calling the `onboarding` tool before proceeding with the task."
|
||||
)
|
||||
else:
|
||||
return f"""The onboarding was already performed, below is the list of available memories.
|
||||
Do not read them immediately, just remember that they exist and that you can read them later, if it is necessary
|
||||
for the current task.
|
||||
Some memories may be based on previous conversations, others may be general for the current project.
|
||||
You should be able to tell which one you need based on the name of the memory.
|
||||
|
||||
{memories}"""
|
||||
|
||||
|
||||
class OnboardingTool(Tool):
|
||||
"""
|
||||
Performs onboarding (identifying the project structure and essential tasks, e.g. for testing or building).
|
||||
"""
|
||||
|
||||
def apply(self) -> str:
|
||||
"""
|
||||
Call this tool if onboarding was not performed yet.
|
||||
You will call this tool at most once per conversation.
|
||||
|
||||
:return: instructions on how to create the onboarding information
|
||||
"""
|
||||
system = platform.system()
|
||||
return self.prompt_factory.create_onboarding_prompt(system=system)
|
||||
|
||||
|
||||
class ThinkAboutCollectedInformationTool(Tool):
|
||||
"""
|
||||
Thinking tool for pondering the completeness of collected information.
|
||||
"""
|
||||
|
||||
def apply(self) -> str:
|
||||
"""
|
||||
Think about the collected information and whether it is sufficient and relevant.
|
||||
This tool should ALWAYS be called after you have completed a non-trivial sequence of searching steps like
|
||||
find_symbol, find_referencing_symbols, search_files_for_pattern, read_file, etc.
|
||||
"""
|
||||
return self.prompt_factory.create_think_about_collected_information()
|
||||
|
||||
|
||||
class ThinkAboutTaskAdherenceTool(Tool):
|
||||
"""
|
||||
Thinking tool for determining whether the agent is still on track with the current task.
|
||||
"""
|
||||
|
||||
def apply(self) -> str:
|
||||
"""
|
||||
Think about the task at hand and whether you are still on track.
|
||||
Especially important if the conversation has been going on for a while and there
|
||||
has been a lot of back and forth.
|
||||
|
||||
This tool should ALWAYS be called before you insert, replace, or delete code.
|
||||
"""
|
||||
return self.prompt_factory.create_think_about_task_adherence()
|
||||
|
||||
|
||||
class ThinkAboutWhetherYouAreDoneTool(Tool):
|
||||
"""
|
||||
Thinking tool for determining whether the task is truly completed.
|
||||
"""
|
||||
|
||||
def apply(self) -> str:
|
||||
"""
|
||||
Whenever you feel that you are done with what the user has asked for, it is important to call this tool.
|
||||
"""
|
||||
return self.prompt_factory.create_think_about_whether_you_are_done()
|
||||
|
||||
|
||||
class SummarizeChangesTool(Tool):
|
||||
"""
|
||||
Provides instructions for summarizing the changes made to the codebase.
|
||||
"""
|
||||
|
||||
def apply(self) -> str:
|
||||
"""
|
||||
Summarize the changes you have made to the codebase.
|
||||
This tool should always be called after you have fully completed any non-trivial coding task,
|
||||
but only after the think_about_whether_you_are_done call.
|
||||
"""
|
||||
return self.prompt_factory.create_summarize_changes()
|
||||
|
||||
|
||||
class PrepareForNewConversationTool(Tool):
|
||||
"""
|
||||
Provides instructions for preparing for a new conversation (in order to continue with the necessary context).
|
||||
"""
|
||||
|
||||
def apply(self) -> str:
|
||||
"""
|
||||
Instructions for preparing for a new conversation. This tool should only be called on explicit user request.
|
||||
"""
|
||||
return self.prompt_factory.create_prepare_for_new_conversation()
|
||||
|
||||
|
||||
class InitialInstructionsTool(Tool, ToolMarkerDoesNotRequireActiveProject):
|
||||
"""
|
||||
Gets the initial instructions for the current project.
|
||||
Should only be used in settings where the system prompt cannot be set,
|
||||
e.g. in clients you have no control over, like Claude Desktop.
|
||||
"""
|
||||
|
||||
def apply(self) -> str:
|
||||
"""
|
||||
Get the initial instructions for the current coding project.
|
||||
You should always call this tool before starting to work (including using any other tool) on any programming task!
|
||||
The only exception is when a user asks you to activate a project, in which case you should call the `activate_project` first
|
||||
instead and then call this tool.
|
||||
"""
|
||||
return self.agent.create_system_prompt()
|
||||
Reference in new issue
Block a user