fix(typing): resolve mypy errors in agent node lifecycle tools

This commit is contained in:
Mark Paronyan committed 2026-09-29 09:46:32 +03:00
1 parent bc27c4814f
commit 780b7f2ab6
5 files changed
+18 -9

No files matched your search

@@ -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"}
@@ -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)
@@ -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)
@@ -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)