mirror of
https://github.com/tiennm99/serena.git
synced 2026-08-18 04:28:14 +00:00
Fix mypy issues
This commit is contained in:
@@ -14,9 +14,10 @@ def serena_version() -> str:
|
||||
version = __version__
|
||||
try:
|
||||
git_status = get_git_status()
|
||||
version += f"-{git_status.commit[:8]}"
|
||||
if not git_status.is_clean:
|
||||
version += "-dirty"
|
||||
if git_status is not None:
|
||||
version += f"-{git_status.commit[:8]}"
|
||||
if not git_status.is_clean:
|
||||
version += "-dirty"
|
||||
except:
|
||||
pass
|
||||
return version
|
||||
|
||||
@@ -3,8 +3,9 @@ import queue
|
||||
import socket
|
||||
import threading
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
from flask import Flask, request, send_from_directory
|
||||
from flask import Flask, Response, request, send_from_directory
|
||||
from pydantic import BaseModel
|
||||
from sensai.util import logging
|
||||
|
||||
@@ -85,16 +86,16 @@ class SerenaDashboardAPI:
|
||||
def _setup_routes(self) -> None:
|
||||
# Static files
|
||||
@self._app.route("/dashboard/<path:filename>")
|
||||
def serve_dashboard(filename):
|
||||
def serve_dashboard(filename: str) -> Response:
|
||||
return send_from_directory(SERENA_DASHBOARD_DIR, filename)
|
||||
|
||||
@self._app.route("/dashboard/")
|
||||
def serve_dashboard_index():
|
||||
def serve_dashboard_index() -> Response:
|
||||
return send_from_directory(SERENA_DASHBOARD_DIR, "index.html")
|
||||
|
||||
# API routes
|
||||
@self._app.route("/get_log_messages", methods=["POST"])
|
||||
def get_log_messages():
|
||||
def get_log_messages() -> dict[str, Any]:
|
||||
request_data = request.get_json()
|
||||
if not request_data:
|
||||
request_log = RequestLog()
|
||||
@@ -105,12 +106,12 @@ class SerenaDashboardAPI:
|
||||
return result.model_dump()
|
||||
|
||||
@self._app.route("/get_tool_names", methods=["GET"])
|
||||
def get_tool_names():
|
||||
def get_tool_names() -> dict[str, Any]:
|
||||
result = self._get_tool_names()
|
||||
return result.model_dump()
|
||||
|
||||
@self._app.route("/shutdown", methods=["PUT"])
|
||||
def shutdown():
|
||||
def shutdown() -> dict[str, str]:
|
||||
self._shutdown()
|
||||
return {"status": "shutting down"}
|
||||
|
||||
|
||||
+13
-5
@@ -184,11 +184,11 @@ class SerenaMCPFactory:
|
||||
# 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_settings: Settings = Settings(lifespan=self.server_lifespan, host=host, port=port)
|
||||
mcp = FastMCP(**mcp_settings.model_dump())
|
||||
return mcp
|
||||
|
||||
@asynccontextmanager
|
||||
@asynccontextmanager # type: ignore
|
||||
@abstractmethod
|
||||
async def server_lifespan(self, mcp_server: FastMCP) -> AsyncIterator[None]:
|
||||
"""Manage server startup and shutdown lifecycle."""
|
||||
@@ -213,6 +213,7 @@ class SerenaMCPFactorySingleProcess(SerenaMCPFactory):
|
||||
self.agent = SerenaAgent(project=self.project, serena_config=serena_config, context=self.context, modes=modes)
|
||||
|
||||
def _iter_tools(self) -> Iterator[ToolInterface]:
|
||||
assert self.agent is not None
|
||||
yield from self.agent.get_exposed_tool_instances()
|
||||
|
||||
@asynccontextmanager
|
||||
@@ -236,12 +237,12 @@ class SerenaMCPFactoryWithProcessIsolation(SerenaMCPFactory):
|
||||
"""
|
||||
super().__init__(context=context, project=project)
|
||||
|
||||
self.active_tool_names: list[str] | None = None
|
||||
self.active_tool_names: set[str] | None = None
|
||||
self.serena_agent_process: ProcessIsolatedSerenaAgent | None = None
|
||||
self.serena_dashboard_process: ProcessIsolatedDashboard | None = None
|
||||
|
||||
@staticmethod
|
||||
def _determine_active_tool_names(context: SerenaAgentContext, project: Project) -> set[str]:
|
||||
def _determine_active_tool_names(context: SerenaAgentContext, project: Project | None) -> set[str]:
|
||||
"""
|
||||
Determine the names of tools that should be included in this session based on the context.
|
||||
"""
|
||||
@@ -306,6 +307,8 @@ class SerenaMCPFactoryWithProcessIsolation(SerenaMCPFactory):
|
||||
)
|
||||
|
||||
def _iter_tools(self) -> Iterator[ToolInterface]:
|
||||
assert self.active_tool_names is not None
|
||||
assert self.serena_agent_process is not None
|
||||
for tool_name in self.active_tool_names:
|
||||
yield ProcessIsolatedTool(process_agent=self.serena_agent_process, tool_name=tool_name)
|
||||
|
||||
@@ -319,12 +322,14 @@ class SerenaMCPFactoryWithProcessIsolation(SerenaMCPFactory):
|
||||
mcp._tool_manager._tools[tool.get_name()] = mcp_tool
|
||||
|
||||
def _instantiate_agent(self, serena_config: SerenaConfig, modes: list[SerenaAgentMode]) -> None:
|
||||
self.project_instance = serena_config.get_project(self.project)
|
||||
if self.project is not None:
|
||||
self.project_instance = serena_config.get_project(self.project)
|
||||
self.serena_agent_process = ProcessIsolatedSerenaAgent(
|
||||
project=self.project, serena_config=serena_config, modes=modes, context=self.context
|
||||
)
|
||||
self.active_tool_names = self._determine_active_tool_names(self.context, self.project_instance)
|
||||
if serena_config.web_dashboard:
|
||||
assert self.active_tool_names is not None
|
||||
self.serena_dashboard_process = ProcessIsolatedDashboard(tool_names=sorted(self.active_tool_names))
|
||||
|
||||
def create_mcp_server(
|
||||
@@ -399,8 +404,10 @@ class SerenaMCPFactoryWithProcessIsolation(SerenaMCPFactory):
|
||||
|
||||
if self.serena_dashboard_process is not None:
|
||||
log.info("Starting dashboard process")
|
||||
assert self.serena_dashboard_process is not None
|
||||
self.serena_dashboard_process.start()
|
||||
log.info("Starting serena agent process")
|
||||
assert self.serena_agent_process is not None
|
||||
self.serena_agent_process.start()
|
||||
|
||||
self._set_mcp_tools(mcp_server)
|
||||
@@ -578,6 +585,7 @@ 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
|
||||
if USE_SOLID_LSP:
|
||||
mcp_factory = SerenaMCPFactorySingleProcess(context=context, project=project_file)
|
||||
else:
|
||||
|
||||
@@ -80,7 +80,7 @@ def _dashboard_worker(
|
||||
port_value.value = port
|
||||
|
||||
# Start Flask server in a thread
|
||||
def run_flask_server():
|
||||
def run_flask_server() -> None:
|
||||
api._app.run(host="0.0.0.0", port=port, debug=False, use_reloader=False, threaded=True)
|
||||
|
||||
server_thread = threading.Thread(target=run_flask_server, daemon=True)
|
||||
@@ -414,10 +414,10 @@ class ProcessIsolatedSerenaAgent:
|
||||
|
||||
# Create communication pipe
|
||||
parent_conn, child_conn = multiprocessing.Pipe()
|
||||
self.conn = parent_conn
|
||||
self.conn = parent_conn # type: ignore
|
||||
|
||||
# Create and start worker process, passing along the dashboard's queue if available
|
||||
worker = SerenaAgentWorker(child_conn)
|
||||
worker = SerenaAgentWorker(child_conn) # type: ignore
|
||||
self.process = multiprocessing.Process(target=worker.run, args=[_global_log_queue])
|
||||
self.process.start()
|
||||
|
||||
|
||||
@@ -51,7 +51,7 @@ def subprocess_check_output(args: list[str], encoding: str = "utf-8", strip: boo
|
||||
}
|
||||
if platform.system() == "Windows":
|
||||
kwargs["creationflags"] = subprocess.CREATE_NO_WINDOW
|
||||
output = subprocess.check_output(args, **kwargs).decode(encoding)
|
||||
output = subprocess.check_output(args, **kwargs).decode(encoding) # type: ignore
|
||||
if strip:
|
||||
output = output.strip()
|
||||
return output
|
||||
|
||||
Reference in New Issue
Block a user