Merge branch '3.1' into mypy-agent-copilot-tools-nornir

gns3server/agent/gns3_copilot/agent/context_manager.py

@@ -140,7 +140,7 @@ def estimate_tool_tokens(tools: list[Any]) -> int:
     for tool in tools:
         try:
             # Build tool schema in OpenAI format
-            tool_schema = {
+            tool_schema: dict[str, Any] = {
                 "type": "function",
                 "function": {
                     "name": tool.name,

gns3server/agent/gns3_copilot/agent/model_factory.py

@@ -110,7 +110,7 @@ def create_base_model(
 
     try:
         # Prepare parameters for init_chat_model
-        init_params = {
+        init_params: dict[str, Any] = {
             "model": config_vars["model_name"],
             "model_provider": config_vars["model_provider"],
             "api_key": config_vars["api_key"],
@@ -177,7 +177,7 @@ def create_title_model(
 
     try:
         # Prepare parameters for init_chat_model
-        init_params = {
+        init_params: dict[str, Any] = {
             "model": config_vars["model_name"],
             "model_provider": config_vars["model_provider"],
             "api_key": config_vars["api_key"],

gns3server/agent/gns3_copilot/agent_service.py

@@ -256,17 +256,16 @@ class AgentService:
         )
 
         # Ensure checkpointer is initialized
-        if not self._checkpointer_conn:
+        checkpointer = await self._get_checkpointer()
-            await self._get_checkpointer()
 
         # Get or create chat session
-        repo = ChatSessionsRepository(self._checkpointer_conn)
+        repo = ChatSessionsRepository(checkpointer.conn)
         session = await repo.get_session_by_thread(session_id)
         is_new_session = session is None
 
         if is_new_session:
             # Create new session
-            copilot_mode = llm_config.get("copilot_mode", "teaching_assistant").lower()
+            copilot_mode = (llm_config or {}).get("copilot_mode", "teaching_assistant").lower()
             session = await repo.create_session(
                 thread_id=session_id,
                 user_id=user_id or "",
@@ -290,7 +289,7 @@ class AgentService:
 
         # Build config - only thread-safe identifiers
         # Determine recursion_limit based on copilot_mode
-        copilot_mode = llm_config.get("copilot_mode", "teaching_assistant").lower()
+        copilot_mode = (llm_config or {}).get("copilot_mode", "teaching_assistant").lower()
         if copilot_mode == "troubleshooting_injection":
             recursion_limit = 100  # Need more recursion depth for fault injection workflow
             log.debug("Using extended recursion_limit for troubleshooting_injection mode: 100")
@@ -453,10 +452,10 @@ class AgentService:
                         yield chunk
                 else:
                     # Use stateless converter for other events
-                    chunk = self._convert_event_to_chunk(event, session_id)
+                    converted = self._convert_event_to_chunk(event, session_id)
-                    if chunk:
+                    if converted:
-                        log.debug("Yielding chunk: type=%s", chunk.get("type"))
+                        log.debug("Yielding chunk: type=%s", converted.get("type"))
-                        yield chunk
+                        yield converted
 
             # Check if stream was aborted and yield tool_end events for aborted tools
             from gns3server.agent.gns3_copilot.agent.gns3_copilot import (
@@ -624,10 +623,9 @@ class AgentService:
         Returns:
             List of session dictionaries
         """
-        if not self._checkpointer_conn:
+        checkpointer = await self._get_checkpointer()
-            await self._get_checkpointer()
 
-        repo = ChatSessionsRepository(self._checkpointer_conn)
+        repo = ChatSessionsRepository(checkpointer.conn)
         sessions = await repo.list_sessions(user_id=user_id, copilot_mode=copilot_mode, limit=limit)
         return [s.to_dict() for s in sessions]
 
@@ -641,10 +639,9 @@ class AgentService:
         Returns:
             True if deleted, False if not found
         """
-        if not self._checkpointer_conn:
+        checkpointer = await self._get_checkpointer()
-            await self._get_checkpointer()
 
-        repo = ChatSessionsRepository(self._checkpointer_conn)
+        repo = ChatSessionsRepository(checkpointer.conn)
         return await repo.delete_session(session_id)
 
     async def rename_session(self, session_id: str, new_title: str) -> Optional[Dict[str, Any]]:
@@ -658,10 +655,9 @@ class AgentService:
         Returns:
             Updated session dictionary or None
         """
-        if not self._checkpointer_conn:
+        checkpointer = await self._get_checkpointer()
-            await self._get_checkpointer()
 
-        repo = ChatSessionsRepository(self._checkpointer_conn)
+        repo = ChatSessionsRepository(checkpointer.conn)
         session = await repo.update_session(thread_id=session_id, title=new_title)
         return session.to_dict() if session else None
 
@@ -676,10 +672,9 @@ class AgentService:
         Returns:
             Updated session dictionary or None
         """
-        if not self._checkpointer_conn:
+        checkpointer = await self._get_checkpointer()
-            await self._get_checkpointer()
 
-        repo = ChatSessionsRepository(self._checkpointer_conn)
+        repo = ChatSessionsRepository(checkpointer.conn)
         session = await repo.pin_session(thread_id=session_id, pinned=pinned)
         return session.to_dict() if session else None
 

gns3server/agent/gns3_copilot/gns3_client/connector_factory.py

@@ -347,7 +347,7 @@ def get_gns3_server_host() -> str:
         host = get_gns3_server_host()
         print(f"GNS3 server host: {host}")
     """
-    url = _detect_url_for_api()
+    url = _detect_url_for_api() or DEFAULT_GNS3_URL
 
     # Extract host from URL
     # URL format: protocol://host:port

gns3server/agent/gns3_copilot/gns3_client/project_inventory.py

@@ -49,7 +49,7 @@ def build_nodes_inventory(nodes: list[dict[str, Any]], server_host: str | None)
     """
     inventory: dict[str, Any] = {}
     for n in nodes:
-        inventory[n.get("name")] = {
+        inventory[n.get("name") or ""] = {
             "server": server_host,
             "name": n.get("name"),
             "node_id": n.get("node_id"),
@@ -68,7 +68,7 @@ def build_nodes_inventory(nodes: list[dict[str, Any]], server_host: str | None)
     return inventory
 
 
-def build_links_summary(nodes: list[dict[str, Any]], links: list[dict[str, Any]]) -> list[dict[str, str]]:
+def build_links_summary(nodes: list[dict[str, Any]], links: list[dict[str, Any]]) -> list[dict[str, Any]]:
     """
     Build a human/LLM-friendly link list resolving node and port names.
 
@@ -76,7 +76,7 @@ def build_links_summary(nodes: list[dict[str, Any]], links: list[dict[str, Any]]
     Links whose endpoints cannot be resolved are skipped, mirroring the
     original Project.links_summary behavior.
     """
-    summary: list[dict[str, str]] = []
+    summary: list[dict[str, Any]] = []
     for link in links:
         if not link.get("nodes"):
             continue

gns3server/agent/gns3_copilot/skills/loader.py

@@ -248,7 +248,7 @@ class SkillsLoader:
                     continue
                 # Use device_type from YAML content as the key
                 # Fallback to filename stem if device_type not present
-                skill_key = skill_data.get("device_type") if isinstance(skill_data, dict) else None
+                skill_key = skill_data.get("device_type")
                 if not skill_key:
                     skill_key = yaml_file.stem
                     logger.warning(f"No device_type in {yaml_file}, using filename '{skill_key}' as key")

gns3server/agent/gns3_copilot/skills/manager.py

@@ -37,8 +37,10 @@ from typing import Optional, Dict, Any
 
 try:
     import git
+
+    GIT_AVAILABLE = True
 except ImportError:
-    git = None
+    GIT_AVAILABLE = False
 
 from gns3server.config import Config
 from .loader import SkillsLoader
@@ -66,7 +68,7 @@ class SkillsManager:
     - Version tracking
     """
 
-    def __init__(self, repo_url: str = None, branch: str = "main", auto_update: bool = False):
+    def __init__(self, repo_url: Optional[str] = None, branch: str = "main", auto_update: bool = False):
         """
         Initialize the skills manager.
 
@@ -90,7 +92,7 @@ class SkillsManager:
         self._repo: Optional["git.Repo"] = None
         self._prompt_count = 0
 
-        if git is None:
+        if not GIT_AVAILABLE:
             logger.warning("GitPython is not installed. Skills management features will be limited.")
 
     def initialize(self) -> bool:
@@ -134,7 +136,7 @@ class SkillsManager:
         - Up to date → nothing
         - Network error → use existing files, log warning
         """
-        if git is None or self._repo is None:
+        if not GIT_AVAILABLE or self._repo is None:
             return
 
         # Check for uncommitted changes
@@ -173,7 +175,7 @@ class SkillsManager:
         Returns:
             True if successful, False otherwise
         """
-        if git is None:
+        if not GIT_AVAILABLE:
             logger.error("GitPython is not installed. Cannot clone repository.")
             return False
 
@@ -347,7 +349,7 @@ class SkillsManager:
         Returns:
             Commit hash as string, or empty string if not available
         """
-        if git is None or self._repo is None:
+        if not GIT_AVAILABLE or self._repo is None:
             try:
                 self._repo = git.Repo(self.local_path)
             except Exception:
@@ -409,7 +411,7 @@ class SkillsManager:
         Returns:
             True if successful, False otherwise
         """
-        if git is None:
+        if not GIT_AVAILABLE:
             logger.error("GitPython is not installed. Cannot rollback.")
             return False
 
@@ -436,7 +438,7 @@ class SkillsManager:
         Returns:
             List of commit information dictionaries
         """
-        if git is None:
+        if not GIT_AVAILABLE:
             return []
 
         try:

gns3server/agent/mcp/__init__.py

@@ -405,22 +405,24 @@ async def project_stats(
 @mcp.tool()
 async def project_update(
     project_id: Annotated[str, Field(description="UUID of the project to update")],
-    name: Annotated[str, Field(description="New project name")] = None,
+    name: Annotated[str | None, Field(description="New project name")] = None,
-    auto_close: Annotated[bool, Field(description="Close project when last client leaves")] = None,
+    auto_close: Annotated[bool | None, Field(description="Close project when last client leaves")] = None,
-    auto_open: Annotated[bool, Field(description="Project opens when GNS3 starts")] = None,
+    auto_open: Annotated[bool | None, Field(description="Project opens when GNS3 starts")] = None,
-    auto_start: Annotated[bool, Field(description="Project starts when opened")] = None,
+    auto_start: Annotated[bool | None, Field(description="Project starts when opened")] = None,
-    scene_width: Annotated[int, Field(description="Width of the drawing area")] = None,
+    scene_width: Annotated[int | None, Field(description="Width of the drawing area")] = None,
-    scene_height: Annotated[int, Field(description="Height of the drawing area")] = None,
+    scene_height: Annotated[int | None, Field(description="Height of the drawing area")] = None,
-    zoom: Annotated[int, Field(description="Zoom of the drawing area")] = None,
+    zoom: Annotated[int | None, Field(description="Zoom of the drawing area")] = None,
-    show_layers: Annotated[bool, Field(description="Show layers on the drawing area")] = None,
+    show_layers: Annotated[bool | None, Field(description="Show layers on the drawing area")] = None,
-    snap_to_grid: Annotated[bool, Field(description="Snap to grid on the drawing area")] = None,
+    snap_to_grid: Annotated[bool | None, Field(description="Snap to grid on the drawing area")] = None,
-    show_grid: Annotated[bool, Field(description="Show the grid on the drawing area")] = None,
+    show_grid: Annotated[bool | None, Field(description="Show the grid on the drawing area")] = None,
-    grid_size: Annotated[int, Field(description="Grid size for the drawing area for nodes")] = None,
+    grid_size: Annotated[int | None, Field(description="Grid size for the drawing area for nodes")] = None,
-    drawing_grid_size: Annotated[int, Field(description="Grid size for the drawing area for drawings")] = None,
+    drawing_grid_size: Annotated[int | None, Field(description="Grid size for the drawing area for drawings")] = None,
-    show_interface_labels: Annotated[bool, Field(description="Show interface labels on the drawing area")] = None,
+    show_interface_labels: Annotated[
+        bool | None, Field(description="Show interface labels on the drawing area")
+    ] = None,
 ) -> list[dict[str, Any]]:
     """Update a project's properties (name, auto_close, auto_open, etc.)."""
-    params = {"project_id": project_id}
+    params: dict[str, Any] = {"project_id": project_id}
     local_vars = {
         "name": name,
         "auto_close": auto_close,
@@ -449,7 +451,7 @@ async def project_duplicate(
     reset_mac_addresses: Annotated[bool, Field(description="Reset MAC addresses for this project")] = False,
 ) -> list[dict[str, Any]]:
     """Duplicate a project."""
-    params = {"project_id": project_id, "name": name}
+    params: dict[str, Any] = {"project_id": project_id, "name": name}
     if reset_mac_addresses:
         params["reset_mac_addresses"] = reset_mac_addresses
     return await asyncio.to_thread(_run_handler_sync, duplicate_project_handler, params)
@@ -523,7 +525,7 @@ async def node_start(
     ] = None,
 ) -> list[dict[str, Any]]:
     """Start one or more nodes. Provide node_id for single, or node_ids for batch."""
-    params = {"project_id": project_id}
+    params: dict[str, Any] = {"project_id": project_id}
     if node_ids:
         params["node_ids"] = node_ids
     else:
@@ -540,7 +542,7 @@ async def node_stop(
     ] = None,
 ) -> list[dict[str, Any]]:
     """Stop one or more nodes. Provide node_id for single, or node_ids for batch."""
-    params = {"project_id": project_id}
+    params: dict[str, Any] = {"project_id": project_id}
     if node_ids:
         params["node_ids"] = node_ids
     else:
@@ -557,7 +559,7 @@ async def node_suspend(
     ] = None,
 ) -> list[dict[str, Any]]:
     """Suspend one or more nodes. Provide node_id for single, or node_ids for batch."""
-    params = {"project_id": project_id}
+    params: dict[str, Any] = {"project_id": project_id}
     if node_ids:
         params["node_ids"] = node_ids
     else:
@@ -635,7 +637,7 @@ async def node_delete(
     ] = None,
 ) -> list[dict[str, Any]]:
     """Delete one or more nodes from a project. Provide node_id for single, or node_ids for batch."""
-    params = {"project_id": project_id}
+    params: dict[str, Any] = {"project_id": project_id}
     if node_ids:
         params["node_ids"] = node_ids
     else:
@@ -756,7 +758,7 @@ async def link_create(
                 "fields": fields,
             },
         )
-    params = {"project_id": project_id, "nodes": nodes, "link_type": link_type, "fields": fields}
+    params: dict[str, Any] = {"project_id": project_id, "nodes": nodes, "link_type": link_type, "fields": fields}
     if filters:
         params["filters"] = filters
     return await asyncio.to_thread(_run_handler_sync, create_link_handler, params)
@@ -771,7 +773,7 @@ async def link_delete(
     ] = None,
 ) -> list[dict[str, Any]]:
     """Delete one or more links from a project."""
-    params = {"project_id": project_id}
+    params: dict[str, Any] = {"project_id": project_id}
     if link_ids:
         params["link_ids"] = link_ids
     else:
@@ -1201,7 +1203,7 @@ async def link_reset(
     while keeping the filter configuration intact. Filters are preserved but
     their internal application state resets.
     """
-    params = {"project_id": project_id}
+    params: dict[str, Any] = {"project_id": project_id}
     if link_ids:
         params["link_ids"] = link_ids
     else:
@@ -1222,7 +1224,7 @@ async def link_capture_start(
     ] = None,
 ) -> list[dict[str, Any]]:
     """Start packet capture on one or more links."""
-    params = {
+    params: dict[str, Any] = {
         "project_id": project_id,
         "data_link_type": data_link_type,
         "capture_file_name": capture_file_name,
@@ -1245,7 +1247,7 @@ async def link_capture_stop(
     ] = None,
 ) -> list[dict[str, Any]]:
     """Stop packet capture on one or more links."""
-    params = {"project_id": project_id}
+    params: dict[str, Any] = {"project_id": project_id}
     if link_ids:
         params["link_ids"] = link_ids
     else:
@@ -1273,7 +1275,7 @@ async def link_capture_download(
       - The ticket expires after 10 minutes: call this tool again to get a
         fresh one; do not reuse an old URL.
     """
-    params = {"project_id": project_id}
+    params: dict[str, Any] = {"project_id": project_id}
     if link_ids:
         params["link_ids"] = link_ids
     else:
@@ -1555,7 +1557,7 @@ async def drawing_update(
     rotation: Annotated[int | None, Field(description="Rotation angle in degrees, -359 to 359")] = None,
 ) -> list[dict[str, Any]]:
     """Update a drawing's properties (svg, position, lock state, rotation, etc.)."""
-    params = {"project_id": project_id, "drawing_id": drawing_id}
+    params: dict[str, Any] = {"project_id": project_id, "drawing_id": drawing_id}
     local_vars = {"svg": svg, "locked": locked, "x": x, "y": y, "z": z, "rotation": rotation}
     for key, val in local_vars.items():
         if val is not None:

gns3server/agent/web_wireshark/manage_wireshark.py

@@ -81,10 +81,7 @@ from gns3server.utils.uuid_validator import validate_uuid
 if __name__ == "__main__":
     sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))))
 
-try:
+from gns3server.agent.web_wireshark.manager import WebWiresharkManager
-    from gns3server.agent.web_wireshark.manager import WebWiresharkManager
-except ImportError:
-    from manager import WebWiresharkManager
 
 logger = logging.getLogger(__name__)
 

gns3server/agent/web_wireshark/manager.py

@@ -84,7 +84,7 @@ class WebWiresharkManager:
             logger.warning(f"Container {container_id[:12]} health check failed: {e}")
             return False
 
-    async def _exec_in_container(self, container_id: str, command: str, timeout: int = None) -> tuple:
+    async def _exec_in_container(self, container_id: str, command: str, timeout: Optional[int] = None) -> tuple:
         """Execute command in container with timeout.
 
         Args:
@@ -197,7 +197,7 @@ class WebWiresharkManager:
             stdout, _ = await proc.communicate()
 
             # Build parent->children mapping
-            children_map = {}
+            children_map: dict[int, list[int]] = {}
             for line in stdout.decode().strip().split("\n"):
                 parts = line.split(None, 2)
                 if len(parts) < 3:
@@ -326,7 +326,7 @@ class WebWiresharkManager:
                 return
 
             # Build parent->children mapping and collect process info
-            children_map = {}  # ppid -> [pid]
+            children_map: dict[int, list[int]] = {}  # ppid -> [pid]
             process_info = {}  # pid -> (ppid, command)
 
             for line in stdout.decode().strip().split("\n"):
@@ -480,7 +480,7 @@ class WebWiresharkManager:
             logger.debug(f"Cannot get URL from Config: {e}")
         return None
 
-    async def _get_container_gateway_ip(self, container_id: str = None) -> Optional[str]:
+    async def _get_container_gateway_ip(self, container_id: Optional[str] = None) -> Optional[str]:
         """Get the Docker bridge gateway IP for container to access host.
 
         Args:
@@ -519,7 +519,7 @@ class WebWiresharkManager:
 
         return None
 
-    async def get_container_ip(self, container_name: str, container_id: str = None) -> Optional[str]:
+    async def get_container_ip(self, container_name: str, container_id: Optional[str] = None) -> Optional[str]:
         """Get the container IP address in the wireshark network.
 
         Args:
@@ -626,7 +626,7 @@ class WebWiresharkManager:
         project_id: str,
         image: str = "gns3/web-wireshark:latest",
         memory: str = "2g",
-        memory_swap: str = None,
+        memory_swap: Optional[str] = None,
         cpus: float = 1.0,
         pids_limit: int = 1000,
     ) -> str:
@@ -935,10 +935,10 @@ class WebWiresharkManager:
         project_id: str,
         link_id: str,
         jwt_token: str,
-        capture_stream_url: str = None,
+        capture_stream_url: Optional[str] = None,
         image: str = "gns3/web-wireshark:latest",
         memory: str = "2g",
-        memory_swap: str = None,
+        memory_swap: Optional[str] = None,
         cpus: float = 1.0,
         pids_limit: int = 1000,
     ):

gns3server/agent/web_wireshark/stats.py

@@ -7,7 +7,7 @@ about Web Wireshark containers and sessions.
 
 import logging
 import subprocess
-from typing import Dict, List, Optional
+from typing import Any, Dict, List, Optional
 
 logger = logging.getLogger(__name__)
 
@@ -30,7 +30,7 @@ async def collect_webwireshark_stats(projects: List) -> Dict:
     """
     from .manager import WebWiresharkManager
 
-    stats = {"total_containers": 0, "running_containers": 0, "active_sessions": 0, "containers": []}
+    stats: Dict[str, Any] = {"total_containers": 0, "running_containers": 0, "active_sessions": 0, "containers": []}
 
     # Create a single manager instance and reuse it
     manager = WebWiresharkManager()

gns3server/api/routes/compute/qemu_nodes.py

@@ -23,7 +23,7 @@ import os
 from fastapi import APIRouter, WebSocket, Depends, Body, Path, status, HTTPException
 from fastapi.encoders import jsonable_encoder
 from fastapi.responses import StreamingResponse
-from typing import Union
+from typing import Any, Union
 from uuid import UUID
 
 from gns3server import schemas
@@ -36,7 +36,9 @@ import logging
 
 log = logging.getLogger(__name__)
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or Qemu node"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Could not find project or Qemu node"}
+}
 
 router = APIRouter(responses=responses)
 
@@ -64,29 +66,29 @@ async def create_qemu_node(project_id: UUID, node_data: schemas.QemuCreate) -> s
     """
 
     qemu = Qemu.instance()
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
+    data = jsonable_encoder(node_data, exclude_unset=True)
-    disk_images_to_reset = set(node_data.pop("disk_images_to_reset", []))
+    disk_images_to_reset = set(data.pop("disk_images_to_reset", []))
     vm = await qemu.create_node(
-        node_data.pop("name"),
+        data.pop("name"),
         str(project_id),
-        node_data.pop("node_id", None),
+        data.pop("node_id", None),
-        linked_clone=node_data.get("linked_clone", True),
+        linked_clone=data.get("linked_clone", True),
-        qemu_path=node_data.pop("qemu_path", None),
+        qemu_path=data.pop("qemu_path", None),
-        console=node_data.pop("console", None),
+        console=data.pop("console", None),
-        console_type=node_data.pop("console_type", "telnet"),
+        console_type=data.pop("console_type", "telnet"),
-        aux=node_data.get("aux"),
+        aux=data.get("aux"),
-        aux_type=node_data.pop("aux_type", "none"),
+        aux_type=data.pop("aux_type", "none"),
-        platform=node_data.pop("platform", None),
+        platform=data.pop("platform", None),
     )
 
     # update the disk image with the backing file if provided
     # this is needed when duplicating a node that uses backed disk images
     drives = ["a", "b", "c", "d"]
     for drive in drives:
-        disk_image_backing_file = node_data.get(f"hd{drive}_disk_image_backing_file")
+        disk_image_backing_file = data.get(f"hd{drive}_disk_image_backing_file")
         if disk_image_backing_file:
             log.debug(f"Updating disk image for drive {drive} with backing file {disk_image_backing_file}")
-            node_data[f"hd{drive}_disk_image"] = disk_image_backing_file
+            data[f"hd{drive}_disk_image"] = disk_image_backing_file
 
     # Validate every explicitly replaced disk before removing its stale
     # overlay. Other unresolved disks may still make this create request fail,
@@ -94,7 +96,7 @@ async def create_qemu_node(project_id: UUID, node_data: schemas.QemuCreate) -> s
     for drive in drives:
         disk_image_property = f"hd{drive}_disk_image"
         if disk_image_property in disk_images_to_reset:
-            replacement_image = node_data.get(disk_image_property)
+            replacement_image = data.get(disk_image_property)
             if replacement_image:
                 vm.manager.get_abs_image_path(replacement_image, vm.working_dir)
             local_disk_name = f"hd{drive}_disk.qcow2"
@@ -107,11 +109,11 @@ async def create_qemu_node(project_id: UUID, node_data: schemas.QemuCreate) -> s
                 log.info(
                     "Removing stale linked-clone disk '%s' before using replacement image '%s'",
                     local_disk,
-                    node_data.get(disk_image_property),
+                    data.get(disk_image_property),
                 )
                 vm.delete_disk_image(local_disk_name)
 
-    for name, value in node_data.items():
+    for name, value in data.items():
         if hasattr(vm, name) and getattr(vm, name) != value:
             setattr(vm, name, value)
 
@@ -133,10 +135,10 @@ async def update_qemu_node(node_data: schemas.QemuUpdate, node: QemuVM = Depends
     Update a Qemu node.
     """
 
-    node_data = jsonable_encoder(node_data, exclude_unset=True)
+    data = jsonable_encoder(node_data, exclude_unset=True)
     # update the console first to avoid issue if updating console type
-    node.console = node_data.pop("console", node.console)
+    node.console = data.pop("console", node.console)
-    for name, value in node_data.items():
+    for name, value in data.items():
         if hasattr(node, name) and getattr(node, name) != value:
             await node.update_property(name, value)
     node.updated()

gns3server/api/routes/controller/links.py

@@ -26,7 +26,7 @@ import aiohttp
 from fastapi import APIRouter, Depends, Request, status, WebSocket
 from fastapi.responses import FileResponse, StreamingResponse
 from fastapi.encoders import jsonable_encoder
-from typing import List, Union
+from typing import Any, List, Union
 from uuid import UUID, uuid4
 
 from gns3server.controller import Controller
@@ -46,7 +46,9 @@ import logging
 
 log = logging.getLogger(__name__)
 
-responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or link"}}
+responses: dict[int | str, dict[str, Any]] = {
+    404: {"model": schemas.ErrorMessage, "description": "Could not find project or link"}
+}
 
 router = APIRouter(responses=responses)
 
@@ -91,7 +93,7 @@ async def get_links(project_id: UUID) -> List[schemas.Link]:
     },
     dependencies=[Depends(has_privilege("Link.Allocate"))],
 )
-async def create_link(project_id: UUID, link_data: schemas.LinkCreate) -> schemas.Link:
+async def create_link(project_id: UUID, link_create: schemas.LinkCreate) -> schemas.Link:
     """
     Create a new link.
 
@@ -100,7 +102,7 @@ async def create_link(project_id: UUID, link_data: schemas.LinkCreate) -> schema
 
     project = await Controller.instance().get_loaded_project(str(project_id))
     link = await project.add_link()
-    link_data = jsonable_encoder(link_data, exclude_unset=True)
+    link_data = jsonable_encoder(link_create, exclude_unset=True)
     if "filters" in link_data:
         await link.update_filters(link_data["filters"])
     if "link_style" in link_data:
@@ -156,14 +158,14 @@ async def get_link(link: Link = Depends(dep_link)) -> schemas.Link:
     response_model_exclude_unset=True,
     dependencies=[Depends(has_privilege("Link.Modify"))],
 )
-async def update_link(link_data: schemas.LinkUpdate, link: Link = Depends(dep_link)) -> schemas.Link:
+async def update_link(link_update: schemas.LinkUpdate, link: Link = Depends(dep_link)) -> schemas.Link:
     """
     Update a link.
 
     Required privilege: Link.Modify
     """
 
-    link_data = jsonable_encoder(link_data, exclude_unset=True)
+    link_data = jsonable_encoder(link_update, exclude_unset=True)
     if "filters" in link_data:
         await link.update_filters(link_data["filters"])
     if "link_style" in link_data:
@@ -293,7 +295,7 @@ async def stream_pcap(request: Request, link: Link = Depends(dep_link)) -> Strea
     pcap_streaming_url = link.pcap_streaming_url()
     headers = multidict.MultiDict(request.headers)
     headers["Host"] = compute.host
-    headers["Router-Host"] = request.client.host
+    headers["Router-Host"] = request.client.host if request.client else ""
     body = await request.body()
 
     async def compute_pcap_stream():
@@ -487,7 +489,7 @@ async def update_marker(marker_name: str, marker_data: schemas.MarkerUpdate, lin
     response_model=Union[schemas.UDPPortInfo, schemas.EthernetPortInfo],
     dependencies=[Depends(has_privilege("Link.Audit"))],
 )
-async def get_iface(link: Link = Depends(dep_link)) -> Union[schemas.UDPPortInfo, schemas.EthernetPortInfo]:
+async def get_iface(link: Link = Depends(dep_link)) -> dict:
     """
     Return iface info for links to Cloud or NAT devices.
 

gns3server/compute/iou/iou_vm.py

@@ -139,8 +139,8 @@ class IOUVM(BaseNode):
     # Class-level caches shared across all IOU VM instances using the same image.
     # These avoid redundant subprocess calls during project loading when multiple
     # IOU nodes use the same image.
-    _loader_cache = {}  # image path -> loader command list
+    _loader_cache: dict[str, list[str]] = {}  # image path -> loader command list
-    _default_values_cache = {}  # image path -> (ram, nvram)
+    _default_values_cache: dict[str, tuple[int, int]] = {}  # image path -> (ram, nvram)
 
     """
     IOU VM implementation.
@@ -464,7 +464,11 @@ class IOUVM(BaseNode):
         )
         self._nvram = nvram
 
-    @BaseNode.name.setter
+    @property
+    def name(self):
+        return self._name
+
+    @name.setter
     def name(self, new_name):
         """
         Sets the name of this IOU VM.
@@ -949,7 +953,11 @@ class IOUVM(BaseNode):
             return True
         return False
 
-    @BaseNode.console_type.setter
+    @property
+    def console_type(self):
+        return self._console_type
+
+    @console_type.setter
     def console_type(self, new_console_type):
         """
         Sets the console type for this IOU VM.

gns3server/controller/link.py

@@ -365,7 +365,21 @@ class Link:
 
         raise NotImplementedError
 
-    async def start_marker(self, name, bpf, tag=None, direction=None, capture_node_id=None, enabled=True):
+    async def start_marker(
+        self,
+        name,
+        bpf,
+        tag=None,
+        direction=None,
+        data_link_type="DLT_EN10MB",
+        capture_node_id=None,
+        color=None,
+        highlight_duration=None,
+        enabled=True,
+        inherited_from=None,
+        dump=True,
+        memory_only=False,
+    ):
         """
         Attach a traffic-insight marker to this link (base — UDPLink overrides).
         """
@@ -377,7 +391,19 @@ class Link:
         """
         raise NotImplementedError
 
-    async def update_marker(self, name, bpf=None, tag=None, enabled=None, direction=_UNSET):
+    async def update_marker(
+        self,
+        name,
+        bpf=None,
+        tag=None,
+        enabled=None,
+        direction=_UNSET,
+        color=None,
+        highlight_duration=None,
+        inherited=False,
+        dump=True,
+        memory_only=False,
+    ):
         """
         Update an existing marker's BPF, tag, or enabled flag.
 

gns3server/controller/topology.py

@@ -53,7 +53,7 @@ GNS3_FILE_FORMAT_REVISION = 10
 
 
 class DynamipsNodeValidation(DynamipsCreate):
-    name: Optional[str] = None
+    name: Optional[str] = None  # type: ignore[assignment]
 
 
 def _check_topology_schema(topo, path):

gns3server/db/models/acl.py

@@ -29,7 +29,7 @@ class ACE(BaseTable):
     __tablename__ = "acl"
 
     ace_id = Column(GUID, primary_key=True, default=generate_uuid)
-    ace_type: str = Column(String)
+    ace_type = Column(String)
     path = Column(String)
     propagate = Column(Boolean, default=True)
     allowed = Column(Boolean, default=True)

gns3server/db/models/base.py

@@ -21,11 +21,10 @@ from fastapi.encoders import jsonable_encoder
 from sqlalchemy import Column, DateTime, func, inspect
 from sqlalchemy.types import TypeDecorator, CHAR, VARCHAR
 from sqlalchemy.dialects.postgresql import UUID
-from sqlalchemy.orm import as_declarative
+from sqlalchemy.orm import DeclarativeBase
 
 
-@as_declarative()
+class Base(DeclarativeBase):
-class Base:
     def asdict(self):
 
         return {c.key: getattr(self, c.key) for c in inspect(self).mapper.column_attrs}

gns3server/schemas/controller/templates/dynamips_templates.py

@@ -30,12 +30,12 @@ from typing import Optional
 from enum import Enum
 
 
-class DynamipsTemplate(TemplateBase):
+class DynamipsTemplateBase(TemplateBase):
     category: Optional[Category] = Category.router
     default_name_format: Optional[str] = "R{0}"
     symbol: Optional[str] = "router"
-    platform: DynamipsPlatform = Field(..., description="Cisco router platform")
+    platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform")
-    image: str = Field(..., description="Path to the IOS image")
+    image: Optional[str] = Field(None, description="Path to the IOS image")
     exec_area: Optional[int] = Field(64, ge=0, description="Exec area value")
     mmap: Optional[bool] = Field(True, description="MMAP feature")
     mac_addr: Optional[str] = Field(
@@ -67,7 +67,12 @@ class DynamipsTemplate(TemplateBase):
     wic2: Optional[DynamipsWics] = Field(None, description="Network module WIC slot 2")
 
 
-class C7200DynamipsTemplate(DynamipsTemplate):
+class DynamipsTemplate(DynamipsTemplateBase):
+    platform: DynamipsPlatform = Field(..., description="Cisco router platform")
+    image: str = Field(..., description="Path to the IOS image")
+
+
+class C7200DynamipsTemplateUpdate(DynamipsTemplateBase):
     ram: Optional[int] = Field(512, gt=0, description="Amount of RAM in MB")
     nvram: Optional[int] = Field(512, gt=0, description="Amount of NVRAM in KB")
     npe: Optional[DynamipsNPE] = Field(DynamipsNPE.npe_400, description="NPE model")
@@ -75,33 +80,33 @@ class C7200DynamipsTemplate(DynamipsTemplate):
     sparsemem: Optional[bool] = Field(True, description="Sparse memory feature")
 
 
-class C7200DynamipsTemplateUpdate(C7200DynamipsTemplate):
+class C7200DynamipsTemplate(C7200DynamipsTemplateUpdate):
-    platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform")
+    platform: DynamipsPlatform = Field(..., description="Cisco router platform")
-    image: Optional[str] = Field(None, description="Path to the IOS image")
+    image: str = Field(..., description="Path to the IOS image")
 
 
-class C3725DynamipsTemplate(DynamipsTemplate):
+class C3725DynamipsTemplateUpdate(DynamipsTemplateBase):
     ram: Optional[int] = Field(128, gt=0, description="Amount of RAM in MB")
     nvram: Optional[int] = Field(256, gt=0, description="Amount of NVRAM in KB")
     iomem: Optional[int] = Field(5, ge=0, le=100, description="I/O memory percentage")
     sparsemem: Optional[bool] = Field(True, description="Sparse memory feature")
 
 
-class C3725DynamipsTemplateUpdate(C3725DynamipsTemplate):
+class C3725DynamipsTemplate(C3725DynamipsTemplateUpdate):
-    platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform")
+    platform: DynamipsPlatform = Field(..., description="Cisco router platform")
-    image: Optional[str] = Field(None, description="Path to the IOS image")
+    image: str = Field(..., description="Path to the IOS image")
 
 
-class C3745DynamipsTemplate(DynamipsTemplate):
+class C3745DynamipsTemplateUpdate(DynamipsTemplateBase):
     ram: Optional[int] = Field(256, gt=0, description="Amount of RAM in MB")
     nvram: Optional[int] = Field(256, gt=0, description="Amount of NVRAM in KB")
     iomem: Optional[int] = Field(5, ge=0, le=100, description="I/O memory percentage")
     sparsemem: Optional[bool] = Field(True, description="Sparse memory feature")
 
 
-class C3745DynamipsTemplateUpdate(C3745DynamipsTemplate):
+class C3745DynamipsTemplate(C3745DynamipsTemplateUpdate):
-    platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform")
+    platform: DynamipsPlatform = Field(..., description="Cisco router platform")
-    image: Optional[str] = Field(None, description="Path to the IOS image")
+    image: str = Field(..., description="Path to the IOS image")
 
 
 class C3600ChassisType(str, Enum):
@@ -110,29 +115,29 @@ class C3600ChassisType(str, Enum):
     chassis_3660 = "3660"
 
 
-class C3600DynamipsTemplate(DynamipsTemplate):
+class C3600DynamipsTemplateUpdate(DynamipsTemplateBase):
-    chassis: Optional[C3600ChassisType] = Field("c3660", description="Chassis type")
+    chassis: Optional[C3600ChassisType] = Field(C3600ChassisType.chassis_3660, description="Chassis type")
     ram: Optional[int] = Field(192, gt=0, description="Amount of RAM in MB")
     nvram: Optional[int] = Field(128, gt=0, description="Amount of NVRAM in KB")
     iomem: Optional[int] = Field(5, ge=0, le=100, description="I/O memory percentage")
     sparsemem: Optional[bool] = Field(True, description="Sparse memory feature")
 
 
-class C3600DynamipsTemplateUpdate(C3600DynamipsTemplate):
+class C3600DynamipsTemplate(C3600DynamipsTemplateUpdate):
-    platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform")
+    platform: DynamipsPlatform = Field(..., description="Cisco router platform")
-    image: Optional[str] = Field(None, description="Path to the IOS image")
+    image: str = Field(..., description="Path to the IOS image")
 
 
-class C2691DynamipsTemplate(DynamipsTemplate):
+class C2691DynamipsTemplateUpdate(DynamipsTemplateBase):
     ram: Optional[int] = Field(192, gt=0, description="Amount of RAM in MB")
     nvram: Optional[int] = Field(256, gt=0, description="Amount of NVRAM in KB")
     iomem: Optional[int] = Field(5, ge=0, le=100, description="I/O memory percentage")
     sparsemem: Optional[bool] = Field(True, description="Sparse memory feature")
 
 
-class C2691DynamipsTemplateUpdate(C2691DynamipsTemplate):
+class C2691DynamipsTemplate(C2691DynamipsTemplateUpdate):
-    platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform")
+    platform: DynamipsPlatform = Field(..., description="Cisco router platform")
-    image: Optional[str] = Field(None, description="Path to the IOS image")
+    image: str = Field(..., description="Path to the IOS image")
 
 
 class C2600ChassisType(str, Enum):
@@ -147,17 +152,17 @@ class C2600ChassisType(str, Enum):
     chassis_2651xm = "2651XM"
 
 
-class C2600DynamipsTemplate(DynamipsTemplate):
+class C2600DynamipsTemplateUpdate(DynamipsTemplateBase):
-    chassis: Optional[C2600ChassisType] = Field("2651XM", description="Chassis type")
+    chassis: Optional[C2600ChassisType] = Field(C2600ChassisType.chassis_2651xm, description="Chassis type")
     ram: Optional[int] = Field(160, gt=0, description="Amount of RAM in MB")
     nvram: Optional[int] = Field(128, gt=0, description="Amount of NVRAM in KB")
     iomem: Optional[int] = Field(15, ge=0, le=100, description="I/O memory percentage")
     sparsemem: Optional[bool] = Field(True, description="Sparse memory feature")
 
 
-class C2600DynamipsTemplateUpdate(C2600DynamipsTemplate):
+class C2600DynamipsTemplate(C2600DynamipsTemplateUpdate):
-    platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform")
+    platform: DynamipsPlatform = Field(..., description="Cisco router platform")
-    image: Optional[str] = Field(None, description="Path to the IOS image")
+    image: str = Field(..., description="Path to the IOS image")
 
 
 class C1700ChassisType(str, Enum):
@@ -168,14 +173,14 @@ class C1700ChassisType(str, Enum):
     chassis_1760 = "1760"
 
 
-class C1700DynamipsTemplate(DynamipsTemplate):
+class C1700DynamipsTemplateUpdate(DynamipsTemplateBase):
-    chassis: Optional[C1700ChassisType] = Field("1760", description="Chassis type")
+    chassis: Optional[C1700ChassisType] = Field(C1700ChassisType.chassis_1760, description="Chassis type")
     ram: Optional[int] = Field(160, gt=0, description="Amount of RAM in MB")
     nvram: Optional[int] = Field(128, gt=0, description="Amount of NVRAM in KB")
     iomem: Optional[int] = Field(15, ge=0, le=100, description="I/O memory percentage")
     sparsemem: Optional[bool] = Field(False, description="Sparse memory feature")
 
 
-class C1700DynamipsTemplateUpdate(C1700DynamipsTemplate):
+class C1700DynamipsTemplate(C1700DynamipsTemplateUpdate):
-    platform: Optional[DynamipsPlatform] = Field(None, description="Cisco router platform")
+    platform: DynamipsPlatform = Field(..., description="Cisco router platform")
-    image: Optional[str] = Field(None, description="Path to the IOS image")
+    image: str = Field(..., description="Path to the IOS image")

gns3server/services/authentication.py

@@ -73,21 +73,23 @@ class AuthService:
         encoded_jwt = jwt.encode({"alg": algorithm}, to_encode, key)
         return encoded_jwt
 
-    def create_access_token(self, username, token_version: int = 0, secret_key: str = None, expires_in: int = 0) -> str:
+    def create_access_token(
+        self, username, token_version: int = 0, secret_key: Optional[str] = None, expires_in: int = 0
+    ) -> str:
 
         if not expires_in:
             expires_in = Config.instance().settings.Controller.jwt_access_token_expire_minutes
         return self._create_token(username, token_version, "access", expires_in, secret_key)
 
     def create_refresh_token(
-        self, username, token_version: int = 0, secret_key: str = None, expires_in: int = 0
+        self, username, token_version: int = 0, secret_key: Optional[str] = None, expires_in: int = 0
     ) -> str:
 
         if not expires_in:
             expires_in = Config.instance().settings.Controller.jwt_refresh_token_expire_minutes
         return self._create_token(username, token_version, "refresh", expires_in, secret_key)
 
-    def get_token_data(self, token: str, secret_key: str = None) -> TokenData:
+    def get_token_data(self, token: str, secret_key: Optional[str] = None) -> TokenData:
 
         def auth_error(detail: str) -> HTTPException:
             return HTTPException(
@@ -105,7 +107,7 @@ class AuthService:
         key = OctKey.import_key(secret_key)
         try:
             payload = jwt.decode(token, key, algorithms=[algorithm])
-            username: str = payload.claims.get("sub")
+            username: Optional[str] = payload.claims.get("sub")
             if username is None:
                 raise auth_error("Invalid token: missing subject claim")
             # Validate the exp claim — joserfc does not validate time-based claims by default
@@ -123,5 +125,5 @@ class AuthService:
             raise auth_error(f"Invalid token ({type(e).__name__})")
         return token_data
 
-    def get_username_from_token(self, token: str, secret_key: str = None) -> Optional[str]:
+    def get_username_from_token(self, token: str, secret_key: Optional[str] = None) -> Optional[str]:
         return self.get_token_data(token, secret_key).username

gns3server/utils/zipfile_zstd/__init__.py

@@ -2,7 +2,7 @@
 from . import _zipfile
 
 from zipfile import *
-from zipfile import (
+from ._zipfile import (
     ZIP_ZSTANDARD,
     ZSTANDARD_VERSION,
 )

gns3server/utils/zipfile_zstd/_zipfile.py

@@ -1,13 +1,18 @@
-import zipfile
+import zipfile as _zipfile_module
 import zstandard as zstd
 import inspect
+from typing import Any
 
 from ._patcher import patch
 
+zipfile: Any = _zipfile_module
 
-zipfile.ZIP_ZSTANDARD = 93
+ZIP_ZSTANDARD = 93
+ZSTANDARD_VERSION = 20
+
+zipfile.ZIP_ZSTANDARD = ZIP_ZSTANDARD
 zipfile.compressor_names[zipfile.ZIP_ZSTANDARD] = "zstandard"
-zipfile.ZSTANDARD_VERSION = 20
+zipfile.ZSTANDARD_VERSION = ZSTANDARD_VERSION
 
 
 @patch(zipfile, "_check_compression")

pyproject.toml

@@ -295,15 +295,8 @@ enable_error_code = ["ignore-without-code", "redundant-expr", "truthy-bool"]
 # Remove modules from this list one small PR at a time. Never add new ones.
 module = [
   "gns3server.agent",  # 2
-  "gns3server.agent.gns3_copilot.agent.context_manager",  # 5
-  "gns3server.agent.gns3_copilot.agent.model_factory",  # 4
-  "gns3server.agent.gns3_copilot.agent_service",  # 8
   "gns3server.agent.gns3_copilot.chat_sessions_repository",  # 8
   "gns3server.agent.gns3_copilot.gns3_client.api_handlers",  # 10
-  "gns3server.agent.gns3_copilot.gns3_client.connector_factory",  # 1
-  "gns3server.agent.gns3_copilot.gns3_client.project_inventory",  # 2
-  "gns3server.agent.gns3_copilot.skills.loader",  # 1
-  "gns3server.agent.gns3_copilot.skills.manager",  # 4
   "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
@@ -312,12 +305,8 @@ module = [
   "gns3server.agent.gns3_copilot.utils.custom_netmiko.scripts.list_netmiko_telnet_devices",  # 1
   "gns3server.agent.gns3_copilot.utils.message_converters",  # 2
   "gns3server.agent.gns3_copilot.utils.parse_tool_content",  # 1
-  "gns3server.agent.mcp",  # 34
   "gns3server.agent.mcp.appliances",  # 1
   "gns3server.agent.web_wireshark.docker_client",  # 5
-  "gns3server.agent.web_wireshark.manage_wireshark",  # 1
-  "gns3server.agent.web_wireshark.manager",  # 9
-  "gns3server.agent.web_wireshark.stats",  # 4
   "gns3server.api.routes.compute.atm_switch_nodes",  # 9
   "gns3server.api.routes.compute.capabilities",  # 1
   "gns3server.api.routes.compute.cloud_nodes",  # 14
@@ -332,7 +321,6 @@ module = [
   "gns3server.api.routes.compute.nat_nodes",  # 10
   "gns3server.api.routes.compute.notifications",  # 4
   "gns3server.api.routes.compute.projects",  # 7
-  "gns3server.api.routes.compute.qemu_nodes",  # 18
   "gns3server.api.routes.compute.virtualbox_nodes",  # 20
   "gns3server.api.routes.compute.vmware_nodes",  # 10
   "gns3server.api.routes.compute.vpcs_nodes",  # 9
@@ -346,7 +334,6 @@ module = [
   "gns3server.api.routes.controller.drawings",  # 1
   "gns3server.api.routes.controller.groups",  # 6
   "gns3server.api.routes.controller.images",  # 9
-  "gns3server.api.routes.controller.links",  # 27
   "gns3server.api.routes.controller.llm_model_configs",  # 32
   "gns3server.api.routes.controller.nodes",  # 13
   "gns3server.api.routes.controller.pools",  # 8
@@ -358,7 +345,6 @@ module = [
   "gns3server.api.routes.controller.users",  # 14
   "gns3server.api.server",  # 2
   "gns3server.compute.dynamips.nodes.router",  # 2
-  "gns3server.compute.iou.iou_vm",  # 4
   "gns3server.compute.iou.utils.iou_export",  # 2
   "gns3server.compute.iou.utils.iou_import",  # 2
   "gns3server.compute.virtualbox.virtualbox_vm",  # 1
@@ -366,14 +352,6 @@ module = [
   "gns3server.controller.appliance",  # 2
   "gns3server.controller.appliance_manager",  # 5
   "gns3server.controller.project",  # 2
-  "gns3server.controller.topology",  # 1
-  "gns3server.db.models.acl",  # 1
-  "gns3server.db.models.images",  # 1
-  "gns3server.db.models.pools",  # 1
-  "gns3server.db.models.privileges",  # 2
-  "gns3server.db.models.roles",  # 1
-  "gns3server.db.models.templates",  # 22
-  "gns3server.db.models.users",  # 3
   "gns3server.db.repositories.api_keys",  # 8
   "gns3server.db.repositories.computes",  # 10
   "gns3server.db.repositories.images",  # 14
@@ -383,7 +361,6 @@ module = [
   "gns3server.db.repositories.templates",  # 6
   "gns3server.db.repositories.users",  # 12
   "gns3server.db.tasks",  # 12
-  "gns3server.db_migrations.env",  # 1
   "gns3server.schemas.compute.atm_switch_nodes",  # 5
   "gns3server.schemas.compute.cloud_nodes",  # 1
   "gns3server.schemas.compute.docker_nodes",  # 2
@@ -401,12 +378,10 @@ module = [
   "gns3server.schemas.controller.projects",  # 2
   "gns3server.schemas.controller.settings",  # 3
   "gns3server.schemas.controller.templates.docker_templates",  # 1
-  "gns3server.schemas.controller.templates.dynamips_templates",  # 17
   "gns3server.schemas.controller.templates.iou_templates",  # 1
   "gns3server.schemas.controller.templates.virtualbox_templates",  # 1
   "gns3server.schemas.controller.templates.vmware_templates",  # 1
   "gns3server.schemas.qemu_disk_image",  # 2
-  "gns3server.services.authentication",  # 5
   "gns3server.services.computes",  # 4
   "gns3server.services.templates",  # 18
   "gns3server.utils.asyncio.aiozipstream",  # 8
@@ -415,8 +390,6 @@ module = [
   "gns3server.utils.images",  # 4
   "gns3server.utils.packet_filter_validation",  # 16
   "gns3server.utils.websocket_to_websocket",  # 12
-  "gns3server.utils.zipfile_zstd",  # 2
   "gns3server.utils.zipfile_zstd._patcher",  # 1
-  "gns3server.utils.zipfile_zstd._zipfile",  # 5
 ]
 ignore_errors = true

tests/agent/mcp/test_tool_params.py

@@ -193,6 +193,9 @@ def _initial_params_keys(fn_node):
                 and isinstance(stmt.value, ast.Dict)
             ):
                 return _dict_literal_keys(stmt.value)
+        if isinstance(stmt, ast.AnnAssign):
+            if isinstance(stmt.target, ast.Name) and stmt.target.id == "params" and isinstance(stmt.value, ast.Dict):
+                return _dict_literal_keys(stmt.value)
     return None