diff --git a/pyproject.toml b/pyproject.toml index b8d5665..7a56126 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -238,6 +238,7 @@ ignore = [ "SIM102", "W291", "W293", + "B009", ] unfixable = [ "F841", diff --git a/scripts/mcp_server.py b/scripts/mcp_server.py index c375575..05497cb 100644 --- a/scripts/mcp_server.py +++ b/scripts/mcp_server.py @@ -1,10 +1,5 @@ -import logging - -from serena.mcp import mcp - -log = logging.getLogger(__name__) - +from serena.mcp import create_mcp_server +mcp = create_mcp_server() if __name__ == "__main__": - log.info("Starting server") mcp.run() diff --git a/src/serena/agent.py b/src/serena/agent.py index 1e9ba03..b36eb12 100644 --- a/src/serena/agent.py +++ b/src/serena/agent.py @@ -7,20 +7,15 @@ import json import os import platform import sys -import traceback -from abc import ABC, abstractmethod +from abc import ABC from collections import defaultdict -from collections.abc import AsyncIterator, Sequence -from contextlib import asynccontextmanager -from dataclasses import dataclass +from collections.abc import Iterator, Sequence +from contextlib import contextmanager from logging import Logger from pathlib import Path -from typing import Any, cast +from typing import Any, TypeVar import yaml -from mcp.server.fastmcp import server -from mcp.server.fastmcp.prompts.base import Message, UserMessage -from mcp.server.fastmcp.server import Context, FastMCP, Settings from sensai.util import logging from sensai.util.string import dict_string @@ -37,68 +32,65 @@ from serena.util.shell import execute_shell_command log = logging.getLogger(__name__) LOG_FORMAT = "%(levelname)-5s %(asctime)-15s %(name)s:%(funcName)s:%(lineno)d - %(message)s" LOG_LEVEL = logging.INFO +TTool = TypeVar("TTool", bound="Tool") -def configure_logging(*args, **kwargs) -> None: # type: ignore - # configure logging to stderr (will be captured by Claude Desktop); stdio is the MCP communication stream and cannot be used! - logging.basicConfig(level=LOG_LEVEL, stream=sys.stderr, format=LOG_FORMAT) +class SerenaAgent: + def __init__(self, project_file_path: str): + if not os.path.exists(project_file_path): + print(f"Project file not found: {project_file_path}", file=sys.stderr) + sys.exit(1) + # read project configuration + with open(project_file_path, encoding="utf-8") as f: + project_config = yaml.safe_load(f) + self.project_config = project_config + self.language = Language(project_config["language"]) + self.project_root = str(Path(project_config["project_root"]).resolve()) -# patch the logging configuration function in fastmcp, because it's hard-coded and broken -server.configure_logging = configure_logging + # enable GUI log window + enable_gui_log = project_config.get("gui_log_window", True) + if enable_gui_log: + if platform.system() == "Darwin": + log.warning("GUI log window is not supported on macOS") + else: + log_handler = GuiLogViewerHandler(GuiLogViewer(title="Serena Logs"), level=LOG_LEVEL, format_string=LOG_FORMAT) + Logger.root.addHandler(log_handler) + log.info( + f"Starting serena server for project {project_file_path} (language={self.language}, root={self.project_root}); " + f"process id={os.getpid()}, parent process id={os.getppid()}" + ) -@dataclass -class SerenaMCPRequestContext: - language_server: SyncLanguageServer - project_root: str - project_config: dict[str, Any] - prompt_factory: PromptFactory + # create and start the language server instance + config = MultilspyConfig(code_language=self.language) + logger = MultilspyLogger() + self.language_server = SyncLanguageServer.create(config, logger, self.project_root) + + self.prompt_factory = PromptFactory() + + memories_dir = os.path.join(self.get_serena_managed_dir(), "memories") + self.memories_manager = MemoriesManager(memories_dir) + + # find all tool classes and instantiate them + self.tools: dict[type[Tool], Tool] = {} + tool_classes = [ + cls for name, cls in inspect.getmembers(sys.modules[__name__], inspect.isclass) if issubclass(cls, Tool) and cls is not Tool + ] + for tool_class in tool_classes: + tool_instance = tool_class(self) + self.tools[tool_class] = tool_instance + + def get_tool(self, tool_class: type[TTool]) -> TTool: + return self.tools[tool_class] # type: ignore def get_serena_managed_dir(self) -> str: return os.path.join(self.project_root, ".serena") - -@asynccontextmanager -async def server_lifespan(mcp_server: FastMCP) -> AsyncIterator[SerenaMCPRequestContext]: - """Manage server startup and shutdown lifecycle.""" - argv = sys.argv[1:] - if len(argv) != 1: - print("\nUsage: mcp_server <.yml project file>", file=sys.stderr) - sys.exit(1) - - project_file = argv[0] - if not os.path.exists(project_file): - print(f"Project file not found: {project_file}", file=sys.stderr) - sys.exit(1) - - # read project configuration - with open(project_file, encoding="utf-8") as f: - project_config = yaml.safe_load(f) - language = Language(project_config["language"]) - project_root = str(Path(project_config["project_root"]).resolve()) - - # enable GUI log window - enable_gui_log = project_config.get("gui_log_window", True) - if enable_gui_log: - if platform.system() == "Darwin": - log.warning("GUI log window is not supported on macOS") - else: - log_handler = GuiLogViewerHandler(GuiLogViewer(title="Serena Logs"), level=LOG_LEVEL, format_string=LOG_FORMAT) - Logger.root.addHandler(log_handler) - - log.info( - f"Starting serena server for project {project_file} (language={language}, root={project_root}); process id={os.getpid()}, parent process id={os.getppid()}" - ) - - # create and start the language server instance - config = MultilspyConfig(code_language=language) - logger = MultilspyLogger() - language_server = SyncLanguageServer.create(config, logger, project_root) - with language_server.start_server(): - yield SerenaMCPRequestContext( - language_server=language_server, project_root=project_root, project_config=project_config, prompt_factory=PromptFactory() - ) + @contextmanager + def start_server(self) -> Iterator[None]: + with self.language_server.start_server(): + yield class MemoriesManager: @@ -131,26 +123,28 @@ class MemoriesManager: return f"Memory file {memory_file_name} deleted." -mcp_settings = Settings(lifespan=server_lifespan) -mcp = FastMCP(**mcp_settings.model_dump()) - - class Component(ABC): - def __init__(self, ctx: Context): - lifespan_context = cast(SerenaMCPRequestContext, ctx.request_context.lifespan_context) - self.langsrv = lifespan_context.language_server - self.project_root = lifespan_context.project_root - self.project_config = lifespan_context.project_config - self.prompt_factory = lifespan_context.prompt_factory - - memories_dir = os.path.join(lifespan_context.get_serena_managed_dir(), "memories") - self.memories_manager = MemoriesManager(memories_dir) + def __init__(self, agent: "SerenaAgent"): + self.agent = agent + self.langsrv = agent.language_server + self.project_root = agent.project_root + self.project_config = agent.project_config + self.prompt_factory = agent.prompt_factory + self.memories_manager = agent.memories_manager _DEFAULT_MAX_ANSWER_LENGTH = int(2e5) class Tool(Component): + def get_name(self) -> str: + name = self.__class__.__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 + @staticmethod def _log_tool_application(frame: Any) -> None: params = {} @@ -164,689 +158,530 @@ class Tool(Component): params[name] = value log.info(f"{tool_name}: {dict_string(params)}") - def execute(self, max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH) -> str: - try: - self._log_tool_application(inspect.currentframe().f_back) # type: ignore - result = self._execute() - 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." - ) - log.info(f"Result: {result}") - return result - except Exception as e: - msg = f"Error executing tool: {e}\n{traceback.format_exc()}" - return msg - - @abstractmethod - def _execute(self) -> str: - pass - - -class SimplePrompt(Component): - def create(self) -> str: - return self._create_prompt() - - @abstractmethod - def _create_prompt(self) -> str: - pass - - -class SequentialPrompt(Component): - def __init__(self, ctx: Context): - super().__init__(ctx) - self.messages: list[Message] = [] - - def create(self) -> list[Message]: - self._add_messages() - return self.messages - - @abstractmethod - def _add_messages(self) -> None: - pass - - def _add_user_message(self, msg: str) -> None: - self.messages.append(UserMessage(content=msg)) - - -@mcp.tool() -def read_file( - ctx: Context, relative_path: str, start_line: int = 0, end_line: int | None = None, max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH -) -> str: - """ - Read the given file or a chunk of it (between start_line and end_line). 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 ctx: the context object, which will be created and provided automatically - :param relative_path: the relative path to the file to read - :param start_line: the start line of the range to read - :param end_line: the end line of the range to read. If None, the entire file will be read. - :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 - """ - - class ReadFileTool(Tool): - def _execute(self) -> str: - result = self.langsrv.retrieve_full_file_content(relative_path) - result_lines = result.splitlines() - if end_line is None: - result_lines = result_lines[start_line:] - else: - result_lines = result_lines[start_line:end_line] - result = "\n".join(result_lines) - return result - - return ReadFileTool(ctx).execute(max_answer_chars=max_answer_chars) - - -@mcp.tool() -def create_text_file(ctx: Context, 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. - - :param ctx: the context object, which will be created and provided automatically - :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 - """ - - class CreateFileTool(Tool): - def _execute(self) -> str: - absolute_path = os.path.join(self.project_root, relative_path) - with open(absolute_path, "w", encoding="utf-8") as f: - f.write(content) - return f"File created: {relative_path}" - - return CreateFileTool(ctx).execute() - - -@mcp.tool() -def list_dir(ctx: Context, relative_path: str, recursive: bool, max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH) -> str: - """ - :param ctx: the context object, which will be created and provided automatically - :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 - """ - - class ListDirTool(Tool): - def _execute(self) -> str: - dirs, files = scan_directory( - os.path.join(self.project_root, relative_path), - relative_to=self.project_root, - recursive=recursive, - ignored_dirs=self.project_config["ignored_dirs"], + @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 json.dumps({"dirs": dirs, "files": files}) - - return ListDirTool(ctx).execute(max_answer_chars=max_answer_chars) + return result -@mcp.tool() -def get_dir_overview(ctx: Context, relative_path: str, max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH) -> str: - """ - Get an overview of the given directory. - For each file in the directory, we list the top-level symbols in the file (name, kind, line). - Use this tool to get a high-level understanding of the code symbols inside a directory. +class ReadFileTool(Tool): + def apply( + self, relative_path: str, start_line: int = 0, end_line: int | None = None, max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH + ) -> str: + """ + Read the given file or a chunk of it (between start_line and end_line). 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 ctx: the context object, which will be created and provided automatically - :param relative_path: the relative path to the 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, kind, line). - """ + :param relative_path: the relative path to the file to read + :param start_line: the start line of the range to read + :param end_line: the end line of the range to read. If None, the entire file will be read. + :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 + """ + result = self.langsrv.retrieve_full_file_content(relative_path) + result_lines = result.splitlines() + if end_line is None: + result_lines = result_lines[start_line:] + else: + result_lines = result_lines[start_line:end_line] + result = "\n".join(result_lines) - class GetDirOverviewTool(Tool): - def _execute(self) -> str: - return json.dumps(self.langsrv.request_dir_overview(relative_path)) - - return GetDirOverviewTool(ctx).execute(max_answer_chars=max_answer_chars) + return self._limit_length(result, max_answer_chars) -@mcp.tool() -def get_document_overview(ctx: Context, relative_path: str, max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH) -> str: - """ - Use this tool to get a high-level understanding of the code symbols in a file. It often makes sense - to call this before targeted reading, searching or editing operations on the code symbols in the file, - as the output will contain a lot of information about names and lines. +class CreateFileTool(Tool): + 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. - :param ctx: the context object, which will be created and provided automatically - :param relative_path: the relative path to the file 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. - :return: a JSON object with the list of tuples (name, kind, line, column) of all top-level symbols in the file. - """ - - class GetDocumentOverviewTool(Tool): - def _execute(self) -> str: - return json.dumps(self.langsrv.request_document_overview(relative_path)) - - return GetDocumentOverviewTool(ctx).execute(max_answer_chars=max_answer_chars) + :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 + """ + absolute_path = os.path.join(self.project_root, relative_path) + with open(absolute_path, "w", encoding="utf-8") as f: + f.write(content) + return f"File created: {relative_path}" -@mcp.tool() -def find_symbol( - ctx: Context, - name: str, - depth: int = 0, - include_body: bool = False, - include_kinds: Sequence[SymbolKind] | None = None, - exclude_kinds: Sequence[SymbolKind] | None = None, - substring_matching: bool = False, - dir_relative_path: str | None = None, - max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH, -) -> str: - """ - Retrieves information on all symbols/code entities, i.e. classes, methods, attributes, variables, etc. - with the given name. - The returned symbol location information can subsequently be used to edit the returned symbols - or to retrieve further information using other tools. - If you already anticipate that you will need to reference children of the symbol (like methods or fields contained in a class), - you can specify a depth > 0. - - :param ctx: the context object, which will be created and provided automatically - :param name: the name of the symbols to find - :param depth: specifies the depth up to which descendants of the symbol are to be retrieved - (e.g. depth 1 will retrieve methods and attributes for the case where the symbol refers to a class). - Provide a non-zero depth if you intend to subsequently query symbols that are contained in the - retrieved symbol. - :param dir_relative_path: pass a directory relative path to only consider symbols within this directory. - If None, the entire codebase will be considered. - :param include_body: whether to include the body of all symbols in the result. - :param include_kinds: an optional list of ints representing the LSP symbol kind. - If provided, only symbols of the given kinds will be included in the result. - 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: If provided, symbols of the given kinds will be excluded from the result. - Takes precedence over include_kinds. - :param substring_matching: whether to use substring matching for the symbol name. - If True, the symbol name will be matched if it contains the given name as a substring. - :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. - :return: a list of symbols (with symbol locations) that match the given name in JSON format - """ - - class FindSymbolTool(Tool): - def _execute(self) -> str: - symbols = SymbolManager(self.langsrv).find_by_name( - name, - include_body=include_body, - include_kinds=include_kinds, - exclude_kinds=exclude_kinds, - substring_matching=substring_matching, - dir_relative_path=dir_relative_path, - ) - symbol_dicts = [s.to_dict(kind=True, location=True, depth=depth, include_body=include_body) for s in symbols] - return json.dumps(symbol_dicts) - - return FindSymbolTool(ctx).execute(max_answer_chars=max_answer_chars) - - -@mcp.tool() -def find_referencing_symbols( - ctx: Context, - relative_path: str, - line: int, - column: int, - include_body: bool = False, - include_kinds: Sequence[SymbolKind] | None = None, - exclude_kinds: Sequence[SymbolKind] | None = None, - max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH, -) -> str: - """ - Finds symbols that reference the symbol at the given location. - 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 ctx: the context object, which will be created and provided automatically - :param relative_path: the relative path to the file containing the symbol - :param line: the line number - :param column: the column - :param include_body: whether to include the body of the symbols in the result - :param include_kinds: an optional list of integers representing the LSP symbol kinds to include. - If provided, only symbols of the given kinds will be included in the result. - :param exclude_kinds: If provided, symbols of the given kinds will be excluded from the result. - Takes precedence over include_kinds. - :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. - :return: a list of JSON objects with the symbols referencing the requested symbol - """ - - class FindReferencingSymbolsTool(Tool): - def _execute(self) -> str: - symbols = SymbolManager(self.langsrv).find_referencing_symbols( - SymbolLocation(relative_path, line, column), - include_body=include_body, - include_kinds=include_kinds, - exclude_kinds=exclude_kinds, - ) - symbol_dicts = [s.to_dict(kind=True, location=True, depth=0, include_body=include_body) for s in symbols] - return json.dumps(symbol_dicts) - - return FindReferencingSymbolsTool(ctx).execute(max_answer_chars=max_answer_chars) - - -@mcp.tool() -def replace_symbol_body( - ctx: Context, - relative_path: str, - line: int, - column: int, - body: str, -) -> str: - """ - Replaces the body of the symbol at the given location. - Important: Do not try to guess symbol locations but instead use the find_symbol tool to get the correct location. - - :param ctx: the context object, which will be created and provided automatically - :param relative_path: the relative path to the file containing the symbol - :param line: the line number - :param column: the column - :param body: the new symbol body. Important: Provide the correct level of indentation - (as the original body). Note that the first line must not be indented (i.e. no leading spaces). - """ - - class ReplaceSymbolBodyTool(Tool): - def _execute(self) -> str: - SymbolManager(self.langsrv).replace_body( - SymbolLocation(relative_path, line, column), - body=body, - ) - return "OK" - - return ReplaceSymbolBodyTool(ctx).execute() - - -@mcp.tool() -def insert_after_symbol( - ctx: Context, - relative_path: str, - line: int, - column: int, - 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 ctx: the context object, which will be created and provided automatically - :param relative_path: the relative path to the file containing the symbol - :param line: the line number - :param column: the column - :param body: the body/content to be inserted - """ - - class AppendAfterSymbolTool(Tool): - def _execute(self) -> str: - SymbolManager(self.langsrv).insert_after( - SymbolLocation(relative_path, line, column), - body=body, - ) - return "OK" - - return AppendAfterSymbolTool(ctx).execute() - - -@mcp.tool() -def insert_before_symbol( - ctx: Context, - relative_path: str, - line: int, - column: int, - 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 ctx: the context object, which will be created and provided automatically - :param relative_path: the relative path to the file containing the symbol - :param line: the line number - :param column: the column - :param body: the body/content to be inserted - """ - - class AppendAfterSymbolTool(Tool): - def _execute(self) -> str: - SymbolManager(self.langsrv).insert_before( - SymbolLocation(relative_path, line, column), - body=body, - ) - return "OK" - - return AppendAfterSymbolTool(ctx).execute() - - -def delete_lines( - ctx: Context, - relative_path: str, - start_line: int, - end_line: int, -) -> str: - """ - Deletes the given lines in the file. An editing operation, rarely used alone but can be useful in combination with - other tools. - - :param ctx: the context object, which will be created and provided automatically - :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 - """ - - class DeleteLinesTool(Tool): - def _execute(self) -> str: - SymbolManager(self.langsrv).delete_lines(relative_path, start_line, end_line) - return "OK" - - return DeleteLinesTool(ctx).execute() - - -def insert_at_line( - ctx: Context, - relative_path: str, - line: int, - content: str, -) -> str: - """ - Inserts the given content at the given line in the file. 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 ctx: the context object, which will be created and provided automatically - :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 body/content to be inserted - """ - - class InsertAtLineTool(Tool): - def _execute(self) -> str: - SymbolManager(self.langsrv).insert_at_line(relative_path, line, content) - return "OK" - - return InsertAtLineTool(ctx).execute() - - -@mcp.tool() -def check_onboarding_performed(ctx: Context) -> str: - """ - Check if onboarding was performed yet. - You should always call this tool in the beginning of the conversation, - before any question about code or the project is asked. - You will call this tool only once per conversation. - """ - memories = json.loads(list_memories(ctx)) - 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." +class ListDirTool(Tool): + def apply(self, relative_path: str, recursive: bool, max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH) -> str: + """ + :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 + """ + dirs, files = scan_directory( + os.path.join(self.project_root, relative_path), + relative_to=self.project_root, + recursive=recursive, + ignored_dirs=self.project_config["ignored_dirs"], ) - else: - return "Onboarding already performed, no need to perform it again." + result = json.dumps({"dirs": dirs, "files": files}) + return self._limit_length(result, max_answer_chars) -@mcp.tool() -def onboarding(ctx: Context) -> str: - """ - Call this tool if onboarding was not performed yet. - You will call this tool at most once per conversation. +class GetDirOverviewTool(Tool): + def apply(self, relative_path: str, max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH) -> str: + """ + Get an overview of the given directory. + For each file in the directory, we list the top-level symbols in the file (name, kind, line). + Use this tool to get a high-level understanding of the code symbols inside a directory. - :param ctx: the context object, which will be created and provided automatically - :return: instructions on how to create the onboarding information - """ - system = platform.system() - - class OnboardingPrompt(SimplePrompt): - def _create_prompt(self) -> str: - return self.prompt_factory.create_onboarding_prompt(system=system) - - return OnboardingPrompt(ctx).create() + :param relative_path: the relative path to the 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, kind, line). + """ + result = json.dumps(self.langsrv.request_dir_overview(relative_path)) + return self._limit_length(result, max_answer_chars) -@mcp.tool() -def write_memory(ctx: Context, memory_file_name: str, content: str, max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH) -> str: - """ - Write some general information about this project that can be useful for future tasks to a memory file. - The information should be short and to the point. - The memory file name should be meaningful, such that from the name you can infer what the information is about. - It is better to have multiple small memory files than to have a single large one because - memories will be read one by one and we only ever want to read relevant memories. +class GetDocumentOverviewTool(Tool): + def apply(self, relative_path: str, max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH) -> str: + """ + Use this tool to get a high-level understanding of the code symbols in a file. It often makes sense + to call this before targeted reading, searching or editing operations on the code symbols in the file, + as the output will contain a lot of information about names and lines. - 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_file_name } is too long. Max length is {max_answer_chars} characters. " - + "Please make the content shorter." + :param relative_path: the relative path to the file 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. + :return: a JSON object with the list of tuples (name, kind, line, column) of all top-level symbols in the file. + """ + result = json.dumps(self.langsrv.request_document_overview(relative_path)) + return self._limit_length(result, max_answer_chars) + + +class FindSymbolTool(Tool): + def apply( + self, + name: str, + depth: int = 0, + include_body: bool = False, + include_kinds: Sequence[SymbolKind] | None = None, + exclude_kinds: Sequence[SymbolKind] | None = None, + substring_matching: bool = False, + dir_relative_path: str | None = None, + max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH, + ) -> str: + """ + Retrieves information on all symbols/code entities, i.e. classes, methods, attributes, variables, etc. + with the given name. + The returned symbol location information can subsequently be used to edit the returned symbols + or to retrieve further information using other tools. + If you already anticipate that you will need to reference children of the symbol (like methods or fields contained in a class), + you can specify a depth > 0. + + :param name: the name of the symbols to find + :param depth: specifies the depth up to which descendants of the symbol are to be retrieved + (e.g. depth 1 will retrieve methods and attributes for the case where the symbol refers to a class). + Provide a non-zero depth if you intend to subsequently query symbols that are contained in the + retrieved symbol. + :param dir_relative_path: pass a directory relative path to only consider symbols within this directory. + If None, the entire codebase will be considered. + :param include_body: whether to include the body of all symbols in the result. + :param include_kinds: an optional list of ints representing the LSP symbol kind. + If provided, only symbols of the given kinds will be included in the result. + 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: If provided, symbols of the given kinds will be excluded from the result. + Takes precedence over include_kinds. + :param substring_matching: whether to use substring matching for the symbol name. + If True, the symbol name will be matched if it contains the given name as a substring. + :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. + :return: a list of symbols (with symbol locations) that match the given name in JSON format + """ + symbols = SymbolManager(self.langsrv).find_by_name( + name, + include_body=include_body, + include_kinds=include_kinds, + exclude_kinds=exclude_kinds, + substring_matching=substring_matching, + dir_relative_path=dir_relative_path, ) - - class WriteMemoryTool(Tool): - def _execute(self) -> str: - return self.memories_manager.save_memory(memory_file_name, content) - - return WriteMemoryTool(ctx).execute() + symbol_dicts = [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) -@mcp.tool() -def read_memory(ctx: Context, memory_file_name: str, max_answer_chars: int = _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 should be able to 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. - """ +class FindReferencingSymbolsTool(Tool): + def apply( + self, + relative_path: str, + line: int, + column: int, + include_body: bool = False, + include_kinds: Sequence[SymbolKind] | None = None, + exclude_kinds: Sequence[SymbolKind] | None = None, + max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH, + ) -> str: + """ + Finds symbols that reference the symbol at the given location. + 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. - class ReadMemoryTool(Tool): - def _execute(self) -> str: - return self.memories_manager.load_memory(memory_file_name) - - return ReadMemoryTool(ctx).execute(max_answer_chars=max_answer_chars) + :param relative_path: the relative path to the file containing the symbol + :param line: the line number + :param column: the column + :param include_body: whether to include the body of the symbols in the result + :param include_kinds: an optional list of integers representing the LSP symbol kinds to include. + If provided, only symbols of the given kinds will be included in the result. + :param exclude_kinds: If provided, symbols of the given kinds will be excluded from the result. + Takes precedence over include_kinds. + :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. + :return: a list of JSON objects with the symbols referencing the requested symbol + """ + symbols = SymbolManager(self.langsrv).find_referencing_symbols( + SymbolLocation(relative_path, line, column), + include_body=include_body, + include_kinds=include_kinds, + exclude_kinds=exclude_kinds, + ) + symbol_dicts = [s.to_dict(kind=True, location=True, depth=0, include_body=include_body) for s in symbols] + result = json.dumps(symbol_dicts) + return self._limit_length(result, max_answer_chars) -@mcp.tool() -def list_memories(ctx: Context) -> str: - """ - List available memories. Any memory can be read using the `read_memory` tool. - """ +class ReplaceSymbolBodyTool(Tool): + def apply( + self, + relative_path: str, + line: int, + column: int, + body: str, + ) -> str: + """ + Replaces the body of the symbol at the given location. + Important: Do not try to guess symbol locations but instead use the find_symbol tool to get the correct location. - class ListMemoriesTool(Tool): - def _execute(self) -> str: - return json.dumps(self.memories_manager.list_memories()) - - return ListMemoriesTool(ctx).execute() + :param relative_path: the relative path to the file containing the symbol + :param line: the line number + :param column: the column + :param body: the new symbol body. Important: Provide the correct level of indentation + (as the original body). Note that the first line must not be indented (i.e. no leading spaces). + """ + SymbolManager(self.langsrv).replace_body( + SymbolLocation(relative_path, line, column), + body=body, + ) + return "OK" -@mcp.tool() -def delete_memory(ctx: Context, 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. - """ +class InsertAfterSymbolTool(Tool): + def apply( + self, + relative_path: str, + line: int, + column: int, + 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. - class DeleteMemoryTool(Tool): - def _execute(self) -> str: - return self.memories_manager.delete_memory(memory_file_name) - - return DeleteMemoryTool(ctx).execute() + :param relative_path: the relative path to the file containing the symbol + :param line: the line number + :param column: the column + :param body: the body/content to be inserted + """ + SymbolManager(self.langsrv).insert_after( + SymbolLocation(relative_path, line, column), + body=body, + ) + return "OK" -@mcp.tool() -def think_about_collected_information(ctx: Context) -> 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. - """ +class InsertBeforeSymbolTool(Tool): + def apply( + self, + relative_path: str, + line: int, + column: int, + 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. - class ThinkAboutCollectedInformationTool(Tool): - def _execute(self) -> str: - return self.prompt_factory.create_think_about_collected_information() - - return ThinkAboutCollectedInformationTool(ctx).execute() + :param relative_path: the relative path to the file containing the symbol + :param line: the line number + :param column: the column + :param body: the body/content to be inserted + """ + SymbolManager(self.langsrv).insert_before( + SymbolLocation(relative_path, line, column), + body=body, + ) + return "OK" -@mcp.tool() -def think_about_task_adherence(ctx: Context) -> 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. +class DeleteLinesTool(Tool): + def apply( + self, + relative_path: str, + start_line: int, + end_line: int, + ) -> str: + """ + Deletes the given lines in the file. An editing operation, rarely used alone but can be useful in combination with + other tools. - This tool should ALWAYS be called before you insert, replace, or delete code. - """ - - class ThinkAboutTaskAdherenceTool(Tool): - def _execute(self) -> str: - return self.prompt_factory.create_think_about_task_adherence() - - return ThinkAboutTaskAdherenceTool(ctx).execute() + :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 + """ + SymbolManager(self.langsrv).delete_lines(relative_path, start_line, end_line) + return "OK" -@mcp.tool() -def think_about_whether_you_are_done(ctx: Context) -> str: - """ - Think about whether you are done with the task. +class InsertAtLineTool(Tool): + def apply( + self, + relative_path: str, + line: int, + content: str, + ) -> str: + """ + Inserts the given content at the given line in the file. 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). - This tool should ALWAYS be called after you have completed a task or a subtask. - """ - - class ThinkAboutWhetherYouAreDoneTool(Tool): - def _execute(self) -> str: - return self.prompt_factory.create_think_about_whether_you_are_done() - - return ThinkAboutWhetherYouAreDoneTool(ctx).execute() + :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 body/content to be inserted + """ + SymbolManager(self.langsrv).insert_at_line(relative_path, line, content) + return "OK" -@mcp.tool() -def summarize_changes(ctx: Context) -> 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 after the think_about_whether_you_are_done call). - """ - - class SummarizeChangesTool(Tool): - def _execute(self) -> str: - return self.prompt_factory.create_summarize_changes() - - return SummarizeChangesTool(ctx).execute() - - -@mcp.tool() -def prepare_for_new_conversation(ctx: Context) -> str: - """ - Instructions for preparing for a new conversation. This tool should only be called on explicit user request. - """ - - class PrepareForNewConversationTool(Tool): - def _execute(self) -> str: - return self.prompt_factory.create_prepare_for_new_conversation() - - return PrepareForNewConversationTool(ctx).execute() - - -@mcp.tool() -def search_files_for_pattern( - ctx: Context, - pattern: str, - context_lines_before: int = 0, - context_lines_after: int = 0, - paths_include_glob: str | None = None, - paths_exclude_glob: str | None = None, - max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH, -) -> str: - """ - Search for a pattern in all code files (and only in code files) in the project. Generally, symbolic operations like find_symbol or find_referencing_symbols - should be preferred if you know which symbols you are looking for. - If you have to look in non-code files (like notebooks, documentation, etc.), you should use the shell_command tool with grep or similar. - This tool can be useful if you are looking for a specific pattern in the codebase that is not a symbol name. - - :param ctx: the context object, which will be created and provided automatically - :param pattern: Regular expression pattern to search for, either as a compiled Pattern or string - :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; if not provided, search globally. - :param paths_exclude_glob: optional glob pattern specifying files to exclude from the search (takes precedence over paths_include_glob). - :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. - :return: A JSON object mapping file paths to lists of matched consecutive lines (with context, if requested). - """ - - class SearchInAllCodeTool(Tool): - def _execute(self) -> str: - matches = self.langsrv.search_files_for_pattern( - pattern=pattern, - context_lines_before=context_lines_before, - context_lines_after=context_lines_after, - paths_include_glob=paths_include_glob, - paths_exclude_glob=paths_exclude_glob, +class CheckOnboardingPerformedTool(Tool): + def apply(self) -> str: + """ + Check if onboarding was performed yet. + You should always call this tool in the beginning of the conversation, + before any question about code or the project is asked. + You will call this tool only once per conversation. + """ + 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." ) - # 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()) - return json.dumps(file_to_matches) - - return SearchInAllCodeTool(ctx).execute(max_answer_chars=max_answer_chars) + else: + return "Onboarding already performed, no need to perform it again." -@mcp.tool() -def shell_command( - ctx: Context, - command: str, - cwd: str | None = None, - capture_stderr: bool = True, - max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH, -) -> str: - """ - Execute a shell command and return its output. +class OnboardingTool(Tool): + def apply(self) -> str: + """ + Call this tool if onboarding was not performed yet. + You will call this tool at most once per conversation. - 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. + :return: instructions on how to create the onboarding information + """ + system = platform.system() + return self.prompt_factory.create_onboarding_prompt(system=system) - 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 ctx: the context object, which will be created and provided automatically - :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 - """ +class WriteMemoryTool(Tool): + def apply(self, memory_file_name: str, content: str, max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH) -> str: + """ + Write some general information about this project that can be useful for future tasks to a memory file. + The information should be short and to the point. + The memory file name should be meaningful, such that from the name you can infer what the information is about. + It is better to have multiple small memory files than to have a single large one because + memories will be read one by one and we only ever want to read relevant memories. - class ExecuteShellCommandTool(Tool): - def _execute(self) -> str: - _cwd = cwd or self.project_root - result = execute_shell_command(command, cwd=_cwd, capture_stderr=capture_stderr) - return result.json() + 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_file_name } is too long. Max length is {max_answer_chars} characters. " + + "Please make the content shorter." + ) - return ExecuteShellCommandTool(ctx).execute(max_answer_chars=max_answer_chars) + return self.memories_manager.save_memory(memory_file_name, content) + + +class ReadMemoryTool(Tool): + def apply(self, memory_file_name: str, max_answer_chars: int = _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 should be able to 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): + 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): + 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) + + +class ThinkAboutCollectedInformationTool(Tool): + 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): + 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): + def apply(self) -> str: + """ + Think about whether you are done with the task. + + This tool should ALWAYS be called after you have completed a task or a subtask. + """ + return self.prompt_factory.create_think_about_whether_you_are_done() + + +class SummarizeChangesTool(Tool): + 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 after the think_about_whether_you_are_done call). + """ + return self.prompt_factory.create_summarize_changes() + + +class PrepareForNewConversationTool(Tool): + 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 SearchInAllCodeTool(Tool): + def apply( + self, + pattern: str, + context_lines_before: int = 0, + context_lines_after: int = 0, + paths_include_glob: str | None = None, + paths_exclude_glob: str | None = None, + max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH, + ) -> str: + """ + Search for a pattern in all code files (and only in code files) in the project. Generally, symbolic operations like find_symbol or find_referencing_symbols + should be preferred if you know which symbols you are looking for. + If you have to look in non-code files (like notebooks, documentation, etc.), you should use the shell_command tool with grep or similar. + This tool can be useful if you are looking for a specific pattern in the codebase that is not a symbol name. + + :param pattern: Regular expression pattern to search for, either as a compiled Pattern or string + :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; if not provided, search globally. + :param paths_exclude_glob: optional glob pattern specifying files to exclude from the search (takes precedence over paths_include_glob). + :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. + :return: A JSON object mapping file paths to lists of matched consecutive lines (with context, if requested). + """ + matches = self.langsrv.search_files_for_pattern( + pattern=pattern, + context_lines_before=context_lines_before, + context_lines_after=context_lines_after, + 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) + + +class ExecuteShellCommandTool(Tool): + def apply( + self, + command: str, + cwd: str | None = None, + capture_stderr: bool = True, + max_answer_chars: int = _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.project_root + result = execute_shell_command(command, cwd=_cwd, capture_stderr=capture_stderr) + result = result.json() + return self._limit_length(result, max_answer_chars) diff --git a/src/serena/mcp.py b/src/serena/mcp.py index 1e9ba03..6ecde00 100644 --- a/src/serena/mcp.py +++ b/src/serena/mcp.py @@ -2,37 +2,20 @@ The Serena Model Context Protocol (MCP) Server """ -import inspect -import json -import os -import platform import sys import traceback -from abc import ABC, abstractmethod -from collections import defaultdict -from collections.abc import AsyncIterator, Sequence +from collections.abc import AsyncIterator from contextlib import asynccontextmanager from dataclasses import dataclass -from logging import Logger -from pathlib import Path -from typing import Any, cast -import yaml from mcp.server.fastmcp import server -from mcp.server.fastmcp.prompts.base import Message, UserMessage -from mcp.server.fastmcp.server import Context, FastMCP, Settings +from mcp.server.fastmcp.server import FastMCP, Settings +from mcp.server.fastmcp.tools.base import Tool as MCPTool +from mcp.server.fastmcp.utilities.func_metadata import func_metadata from sensai.util import logging -from sensai.util.string import dict_string +from sensai.util.helper import mark_used -from multilspy import SyncLanguageServer -from multilspy.multilspy_config import Language, MultilspyConfig -from multilspy.multilspy_logger import MultilspyLogger -from multilspy.multilspy_types import SymbolKind -from serena.gui_log_viewer import GuiLogViewer, GuiLogViewerHandler -from serena.llm.prompt_factory import PromptFactory -from serena.symbol import SymbolLocation, SymbolManager -from serena.util.file_system import scan_directory -from serena.util.shell import execute_shell_command +from serena.agent import SerenaAgent, Tool log = logging.getLogger(__name__) LOG_FORMAT = "%(levelname)-5s %(asctime)-15s %(name)s:%(funcName)s:%(lineno)d - %(message)s" @@ -50,803 +33,64 @@ server.configure_logging = configure_logging @dataclass class SerenaMCPRequestContext: - language_server: SyncLanguageServer - project_root: str - project_config: dict[str, Any] - prompt_factory: PromptFactory - - def get_serena_managed_dir(self) -> str: - return os.path.join(self.project_root, ".serena") + agent: SerenaAgent -@asynccontextmanager -async def server_lifespan(mcp_server: FastMCP) -> AsyncIterator[SerenaMCPRequestContext]: - """Manage server startup and shutdown lifecycle.""" +def make_tool( + tool: Tool, +) -> MCPTool: + """Create a Tool from a function.""" + from mcp.server.fastmcp import Context + + func_name = tool.get_name() + + apply_fn = getattr(tool, "apply") + if apply_fn is None: + raise ValueError(f"Tool does not have an apply method: {tool}") + + func_doc = apply_fn.__doc__ or "" + is_async = False + + func_arg_metadata = func_metadata(apply_fn) + parameters = func_arg_metadata.arg_model.model_json_schema() + + def execute_fn(ctx: Context, *args, **kwargs) -> str: # type: ignore + mark_used(ctx) + try: + return apply_fn(*args, **kwargs) + except Exception as e: + msg = f"Error executing tool: {e}\n{traceback.format_exc()}" + return msg + + return MCPTool( + fn=execute_fn, + name=func_name, + description=func_doc, + parameters=parameters, + fn_metadata=func_arg_metadata, + is_async=is_async, + context_kwarg="ctx", + ) + + +def create_mcp_server() -> FastMCP: argv = sys.argv[1:] if len(argv) != 1: print("\nUsage: mcp_server <.yml project file>", file=sys.stderr) sys.exit(1) - project_file = argv[0] - if not os.path.exists(project_file): - print(f"Project file not found: {project_file}", file=sys.stderr) - sys.exit(1) + project_file_path = argv[0] + agent = SerenaAgent(project_file_path) - # read project configuration - with open(project_file, encoding="utf-8") as f: - project_config = yaml.safe_load(f) - language = Language(project_config["language"]) - project_root = str(Path(project_config["project_root"]).resolve()) + @asynccontextmanager + async def server_lifespan(mcp_server: FastMCP) -> AsyncIterator[SerenaMCPRequestContext]: + """Manage server startup and shutdown lifecycle.""" + with agent.start_server(): + yield SerenaMCPRequestContext(agent=agent) - # enable GUI log window - enable_gui_log = project_config.get("gui_log_window", True) - if enable_gui_log: - if platform.system() == "Darwin": - log.warning("GUI log window is not supported on macOS") - else: - log_handler = GuiLogViewerHandler(GuiLogViewer(title="Serena Logs"), level=LOG_LEVEL, format_string=LOG_FORMAT) - Logger.root.addHandler(log_handler) + mcp_settings = Settings(lifespan=server_lifespan) + mcp = FastMCP(**mcp_settings.model_dump()) + for tool in agent.tools.values(): + mcp._tool_manager._tools[tool.get_name()] = make_tool(tool) - log.info( - f"Starting serena server for project {project_file} (language={language}, root={project_root}); process id={os.getpid()}, parent process id={os.getppid()}" - ) - - # create and start the language server instance - config = MultilspyConfig(code_language=language) - logger = MultilspyLogger() - language_server = SyncLanguageServer.create(config, logger, project_root) - with language_server.start_server(): - yield SerenaMCPRequestContext( - language_server=language_server, project_root=project_root, project_config=project_config, prompt_factory=PromptFactory() - ) - - -class MemoriesManager: - def __init__(self, memory_dir: str): - self._memory_dir = Path(memory_dir) - self._memory_dir.mkdir(parents=True, exist_ok=True) - - def _get_memory_file_path(self, memory_file_name: str) -> Path: - return self._memory_dir / memory_file_name - - def load_memory(self, memory_file_name: str) -> str: - memory_file_path = self._get_memory_file_path(memory_file_name) - if not memory_file_path.exists(): - return f"Memory file {memory_file_name} not found, consider creating it with the `write_memory` tool if you need it." - with open(memory_file_path, encoding="utf-8") as f: - return f.read() - - def save_memory(self, memory_file_name: str, content: str) -> str: - memory_file_path = self._get_memory_file_path(memory_file_name) - with open(memory_file_path, "w", encoding="utf-8") as f: - f.write(content) - return f"Memory file {memory_file_name} written." - - def list_memories(self) -> list[str]: - return [f.name for f in self._memory_dir.iterdir() if f.is_file()] - - def delete_memory(self, memory_file_name: str) -> str: - memory_file_path = self._get_memory_file_path(memory_file_name) - memory_file_path.unlink() - return f"Memory file {memory_file_name} deleted." - - -mcp_settings = Settings(lifespan=server_lifespan) -mcp = FastMCP(**mcp_settings.model_dump()) - - -class Component(ABC): - def __init__(self, ctx: Context): - lifespan_context = cast(SerenaMCPRequestContext, ctx.request_context.lifespan_context) - self.langsrv = lifespan_context.language_server - self.project_root = lifespan_context.project_root - self.project_config = lifespan_context.project_config - self.prompt_factory = lifespan_context.prompt_factory - - memories_dir = os.path.join(lifespan_context.get_serena_managed_dir(), "memories") - self.memories_manager = MemoriesManager(memories_dir) - - -_DEFAULT_MAX_ANSWER_LENGTH = int(2e5) - - -class Tool(Component): - @staticmethod - def _log_tool_application(frame: Any) -> None: - params = {} - tool_name = None - for name, value in frame.f_locals.items(): - if name == "ctx": - continue - if name.endswith("Tool"): - tool_name = name - continue - params[name] = value - log.info(f"{tool_name}: {dict_string(params)}") - - def execute(self, max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH) -> str: - try: - self._log_tool_application(inspect.currentframe().f_back) # type: ignore - result = self._execute() - 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." - ) - log.info(f"Result: {result}") - return result - except Exception as e: - msg = f"Error executing tool: {e}\n{traceback.format_exc()}" - return msg - - @abstractmethod - def _execute(self) -> str: - pass - - -class SimplePrompt(Component): - def create(self) -> str: - return self._create_prompt() - - @abstractmethod - def _create_prompt(self) -> str: - pass - - -class SequentialPrompt(Component): - def __init__(self, ctx: Context): - super().__init__(ctx) - self.messages: list[Message] = [] - - def create(self) -> list[Message]: - self._add_messages() - return self.messages - - @abstractmethod - def _add_messages(self) -> None: - pass - - def _add_user_message(self, msg: str) -> None: - self.messages.append(UserMessage(content=msg)) - - -@mcp.tool() -def read_file( - ctx: Context, relative_path: str, start_line: int = 0, end_line: int | None = None, max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH -) -> str: - """ - Read the given file or a chunk of it (between start_line and end_line). 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 ctx: the context object, which will be created and provided automatically - :param relative_path: the relative path to the file to read - :param start_line: the start line of the range to read - :param end_line: the end line of the range to read. If None, the entire file will be read. - :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 - """ - - class ReadFileTool(Tool): - def _execute(self) -> str: - result = self.langsrv.retrieve_full_file_content(relative_path) - result_lines = result.splitlines() - if end_line is None: - result_lines = result_lines[start_line:] - else: - result_lines = result_lines[start_line:end_line] - result = "\n".join(result_lines) - return result - - return ReadFileTool(ctx).execute(max_answer_chars=max_answer_chars) - - -@mcp.tool() -def create_text_file(ctx: Context, 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. - - :param ctx: the context object, which will be created and provided automatically - :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 - """ - - class CreateFileTool(Tool): - def _execute(self) -> str: - absolute_path = os.path.join(self.project_root, relative_path) - with open(absolute_path, "w", encoding="utf-8") as f: - f.write(content) - return f"File created: {relative_path}" - - return CreateFileTool(ctx).execute() - - -@mcp.tool() -def list_dir(ctx: Context, relative_path: str, recursive: bool, max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH) -> str: - """ - :param ctx: the context object, which will be created and provided automatically - :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 - """ - - class ListDirTool(Tool): - def _execute(self) -> str: - dirs, files = scan_directory( - os.path.join(self.project_root, relative_path), - relative_to=self.project_root, - recursive=recursive, - ignored_dirs=self.project_config["ignored_dirs"], - ) - return json.dumps({"dirs": dirs, "files": files}) - - return ListDirTool(ctx).execute(max_answer_chars=max_answer_chars) - - -@mcp.tool() -def get_dir_overview(ctx: Context, relative_path: str, max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH) -> str: - """ - Get an overview of the given directory. - For each file in the directory, we list the top-level symbols in the file (name, kind, line). - Use this tool to get a high-level understanding of the code symbols inside a directory. - - :param ctx: the context object, which will be created and provided automatically - :param relative_path: the relative path to the 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, kind, line). - """ - - class GetDirOverviewTool(Tool): - def _execute(self) -> str: - return json.dumps(self.langsrv.request_dir_overview(relative_path)) - - return GetDirOverviewTool(ctx).execute(max_answer_chars=max_answer_chars) - - -@mcp.tool() -def get_document_overview(ctx: Context, relative_path: str, max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH) -> str: - """ - Use this tool to get a high-level understanding of the code symbols in a file. It often makes sense - to call this before targeted reading, searching or editing operations on the code symbols in the file, - as the output will contain a lot of information about names and lines. - - :param ctx: the context object, which will be created and provided automatically - :param relative_path: the relative path to the file 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. - :return: a JSON object with the list of tuples (name, kind, line, column) of all top-level symbols in the file. - """ - - class GetDocumentOverviewTool(Tool): - def _execute(self) -> str: - return json.dumps(self.langsrv.request_document_overview(relative_path)) - - return GetDocumentOverviewTool(ctx).execute(max_answer_chars=max_answer_chars) - - -@mcp.tool() -def find_symbol( - ctx: Context, - name: str, - depth: int = 0, - include_body: bool = False, - include_kinds: Sequence[SymbolKind] | None = None, - exclude_kinds: Sequence[SymbolKind] | None = None, - substring_matching: bool = False, - dir_relative_path: str | None = None, - max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH, -) -> str: - """ - Retrieves information on all symbols/code entities, i.e. classes, methods, attributes, variables, etc. - with the given name. - The returned symbol location information can subsequently be used to edit the returned symbols - or to retrieve further information using other tools. - If you already anticipate that you will need to reference children of the symbol (like methods or fields contained in a class), - you can specify a depth > 0. - - :param ctx: the context object, which will be created and provided automatically - :param name: the name of the symbols to find - :param depth: specifies the depth up to which descendants of the symbol are to be retrieved - (e.g. depth 1 will retrieve methods and attributes for the case where the symbol refers to a class). - Provide a non-zero depth if you intend to subsequently query symbols that are contained in the - retrieved symbol. - :param dir_relative_path: pass a directory relative path to only consider symbols within this directory. - If None, the entire codebase will be considered. - :param include_body: whether to include the body of all symbols in the result. - :param include_kinds: an optional list of ints representing the LSP symbol kind. - If provided, only symbols of the given kinds will be included in the result. - 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: If provided, symbols of the given kinds will be excluded from the result. - Takes precedence over include_kinds. - :param substring_matching: whether to use substring matching for the symbol name. - If True, the symbol name will be matched if it contains the given name as a substring. - :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. - :return: a list of symbols (with symbol locations) that match the given name in JSON format - """ - - class FindSymbolTool(Tool): - def _execute(self) -> str: - symbols = SymbolManager(self.langsrv).find_by_name( - name, - include_body=include_body, - include_kinds=include_kinds, - exclude_kinds=exclude_kinds, - substring_matching=substring_matching, - dir_relative_path=dir_relative_path, - ) - symbol_dicts = [s.to_dict(kind=True, location=True, depth=depth, include_body=include_body) for s in symbols] - return json.dumps(symbol_dicts) - - return FindSymbolTool(ctx).execute(max_answer_chars=max_answer_chars) - - -@mcp.tool() -def find_referencing_symbols( - ctx: Context, - relative_path: str, - line: int, - column: int, - include_body: bool = False, - include_kinds: Sequence[SymbolKind] | None = None, - exclude_kinds: Sequence[SymbolKind] | None = None, - max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH, -) -> str: - """ - Finds symbols that reference the symbol at the given location. - 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 ctx: the context object, which will be created and provided automatically - :param relative_path: the relative path to the file containing the symbol - :param line: the line number - :param column: the column - :param include_body: whether to include the body of the symbols in the result - :param include_kinds: an optional list of integers representing the LSP symbol kinds to include. - If provided, only symbols of the given kinds will be included in the result. - :param exclude_kinds: If provided, symbols of the given kinds will be excluded from the result. - Takes precedence over include_kinds. - :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. - :return: a list of JSON objects with the symbols referencing the requested symbol - """ - - class FindReferencingSymbolsTool(Tool): - def _execute(self) -> str: - symbols = SymbolManager(self.langsrv).find_referencing_symbols( - SymbolLocation(relative_path, line, column), - include_body=include_body, - include_kinds=include_kinds, - exclude_kinds=exclude_kinds, - ) - symbol_dicts = [s.to_dict(kind=True, location=True, depth=0, include_body=include_body) for s in symbols] - return json.dumps(symbol_dicts) - - return FindReferencingSymbolsTool(ctx).execute(max_answer_chars=max_answer_chars) - - -@mcp.tool() -def replace_symbol_body( - ctx: Context, - relative_path: str, - line: int, - column: int, - body: str, -) -> str: - """ - Replaces the body of the symbol at the given location. - Important: Do not try to guess symbol locations but instead use the find_symbol tool to get the correct location. - - :param ctx: the context object, which will be created and provided automatically - :param relative_path: the relative path to the file containing the symbol - :param line: the line number - :param column: the column - :param body: the new symbol body. Important: Provide the correct level of indentation - (as the original body). Note that the first line must not be indented (i.e. no leading spaces). - """ - - class ReplaceSymbolBodyTool(Tool): - def _execute(self) -> str: - SymbolManager(self.langsrv).replace_body( - SymbolLocation(relative_path, line, column), - body=body, - ) - return "OK" - - return ReplaceSymbolBodyTool(ctx).execute() - - -@mcp.tool() -def insert_after_symbol( - ctx: Context, - relative_path: str, - line: int, - column: int, - 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 ctx: the context object, which will be created and provided automatically - :param relative_path: the relative path to the file containing the symbol - :param line: the line number - :param column: the column - :param body: the body/content to be inserted - """ - - class AppendAfterSymbolTool(Tool): - def _execute(self) -> str: - SymbolManager(self.langsrv).insert_after( - SymbolLocation(relative_path, line, column), - body=body, - ) - return "OK" - - return AppendAfterSymbolTool(ctx).execute() - - -@mcp.tool() -def insert_before_symbol( - ctx: Context, - relative_path: str, - line: int, - column: int, - 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 ctx: the context object, which will be created and provided automatically - :param relative_path: the relative path to the file containing the symbol - :param line: the line number - :param column: the column - :param body: the body/content to be inserted - """ - - class AppendAfterSymbolTool(Tool): - def _execute(self) -> str: - SymbolManager(self.langsrv).insert_before( - SymbolLocation(relative_path, line, column), - body=body, - ) - return "OK" - - return AppendAfterSymbolTool(ctx).execute() - - -def delete_lines( - ctx: Context, - relative_path: str, - start_line: int, - end_line: int, -) -> str: - """ - Deletes the given lines in the file. An editing operation, rarely used alone but can be useful in combination with - other tools. - - :param ctx: the context object, which will be created and provided automatically - :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 - """ - - class DeleteLinesTool(Tool): - def _execute(self) -> str: - SymbolManager(self.langsrv).delete_lines(relative_path, start_line, end_line) - return "OK" - - return DeleteLinesTool(ctx).execute() - - -def insert_at_line( - ctx: Context, - relative_path: str, - line: int, - content: str, -) -> str: - """ - Inserts the given content at the given line in the file. 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 ctx: the context object, which will be created and provided automatically - :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 body/content to be inserted - """ - - class InsertAtLineTool(Tool): - def _execute(self) -> str: - SymbolManager(self.langsrv).insert_at_line(relative_path, line, content) - return "OK" - - return InsertAtLineTool(ctx).execute() - - -@mcp.tool() -def check_onboarding_performed(ctx: Context) -> str: - """ - Check if onboarding was performed yet. - You should always call this tool in the beginning of the conversation, - before any question about code or the project is asked. - You will call this tool only once per conversation. - """ - memories = json.loads(list_memories(ctx)) - 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 "Onboarding already performed, no need to perform it again." - - -@mcp.tool() -def onboarding(ctx: Context) -> str: - """ - Call this tool if onboarding was not performed yet. - You will call this tool at most once per conversation. - - :param ctx: the context object, which will be created and provided automatically - :return: instructions on how to create the onboarding information - """ - system = platform.system() - - class OnboardingPrompt(SimplePrompt): - def _create_prompt(self) -> str: - return self.prompt_factory.create_onboarding_prompt(system=system) - - return OnboardingPrompt(ctx).create() - - -@mcp.tool() -def write_memory(ctx: Context, memory_file_name: str, content: str, max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH) -> str: - """ - Write some general information about this project that can be useful for future tasks to a memory file. - The information should be short and to the point. - The memory file name should be meaningful, such that from the name you can infer what the information is about. - It is better to have multiple small memory files 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_file_name } is too long. Max length is {max_answer_chars} characters. " - + "Please make the content shorter." - ) - - class WriteMemoryTool(Tool): - def _execute(self) -> str: - return self.memories_manager.save_memory(memory_file_name, content) - - return WriteMemoryTool(ctx).execute() - - -@mcp.tool() -def read_memory(ctx: Context, memory_file_name: str, max_answer_chars: int = _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 should be able to 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. - """ - - class ReadMemoryTool(Tool): - def _execute(self) -> str: - return self.memories_manager.load_memory(memory_file_name) - - return ReadMemoryTool(ctx).execute(max_answer_chars=max_answer_chars) - - -@mcp.tool() -def list_memories(ctx: Context) -> str: - """ - List available memories. Any memory can be read using the `read_memory` tool. - """ - - class ListMemoriesTool(Tool): - def _execute(self) -> str: - return json.dumps(self.memories_manager.list_memories()) - - return ListMemoriesTool(ctx).execute() - - -@mcp.tool() -def delete_memory(ctx: Context, 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. - """ - - class DeleteMemoryTool(Tool): - def _execute(self) -> str: - return self.memories_manager.delete_memory(memory_file_name) - - return DeleteMemoryTool(ctx).execute() - - -@mcp.tool() -def think_about_collected_information(ctx: Context) -> 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. - """ - - class ThinkAboutCollectedInformationTool(Tool): - def _execute(self) -> str: - return self.prompt_factory.create_think_about_collected_information() - - return ThinkAboutCollectedInformationTool(ctx).execute() - - -@mcp.tool() -def think_about_task_adherence(ctx: Context) -> 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. - """ - - class ThinkAboutTaskAdherenceTool(Tool): - def _execute(self) -> str: - return self.prompt_factory.create_think_about_task_adherence() - - return ThinkAboutTaskAdherenceTool(ctx).execute() - - -@mcp.tool() -def think_about_whether_you_are_done(ctx: Context) -> str: - """ - Think about whether you are done with the task. - - This tool should ALWAYS be called after you have completed a task or a subtask. - """ - - class ThinkAboutWhetherYouAreDoneTool(Tool): - def _execute(self) -> str: - return self.prompt_factory.create_think_about_whether_you_are_done() - - return ThinkAboutWhetherYouAreDoneTool(ctx).execute() - - -@mcp.tool() -def summarize_changes(ctx: Context) -> 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 after the think_about_whether_you_are_done call). - """ - - class SummarizeChangesTool(Tool): - def _execute(self) -> str: - return self.prompt_factory.create_summarize_changes() - - return SummarizeChangesTool(ctx).execute() - - -@mcp.tool() -def prepare_for_new_conversation(ctx: Context) -> str: - """ - Instructions for preparing for a new conversation. This tool should only be called on explicit user request. - """ - - class PrepareForNewConversationTool(Tool): - def _execute(self) -> str: - return self.prompt_factory.create_prepare_for_new_conversation() - - return PrepareForNewConversationTool(ctx).execute() - - -@mcp.tool() -def search_files_for_pattern( - ctx: Context, - pattern: str, - context_lines_before: int = 0, - context_lines_after: int = 0, - paths_include_glob: str | None = None, - paths_exclude_glob: str | None = None, - max_answer_chars: int = _DEFAULT_MAX_ANSWER_LENGTH, -) -> str: - """ - Search for a pattern in all code files (and only in code files) in the project. Generally, symbolic operations like find_symbol or find_referencing_symbols - should be preferred if you know which symbols you are looking for. - If you have to look in non-code files (like notebooks, documentation, etc.), you should use the shell_command tool with grep or similar. - This tool can be useful if you are looking for a specific pattern in the codebase that is not a symbol name. - - :param ctx: the context object, which will be created and provided automatically - :param pattern: Regular expression pattern to search for, either as a compiled Pattern or string - :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; if not provided, search globally. - :param paths_exclude_glob: optional glob pattern specifying files to exclude from the search (takes precedence over paths_include_glob). - :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. - :return: A JSON object mapping file paths to lists of matched consecutive lines (with context, if requested). - """ - - class SearchInAllCodeTool(Tool): - def _execute(self) -> str: - matches = self.langsrv.search_files_for_pattern( - pattern=pattern, - context_lines_before=context_lines_before, - context_lines_after=context_lines_after, - 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()) - return json.dumps(file_to_matches) - - return SearchInAllCodeTool(ctx).execute(max_answer_chars=max_answer_chars) - - -@mcp.tool() -def shell_command( - ctx: Context, - command: str, - cwd: str | None = None, - capture_stderr: bool = True, - max_answer_chars: int = _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 ctx: the context object, which will be created and provided automatically - :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 - """ - - class ExecuteShellCommandTool(Tool): - def _execute(self) -> str: - _cwd = cwd or self.project_root - result = execute_shell_command(command, cwd=_cwd, capture_stderr=capture_stderr) - return result.json() - - return ExecuteShellCommandTool(ctx).execute(max_answer_chars=max_answer_chars) + return mcp