tiennm99/serena
1
0
Fork 0
mirror of https://github.com/tiennm99/serena.git synced 2026-09-03 10:23:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add MCP factory which runs the agent and its LS in the same process again

Browse Source 
(isolation no longer required when using solidlsp)
This commit is contained in:
Dominik Jain
2025-06-21 21:36:32 +02:00
parent aba7184e27
commit d2c13fc081
2 changed files with 169 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/serena/agent.py
+4 -5
View File
@@ -736,11 +736,10 @@ class SerenaAgent:
:param modes: list of modes in which the agent is operating (they will be combined), None for default modes.
The modes may adjust prompts, tool availability, and tool descriptions.
:param serena_config: the Serena configuration or None to read the configuration from the default location.
:param enable_web_dashboard: Whether to enable the web dashboard. If not specified, will take the value from the serena configuration.
:param enable_gui_log_window: Whether to enable the GUI log window. It currently does not work on macOS, and setting this to True will be ignored then.
If not specified, will take the value from the serena configuration.
: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.
:param enable_web_dashboard: whether to enable the web dashboard; If None, will take the value from the Serena configuration.
:param enable_gui_log_window: whether to enable the GUI log window; If None, will take the value from the Serena configuration.
:param log_level: the log level for the GUI log window; If None, will take the value from the serena configuration.
:param tool_timeout: the timeout in seconds for tool execution. If None, will take the value from the serena configuration.
"""
# obtain serena configuration using the decoupled factory function
self.serena_config = create_serena_config(
src/serena/mcp.py
+165 -2
View File
@@ -9,6 +9,7 @@ import signal
import sys
import threading
import time
from abc import abstractmethod
from collections.abc import AsyncIterator, Iterator, Sequence
from contextlib import asynccontextmanager
from dataclasses import dataclass
@@ -35,7 +36,7 @@ from serena.agent import (
show_fatal_exception_safe,
)
from serena.config import RegisteredContext, SerenaAgentContext, SerenaAgentMode
from serena.constants import DEFAULT_CONTEXT, DEFAULT_MODES
from serena.constants import DEFAULT_CONTEXT, DEFAULT_MODES, USE_SOLID_LSP
from serena.process_isolated_agent import (
ProcessIsolatedDashboard,
ProcessIsolatedSerenaAgent,
@@ -77,6 +78,164 @@ class SerenaMCPFactory:
self.context = SerenaAgentContext.load(context)
self.project = project
@staticmethod
def _make_mcp_tool(tool: ToolInterface) -> MCPTool:
func_name = tool.get_name()
func_doc = tool.get_apply_docstring() or ""
func_arg_metadata = tool.get_apply_fn_metadata()
is_async = False
parameters = func_arg_metadata.arg_model.model_json_schema()
docstring = docstring_parser.parse(func_doc)
# Mount the tool description as a combination of the docstring description and
# the return value description, if it exists.
if docstring.description:
func_doc = f"{docstring.description.strip().strip('.')}."
else:
func_doc = ""
if docstring.returns and (docstring_returns_descr := docstring.returns.description):
# Only add a space before "Returns" if func_doc is not empty
prefix = " " if func_doc else ""
func_doc = f"{func_doc}{prefix}Returns {docstring_returns_descr.strip().strip('.')}."
# Parse the parameter descriptions from the docstring and add pass its description
# to the parameter schema.
docstring_params = {param.arg_name: param for param in docstring.params}
parameters_properties: dict[str, dict[str, Any]] = parameters["properties"]
for parameter, properties in parameters_properties.items():
if (param_doc := docstring_params.get(parameter)) and param_doc.description:
param_desc = f"{param_doc.description.strip().strip('.') + '.'}"
properties["description"] = param_desc[0].upper() + param_desc[1:]
def execute_fn(**kwargs) -> str: # type: ignore
return tool.apply_ex(log_call=True, catch_exceptions=True, **kwargs)
return MCPTool(
fn=execute_fn,
name=func_name,
description=func_doc,
parameters=parameters,
fn_metadata=func_arg_metadata,
is_async=is_async,
context_kwarg=None,
annotations=None,
)
@abstractmethod
def _iter_tools(self) -> Iterator[ToolInterface]:
pass
# noinspection PyProtectedMember
def _set_mcp_tools(self, mcp: FastMCP) -> None:
"""Update the tools in the MCP server"""
if mcp is not None:
mcp._tool_manager._tools = {}
for tool in self._iter_tools():
mcp_tool = self._make_mcp_tool(tool)
mcp._tool_manager._tools[tool.get_name()] = mcp_tool
@abstractmethod
def _instantiate_agent(self, serena_config: SerenaConfig, modes: list[SerenaAgentMode]) -> None:
pass
def create_mcp_server(
self,
host: str = "0.0.0.0",
port: int = 8000,
modes: Sequence[str] = DEFAULT_MODES,
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,
) -> FastMCP:
"""
Create an MCP server with process-isolated SerenaAgent to prevent asyncio contamination.
:param host: The host to bind to
:param port: The port to bind to
:param modes: List of mode names or paths to mode files
:param enable_web_dashboard: Whether to enable the web dashboard. If not specified, will take the value from the serena configuration.
:param enable_gui_log_window: Whether to enable the GUI log window. It currently does not work on macOS, and setting this to True will be ignored then.
If not specified, will take the value from the serena configuration.
:param log_level: Log level. If not specified, will take the value from the serena configuration.
:param trace_lsp_communication: Whether to trace the communication between Serena and the language servers.
This is useful for debugging language server issues.
:param tool_timeout: Timeout in seconds for tool execution. If not specified, will take the value from the serena configuration.
"""
try:
serena_config = create_serena_config(
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,
)
modes_instances = [SerenaAgentMode.load(mode) for mode in modes]
self._instantiate_agent(serena_config, modes_instances)
except Exception as e:
show_fatal_exception_safe(e)
raise
# Override model_config to disable the use of `.env` files for reading settings, because user projects are likely to contain
# `.env` files (e.g. containing LOG_LEVEL) that are not supposed to override the MCP settings;
# retain only FASTMCP_ prefix for already set environment variables.
Settings.model_config = SettingsConfigDict(env_prefix="FASTMCP_")
mcp_settings = Settings(lifespan=self.server_lifespan, host=host, port=port)
mcp = FastMCP(**mcp_settings.model_dump())
return mcp
@asynccontextmanager
@abstractmethod
async def server_lifespan(self, mcp_server: FastMCP) -> AsyncIterator[None]:
"""Manage server startup and shutdown lifecycle."""
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):
"""
: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
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)
def _iter_tools(self) -> Iterator[ToolInterface]:
yield from self.agent.get_exposed_tool_instances()
@asynccontextmanager
async def server_lifespan(self, mcp_server: FastMCP) -> AsyncIterator[None]:
self._set_mcp_tools(mcp_server)
yield
class SerenaMCPFactoryWithProcessIsolation(SerenaMCPFactory):
"""
MCP server factory with process isolation for the SerenaAgent and its language server; they run in a separate process
from the MCP server.
"""
def __init__(self, context: str = DEFAULT_CONTEXT, project: str | 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.active_tool_names: list[str] | None = None
self.serena_agent_process: ProcessIsolatedSerenaAgent | None = None
self.serena_dashboard_process: ProcessIsolatedDashboard | None = None
@@ -419,7 +578,11 @@ def start_mcp_server(
# This is for backward compatibility with the old CLI, should be removed in the future!
project_file = project_file_arg if project_file_arg is not None else project
mcp_factory = SerenaMCPFactory(context=context, project=project_file)
if USE_SOLID_LSP:
mcp_factory = SerenaMCPFactorySingleProcess(context=context, project=project_file)
else:
# using multilspy requires process isolation to prevent asyncio contamination
mcp_factory = SerenaMCPFactoryWithProcessIsolation(context=context, project=project_file)
# Use process isolation by default to prevent asyncio event loop contamination
mcp_server = mcp_factory.create_mcp_server(