mirror of
https://github.com/tiennm99/serena.git
synced 2026-09-03 10:23:43 +00:00
Support for multiple usage contexts and modes
The context and mode system allows for tool selection and prompt adjustments, thus customizing the behavior of the agent to the user's needs.
This commit is contained in:
committed by
Michael Panchenko
parent
f9cea02674
commit
0884fc7a8c
@@ -2,6 +2,8 @@
|
||||
# black: skip
|
||||
# mypy: ignore-errors
|
||||
|
||||
from typing import List, Optional
|
||||
|
||||
from .multilang_prompt import MultiLangContainer, MultiLangPromptTemplateCollection, PromptList
|
||||
|
||||
|
||||
@@ -12,6 +14,8 @@ class PromptFactory:
|
||||
self.lang_shortcode = lang_shortcode
|
||||
self.collection = MultiLangPromptTemplateCollection()
|
||||
self.fallback_mode = fallback_mode
|
||||
self.context_extension = ""
|
||||
self.mode_extensions = []
|
||||
|
||||
def _format_prompt(self, prompt_name: str, kwargs) -> str:
|
||||
del kwargs["self"]
|
||||
@@ -21,5 +25,33 @@ class PromptFactory:
|
||||
def _get_list(self, prompt_name: str) -> PromptList:
|
||||
mpl = self.collection.get_multilang_prompt_list(prompt_name)
|
||||
return mpl.get_item(self.lang_shortcode, self.fallback_mode)
|
||||
|
||||
def set_context(self, context_extension: str) -> None:
|
||||
"""Set the context extension for the system prompt."""
|
||||
self.context_extension = context_extension
|
||||
|
||||
def set_modes(self, mode_extensions: List[str]) -> None:
|
||||
"""Set the mode extensions for the system prompt."""
|
||||
self.mode_extensions = mode_extensions
|
||||
|
||||
# methods
|
||||
def create_system_prompt(self) -> str:
|
||||
"""Create the system prompt with context and mode extensions."""
|
||||
base_prompt = self._format_prompt("system_prompt", locals())
|
||||
|
||||
# Add context and mode extensions
|
||||
extensions = []
|
||||
if self.context_extension:
|
||||
extensions.append(self.context_extension)
|
||||
|
||||
for mode_extension in self.mode_extensions:
|
||||
extensions.append(mode_extension)
|
||||
|
||||
# If no extensions, return the base prompt
|
||||
if not extensions:
|
||||
return base_prompt
|
||||
|
||||
# Combine the base prompt with the extensions
|
||||
combined_prompt = base_prompt + "\n\n" + "\n\n".join(extensions)
|
||||
return combined_prompt
|
||||
|
||||
# methods
|
||||
@@ -0,0 +1,8 @@
|
||||
name: agent
|
||||
description: All tools except InitialInstructionsTool for agent context
|
||||
system_prompt_extension: |
|
||||
You are running in agent context where the system prompt is provided externally. You should use symbolic
|
||||
tools when possible for code understanding and modification.
|
||||
excluded_tools:
|
||||
- initial_instructions
|
||||
tool_description_overrides: {}
|
||||
@@ -0,0 +1,7 @@
|
||||
name: desktop-app
|
||||
description: All tools included for desktop app context
|
||||
system_prompt_extension: |
|
||||
You are running in desktop app context with all tools available. You should use the symbolic tools
|
||||
when possible, but you also have access to file and shell operations for more complex tasks.
|
||||
excluded_tools: []
|
||||
tool_description_overrides: {}
|
||||
@@ -0,0 +1,13 @@
|
||||
name: ide-assistant
|
||||
description: Non-symbolic editing tools and general shell tool are excluded
|
||||
system_prompt_extension: |
|
||||
You are running in IDE assistant context where file operations and shell commands are handled by the IDE.
|
||||
You should exclusively use symbolic tools for exploring and modifying the code, as the IDE handles
|
||||
file-level operations.
|
||||
excluded_tools:
|
||||
- create_text_file
|
||||
- delete_lines
|
||||
- replace_lines
|
||||
- insert_at_line
|
||||
- execute_shell_command
|
||||
tool_description_overrides: {}
|
||||
@@ -0,0 +1,25 @@
|
||||
name: editing
|
||||
description: All tools, with detailed instructions for code editing
|
||||
system_prompt_extension: |
|
||||
You are operating in editing mode. Your task is to implement the requested changes while adhering to the project's
|
||||
code style and patterns. Use symbolic editing tools whenever possible for precise code modifications.
|
||||
excluded_tools: []
|
||||
tool_description_overrides:
|
||||
replace_symbol_body: |
|
||||
Replaces the body of the symbol at the given location. This is a powerful tool for refactoring an entire class, method,
|
||||
or function while maintaining its interface.
|
||||
|
||||
Example: To refactor a method to improve performance or readability, first find the symbol using find_symbol,
|
||||
then use this tool to replace the entire implementation with an improved version.
|
||||
insert_after_symbol: |
|
||||
Inserts the given body/content after the end of the definition of the given symbol. This is ideal for adding
|
||||
new methods to a class, new functions to a module, or new fields to a class.
|
||||
|
||||
Example: To add a new method to an existing class, first find the class using find_symbol,
|
||||
then use this tool to insert the new method at the appropriate location.
|
||||
insert_before_symbol: |
|
||||
Inserts the given body/content before the beginning of the definition of the given symbol. This is useful
|
||||
for adding imports, new classes, or documentation above an existing symbol.
|
||||
|
||||
Example: To add missing imports at the top of a file, find the first symbol in the file using get_symbols_overview,
|
||||
then use this tool to insert the imports before that symbol.
|
||||
@@ -0,0 +1,13 @@
|
||||
name: interactive
|
||||
description: Interactive mode for clarification and step-by-step work
|
||||
system_prompt_extension: |
|
||||
You are operating in interactive mode. You should engage with the user throughout the task, asking for clarification
|
||||
whenever anything is unclear, insufficiently specified, or ambiguous.
|
||||
|
||||
Break down complex tasks into smaller steps and explain your thinking at each stage. When you're uncertain about
|
||||
a decision, present options to the user and ask for guidance rather than making assumptions.
|
||||
|
||||
Focus on providing informative results for intermediate steps so the user can follow along with your progress and
|
||||
provide feedback as needed.
|
||||
excluded_tools: []
|
||||
tool_description_overrides: {}
|
||||
@@ -0,0 +1,14 @@
|
||||
name: one-shot
|
||||
description: Focus on completely finishing a task without interaction
|
||||
system_prompt_extension: |
|
||||
You are operating in one-shot mode. Your goal is to complete the entire task autonomously without further user interaction.
|
||||
You should assume auto-approval for all tools and continue working until the task is completely finished.
|
||||
|
||||
If the task is planning, your final result should be a comprehensive plan. If the task is coding, your final result
|
||||
should be working code with tests passing and all requirements fulfilled. Complete all necessary steps including
|
||||
linting, formatting, and testing to ensure your solution is production-ready.
|
||||
|
||||
Only abort the task if absolutely necessary, such as when critical information is missing that cannot be inferred
|
||||
from the codebase.
|
||||
excluded_tools: []
|
||||
tool_description_overrides: {}
|
||||
@@ -0,0 +1,29 @@
|
||||
name: planning
|
||||
description: Only read-only tools, focused on analysis and planning
|
||||
system_prompt_extension: |
|
||||
You are operating in planning mode. Your task is to analyze code and create a comprehensive plan but not write any code.
|
||||
Focus on understanding the existing codebase structure, architecture, and functionality to create detailed planning
|
||||
documents that can be used for future implementation.
|
||||
excluded_tools:
|
||||
- create_text_file
|
||||
- replace_symbol_body
|
||||
- insert_after_symbol
|
||||
- insert_before_symbol
|
||||
- delete_lines
|
||||
- replace_lines
|
||||
- insert_at_line
|
||||
- execute_shell_command
|
||||
tool_description_overrides:
|
||||
find_symbol: |
|
||||
Retrieves information on all symbols/code entities, i.e. classes, methods, attributes, variables, etc.
|
||||
with the given name. In planning mode, this tool is essential for understanding the codebase structure.
|
||||
Use this tool to explore the codebase thoroughly before creating a comprehensive plan.
|
||||
|
||||
Example: To understand class hierarchies, use this tool to find base classes, then follow up with find_referencing_symbols
|
||||
to locate derived classes.
|
||||
find_referencing_symbols: |
|
||||
Finds symbols that reference the symbol at the given location. In planning mode, this tool is crucial for
|
||||
understanding dependencies and relationships between different parts of the codebase.
|
||||
|
||||
Example: To understand how a service class is used throughout the application, find the service class symbol first,
|
||||
then use this tool to identify all consumers of the service.
|
||||
+191
-12
@@ -15,6 +15,10 @@ from logging import Logger
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Any, Self, TypeVar, Union, cast
|
||||
|
||||
# Import here so we have the type
|
||||
if TYPE_CHECKING:
|
||||
from serena.util.config_loader import ConfigData
|
||||
|
||||
import yaml
|
||||
from sensai.util import logging
|
||||
from sensai.util.logging import FallbackHandler
|
||||
@@ -182,12 +186,20 @@ class LinesRead:
|
||||
|
||||
|
||||
class SerenaAgent:
|
||||
def __init__(self, project_file_path: str | None = None, project_activation_callback: Callable[[], None] | None = None):
|
||||
def __init__(
|
||||
self,
|
||||
project_file_path: str | None = None,
|
||||
project_activation_callback: Callable[[], None] | None = None,
|
||||
context: str | None = None,
|
||||
modes: list[str] | None = None,
|
||||
):
|
||||
"""
|
||||
:param project_file_path: the configuration file (.yml) of the project to load immediately;
|
||||
if None, do not load any project (must use project selection tool to activate a project).
|
||||
If a project is provided, the corresponding language server will be started.
|
||||
:param project_activation_callback: a callback function to be called when a project is activated.
|
||||
:param context: the context name or path to context file to use
|
||||
:param modes: list of mode names or paths to mode files to use
|
||||
"""
|
||||
# obtain serena configuration
|
||||
self.serena_config = SerenaConfig()
|
||||
@@ -212,9 +224,30 @@ class SerenaAgent:
|
||||
log.info(f"Starting Serena server (version={serena_version()}, process id={os.getpid()}, parent process id={os.getppid()})")
|
||||
log.info("Available projects: {}".format(", ".join(self.serena_config.project_names)))
|
||||
|
||||
# Initialize the prompt factory
|
||||
self.prompt_factory = PromptFactory()
|
||||
self._project_activation_callback = project_activation_callback
|
||||
|
||||
# Load context and mode configuration
|
||||
from serena.util.config_loader import ConfigData, ConfigLoader
|
||||
|
||||
self.config_loader = ConfigLoader()
|
||||
self.current_context: ConfigData | None = None
|
||||
self.current_modes: list[ConfigData] = []
|
||||
|
||||
# Set context and modes if provided
|
||||
if context is not None:
|
||||
self.set_context(context)
|
||||
else:
|
||||
# Default to desktop-app context if none provided
|
||||
self.set_context("desktop-app")
|
||||
|
||||
if modes is not None:
|
||||
self.set_modes(modes)
|
||||
else:
|
||||
# Default to interactive mode if none provided
|
||||
self.set_modes(["interactive"])
|
||||
|
||||
# project-specific instances, which will be initialized upon project activation
|
||||
self.project_config: ProjectConfig | None = None
|
||||
self.language_server: SyncLanguageServer | None = None
|
||||
@@ -231,7 +264,9 @@ class SerenaAgent:
|
||||
log.info(f"Excluding tool '{tool_instance.get_name()}' because project activation is disabled in configuration")
|
||||
continue
|
||||
self._all_tools[tool_class] = tool_instance
|
||||
self._active_tools = dict(self._all_tools)
|
||||
|
||||
# Apply context and mode tool configurations
|
||||
self._update_active_tools()
|
||||
log.info(f"Loaded tools ({len(self._all_tools)}): {', '.join([tool.get_name() for tool in self._all_tools.values()])}")
|
||||
|
||||
# If GUI log window is enabled, set the tool names for highlighting
|
||||
@@ -270,25 +305,149 @@ class SerenaAgent:
|
||||
# When project activation is not enabled, we only expose the active tools
|
||||
return list(self._active_tools.values())
|
||||
|
||||
def set_context(self, context: str) -> None:
|
||||
"""
|
||||
Set the current context configuration.
|
||||
|
||||
:param context: Name or path of the context to use
|
||||
"""
|
||||
try:
|
||||
context_config = self.config_loader.get_context(context)
|
||||
self.current_context = context_config
|
||||
|
||||
# Update the prompt factory with the new context
|
||||
self.prompt_factory.set_context(context_config.system_prompt_extension)
|
||||
|
||||
# Update tool configurations
|
||||
self._update_active_tools()
|
||||
|
||||
log.info(f"Set context to '{context_config.name}': {context_config.description}")
|
||||
except Exception as e:
|
||||
log.error(f"Failed to set context '{context}': {e}")
|
||||
raise
|
||||
|
||||
def set_modes(self, modes: list[str]) -> None:
|
||||
"""
|
||||
Set the current mode configurations.
|
||||
|
||||
:param modes: List of mode names or paths to use
|
||||
"""
|
||||
try:
|
||||
# Check for tool activation conflicts between modes
|
||||
mode_configs = [self.config_loader.get_mode(mode) for mode in modes]
|
||||
|
||||
# Check for conflicts in tool exclusions between modes
|
||||
self._check_mode_conflicts(mode_configs)
|
||||
|
||||
self.current_modes = mode_configs
|
||||
|
||||
# Update the prompt factory with the new modes
|
||||
mode_extensions = [mode.system_prompt_extension for mode in mode_configs]
|
||||
self.prompt_factory.set_modes(mode_extensions)
|
||||
|
||||
# Update tool configurations
|
||||
self._update_active_tools()
|
||||
|
||||
mode_names = [mode.name for mode in mode_configs]
|
||||
log.info(f"Set modes to {mode_names}")
|
||||
except Exception as e:
|
||||
log.error(f"Failed to set modes {modes}: {e}")
|
||||
raise
|
||||
|
||||
def _check_mode_conflicts(self, mode_configs: list["ConfigData"]) -> None:
|
||||
"""
|
||||
Check for conflicts in tool exclusions between modes.
|
||||
|
||||
:param mode_configs: List of mode configurations to check
|
||||
:raises ValueError: If there are conflicts between modes
|
||||
"""
|
||||
if not mode_configs:
|
||||
return
|
||||
|
||||
# Check for conflicts in tool exclusions
|
||||
tools_excluded_by_mode: dict[str, str] = {}
|
||||
for mode in mode_configs:
|
||||
for tool in mode.excluded_tools:
|
||||
if tool in tools_excluded_by_mode:
|
||||
# Another mode already excludes this tool
|
||||
other_mode = tools_excluded_by_mode[tool]
|
||||
if other_mode != mode.name:
|
||||
raise ValueError(
|
||||
f"Conflict between modes: Tool '{tool}' is excluded in mode '{other_mode}' but also in mode '{mode.name}'"
|
||||
)
|
||||
else:
|
||||
tools_excluded_by_mode[tool] = mode.name
|
||||
|
||||
def _update_active_tools(self) -> None:
|
||||
"""
|
||||
Update the active tools based on context, modes, and project configuration.
|
||||
|
||||
Priority order:
|
||||
1. Project-specific exclusions (highest priority)
|
||||
2. Context exclusions
|
||||
3. Mode exclusions
|
||||
"""
|
||||
# Start with all tools
|
||||
self._active_tools = dict(self._all_tools)
|
||||
|
||||
# Collect all excluded tools
|
||||
excluded_tools = set()
|
||||
|
||||
# Apply mode exclusions
|
||||
for mode in self.current_modes:
|
||||
excluded_tools.update(mode.excluded_tools)
|
||||
|
||||
# Apply context exclusions (overrides mode exclusions)
|
||||
if self.current_context:
|
||||
excluded_tools.update(self.current_context.excluded_tools)
|
||||
|
||||
# Apply the exclusions
|
||||
if excluded_tools:
|
||||
self._active_tools = {key: tool for key, tool in self._active_tools.items() if tool.get_name() not in excluded_tools}
|
||||
log.info(f"Tools excluded by context/mode: {sorted(excluded_tools)}")
|
||||
|
||||
# Apply tool description overrides from context and modes
|
||||
if self.current_context:
|
||||
self._apply_tool_description_overrides(self.current_context.tool_description_overrides)
|
||||
|
||||
for mode in self.current_modes:
|
||||
self._apply_tool_description_overrides(mode.tool_description_overrides)
|
||||
|
||||
log.info(f"Active tools after context/mode ({len(self._active_tools)}): {', '.join(self.get_active_tool_names())}")
|
||||
|
||||
def _apply_tool_description_overrides(self, overrides: dict[str, str]) -> None:
|
||||
"""
|
||||
Apply tool description overrides from context or mode configurations.
|
||||
|
||||
:param overrides: Dictionary of tool name to description override
|
||||
"""
|
||||
for tool_name, description in overrides.items():
|
||||
for tool_class, tool in self._all_tools.items():
|
||||
if tool.get_name() == tool_name:
|
||||
# Use monkey patching to override the docstring
|
||||
# This is hacky but effective for this use case
|
||||
tool.__class__.__doc__ = description
|
||||
break
|
||||
|
||||
def activate_project(self, project_config: ProjectConfig) -> None:
|
||||
log.info(f"Activating {project_config}")
|
||||
self.project_config = project_config
|
||||
|
||||
# handle project-specific tool exclusions (if any)
|
||||
# handle project-specific tool exclusions (if any) - highest priority
|
||||
excluded_by_project = set()
|
||||
if self.project_config.excluded_tools:
|
||||
self._active_tools = {
|
||||
key: tool for key, tool in self._all_tools.items() if tool.get_name() not in project_config.excluded_tools
|
||||
}
|
||||
log.info(f"Active tools after exclusions ({len(self._active_tools)}): {', '.join(self.get_active_tool_names())}")
|
||||
else:
|
||||
self._active_tools = dict(self._all_tools)
|
||||
excluded_by_project = self.project_config.excluded_tools
|
||||
self._active_tools = {key: tool for key, tool in self._active_tools.items() if tool.get_name() not in excluded_by_project}
|
||||
log.info(f"Tools excluded by project: {sorted(excluded_by_project)}")
|
||||
log.info(f"Active tools after project exclusions ({len(self._active_tools)}): {', '.join(self.get_active_tool_names())}")
|
||||
|
||||
# if read_only mode is enabled, exclude all editing tools
|
||||
if self.project_config.read_only:
|
||||
editing_tools_before = {key for key, tool in self._active_tools.items() if key.can_edit()}
|
||||
self._active_tools = {key: tool for key, tool in self._active_tools.items() if not key.can_edit()}
|
||||
log.info(
|
||||
f"Project is in read-only mode. Editing tools excluded. Active tools ({len(self._active_tools)}): {', '.join(self.get_active_tool_names())}"
|
||||
)
|
||||
editing_tools_excluded = {self._all_tools[key].get_name() for key in editing_tools_before}
|
||||
log.info(f"Editing tools excluded due to read-only mode: {sorted(editing_tools_excluded)}")
|
||||
log.info(f"Project is in read-only mode. Active tools ({len(self._active_tools)}): {', '.join(self.get_active_tool_names())}")
|
||||
|
||||
# start the language server
|
||||
self.reset_language_server()
|
||||
@@ -1213,6 +1372,26 @@ class PrepareForNewConversationTool(Tool):
|
||||
return self.prompt_factory.create_prepare_for_new_conversation()
|
||||
|
||||
|
||||
class SetModesTool(Tool, ToolMarkerDoesNotRequireActiveProject):
|
||||
"""
|
||||
Changes the current operating modes of the agent.
|
||||
"""
|
||||
|
||||
def apply(self, modes: list[str]) -> str:
|
||||
"""
|
||||
Changes the current operating modes of the agent.
|
||||
|
||||
:param modes: List of mode names to switch to (e.g. ["planning"], ["editing"], ["one-shot"], ["interactive"]),
|
||||
or paths to custom mode configuration files
|
||||
:return: Message indicating success or failure
|
||||
"""
|
||||
try:
|
||||
self.agent.set_modes(modes)
|
||||
return f"Successfully set modes to: {modes}"
|
||||
except Exception as e:
|
||||
return f"Failed to set modes: {e}"
|
||||
|
||||
|
||||
class SearchForPatternTool(Tool):
|
||||
"""
|
||||
Performs a search for a pattern in the project.
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
# black: skip
|
||||
# mypy: ignore-errors
|
||||
|
||||
from typing import List, Optional
|
||||
|
||||
from .multilang_prompt import MultiLangContainer, MultiLangPromptTemplateCollection, PromptList
|
||||
|
||||
|
||||
@@ -12,6 +14,8 @@ class PromptFactory:
|
||||
self.lang_shortcode = lang_shortcode
|
||||
self.collection = MultiLangPromptTemplateCollection()
|
||||
self.fallback_mode = fallback_mode
|
||||
self.context_extension = ""
|
||||
self.mode_extensions = []
|
||||
|
||||
def _format_prompt(self, prompt_name: str, kwargs) -> str:
|
||||
del kwargs["self"]
|
||||
@@ -21,24 +25,54 @@ class PromptFactory:
|
||||
def _get_list(self, prompt_name: str) -> PromptList:
|
||||
mpl = self.collection.get_multilang_prompt_list(prompt_name)
|
||||
return mpl.get_item(self.lang_shortcode, self.fallback_mode)
|
||||
|
||||
def create_onboarding_prompt(self, *, system) -> str:
|
||||
return self._format_prompt("onboarding_prompt", locals())
|
||||
|
||||
def create_think_about_collected_information(self) -> str:
|
||||
return self._format_prompt("think_about_collected_information", locals())
|
||||
|
||||
def create_think_about_task_adherence(self) -> str:
|
||||
return self._format_prompt("think_about_task_adherence", locals())
|
||||
|
||||
def create_think_about_whether_you_are_done(self) -> str:
|
||||
return self._format_prompt("think_about_whether_you_are_done", locals())
|
||||
|
||||
def create_summarize_changes(self) -> str:
|
||||
return self._format_prompt("summarize_changes", locals())
|
||||
|
||||
def create_prepare_for_new_conversation(self) -> str:
|
||||
return self._format_prompt("prepare_for_new_conversation", locals())
|
||||
|
||||
def set_context(self, context_extension: str) -> None:
|
||||
"""Set the context extension for the system prompt."""
|
||||
self.context_extension = context_extension
|
||||
|
||||
def set_modes(self, mode_extensions: List[str]) -> None:
|
||||
"""Set the mode extensions for the system prompt."""
|
||||
self.mode_extensions = mode_extensions
|
||||
|
||||
def create_system_prompt(self) -> str:
|
||||
return self._format_prompt("system_prompt", locals())
|
||||
"""Create the system prompt with context and mode extensions."""
|
||||
base_prompt = self._format_prompt("system_prompt", locals())
|
||||
|
||||
# Add context and mode extensions
|
||||
extensions = []
|
||||
if self.context_extension:
|
||||
extensions.append(self.context_extension)
|
||||
|
||||
for mode_extension in self.mode_extensions:
|
||||
extensions.append(mode_extension)
|
||||
|
||||
# If no extensions, return the base prompt
|
||||
if not extensions:
|
||||
return base_prompt
|
||||
|
||||
# Combine the base prompt with the extensions
|
||||
combined_prompt = base_prompt + "\n\n" + "\n\n".join(extensions)
|
||||
return combined_prompt
|
||||
|
||||
def create_onboarding_prompt(self, *, system) -> str:
|
||||
return self._format_prompt('onboarding_prompt', locals())
|
||||
|
||||
def create_think_about_collected_information(self) -> str:
|
||||
return self._format_prompt('think_about_collected_information', locals())
|
||||
|
||||
def create_think_about_task_adherence(self) -> str:
|
||||
return self._format_prompt('think_about_task_adherence', locals())
|
||||
|
||||
def create_think_about_whether_you_are_done(self) -> str:
|
||||
return self._format_prompt('think_about_whether_you_are_done', locals())
|
||||
|
||||
def create_summarize_changes(self) -> str:
|
||||
return self._format_prompt('summarize_changes', locals())
|
||||
|
||||
def create_prepare_for_new_conversation(self) -> str:
|
||||
return self._format_prompt('prepare_for_new_conversation', locals())
|
||||
|
||||
def create_system_prompt(self) -> str:
|
||||
return self._format_prompt('system_prompt', locals())
|
||||
|
||||
|
||||
+45
-3
@@ -93,13 +93,17 @@ def make_tool(
|
||||
)
|
||||
|
||||
|
||||
def create_mcp_server(project_file_path: str | None, host: str = "0.0.0.0", port: int = 8000) -> FastMCP:
|
||||
def create_mcp_server(
|
||||
project_file_path: str | None, host: str = "0.0.0.0", port: int = 8000, context: str | None = None, modes: list[str] | None = None
|
||||
) -> FastMCP:
|
||||
"""
|
||||
Create an MCP server.
|
||||
|
||||
:param project_file_path: The path to the project file, or None.
|
||||
:param host: The host to bind to
|
||||
:param port: The port to bind to
|
||||
:param context: The context name or path to context file
|
||||
:param modes: List of mode names or paths to mode files
|
||||
"""
|
||||
mcp: FastMCP | None = None
|
||||
|
||||
@@ -108,6 +112,8 @@ def create_mcp_server(project_file_path: str | None, host: str = "0.0.0.0", port
|
||||
project_file_path,
|
||||
# Callback disabled for the time being (see above)
|
||||
# project_activation_callback=update_tools
|
||||
context=context,
|
||||
modes=modes,
|
||||
)
|
||||
except Exception as e:
|
||||
show_fatal_exception_safe(e)
|
||||
@@ -157,6 +163,23 @@ def create_mcp_server(project_file_path: str | None, host: str = "0.0.0.0", port
|
||||
required=False,
|
||||
default=None,
|
||||
)
|
||||
@click.option(
|
||||
"--context",
|
||||
type=str,
|
||||
default=None,
|
||||
help="Context to use. This can be a name of a built-in context ('desktop-app', 'agent', 'ide-assistant') "
|
||||
"or a path to a custom context YAML file. Defaults to 'desktop-app' if not specified.",
|
||||
)
|
||||
@click.option(
|
||||
"--mode",
|
||||
"modes",
|
||||
type=str,
|
||||
multiple=True,
|
||||
default=[],
|
||||
help="Mode(s) to use. This can be names of built-in modes ('planning', 'editing', 'one-shot', 'interactive') "
|
||||
"or paths to custom mode YAML files. Can be specified multiple times to combine modes. "
|
||||
"Defaults to 'interactive' if not specified.",
|
||||
)
|
||||
@click.option(
|
||||
"--transport",
|
||||
type=click.Choice(["stdio", "sse"]),
|
||||
@@ -179,16 +202,28 @@ def create_mcp_server(project_file_path: str | None, host: str = "0.0.0.0", port
|
||||
help="Port to bind to (for SSE transport).",
|
||||
)
|
||||
def start_mcp_server(
|
||||
project_file_opt: str | None, project_file_arg: str | None, transport: Literal["stdio", "sse"], host: str, port: int
|
||||
project_file_opt: str | None,
|
||||
project_file_arg: str | None,
|
||||
context: str | None,
|
||||
modes: tuple[str, ...],
|
||||
transport: Literal["stdio", "sse"],
|
||||
host: str,
|
||||
port: int,
|
||||
) -> None:
|
||||
"""Starts the Serena MCP server.
|
||||
|
||||
Accepts the project file path either via the --project-file option or as a positional argument.
|
||||
|
||||
Use --context to specify the execution environment and --mode to specify behavior mode(s).
|
||||
"""
|
||||
# Prioritize the positional argument if provided
|
||||
# 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_file_opt
|
||||
mcp_server = create_mcp_server(project_file_path=project_file, host=host, port=port)
|
||||
|
||||
# Convert modes tuple to list
|
||||
modes_list = list(modes) if modes else None
|
||||
|
||||
mcp_server = create_mcp_server(project_file_path=project_file, host=host, port=port, context=context, modes=modes_list)
|
||||
|
||||
# log after server creation such that the log appears in the GUI
|
||||
if project_file_arg is not None:
|
||||
@@ -197,4 +232,11 @@ def start_mcp_server(
|
||||
"Please pass the project file path via the `--project-file` option instead.\n"
|
||||
f"Used path: {project_file}"
|
||||
)
|
||||
|
||||
# Log selected context and modes
|
||||
if context:
|
||||
log.info(f"Using context: {context}")
|
||||
if modes:
|
||||
log.info(f"Using modes: {', '.join(modes)}")
|
||||
|
||||
mcp_server.run(transport=transport)
|
||||
|
||||
@@ -0,0 +1,131 @@
|
||||
"""
|
||||
Context and Mode configuration loader
|
||||
"""
|
||||
|
||||
import os
|
||||
import pathlib
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
from sensai.util import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class ConfigData:
|
||||
"""Base class for context and mode configurations."""
|
||||
|
||||
name: str
|
||||
description: str
|
||||
system_prompt_extension: str
|
||||
excluded_tools: set[str]
|
||||
tool_description_overrides: dict[str, str]
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any]) -> "ConfigData":
|
||||
"""Create a ConfigData instance from a dictionary."""
|
||||
return cls(
|
||||
name=data.get("name", ""),
|
||||
description=data.get("description", ""),
|
||||
system_prompt_extension=data.get("system_prompt_extension", ""),
|
||||
excluded_tools=set(data.get("excluded_tools", [])),
|
||||
tool_description_overrides=data.get("tool_description_overrides", {}),
|
||||
)
|
||||
|
||||
|
||||
class ConfigLoader:
|
||||
"""Handles loading of context and mode configurations."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Initialize the config loader."""
|
||||
self.serena_root = self._find_serena_root()
|
||||
self.contexts_dir = os.path.join(self.serena_root, "prompts", "contexts")
|
||||
self.modes_dir = os.path.join(self.serena_root, "prompts", "modes")
|
||||
|
||||
# Ensure directories exist
|
||||
os.makedirs(self.contexts_dir, exist_ok=True)
|
||||
os.makedirs(self.modes_dir, exist_ok=True)
|
||||
|
||||
# Cache loaded configs
|
||||
self.context_cache: dict[str, ConfigData] = {}
|
||||
self.mode_cache: dict[str, ConfigData] = {}
|
||||
|
||||
def _find_serena_root(self) -> str:
|
||||
"""Find the root directory of Serena."""
|
||||
current_dir = pathlib.Path(__file__).parent.parent.parent.parent
|
||||
return str(current_dir)
|
||||
|
||||
def load_config_from_file(self, file_path: str | Path) -> ConfigData:
|
||||
"""Load a configuration from a file."""
|
||||
path = Path(file_path)
|
||||
if not path.exists():
|
||||
raise FileNotFoundError(f"Config file not found: {file_path}")
|
||||
|
||||
with open(path, encoding="utf-8") as f:
|
||||
data = yaml.safe_load(f)
|
||||
return ConfigData.from_dict(data)
|
||||
|
||||
def get_context(self, context_name_or_path: str) -> ConfigData:
|
||||
"""
|
||||
Get the context configuration by name or path.
|
||||
|
||||
:param context_name_or_path: Either a context name (will be looked up in contexts directory)
|
||||
or a path to a YAML file
|
||||
:return: The context configuration
|
||||
"""
|
||||
# Return from cache if available
|
||||
if context_name_or_path in self.context_cache:
|
||||
return self.context_cache[context_name_or_path]
|
||||
|
||||
# Check if it's a file path
|
||||
if os.path.isfile(context_name_or_path):
|
||||
config = self.load_config_from_file(context_name_or_path)
|
||||
self.context_cache[context_name_or_path] = config
|
||||
return config
|
||||
|
||||
# Check if it's a known context name
|
||||
context_file = os.path.join(self.contexts_dir, f"{context_name_or_path}.yml")
|
||||
if os.path.isfile(context_file):
|
||||
config = self.load_config_from_file(context_file)
|
||||
self.context_cache[context_name_or_path] = config
|
||||
return config
|
||||
|
||||
raise ValueError(f"Context not found: {context_name_or_path}. Please provide a valid context name or file path.")
|
||||
|
||||
def get_mode(self, mode_name_or_path: str) -> ConfigData:
|
||||
"""
|
||||
Get the mode configuration by name or path.
|
||||
|
||||
:param mode_name_or_path: Either a mode name (will be looked up in modes directory)
|
||||
or a path to a YAML file
|
||||
:return: The mode configuration
|
||||
"""
|
||||
# Return from cache if available
|
||||
if mode_name_or_path in self.mode_cache:
|
||||
return self.mode_cache[mode_name_or_path]
|
||||
|
||||
# Check if it's a file path
|
||||
if os.path.isfile(mode_name_or_path):
|
||||
config = self.load_config_from_file(mode_name_or_path)
|
||||
self.mode_cache[mode_name_or_path] = config
|
||||
return config
|
||||
|
||||
# Check if it's a known mode name
|
||||
mode_file = os.path.join(self.modes_dir, f"{mode_name_or_path}.yml")
|
||||
if os.path.isfile(mode_file):
|
||||
config = self.load_config_from_file(mode_file)
|
||||
self.mode_cache[mode_name_or_path] = config
|
||||
return config
|
||||
|
||||
raise ValueError(f"Mode not found: {mode_name_or_path}. Please provide a valid mode name or file path.")
|
||||
|
||||
def list_available_contexts(self) -> list[str]:
|
||||
"""List all available context names."""
|
||||
return [f.stem for f in Path(self.contexts_dir).glob("*.yml")]
|
||||
|
||||
def list_available_modes(self) -> list[str]:
|
||||
"""List all available mode names."""
|
||||
return [f.stem for f in Path(self.modes_dir).glob("*.yml")]
|
||||
@@ -0,0 +1,100 @@
|
||||
"""
|
||||
Tests for the configuration loader for contexts and modes.
|
||||
"""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
from serena.util.config_loader import ConfigData, ConfigLoader
|
||||
|
||||
|
||||
def test_config_data_from_dict():
|
||||
"""Test creating ConfigData from a dictionary."""
|
||||
data = {
|
||||
"name": "test-context",
|
||||
"description": "Test context",
|
||||
"system_prompt_extension": "You are in test context",
|
||||
"excluded_tools": ["tool1", "tool2"],
|
||||
"tool_description_overrides": {"tool3": "Override desc"},
|
||||
}
|
||||
|
||||
config = ConfigData.from_dict(data)
|
||||
assert config.name == "test-context"
|
||||
assert config.description == "Test context"
|
||||
assert config.system_prompt_extension == "You are in test context"
|
||||
assert config.excluded_tools == {"tool1", "tool2"}
|
||||
assert config.tool_description_overrides == {"tool3": "Override desc"}
|
||||
|
||||
|
||||
def test_config_loader_default_dirs():
|
||||
"""Test that ConfigLoader initializes the default directories."""
|
||||
loader = ConfigLoader()
|
||||
assert os.path.exists(loader.contexts_dir)
|
||||
assert os.path.exists(loader.modes_dir)
|
||||
|
||||
|
||||
def test_load_config_from_file():
|
||||
"""Test loading a configuration from a file."""
|
||||
with tempfile.NamedTemporaryFile(suffix=".yml", mode="w+", delete=False) as f:
|
||||
f.write(
|
||||
"""
|
||||
name: test-config
|
||||
description: Test config from file
|
||||
system_prompt_extension: Test from file
|
||||
excluded_tools:
|
||||
- tool1
|
||||
- tool2
|
||||
tool_description_overrides:
|
||||
tool3: Override from file
|
||||
"""
|
||||
)
|
||||
f.flush()
|
||||
|
||||
try:
|
||||
loader = ConfigLoader()
|
||||
config = loader.load_config_from_file(f.name)
|
||||
|
||||
assert config.name == "test-config"
|
||||
assert config.description == "Test config from file"
|
||||
assert config.system_prompt_extension == "Test from file"
|
||||
assert config.excluded_tools == {"tool1", "tool2"}
|
||||
assert config.tool_description_overrides == {"tool3": "Override from file"}
|
||||
finally:
|
||||
os.unlink(f.name)
|
||||
|
||||
|
||||
def test_get_context():
|
||||
"""Test getting a context by name."""
|
||||
loader = ConfigLoader()
|
||||
|
||||
# This assumes contexts/desktop-app.yml exists in the project
|
||||
config = loader.get_context("desktop-app")
|
||||
assert config.name == "desktop-app"
|
||||
|
||||
# Test cache functionality
|
||||
assert "desktop-app" in loader.context_cache
|
||||
|
||||
|
||||
def test_get_mode():
|
||||
"""Test getting a mode by name."""
|
||||
loader = ConfigLoader()
|
||||
|
||||
# This assumes modes/interactive.yml exists in the project
|
||||
config = loader.get_mode("interactive")
|
||||
assert config.name == "interactive"
|
||||
|
||||
# Test cache functionality
|
||||
assert "interactive" in loader.mode_cache
|
||||
|
||||
|
||||
def test_list_available():
|
||||
"""Test listing available contexts and modes."""
|
||||
loader = ConfigLoader()
|
||||
|
||||
contexts = loader.list_available_contexts()
|
||||
assert len(contexts) > 0
|
||||
assert "desktop-app" in contexts
|
||||
|
||||
modes = loader.list_available_modes()
|
||||
assert len(modes) > 0
|
||||
assert "interactive" in modes
|
||||
Reference in New Issue
Block a user