Merge pull request #2915 from markparonyan/mypy-agent

fix(typing): resolve mypy errors in gns3server.agent

gns3server/agent/__init__.py

@@ -57,7 +57,7 @@ except ImportError as e:
     AI_COPILOT_AVAILABLE = False
 
     # Provide stub functions that raise helpful errors
-    async def get_project_agent_manager():
+    async def get_project_agent_manager() -> "ProjectAgentManager":
         """
         Get the global ProjectAgentManager singleton instance.
 
@@ -68,7 +68,7 @@ except ImportError as e:
             "AI Copilot is not available. Install AI dependencies with: pip install gns3-server[ai-features]"
         )
 
-    class ProjectAgentManager:
+    class ProjectAgentManager:  # type: ignore[no-redef]
         """
         Stub class for ProjectAgentManager when AI dependencies are not installed.
         """

gns3server/agent/gns3_copilot/chat_sessions_repository.py

@@ -125,7 +125,7 @@ class ChatSessionsRepository:
         project_id: str,
         title: str = "New Conversation",
         copilot_mode: Optional[str] = None,
-    ) -> ChatSession:
+    ) -> Optional[ChatSession]:
         """
         Create a new chat session.
 
@@ -158,6 +158,8 @@ class ChatSessionsRepository:
         session_id = cursor.lastrowid
         log.info("Created chat session: id=%s, thread_id=%s, copilot_mode=%s", session_id, thread_id, copilot_mode)
 
+        if session_id is None:
+            return None
         return await self.get_session_by_id(session_id)
 
     async def get_session_by_id(self, session_id: int) -> Optional[ChatSession]:
@@ -214,7 +216,7 @@ class ChatSessionsRepository:
             List of ChatSession
         """
         query = "SELECT * FROM chat_sessions"
-        params = []
+        params: List[Any] = []
 
         conditions = []
         if user_id:
@@ -268,7 +270,7 @@ class ChatSessionsRepository:
             Updated ChatSession or None
         """
         updates = []
-        params = []
+        params: List[Any] = []
 
         now = datetime.utcnow().isoformat()
 

gns3server/agent/gns3_copilot/gns3_client/api_handlers.py

@@ -162,14 +162,14 @@ def build_gns3_ctx(jwt_token: str | None = None, url: str | None = None) -> dict
     }
 
 
-def _filter_node_response(node: dict, fields: list[str] = None) -> dict:
+def _filter_node_response(node: dict, fields: list[str] | None = None) -> dict:
     """Filter node response to only include requested fields."""
     if not fields:
         fields = ["node_id", "name", "node_type", "status", "console"]
     return {k: node[k] for k in fields if k in node}
 
 
-def _filter_link_response(link: dict, fields: list[str] = None) -> dict:
+def _filter_link_response(link: dict, fields: list[str] | None = None) -> dict:
     """Filter link response to only include requested fields."""
     if not fields:
         fields = LINK_DEFAULT_FIELDS
@@ -278,7 +278,7 @@ def _batch_lifecycle(project_id, node_ids, action, conn, action_label):
         return list(pool.map(_act, node_ids))
 
 
-def start_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
+def start_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any] | list[dict[str, Any]]:
     project_id = params.get("project_id")
     if not project_id:
         return {"error": "project_id is required"}
@@ -296,7 +296,7 @@ def start_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict
     return {"message": f"Node {node_id} started", "node_id": node_id}
 
 
-def stop_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
+def stop_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any] | list[dict[str, Any]]:
     project_id = params.get("project_id")
     if not project_id:
         return {"error": "project_id is required"}
@@ -314,7 +314,7 @@ def stop_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[
     return {"message": f"Node {node_id} stopped", "node_id": node_id}
 
 
-def suspend_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
+def suspend_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any] | list[dict[str, Any]]:
     project_id = params.get("project_id")
     if not project_id:
         return {"error": "project_id is required"}
@@ -332,7 +332,7 @@ def suspend_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> di
     return {"message": f"Node {node_id} suspended", "node_id": node_id}
 
 
-def create_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
+def create_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any] | list[dict[str, Any]]:
 
     project_id = params.get("project_id")
     if not project_id:
@@ -401,7 +401,7 @@ def create_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dic
     return _filter_node_response(resp, fields)
 
 
-def delete_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
+def delete_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any] | list[dict[str, Any]]:
     project_id = params.get("project_id")
     if not project_id:
         return {"error": "project_id is required"}
@@ -706,7 +706,7 @@ def available_filters_handler(params: dict[str, Any], gns3_ctx: dict[str, Any])
     return conn.http_call("get", url).json()
 
 
-def create_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
+def create_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any] | list[dict[str, Any]]:
     project_id = params.get("project_id")
     if not project_id:
         return {"error": "project_id is required"}
@@ -762,7 +762,7 @@ def create_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dic
     return _filter_link_response(resp, fields)
 
 
-def delete_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
+def delete_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any] | list[dict[str, Any]]:
     project_id = params.get("project_id")
     if not project_id:
         return {"error": "project_id is required"}
@@ -809,7 +809,7 @@ def update_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dic
 # ── Link capture / reset handlers ──────────────────────────────────────
 
 
-def reset_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
+def reset_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any] | list[dict[str, Any]]:
     project_id = params.get("project_id")
     if not project_id:
         return {"error": "project_id is required"}

gns3server/agent/gns3_copilot/tools_v2/gns3_create_link.py

@@ -222,6 +222,8 @@ class GNS3LinkTool(BaseTool):
                         },
                         gns3_ctx,
                     )
+                    if isinstance(link_resp, list):
+                        raise RuntimeError("Unexpected batch response when creating a single link")
                     if "error" in link_resp:
                         raise RuntimeError(link_resp["error"])
 

gns3server/agent/gns3_copilot/tools_v2/gns3_create_node.py

@@ -230,6 +230,8 @@ class GNS3CreateNodeTool(BaseTool):
                         },
                         gns3_ctx,
                     )
+                    if isinstance(created, list):
+                        raise RuntimeError("Unexpected batch response when creating a single node")
                     if "error" in created:
                         raise RuntimeError(created["error"])
                     node_info = {

gns3server/agent/gns3_copilot/tools_v2/gns3_start_node.py

@@ -136,7 +136,11 @@ class GNS3StartNodeTool(BaseTool):
             results = []
             known_ids = [nid for nid in node_ids if nid in nodes_by_id]
             start_results = start_node_handler({"project_id": project_id, "node_ids": known_ids}, gns3_ctx)
-            start_errors = {r["node_id"]: r.get("error") for r in start_results if r.get("status") == "error"}
+            start_errors = (
+                {r["node_id"]: r.get("error") for r in start_results if r.get("status") == "error"}
+                if isinstance(start_results, list)
+                else {}
+            )
 
             # Get immediate status (likely 'starting' or 'stopped') — one call
             listing = get_nodes_handler({"project_id": project_id}, gns3_ctx)

gns3server/agent/gns3_copilot/tools_v2/gns3_stop_node.py

@@ -124,7 +124,11 @@ class GNS3StopNodeTool(BaseTool):
             results = []
             known_ids = [nid for nid in node_ids if nid in nodes_by_id]
             stop_results = stop_node_handler({"project_id": project_id, "node_ids": known_ids}, gns3_ctx)
-            stop_errors = {r["node_id"]: r.get("error") for r in stop_results if r.get("status") == "error"}
+            stop_errors = (
+                {r["node_id"]: r.get("error") for r in stop_results if r.get("status") == "error"}
+                if isinstance(stop_results, list)
+                else {}
+            )
 
             # Get updated status — one call
             listing = get_nodes_handler({"project_id": project_id}, gns3_ctx)

gns3server/agent/gns3_copilot/tools_v2/gns3_suspend_node.py

@@ -130,7 +130,11 @@ class GNS3SuspendNodeTool(BaseTool):
             results = []
             known_ids = [nid for nid in node_ids if nid in nodes_by_id]
             suspend_results = suspend_node_handler({"project_id": project_id, "node_ids": known_ids}, gns3_ctx)
-            suspend_errors = {r["node_id"]: r.get("error") for r in suspend_results if r.get("status") == "error"}
+            suspend_errors = (
+                {r["node_id"]: r.get("error") for r in suspend_results if r.get("status") == "error"}
+                if isinstance(suspend_results, list)
+                else {}
+            )
 
             # Get updated status — one call
             listing = get_nodes_handler({"project_id": project_id}, gns3_ctx)

pyproject.toml

@@ -294,12 +294,6 @@ enable_error_code = ["ignore-without-code", "redundant-expr", "truthy-bool"]
 # Baseline: modules with existing type errors (error count at baseline time).
 # Remove modules from this list one small PR at a time. Never add new ones.
 module = [
-  "gns3server.agent",  # 2
-  "gns3server.agent.gns3_copilot.chat_sessions_repository",  # 8
-  "gns3server.agent.gns3_copilot.gns3_client.api_handlers",  # 10
-  "gns3server.agent.gns3_copilot.tools_v2.gns3_start_node",  # 3
-  "gns3server.agent.gns3_copilot.tools_v2.gns3_stop_node",  # 3
-  "gns3server.agent.gns3_copilot.tools_v2.gns3_suspend_node",  # 3
   "gns3server.agent.gns3_copilot.utils.custom_netmiko.scripts.list_netmiko_telnet_devices",  # 1
   "gns3server.agent.mcp.appliances",  # 1
   "gns3server.agent.web_wireshark.docker_client",  # 5