Fixes in SerenaAgent isolation in process creation

make_tool previously depended on tool instances in the main process, but the tool instances are no longer available there

To solve this, the tool interface was adjusted to provide static information which is used in the ProcessIsolatedTool wrapper.

I also tried another approach before, where ProcessIsolatedTool would accept tool metadata at init, but this lead to serialization issues that couldn't be easily solved (mcp's FuncMetadata is not serializable, despite being a BaseModel...).

Tests for checking that the wrapped tool instance has the same metadata as the wrapper have been added.
This commit is contained in:
Michael Panchenko committed 2025-06-15 15:15:23 +02:00
1 parent 4a4b665d60
commit 37d1004460
7 files changed
+414 -108

No files matched your search

+153 -36
View File
@@ -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()}")
+2 -2
View File
@@ -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
+34 -34
View File
@@ -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()
+30 -30
View File
@@ -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
@@ -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}"
+2 -2
View File
@@ -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)
+33 -4
View File
@@ -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: