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

This commit is contained in:
Mark Paronyan 2026-09-28 06:45:50 +00:00
parent 8df25dec32
commit 6ffa39642f
No known key found for this signature in database
GPG Key ID: 8DC7338A87FE22F8
3 changed files with 31 additions and 27 deletions

View File

@ -405,22 +405,24 @@ async def project_stats(
@mcp.tool() @mcp.tool()
async def project_update( async def project_update(
project_id: Annotated[str, Field(description="UUID of the project to 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]]: ) -> list[dict[str, Any]]:
"""Update a project's properties (name, auto_close, auto_open, etc.).""" """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 = { local_vars = {
"name": name, "name": name,
"auto_close": auto_close, "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, reset_mac_addresses: Annotated[bool, Field(description="Reset MAC addresses for this project")] = False,
) -> list[dict[str, Any]]: ) -> list[dict[str, Any]]:
"""Duplicate a project.""" """Duplicate a project."""
params = {"project_id": project_id, "name": name} params: dict[str, Any] = {"project_id": project_id, "name": name}
if reset_mac_addresses: if reset_mac_addresses:
params["reset_mac_addresses"] = reset_mac_addresses params["reset_mac_addresses"] = reset_mac_addresses
return await asyncio.to_thread(_run_handler_sync, duplicate_project_handler, params) return await asyncio.to_thread(_run_handler_sync, duplicate_project_handler, params)
@ -523,7 +525,7 @@ async def node_start(
] = None, ] = None,
) -> list[dict[str, Any]]: ) -> list[dict[str, Any]]:
"""Start one or more nodes. Provide node_id for single, or node_ids for batch.""" """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: if node_ids:
params["node_ids"] = node_ids params["node_ids"] = node_ids
else: else:
@ -540,7 +542,7 @@ async def node_stop(
] = None, ] = None,
) -> list[dict[str, Any]]: ) -> list[dict[str, Any]]:
"""Stop one or more nodes. Provide node_id for single, or node_ids for batch.""" """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: if node_ids:
params["node_ids"] = node_ids params["node_ids"] = node_ids
else: else:
@ -557,7 +559,7 @@ async def node_suspend(
] = None, ] = None,
) -> list[dict[str, Any]]: ) -> list[dict[str, Any]]:
"""Suspend one or more nodes. Provide node_id for single, or node_ids for batch.""" """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: if node_ids:
params["node_ids"] = node_ids params["node_ids"] = node_ids
else: else:
@ -635,7 +637,7 @@ async def node_delete(
] = None, ] = None,
) -> list[dict[str, Any]]: ) -> list[dict[str, Any]]:
"""Delete one or more nodes from a project. Provide node_id for single, or node_ids for batch.""" """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: if node_ids:
params["node_ids"] = node_ids params["node_ids"] = node_ids
else: else:
@ -756,7 +758,7 @@ async def link_create(
"fields": fields, "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: if filters:
params["filters"] = filters params["filters"] = filters
return await asyncio.to_thread(_run_handler_sync, create_link_handler, params) return await asyncio.to_thread(_run_handler_sync, create_link_handler, params)
@ -771,7 +773,7 @@ async def link_delete(
] = None, ] = None,
) -> list[dict[str, Any]]: ) -> list[dict[str, Any]]:
"""Delete one or more links from a project.""" """Delete one or more links from a project."""
params = {"project_id": project_id} params: dict[str, Any] = {"project_id": project_id}
if link_ids: if link_ids:
params["link_ids"] = link_ids params["link_ids"] = link_ids
else: else:
@ -1201,7 +1203,7 @@ async def link_reset(
while keeping the filter configuration intact. Filters are preserved but while keeping the filter configuration intact. Filters are preserved but
their internal application state resets. their internal application state resets.
""" """
params = {"project_id": project_id} params: dict[str, Any] = {"project_id": project_id}
if link_ids: if link_ids:
params["link_ids"] = link_ids params["link_ids"] = link_ids
else: else:
@ -1222,7 +1224,7 @@ async def link_capture_start(
] = None, ] = None,
) -> list[dict[str, Any]]: ) -> list[dict[str, Any]]:
"""Start packet capture on one or more links.""" """Start packet capture on one or more links."""
params = { params: dict[str, Any] = {
"project_id": project_id, "project_id": project_id,
"data_link_type": data_link_type, "data_link_type": data_link_type,
"capture_file_name": capture_file_name, "capture_file_name": capture_file_name,
@ -1245,7 +1247,7 @@ async def link_capture_stop(
] = None, ] = None,
) -> list[dict[str, Any]]: ) -> list[dict[str, Any]]:
"""Stop packet capture on one or more links.""" """Stop packet capture on one or more links."""
params = {"project_id": project_id} params: dict[str, Any] = {"project_id": project_id}
if link_ids: if link_ids:
params["link_ids"] = link_ids params["link_ids"] = link_ids
else: else:
@ -1273,7 +1275,7 @@ async def link_capture_download(
- The ticket expires after 10 minutes: call this tool again to get a - The ticket expires after 10 minutes: call this tool again to get a
fresh one; do not reuse an old URL. fresh one; do not reuse an old URL.
""" """
params = {"project_id": project_id} params: dict[str, Any] = {"project_id": project_id}
if link_ids: if link_ids:
params["link_ids"] = link_ids params["link_ids"] = link_ids
else: else:
@ -1555,7 +1557,7 @@ async def drawing_update(
rotation: Annotated[int | None, Field(description="Rotation angle in degrees, -359 to 359")] = None, rotation: Annotated[int | None, Field(description="Rotation angle in degrees, -359 to 359")] = None,
) -> list[dict[str, Any]]: ) -> list[dict[str, Any]]:
"""Update a drawing's properties (svg, position, lock state, rotation, etc.).""" """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} local_vars = {"svg": svg, "locked": locked, "x": x, "y": y, "z": z, "rotation": rotation}
for key, val in local_vars.items(): for key, val in local_vars.items():
if val is not None: if val is not None:

View File

@ -315,7 +315,6 @@ module = [
"gns3server.agent.gns3_copilot.utils.custom_netmiko.scripts.list_netmiko_telnet_devices", # 1 "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.message_converters", # 2
"gns3server.agent.gns3_copilot.utils.parse_tool_content", # 1 "gns3server.agent.gns3_copilot.utils.parse_tool_content", # 1
"gns3server.agent.mcp", # 34
"gns3server.agent.mcp.appliances", # 1 "gns3server.agent.mcp.appliances", # 1
"gns3server.agent.web_wireshark.docker_client", # 5 "gns3server.agent.web_wireshark.docker_client", # 5
"gns3server.agent.web_wireshark.manage_wireshark", # 1 "gns3server.agent.web_wireshark.manage_wireshark", # 1

View File

@ -193,6 +193,9 @@ def _initial_params_keys(fn_node):
and isinstance(stmt.value, ast.Dict) and isinstance(stmt.value, ast.Dict)
): ):
return _dict_literal_keys(stmt.value) 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 return None