diff --git a/src/serena/agent.py b/src/serena/agent.py index 75d0e77c..ebb8a36a 100644 --- a/src/serena/agent.py +++ b/src/serena/agent.py @@ -21,9 +21,10 @@ from functools import cached_property from logging import Logger from pathlib import Path from types import TracebackType -from typing import TYPE_CHECKING, Any, Literal, Protocol, Self, TypeVar, Union, cast +from typing import TYPE_CHECKING, Any, Literal, Self, TypeVar, Union, cast import yaml +from mcp.server.fastmcp.utilities.func_metadata import FuncMetadata, func_metadata from overrides import override from pathspec import PathSpec from ruamel.yaml.comments import CommentedMap @@ -548,6 +549,77 @@ class MemoriesManagerMDFilesInProject(MemoriesManager): return f"Memory {name} deleted." +def create_serena_config( + project: str | None = None, + serena_config: SerenaConfigBase | None = None, + context: SerenaAgentContext | None = None, + modes: list[SerenaAgentMode] | None = None, + enable_web_dashboard: bool | None = None, + enable_gui_log_window: bool | None = None, + log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] | None = None, + trace_lsp_communication: bool | None = None, + tool_timeout: float | None = None, +) -> SerenaConfig: + """ + Create a SerenaConfig instance without instantiating a full SerenaAgent. + + This function extracts the configuration creation logic from SerenaAgent.__init__ + to allow creating configurations independently for process isolation and other use cases. + + :param project: the project to register/activate or None + :param serena_config: the base Serena configuration or None to read from default location + :param context: the context in which the agent will operate + :param modes: list of modes in which the agent will operate + :param enable_web_dashboard: Whether to enable the web dashboard + :param enable_gui_log_window: Whether to enable the GUI log window + :param log_level: Log level + :param trace_lsp_communication: Whether to trace LSP communication + :param tool_timeout: Timeout in seconds for tool execution + :return: A fully configured SerenaConfig instance + """ + # obtain serena configuration + if serena_config is not None: + # If a complete SerenaConfig is provided, use it directly + if isinstance(serena_config, SerenaConfig): + config = serena_config + else: + # For SerenaConfigBase instances (like test configs), create an in-memory SerenaConfig + # that preserves the base config attributes without loading from file + from ruamel.yaml.comments import CommentedMap + + config = SerenaConfig.__new__(SerenaConfig) # Create without calling __init__ + # Initialize basic attributes from base config + config.projects = getattr(serena_config, "projects", []) + config.gui_log_window_enabled = serena_config.gui_log_window_enabled + config.log_level = serena_config.log_level + config.trace_lsp_communication = serena_config.trace_lsp_communication + config.web_dashboard = serena_config.web_dashboard + config.tool_timeout = serena_config.tool_timeout + # Set empty yaml for in-memory config + config.loaded_commented_yaml = CommentedMap() + else: + config = SerenaConfig.from_config_file() + + # Apply parameter overrides + if enable_web_dashboard is not None: + config.web_dashboard = enable_web_dashboard + if enable_gui_log_window is not None: + config.gui_log_window_enabled = enable_gui_log_window + if log_level is not None: + log_level = cast(Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], log_level.upper()) + # transform to int + config.log_level = logging.getLevelNamesMapping()[log_level] + if trace_lsp_communication is not None: + config.trace_lsp_communication = trace_lsp_communication + if tool_timeout is not None: + config.tool_timeout = tool_timeout + + # Note: Project registration/activation is handled separately by the caller + # since it involves complex logic that may require the full agent context + + return config + + class SerenaAgent: def __init__( self, @@ -577,20 +649,18 @@ class SerenaAgent: :param log_level: Log level for the GUI log window. If not specified, will take the value from the serena configuration. :param tool_timeout: Timeout in seconds for tool execution. If not specified, will take the value from the serena configuration. """ - # obtain serena configuration - self.serena_config = serena_config or SerenaConfig.from_config_file() - if enable_web_dashboard is not None: - self.serena_config.web_dashboard = enable_web_dashboard - if enable_gui_log_window is not None: - self.serena_config.gui_log_window_enabled = enable_gui_log_window - if log_level is not None: - log_level = cast(Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], log_level.upper()) - # transform to int - self.serena_config.log_level = logging.getLevelNamesMapping()[log_level] - if trace_lsp_communication is not None: - self.serena_config.trace_lsp_communication = trace_lsp_communication - if tool_timeout is not None: - self.serena_config.tool_timeout = tool_timeout + # obtain serena configuration using the decoupled factory function + self.serena_config = create_serena_config( + project=None, # Don't pass project here, handle activation separately + serena_config=serena_config, + context=context, + modes=modes, + enable_web_dashboard=enable_web_dashboard, + enable_gui_log_window=enable_gui_log_window, + log_level=log_level, + trace_lsp_communication=trace_lsp_communication, + tool_timeout=tool_timeout, + ) # adjust log level serena_log_level = self.serena_config.log_level @@ -617,7 +687,7 @@ class SerenaAgent: # instantiate all tool classes self._all_tools: dict[type[Tool], Tool] = {tool_class: tool_class(self) for tool_class in ToolRegistry.get_all_tool_classes()} - tool_names = [tool.get_name() for tool in self._all_tools.values()] + tool_names = [tool.get_name_from_cls() for tool in self._all_tools.values()] # If GUI log window is enabled, set the tool names for highlighting if self._gui_log_handler is not None: @@ -654,7 +724,7 @@ class SerenaAgent: modes = SerenaAgentMode.load_default_modes() self._context = context self._modes = modes - log.info(f"Loaded tools ({len(self._all_tools)}): {', '.join([tool.get_name() for tool in self._all_tools.values()])}") + log.info(f"Loaded tools ({len(self._all_tools)}): {', '.join([tool.get_name_from_cls() for tool in self._all_tools.values()])}") self._active_tools: dict[type[Tool], Tool] = {} self._update_active_tools() @@ -839,7 +909,7 @@ class SerenaAgent: """ :return: the list of names of the active tools for the current project """ - return sorted([tool.get_name() for tool in self.get_active_tool_classes()]) + return sorted([tool.get_name_from_cls() for tool in self.get_active_tool_classes()]) def tool_is_active(self, tool_class: type["Tool"] | str) -> bool: """ @@ -885,7 +955,7 @@ class SerenaAgent: result_str += " " + ", ".join(chunk) + "\n" # Available but not active tools - all_tool_names = sorted([tool.get_name() for tool in self._all_tools.values()]) + all_tool_names = sorted([tool.get_name_from_cls() for tool in self._all_tools.values()]) inactive_tool_names = [tool for tool in all_tool_names if tool not in active_tool_names] if inactive_tool_names: result_str += "Available but not active tools:\n" @@ -1026,23 +1096,31 @@ class ToolMarkerDoesNotRequireActiveProject: pass -class ToolProtocol(Protocol): +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.""" ... - def apply(self, **kwargs: Any) -> str: - """The application logic of the tool, docstring will be set here and used for MCP tool description.""" + @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): +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. @@ -1054,7 +1132,7 @@ class Tool(Component): # and to validate the tool call arguments. @classmethod - def get_name(cls) -> str: + def get_name_from_cls(cls) -> str: name = cls.__name__ if name.endswith("Tool"): name = name[:-4] @@ -1062,6 +1140,9 @@ class Tool(Component): 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: @@ -1084,12 +1165,48 @@ class Tool(Component): return "" return docstring.strip() - def get_function_description(self) -> str: - apply_fn = self.get_apply_fn() + @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 docstring is None: - raise Exception(f"Missing docstring for {self}") - return docstring + 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 = {} @@ -1101,7 +1218,7 @@ class Tool(Component): params.update(value) else: params[param] = value - log.info(f"{self.get_name()}: {dict_string(params)}") + log.info(f"{self.get_name_from_cls()}: {dict_string(params)}") @staticmethod def _limit_length(result: str, max_answer_chars: int) -> str: @@ -1123,9 +1240,9 @@ class Tool(Component): try: if not self.is_active(): - return f"Error: Tool '{self.get_name()}' is not active. Active tools: {self.agent.get_active_tool_names()}" + 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()} is active: {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()) @@ -1143,7 +1260,7 @@ class Tool(Component): # apply the actual tool with a timeout execution_fn = lambda: apply_fn(**kwargs) - execution_result = execute_with_timeout(execution_fn, self.agent.serena_config.tool_timeout, self.get_name()) + execution_result = execute_with_timeout(execution_fn, self.agent.serena_config.tool_timeout, self.get_name_from_cls()) if execution_result.status == ExecutionResult.Status.SUCCESS: result = cast(str, execution_result.result_value) else: @@ -1696,7 +1813,7 @@ class DeleteLinesTool(Tool, ToolMarkerCanEdit): """ 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()}` first to read exactly the affected lines." + 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 @@ -2158,7 +2275,7 @@ def _iter_tool_classes(same_module_only: bool = True) -> Generator[type[Tool], N yield tool_class -_TOOL_REGISTRY_DICT: dict[str, type[Tool]] = {tool_class.get_name(): tool_class for tool_class in _iter_tool_classes()} +_TOOL_REGISTRY_DICT: dict[str, type[Tool]] = {tool_class.get_name_from_cls(): tool_class for tool_class in _iter_tool_classes()} """maps tool name to the corresponding tool class""" @@ -2194,7 +2311,7 @@ class ToolRegistry: tool_dict: dict[str, type[Tool] | Tool] = {} for tool_class in tools: - tool_dict[tool_class.get_name()] = tool_class + 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()}") diff --git a/src/serena/agno.py b/src/serena/agno.py index c9150693..0bb0e70f 100644 --- a/src/serena/agno.py +++ b/src/serena/agno.py @@ -25,7 +25,7 @@ class SerenaAgnoToolkit(Toolkit): def __init__(self, serena_agent: SerenaAgent): super().__init__("Serena") for tool in serena_agent.get_exposed_tool_instances(): - self.functions[tool.get_name()] = self._create_agno_function(tool) + self.functions[tool.get_name_from_cls()] = self._create_agno_function(tool) log.info("Agno agent functions: %s", list(self.functions.keys())) @staticmethod @@ -39,7 +39,7 @@ class SerenaAgnoToolkit(Toolkit): return tool.apply_ex(log_call=True, catch_exceptions=True, **kwargs) function = Function.from_callable(tool.get_apply_fn()) - function.name = tool.get_name() + function.name = tool.get_name_from_cls() function.entrypoint = entrypoint function.skip_entrypoint_processing = True return function diff --git a/src/serena/mcp.py b/src/serena/mcp.py index f499b3d5..db188a2f 100644 --- a/src/serena/mcp.py +++ b/src/serena/mcp.py @@ -8,21 +8,23 @@ from contextlib import asynccontextmanager from dataclasses import dataclass from logging import Formatter, Logger, StreamHandler from pathlib import Path -from typing import Any, Literal +from typing import TYPE_CHECKING, Any, Literal import click # Add click import import docstring_parser from mcp.server.fastmcp import server 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.helper import mark_used -from serena.agent import SerenaAgent, SerenaConfig, ToolProtocol, show_fatal_exception_safe +from serena.agent import SerenaAgent, ToolInterface, create_serena_config, show_fatal_exception_safe from serena.config import SerenaAgentContext, SerenaAgentMode from serena.constants import DEFAULT_CONTEXT, DEFAULT_MODES -from serena.process_isolated_agent import ProcessIsolatedSerenaAgent, ProcessIsolatedTool + +# Import for type annotations +if TYPE_CHECKING: + from serena.process_isolated_agent import ProcessIsolatedSerenaAgent log = logging.getLogger(__name__) LOG_FORMAT = "%(levelname)-5s %(asctime)-15s %(name)s:%(funcName)s:%(lineno)d - %(message)s" @@ -47,18 +49,12 @@ class SerenaMCPRequestContext: def make_tool( - tool: ToolProtocol, + tool: ToolInterface, ) -> MCPTool: 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 "" + func_doc = tool.get_apply_docstring() or "" + func_arg_metadata = tool.get_apply_fn_metadata() is_async = False - - func_arg_metadata = func_metadata(apply_fn) parameters = func_arg_metadata.arg_model.model_json_schema() docstring = docstring_parser.parse(func_doc) @@ -94,6 +90,7 @@ def make_tool( fn_metadata=func_arg_metadata, is_async=is_async, context_kwarg=None, + annotations=None, ) @@ -108,7 +105,7 @@ def create_mcp_server_and_agent( log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] | None = None, trace_lsp_communication: bool | None = None, tool_timeout: float | None = None, -) -> tuple[FastMCP, SerenaAgent]: +) -> tuple[FastMCP, "ProcessIsolatedSerenaAgent"]: """ Create an MCP server (legacy - now uses process isolation). @@ -141,7 +138,7 @@ def create_mcp_server_and_agent( tool_timeout=tool_timeout, ) # Return a dummy agent for compatibility - return mcp, process_agent # type: ignore + return mcp, process_agent def create_mcp_server_and_process_isolated_agent( @@ -155,7 +152,7 @@ def create_mcp_server_and_process_isolated_agent( log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] | None = None, trace_lsp_communication: bool | None = None, tool_timeout: float | None = None, -) -> tuple[FastMCP, ProcessIsolatedSerenaAgent]: +) -> tuple[FastMCP, "ProcessIsolatedSerenaAgent"]: """ Create an MCP server with process-isolated SerenaAgent to prevent asyncio contamination. @@ -179,8 +176,8 @@ def create_mcp_server_and_process_isolated_agent( modes_instances = [SerenaAgentMode.load(mode) for mode in modes] try: - # First create a regular agent to get the config - temp_agent = SerenaAgent( + # Create configuration without instantiating a full SerenaAgent + serena_config = create_serena_config( project=project, context=context_instance, modes=modes_instances, @@ -191,15 +188,10 @@ def create_mcp_server_and_process_isolated_agent( tool_timeout=tool_timeout, ) - # Create process-isolated agent with the same config - if not isinstance(temp_agent.serena_config, SerenaConfig): - raise TypeError("Expected SerenaConfig instance") - process_agent = ProcessIsolatedSerenaAgent(temp_agent.serena_config) + # Create process-isolated agent with the configuration + from serena.process_isolated_agent import ProcessIsolatedSerenaAgent - # Clean up temp agent - if temp_agent.language_server is not None: - temp_agent.language_server.stop() - del temp_agent + process_agent = ProcessIsolatedSerenaAgent(serena_config) except Exception as e: show_fatal_exception_safe(e) @@ -209,23 +201,29 @@ def create_mcp_server_and_process_isolated_agent( """Update the tools in the MCP server - adapted for process isolation.""" nonlocal mcp, process_agent - # Get tool instances data from process-isolated agent + # Check if process agent is running + if process_agent.process is None or not process_agent.process.is_alive(): + log.debug("Process agent not running yet, skipping tool update") + return + + # Get tool names from process-isolated agent try: - tool_instances_data = process_agent.get_tool_instances() + tool_names = process_agent.get_tool_instances() except Exception as e: - log.error(f"Failed to get tool instances from process agent: {e}") + log.error(f"Failed to get tool names from process agent: {e}") return if mcp is not None: + from serena.process_isolated_agent import ProcessIsolatedTool + mcp._tool_manager._tools = {} - for tool_data in tool_instances_data: + for tool_name in tool_names: # Create ProcessIsolatedTool instance that can be used with make_tool - tool_instance = ProcessIsolatedTool( - process_agent=process_agent, tool_name=tool_data["name"], tool_doc=tool_data["apply_method_doc"] - ) + # Metadata is extracted from ToolRegistry using classmethods + tool_instance = ProcessIsolatedTool(process_agent=process_agent, tool_name=tool_name) # Use the original make_tool function to get full docstring parsing mcp_tool = make_tool(tool_instance) - mcp._tool_manager._tools[tool_data["name"]] = mcp_tool + mcp._tool_manager._tools[tool_name] = mcp_tool @asynccontextmanager async def server_lifespan(mcp_server: FastMCP) -> AsyncIterator[None]: @@ -235,6 +233,8 @@ def create_mcp_server_and_process_isolated_agent( try: process_agent.start() + # Update tools now that the process agent is running + update_tools() yield finally: process_agent.stop() diff --git a/src/serena/process_isolated_agent.py b/src/serena/process_isolated_agent.py index de562a11..543b5ec2 100644 --- a/src/serena/process_isolated_agent.py +++ b/src/serena/process_isolated_agent.py @@ -15,7 +15,9 @@ import time import traceback from typing import Any, Self -from serena.agent import SerenaAgent, SerenaConfig, ToolProtocol +from mcp.server.fastmcp.utilities.func_metadata import FuncMetadata + +from serena.agent import SerenaAgent, SerenaConfig, SerenaConfigBase, Tool, ToolInterface, ToolRegistry log = logging.getLogger(__name__) @@ -110,7 +112,7 @@ class SerenaAgentWorker: # Get the tool by name tool = None for tool_instance in self.agent._active_tools.values(): - if tool_instance.get_name() == tool_name: + if tool_instance.get_name_from_cls() == tool_name: tool = tool_instance break @@ -159,23 +161,15 @@ class SerenaAgentWorker: return {"id": request_id, "error": str(e), "traceback": traceback.format_exc()} def _get_tool_instances(self, request_id: str) -> dict[str, Any]: - """Get exposed tool instances for MCP tool creation.""" + """Get exposed tool names for MCP tool creation.""" if self.agent is None: return {"id": request_id, "error": "SerenaAgent not initialized"} try: tool_instances = self.agent.get_exposed_tool_instances() - # Serialize tool information that can be used to recreate tools on the MCP side - tool_data = [] - for tool in tool_instances: - tool_data.append( - { - "name": tool.get_name(), - "class_name": tool.__class__.__name__, - "apply_method_doc": getattr(tool, "apply").__doc__, - } - ) - return {"id": request_id, "result": tool_data} + # Return only tool names - metadata will be reconstructed from ToolRegistry + tool_names = [tool.get_name_from_cls() for tool in tool_instances] + return {"id": request_id, "result": tool_names} except Exception as e: return {"id": request_id, "error": str(e), "traceback": traceback.format_exc()} @@ -193,7 +187,7 @@ class SerenaAgentWorker: class ProcessIsolatedSerenaAgent: """Process-isolated wrapper for SerenaAgent that prevents asyncio contamination.""" - def __init__(self, serena_config: SerenaConfig): + def __init__(self, serena_config: SerenaConfigBase): self.serena_config = serena_config self.process: multiprocessing.Process | None = None self.request_queue: multiprocessing.Queue | None = None @@ -320,8 +314,8 @@ class ProcessIsolatedSerenaAgent: """Reset the language server.""" self._make_request_with_result("reset_language_server") - def get_tool_instances(self) -> list[dict[str, Any]]: - """Get tool instances data for MCP tool creation.""" + def get_tool_instances(self) -> list[str]: + """Get tool names for MCP tool creation.""" return self._make_request_with_result("get_tool_instances") def __enter__(self) -> Self: @@ -332,30 +326,36 @@ class ProcessIsolatedSerenaAgent: self.stop() -class ProcessIsolatedTool(ToolProtocol): +class ProcessIsolatedTool(ToolInterface): """A clean tool wrapper that delegates to ProcessIsolatedSerenaAgent.""" - def __init__(self, process_agent: ProcessIsolatedSerenaAgent, tool_name: str, tool_doc: str | None = None): + def __init__(self, process_agent: ProcessIsolatedSerenaAgent, tool_name: str): self.process_agent = process_agent - self.tool_name = tool_name - self._tool_doc = tool_doc - # Set the docstring for the apply method if provided - if tool_doc: - self.apply.__doc__ = tool_doc + self._tool_name = tool_name - def apply(self, **kwargs: Any) -> str: - """Apply the tool through the process-isolated agent.""" - return self.process_agent.tool_call(self.tool_name, **kwargs) + @property + def _tool_class(self) -> type[Tool]: + return ToolRegistry.get_tool_class_by_name(self._tool_name) def get_name(self) -> str: """Get the tool name for this process-isolated tool.""" - return self.tool_name + return self._tool_name + + def get_apply_docstring(self) -> str: + """Get the docstring for the apply method.""" + # in the actual tool, this is a classmethod + return self._tool_class.get_apply_docstring_from_cls() + + def get_apply_fn_metadata(self) -> FuncMetadata: + """Get the metadata for the apply method.""" + # in the actual tool, this is a classmethod + return self._tool_class.get_apply_fn_metadata_from_cls() def apply_ex(self, log_call: bool = True, catch_exceptions: bool = True, **kwargs: Any) -> str: """Apply the tool with logging and exception handling.""" try: - return self.apply(**kwargs) + return self.process_agent.tool_call(self._tool_name, **kwargs) except Exception as e: if catch_exceptions: - return f"Error executing tool {self.tool_name}: {e!s}" + return f"Error executing tool {self._tool_name}: {e!s}" raise diff --git a/test/serena/test_make_tool_process_isolation.py b/test/serena/test_make_tool_process_isolation.py new file mode 100644 index 00000000..f0fdda26 --- /dev/null +++ b/test/serena/test_make_tool_process_isolation.py @@ -0,0 +1,160 @@ +"""Tests for make_tool consistency between regular tools and ProcessIsolatedTool.""" + +import pytest +from mcp.server.fastmcp.tools.base import Tool as MCPTool + +from serena.agent import SerenaAgent, ToolRegistry +from serena.mcp import make_tool +from serena.process_isolated_agent import ProcessIsolatedSerenaAgent, ProcessIsolatedTool +from test.serena.test_serena_agent import SerenaConfigForTests + + +@pytest.fixture +def in_memory_config(): + """Create an in-memory configuration for tests.""" + return SerenaConfigForTests() + + +@pytest.fixture +def regular_agent(in_memory_config): + """Create a regular SerenaAgent for comparison.""" + return SerenaAgent(serena_config=in_memory_config) + + +@pytest.fixture +def process_isolated_agent(in_memory_config): + """Create a ProcessIsolatedSerenaAgent for comparison.""" + agent = ProcessIsolatedSerenaAgent(in_memory_config) + agent.start() + yield agent + agent.stop() + + +class TestMakeToolProcessIsolation: + """Test that make_tool produces identical metadata for regular and process-isolated tools.""" + + @pytest.mark.parametrize("tool_name", ToolRegistry.get_tool_names()) + def test_make_tool_metadata_consistency( + self, tool_name: str, regular_agent: SerenaAgent, process_isolated_agent: ProcessIsolatedSerenaAgent + ): + """Test that make_tool produces identical metadata for regular and process-isolated tools.""" + # Get regular tool instance + tool_class = ToolRegistry.get_tool_class_by_name(tool_name) + regular_tool = regular_agent.get_tool(tool_class) + + # Get ProcessIsolatedTool instance + isolated_tool = ProcessIsolatedTool(process_isolated_agent, tool_name) + + # Create MCP tools from both + regular_mcp_tool = make_tool(regular_tool) + isolated_mcp_tool = make_tool(isolated_tool) + + # Verify both are MCPTool instances + assert isinstance(regular_mcp_tool, MCPTool) + assert isinstance(isolated_mcp_tool, MCPTool) + + # Test name consistency + assert regular_mcp_tool.name == isolated_mcp_tool.name + assert regular_mcp_tool.name == tool_name + + # Test description consistency + assert regular_mcp_tool.description == isolated_mcp_tool.description, ( + f"Tool {tool_name}: descriptions differ\n" + f"Regular: {regular_mcp_tool.description}\n" + f"Isolated: {isolated_mcp_tool.description}" + ) + + # Test parameters schema consistency + assert regular_mcp_tool.parameters == isolated_mcp_tool.parameters, ( + f"Tool {tool_name}: parameter schemas differ\n" + f"Regular: {regular_mcp_tool.parameters}\n" + f"Isolated: {isolated_mcp_tool.parameters}" + ) + + # Test function metadata consistency (compare schemas, not class objects) + regular_schema = regular_mcp_tool.fn_metadata.arg_model.model_json_schema() + isolated_schema = isolated_mcp_tool.fn_metadata.arg_model.model_json_schema() + assert ( + regular_schema == isolated_schema + ), f"Tool {tool_name}: function metadata schemas differ\nRegular: {regular_schema}\nIsolated: {isolated_schema}" + + # Test async flag consistency + assert regular_mcp_tool.is_async == isolated_mcp_tool.is_async + + # Test context kwarg consistency + assert regular_mcp_tool.context_kwarg == isolated_mcp_tool.context_kwarg + + @pytest.mark.parametrize("tool_name", ToolRegistry.get_tool_names()[:5]) # Test first 5 tools for faster execution + def test_tool_protocol_methods_consistency( + self, tool_name: str, regular_agent: SerenaAgent, process_isolated_agent: ProcessIsolatedSerenaAgent + ): + """Test that Tool methods return identical results for regular and process-isolated tools.""" + # Get regular tool instance + tool_class = ToolRegistry.get_tool_class_by_name(tool_name) + regular_tool = regular_agent.get_tool(tool_class) + + # Get ProcessIsolatedTool instance + isolated_tool = ProcessIsolatedTool(process_isolated_agent, tool_name) + + # Test get_name() + assert regular_tool.get_name_from_cls() == isolated_tool.get_name() + assert regular_tool.get_name_from_cls() == tool_name + + # Test get_apply_docstring() + regular_docstring = regular_tool.get_apply_docstring() + isolated_docstring = isolated_tool.get_apply_docstring() + assert ( + regular_docstring == isolated_docstring + ), f"Tool {tool_name}: docstrings differ\nRegular: {regular_docstring}\nIsolated: {isolated_docstring}" + + # Test get_apply_fn_metadata() + regular_metadata = regular_tool.get_apply_fn_metadata() + isolated_metadata = isolated_tool.get_apply_fn_metadata() + + # Compare metadata properties (compare schemas, not class objects) + regular_schema = regular_metadata.arg_model.model_json_schema() + isolated_schema = isolated_metadata.arg_model.model_json_schema() + assert ( + regular_schema == isolated_schema + ), f"Tool {tool_name}: metadata schemas differ\nRegular: {regular_schema}\nIsolated: {isolated_schema}" + + def test_process_isolated_tool_uses_tool_registry(self, process_isolated_agent: ProcessIsolatedSerenaAgent): + """Test that ProcessIsolatedTool correctly uses ToolRegistry for metadata.""" + tool_name = ToolRegistry.get_tool_names()[0] # Use first available tool + isolated_tool = ProcessIsolatedTool(process_isolated_agent, tool_name) + + # Verify that the tool uses ToolRegistry + assert isolated_tool._tool_class == ToolRegistry.get_tool_class_by_name(tool_name) + + # Verify that metadata comes from the tool class + expected_docstring = isolated_tool._tool_class.get_apply_docstring_from_cls() + expected_metadata = isolated_tool._tool_class.get_apply_fn_metadata_from_cls() + + assert isolated_tool.get_apply_docstring() == expected_docstring + # Compare schemas, not class objects + isolated_schema = isolated_tool.get_apply_fn_metadata().arg_model.model_json_schema() + expected_schema = expected_metadata.arg_model.model_json_schema() + assert isolated_schema == expected_schema + + def test_tool_registry_completeness(self, regular_agent: SerenaAgent, process_isolated_agent: ProcessIsolatedSerenaAgent): + """Test that all tools are available in both agents and the registry.""" + # Get tool names from both agents + regular_active_tools = set(regular_agent.get_active_tool_names()) + regular_all_tools = set(tool.get_name_from_cls() for tool in regular_agent.get_exposed_tool_instances()) + isolated_tool_names = set(process_isolated_agent.get_tool_instances()) + + # The process isolated agent should have all tools (exposed, not just active) + assert regular_all_tools == isolated_tool_names, ( + f"Tool sets differ:\n" + f"Regular exposed only: {regular_all_tools - isolated_tool_names}\n" + f"Isolated only: {isolated_tool_names - regular_all_tools}" + ) + + # Active tools should be a subset of all tools + assert regular_active_tools.issubset( + regular_all_tools + ), f"Some active tools not in exposed tools: {regular_active_tools - regular_all_tools}" + + # All tools should be in the registry + registry_tool_names = set(ToolRegistry.get_tool_names()) + assert regular_all_tools.issubset(registry_tool_names), f"Some tools not in registry: {regular_all_tools - registry_tool_names}" diff --git a/test/serena/test_mcp.py b/test/serena/test_mcp.py index f179b857..eed27a7f 100644 --- a/test/serena/test_mcp.py +++ b/test/serena/test_mcp.py @@ -224,7 +224,7 @@ def test_make_tool_missing_apply() -> None: """, "", ), - ("", ""), + ("Description without params.", "Description without params."), ], ) def test_make_tool_descriptions(docstring, expected_description) -> None: @@ -287,7 +287,7 @@ def test_make_tool_all_tools(tool_class) -> None: # Basic validation assert isinstance(mcp_tool, MCPTool) - assert mcp_tool.name == tool_class.get_name() + assert mcp_tool.name == tool_class.get_name_from_cls() # The description should be a string (either from docstring or default) assert isinstance(mcp_tool.description, str) diff --git a/test/serena/test_serena_agent.py b/test/serena/test_serena_agent.py index f1208dd3..49a47d88 100644 --- a/test/serena/test_serena_agent.py +++ b/test/serena/test_serena_agent.py @@ -6,7 +6,7 @@ from dataclasses import dataclass import pytest from multilspy.multilspy_config import Language -from serena.agent import FindReferencingSymbolsTool, FindSymbolTool, SerenaAgent, SerenaConfigBase +from serena.agent import FindReferencingSymbolsTool, FindSymbolTool, Project, ProjectConfig, SerenaAgent, SerenaConfigBase from test.conftest import LanguageParamRequest, get_repo_path @@ -19,17 +19,46 @@ class SerenaConfigForTests(SerenaConfigBase): gui_log_window_enabled: bool = False web_dashboard: bool = False + def __post_init__(self): + # Initialize with empty projects list if not already set + if not hasattr(self, "projects") or self.projects is None: + self.projects = [] + @pytest.fixture def serena_config(): - return SerenaConfigForTests() + """Create an in-memory configuration for tests with test repositories pre-registered.""" + # Create test projects for all supported languages + test_projects = [] + for language in [Language.PYTHON, Language.GO, Language.JAVA, Language.RUST, Language.TYPESCRIPT, Language.PHP]: + repo_path = get_repo_path(language) + if repo_path.exists(): + project_name = f"test_repo_{language}" + project = Project( + project_root=str(repo_path), + project_config=ProjectConfig( + project_name=project_name, + language=language, + ignored_paths=[], + excluded_tools=set(), + read_only=False, + ignore_all_files_in_gitignore=True, + initial_prompt="", + encoding="utf-8", + ), + ) + test_projects.append(project) + + config = SerenaConfigForTests() + config.projects = test_projects + return config @pytest.fixture def serena_agent(request: LanguageParamRequest, serena_config) -> SerenaAgent: language = Language(request.param) - repo = get_repo_path(language) - return SerenaAgent(project=str(repo), serena_config=serena_config) + project_name = f"test_repo_{language}" + return SerenaAgent(project=project_name, serena_config=serena_config) class TestSerenaAgent: