Merge pull request #881 from oraios/mux

Changes from mux

src/serena/agent.py

@@ -107,26 +107,35 @@ class ToolSet:
         registry = ToolRegistry()
         tool_names = set(self._tool_names)
         for definition in tool_inclusion_definitions:
-            included_tools = []
-            excluded_tools = []
-            for included_tool in definition.included_optional_tools:
-                included_tool = get_updated_tool_name(included_tool)
-                if not registry.is_valid_tool_name(included_tool):
-                    raise ValueError(f"Invalid tool name '{included_tool}' provided for inclusion")
-                if included_tool not in tool_names:
-                    tool_names.add(included_tool)
-                    included_tools.append(included_tool)
-            for excluded_tool in definition.excluded_tools:
-                excluded_tool = get_updated_tool_name(excluded_tool)
-                if not registry.is_valid_tool_name(excluded_tool):
-                    raise ValueError(f"Invalid tool name '{excluded_tool}' provided for exclusion")
-                if excluded_tool in tool_names:
-                    tool_names.remove(excluded_tool)
-                    excluded_tools.append(excluded_tool)
-            if included_tools:
-                log.info(f"{definition} included {len(included_tools)} tools: {', '.join(included_tools)}")
-            if excluded_tools:
-                log.info(f"{definition} excluded {len(excluded_tools)} tools: {', '.join(excluded_tools)}")
+            if definition.is_fixed_tool_set():
+                tool_names = set()
+                for fixed_tool in definition.fixed_tools:
+                    fixed_tool = get_updated_tool_name(fixed_tool)
+                    if not registry.is_valid_tool_name(fixed_tool):
+                        raise ValueError(f"Invalid tool name '{fixed_tool}' provided for fixed tool set")
+                    tool_names.add(fixed_tool)
+                log.info(f"{definition} defined a fixed tool set with {len(tool_names)} tools: {', '.join(tool_names)}")
+            else:
+                included_tools = []
+                excluded_tools = []
+                for included_tool in definition.included_optional_tools:
+                    included_tool = get_updated_tool_name(included_tool)
+                    if not registry.is_valid_tool_name(included_tool):
+                        raise ValueError(f"Invalid tool name '{included_tool}' provided for inclusion")
+                    if included_tool not in tool_names:
+                        tool_names.add(included_tool)
+                        included_tools.append(included_tool)
+                for excluded_tool in definition.excluded_tools:
+                    excluded_tool = get_updated_tool_name(excluded_tool)
+                    if not registry.is_valid_tool_name(excluded_tool):
+                        raise ValueError(f"Invalid tool name '{excluded_tool}' provided for exclusion")
+                    if excluded_tool in tool_names:
+                        tool_names.remove(excluded_tool)
+                        excluded_tools.append(excluded_tool)
+                if included_tools:
+                    log.info(f"{definition} included {len(included_tools)} tools: {', '.join(included_tools)}")
+                if excluded_tools:
+                    log.info(f"{definition} excluded {len(excluded_tools)} tools: {', '.join(excluded_tools)}")
         return ToolSet(tool_names)
 
     def without_editing_tools(self) -> "ToolSet":

src/serena/cli.py

@@ -27,7 +27,7 @@ from serena.constants import (
     SERENAS_OWN_CONTEXT_YAMLS_DIR,
     SERENAS_OWN_MODE_YAMLS_DIR,
 )
-from serena.mcp import SerenaMCPFactory, SerenaMCPFactorySingleProcess
+from serena.mcp import SerenaMCPFactory
 from serena.project import Project
 from serena.tools import FindReferencingSymbolsTool, FindSymbolTool, GetSymbolsOverviewTool, SearchForPatternTool, ToolRegistry
 from serena.util.logging import MemoryLogHandler
@@ -237,7 +237,7 @@ class TopLevelCommands(AutoRegisteringGroup):
             log.info("Auto-detected project root: %s", project)
 
         project_file = project_file_arg or project
