Make Agno agent work

scripts/agno_agent.py

@@ -1,32 +1,42 @@
+import os
+from logging import Logger
+
 from agno.agent import Agent
 from agno.models.anthropic import Claude
+from agno.models.google import Gemini
 from agno.playground import Playground, serve_playground_app
 from dotenv import load_dotenv
+from sensai.util import logging
+from sensai.util.helper import mark_used
 
 from serena.agent import SerenaAgent
 from serena.agno import SerenaAgnoToolkit
 
+os.chdir(os.path.dirname(os.path.abspath(__file__)))
+mark_used(Gemini, Claude)
 load_dotenv()
 
 project_file_path = "../myproject.yml"
-serena_agent = SerenaAgent(project_file_path)
+serena_agent = SerenaAgent(project_file_path, start_language_server=True)
+
 model = Claude(id="claude-3-7-sonnet-20250219")
+# model = Gemini(id="gemini-2.5-pro-exp-03-25")
 
 agno_agent = Agent(
     name="Serena",
     model=model,
     description="A fully-featured coding assistant",
     tools=[SerenaAgnoToolkit(serena_agent)],  # type: ignore
-    show_tool_calls=True,
+    show_tool_calls=False,
     markdown=True,
     system_message="",  # Todo
-    read_tool_call_history=True,
+    read_tool_call_history=False,
     telemetry=False,
 )
 
-
-## The app object should be in the module scope so that the server can access it for hot reloading
+# The app object must be in the module scope so that the server can access it for hot reloading
 app = Playground(agents=[agno_agent]).get_app()
 
 if __name__ == "__main__":
-    serve_playground_app("agno_agent:app", reload=True)
+    Logger.root.setLevel(logging.INFO)
+    serve_playground_app("agno_agent:app", reload=False)

src/serena/agent.py

@@ -37,7 +37,14 @@ TTool = TypeVar("TTool", bound="Tool")
 
 
 class SerenaAgent:
-    def __init__(self, project_file_path: str):
+    def __init__(self, project_file_path: str, start_language_server: bool = False):
+        """
+        :param project_file_path: the project configuration file path (.yml)
+        :param start_language_server: whether to start the language server immediately and manage its
+            lifecycle internally
+        """
+        self._start_language_server = start_language_server
+
         if not os.path.exists(project_file_path):
             print(f"Project file not found: {project_file_path}", file=sys.stderr)
             sys.exit(1)
@@ -51,12 +58,13 @@ class SerenaAgent:
 
         # enable GUI log window
         enable_gui_log = project_config.get("gui_log_window", True)
+        self._gui_log_handler = None
         if enable_gui_log:
             if platform.system() == "Darwin":
                 log.warning("GUI log window is not supported on macOS")
             else:
-                log_handler = GuiLogViewerHandler(GuiLogViewer(title="Serena Logs"), level=LOG_LEVEL, format_string=LOG_FORMAT)
-                Logger.root.addHandler(log_handler)
+                self._gui_log_handler = GuiLogViewerHandler(GuiLogViewer(title="Serena Logs"), level=LOG_LEVEL, format_string=LOG_FORMAT)
+                Logger.root.addHandler(self._gui_log_handler)
 
         log.info(
             f"Starting serena server for project {project_file_path} (language={self.language}, root={self.project_root}); "
@@ -83,14 +91,37 @@ class SerenaAgent:
             self.tools[tool_class] = tool_instance
         log.info(f"Loaded tools: {', '.join([tool.get_name() for tool in self.tools.values()])}")
 
+        # start the language server if requested
+        if self._start_language_server:
+            log.info("Starting the language server ...")
+            self.language_server.start()
+
     def get_tool(self, tool_class: type[TTool]) -> TTool:
         return self.tools[tool_class]  # type: ignore
 
     def get_serena_managed_dir(self) -> str:
         return os.path.join(self.project_root, ".serena")
 
+    def __del__(self) -> None:
+        """
+        Destructor to clean up the language server instance and GUI logger
+        """
+        log.info("SerenaAgent is shutting down ...")
+        if self._start_language_server:
+            log.info("Stopping the language server ...")
+            self.language_server.stop()
+        if self._gui_log_handler:
+            log.info("Stopping the GUI log window ...")
+            self._gui_log_handler.stop_viewer()
+            Logger.root.removeHandler(self._gui_log_handler)
+
     @contextmanager
-    def start_server(self) -> Iterator[None]:
+    def language_server_lifecycle_context(self) -> Iterator[None]:
+        """
+        Context manager for the language server's lifecycle
+        """
+        if self._start_language_server:
+            raise Exception("This context manager can only be used if the instance is created with start_language_server=True")
         with self.language_server.start_server():
             yield
 
@@ -139,8 +170,9 @@ _DEFAULT_MAX_ANSWER_LENGTH = int(2e5)
 
 
 class Tool(Component):
-    def get_name(self) -> str:
-        name = self.__class__.__name__
+    @classmethod
+    def get_name(cls) -> str:
+        name = cls.__name__
         if name.endswith("Tool"):
             name = name[:-4]
         # convert to snake_case
@@ -232,7 +264,7 @@ class ReadFileTool(Tool):
         return self._limit_length(result, max_answer_chars)
 
 
-class CreateFileTool(Tool):
+class CreateTextFileTool(Tool):
     def apply(self, relative_path: str, content: str) -> str:
         """
         Write a new file (or overwrite an existing file). For existing files, it is strongly recommended

src/serena/agno.py

@@ -1,18 +1,29 @@
+import logging
+from typing import Any
+
 from agno.tools import Function, Toolkit
 
 from serena.agent import SerenaAgent, Tool
 
+log = logging.getLogger(__name__)
+
 
 class SerenaAgnoToolkit(Toolkit):
     def __init__(self, serena_agent: SerenaAgent):
         super().__init__("Serena")
         for tool in serena_agent.tools.values():
             self.functions[tool.get_name()] = self._create_agno_function(tool)
+        log.info("Agno agent functions: %s", list(self.functions.keys()))
 
     @staticmethod
     def _create_agno_function(tool: Tool) -> Function:
-        def entrypoint(**kwargs):  # type: ignore
-            tool.apply_ex(log_call=True, catch_exceptions=True, **kwargs)
+        def entrypoint(**kwargs: Any) -> str:
+            if "kwargs" in kwargs:
+                # Agno sometimes passes a kwargs argument explicitly, so we merge it
+                kwargs.update(kwargs["kwargs"])
+                del kwargs["kwargs"]
+            log.info(f"Calling tool {tool}")
+            return tool.apply_ex(log_call=True, catch_exceptions=True, **kwargs)
 
         function = Function.from_callable(tool.get_apply_fn())
         function.name = tool.get_name()

src/serena/mcp.py

@@ -84,7 +84,7 @@ def create_mcp_server() -> FastMCP:
     @asynccontextmanager
     async def server_lifespan(mcp_server: FastMCP) -> AsyncIterator[SerenaMCPRequestContext]:
         """Manage server startup and shutdown lifecycle."""
-        with agent.start_server():
+        with agent.language_server_lifecycle_context():
             yield SerenaMCPRequestContext(agent=agent)
 
     mcp_settings = Settings(lifespan=server_lifespan)