-        factory = SerenaMCPFactorySingleProcess(context=context, project=project_file, memory_log_handler=memory_log_handler)
+        factory = SerenaMCPFactory(context=context, project=project_file, memory_log_handler=memory_log_handler)
         server = factory.create_mcp_server(
             host=host,
             port=port,

src/serena/config/serena_config.py

@@ -5,7 +5,7 @@ The Serena Model Context Protocol (MCP) Server
 import dataclasses
 import os
 import shutil
-from collections.abc import Iterable
+from collections.abc import Sequence
 from copy import deepcopy
 from dataclasses import dataclass, field
 from datetime import datetime
@@ -95,8 +95,32 @@ class SerenaPaths:
 
 @dataclass
 class ToolInclusionDefinition:
-    excluded_tools: Iterable[str] = ()
-    included_optional_tools: Iterable[str] = ()
+    """
+    Defines which tools to include/exclude in Serena's operation.
+    This can mean either
+      * defining exclusions/inclusions to apply to an existing set of tools [incremental mode], or
+      * defining a fixed set of tools to use [fixed mode].
+    """
+
+    excluded_tools: Sequence[str] = ()
+    """
+    the names of tools to exclude from use [incremental mode]
+    """
+    included_optional_tools: Sequence[str] = ()
+    """
+    the names of optional tools to include [incremental mode]
+    """
+    fixed_tools: Sequence[str] = ()
+    """
+    the names of tools to use as a fixed set of tools [fixed mode]
+    """
+
+    def is_fixed_tool_set(self) -> bool:
+        num_fixed = len(self.fixed_tools)
+        num_incremental = len(self.excluded_tools) + len(self.included_optional_tools)
+        if num_fixed > 0 and num_incremental > 0:
+            raise ValueError("Cannot use both fixed_tools and excluded_tools/included_optional_tools at the same time.")
+        return num_fixed > 0
 
 
 class SerenaConfigError(Exception):

src/serena/mcp.py

@@ -3,7 +3,6 @@ The Serena Model Context Protocol (MCP) Server
 """
 
 import sys
-from abc import abstractmethod
 from collections.abc import AsyncIterator, Iterator, Sequence
 from contextlib import asynccontextmanager
 from copy import deepcopy
@@ -49,15 +48,22 @@ class SerenaMCPRequestContext:
 
 
 class SerenaMCPFactory:
-    def __init__(self, context: str = DEFAULT_CONTEXT, project: str | None = None):
+    """
+    Factory for the creation of the Serena MCP server with an associated SerenaAgent.
+    """
+
+    def __init__(self, context: str = DEFAULT_CONTEXT, project: str | None = None, memory_log_handler: MemoryLogHandler | None = None):
         """
         :param context: The context name or path to context file
         :param project: Either an absolute path to the project directory or a name of an already registered project.
             If the project passed here hasn't been registered yet, it will be registered automatically and can be activated by its name
             afterward.
+        :param memory_log_handler: the in-memory log handler to use for the agent's logging
         """
         self.context = SerenaAgentContext.load(context)
         self.project = project
+        self.agent: SerenaAgent | None = None
+        self.memory_log_handler = memory_log_handler
 
     @staticmethod
     def _sanitize_for_openai_tools(schema: dict) -> dict:
@@ -229,9 +235,9 @@ class SerenaMCPFactory:
             title=None,
         )
 
-    @abstractmethod
     def _iter_tools(self) -> Iterator[Tool]:
-        pass
+        assert self.agent is not None
+        yield from self.agent.get_exposed_tool_instances()
 
     # noinspection PyProtectedMember
     def _set_mcp_tools(self, mcp: FastMCP, openai_tool_compatible: bool = False) -> None:
@@ -243,9 +249,13 @@ class SerenaMCPFactory:
                 mcp._tool_manager._tools[tool.get_name()] = mcp_tool
             log.info(f"Starting MCP server with {len(mcp._tool_manager._tools)} tools: {list(mcp._tool_manager._tools.keys())}")
 
-    @abstractmethod
-    def _instantiate_agent(self, serena_config: SerenaConfig, modes: list[SerenaAgentMode]) -> None:
-        pass
+    def _create_serena_agent(self, serena_config: SerenaConfig, modes: list[SerenaAgentMode]) -> SerenaAgent:
+        return SerenaAgent(
+            project=self.project, serena_config=serena_config, context=self.context, modes=modes, memory_log_handler=self.memory_log_handler
+        )
+
+    def _create_default_serena_config(self) -> SerenaConfig:
+        return SerenaConfig.from_config_file()
 
     def create_mcp_server(
         self,
@@ -275,7 +285,7 @@ class SerenaMCPFactory:
         :param tool_timeout: Timeout in seconds for tool execution. If not specified, will take the value from the serena configuration.
         """
         try:
-            config = SerenaConfig.from_config_file()
+            config = self._create_default_serena_config()
 
             # update configuration with the provided parameters
             if enable_web_dashboard is not None:
@@ -293,7 +303,7 @@ class SerenaMCPFactory:
                 config.language_backend = language_backend
 
             modes_instances = [SerenaAgentMode.load(mode) for mode in modes]
-            self._instantiate_agent(config, modes_instances)
+            self.agent = self._create_serena_agent(config, modes_instances)
 
         except Exception as e:
             show_fatal_exception_safe(e)
@@ -308,49 +318,13 @@ class SerenaMCPFactory:
         return mcp
 
     @asynccontextmanager
-    @abstractmethod
     async def server_lifespan(self, mcp_server: FastMCP) -> AsyncIterator[None]:
         """Manage server startup and shutdown lifecycle."""
-        yield None  # ensures MyPy understands we yield None
-
-    @abstractmethod
-    def _get_initial_instructions(self) -> str:
-        pass
-
-
-class SerenaMCPFactorySingleProcess(SerenaMCPFactory):
-    """
-    MCP server factory where the SerenaAgent and its language server run in the same process as the MCP server
-    """
-
-    def __init__(self, context: str = DEFAULT_CONTEXT, project: str | None = None, memory_log_handler: MemoryLogHandler | None = None):
-        """
-        :param context: The context name or path to context file
-        :param project: Either an absolute path to the project directory or a name of an already registered project.
-            If the project passed here hasn't been registered yet, it will be registered automatically and can be activated by its name
-            afterward.
-        """
-        super().__init__(context=context, project=project)
-        self.agent: SerenaAgent | None = None
-        self.memory_log_handler = memory_log_handler
-
-    def _instantiate_agent(self, serena_config: SerenaConfig, modes: list[SerenaAgentMode]) -> None:
-        self.agent = SerenaAgent(
-            project=self.project, serena_config=serena_config, context=self.context, modes=modes, memory_log_handler=self.memory_log_handler
-        )
-
-    def _iter_tools(self) -> Iterator[Tool]:
-        assert self.agent is not None
-        yield from self.agent.get_exposed_tool_instances()
-
-    def _get_initial_instructions(self) -> str:
-        assert self.agent is not None
-        # we don't use the tool (which at the time of writing calls this method), since the tool may be disabled by the config
-        return self.agent.create_system_prompt()
-
-    @asynccontextmanager
-    async def server_lifespan(self, mcp_server: FastMCP) -> AsyncIterator[None]:
         openai_tool_compatible = self.context.name in ["chatgpt", "codex", "oaicompat-agent"]
         self._set_mcp_tools(mcp_server, openai_tool_compatible=openai_tool_compatible)
         log.info("MCP server lifetime setup complete")
         yield
+
+    def _get_initial_instructions(self) -> str:
+        assert self.agent is not None
+        return self.agent.create_system_prompt()

src/serena/tools/tools_base.py

@@ -349,12 +349,15 @@ class RegisteredTool:
     tool_name: str
 
 
+tool_packages = ["serena.tools"]
+
+
 @singleton
 class ToolRegistry:
     def __init__(self) -> None:
         self._tool_dict: dict[str, RegisteredTool] = {}
         for cls in iter_subclasses(Tool):
-            if not cls.__module__.startswith("serena.tools"):
+            if not any(cls.__module__.startswith(pkg) for pkg in tool_packages):
                 continue
             is_optional = issubclass(cls, ToolMarkerOptional)
             name = cls.get_name_from_cls()