diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index 51bdc2f30..dda88f907 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -72,6 +72,14 @@ from .templates import ( from .computes import ( list_computes_handler, get_compute_handler, get_compute_images_handler, ) +from .snapshots import ( + get_snapshots_handler, create_snapshot_handler, + delete_snapshot_handler, restore_snapshot_handler, +) +from .drawings import ( + get_drawings_handler, create_drawing_handler, + get_drawing_handler, update_drawing_handler, delete_drawing_handler, +) log = logging.getLogger(__name__) @@ -712,6 +720,123 @@ async def download_capture_file( }) +# ── Snapshot tools ───────────────────────────────────────────────────── + + +@mcp.tool() +async def get_snapshots( + project_id: Annotated[str, Field(description="UUID of the project")], +) -> list[dict[str, Any]]: + """List all snapshots of a project.""" + return await asyncio.to_thread(_run_handler_sync, get_snapshots_handler, { + "project_id": project_id, + }) + + +@mcp.tool() +async def create_snapshot( + project_id: Annotated[str, Field(description="UUID of the project")], + name: Annotated[str, Field(description="Name for the new snapshot")], +) -> list[dict[str, Any]]: + """Create a new snapshot of a project.""" + return await asyncio.to_thread(_run_handler_sync, create_snapshot_handler, { + "project_id": project_id, "name": name, + }) + + +@mcp.tool() +async def delete_snapshot( + project_id: Annotated[str, Field(description="UUID of the project")], + snapshot_id: Annotated[str, Field(description="UUID of the snapshot to delete")], +) -> list[dict[str, Any]]: + """Delete a snapshot from a project. Cannot be undone.""" + return await asyncio.to_thread(_run_handler_sync, delete_snapshot_handler, { + "project_id": project_id, "snapshot_id": snapshot_id, + }) + + +@mcp.tool() +async def restore_snapshot( + project_id: Annotated[str, Field(description="UUID of the project")], + snapshot_id: Annotated[str, Field(description="UUID of the snapshot to restore")], +) -> list[dict[str, Any]]: + """Restore a project to a previous snapshot state. The project may be closed and reopened.""" + return await asyncio.to_thread(_run_handler_sync, restore_snapshot_handler, { + "project_id": project_id, "snapshot_id": snapshot_id, + }) + + +# ── Drawing tools ────────────────────────────────────────────────────── + + +@mcp.tool() +async def get_drawings( + project_id: Annotated[str, Field(description="UUID of the project")], +) -> list[dict[str, Any]]: + """List all drawings (labels, shapes, images) on a project canvas.""" + return await asyncio.to_thread(_run_handler_sync, get_drawings_handler, { + "project_id": project_id, + }) + + +@mcp.tool() +async def create_drawing( + project_id: Annotated[str, Field(description="UUID of the project")], + svg: Annotated[str, Field(description="SVG content for the drawing")], + x: Annotated[int, Field(description="X coordinate (default: 0)")] = 0, + y: Annotated[int, Field(description="Y coordinate (default: 0)")] = 0, + z: Annotated[int, Field(description="Z layer (default: 0)")] = 0, + locked: Annotated[bool, Field(description="Lock the drawing (default: false)")] = False, + rotation: Annotated[int, Field(description="Rotation angle in degrees, -359 to 359 (default: 0)")] = 0, +) -> list[dict[str, Any]]: + """Create a new drawing (label, shape, or image) on a project canvas.""" + return await asyncio.to_thread(_run_handler_sync, create_drawing_handler, { + "project_id": project_id, "svg": svg, "x": x, "y": y, "z": z, + "locked": locked, "rotation": rotation, + }) + + +@mcp.tool() +async def get_drawing( + project_id: Annotated[str, Field(description="UUID of the project")], + drawing_id: Annotated[str, Field(description="UUID of the drawing")], +) -> list[dict[str, Any]]: + """Get detailed information about a specific drawing.""" + return await asyncio.to_thread(_run_handler_sync, get_drawing_handler, { + "project_id": project_id, "drawing_id": drawing_id, + }) + + +@mcp.tool() +async def update_drawing( + project_id: Annotated[str, Field(description="UUID of the project")], + drawing_id: Annotated[str, Field(description="UUID of the drawing")], + svg: Annotated[str | None, Field(description="New SVG content")] = None, + locked: Annotated[bool | None, Field(description="Lock or unlock the drawing")] = None, + x: Annotated[int | None, Field(description="New X coordinate")] = None, + y: Annotated[int | None, Field(description="New Y coordinate")] = None, + z: Annotated[int | None, Field(description="New Z layer")] = None, +) -> list[dict[str, Any]]: + """Update a drawing's properties (svg, position, lock state, etc.).""" + params = {"project_id": project_id, "drawing_id": drawing_id} + local_vars = {"svg": svg, "locked": locked, "x": x, "y": y, "z": z} + for key, val in local_vars.items(): + if val is not None: + params[key] = val + return await asyncio.to_thread(_run_handler_sync, update_drawing_handler, params) + + +@mcp.tool() +async def delete_drawing( + project_id: Annotated[str, Field(description="UUID of the project")], + drawing_id: Annotated[str, Field(description="UUID of the drawing to delete")], +) -> list[dict[str, Any]]: + """Delete a drawing from a project canvas. Cannot be undone.""" + return await asyncio.to_thread(_run_handler_sync, delete_drawing_handler, { + "project_id": project_id, "drawing_id": drawing_id, + }) + + # ── Auth‑wrapped SSE app ────────────────────────────────────────────── def _make_auth_wrapper(inner_app): diff --git a/gns3server/api/routes/mcp/drawings.py b/gns3server/api/routes/mcp/drawings.py new file mode 100644 index 000000000..6e47a2e02 --- /dev/null +++ b/gns3server/api/routes/mcp/drawings.py @@ -0,0 +1,96 @@ +# +# Copyright (C) 2026 GNS3 Technologies Inc. +# Author: Yue Guobin +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +""" +MCP tool handlers for GNS3 drawing management. +""" + +from typing import Any + +import logging + +log = logging.getLogger(__name__) + + +# ── Helper ───────────────────────────────────────────────────────────────── + +def _get_connector(gns3_ctx: dict[str, Any]): + from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector + return Gns3Connector( + url=gns3_ctx["server_url"], + jwt_token=gns3_ctx["jwt_token"], + api_version=3, + verify=False, + ) + + +# ── Tool handlers ────────────────────────────────────────────────────────── + +def get_drawings_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + if not project_id: + return {"error": "project_id is required"} + conn = _get_connector(gns3_ctx) + drawings = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/drawings").json() + return {"drawings": drawings, "count": len(drawings)} + + +def create_drawing_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + svg = params.get("svg") + if not project_id or not svg: + return {"error": "project_id and svg are required"} + conn = _get_connector(gns3_ctx) + data = { + "svg": svg, + "x": params.get("x", 0), + "y": params.get("y", 0), + "z": params.get("z", 0), + "locked": params.get("locked", False), + "rotation": params.get("rotation", 0), + } + result = conn.http_call("post", f"{conn.base_url}/projects/{project_id}/drawings", json_data=data).json() + return {"message": "Drawing created", "drawing": result} + + +def get_drawing_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + drawing_id = params.get("drawing_id") + if not project_id or not drawing_id: + return {"error": "project_id and drawing_id are required"} + conn = _get_connector(gns3_ctx) + return conn.http_call("get", f"{conn.base_url}/projects/{project_id}/drawings/{drawing_id}").json() + + +def update_drawing_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + drawing_id = params.get("drawing_id") + if not project_id or not drawing_id: + return {"error": "project_id and drawing_id are required"} + conn = _get_connector(gns3_ctx) + data = {k: v for k, v in params.items() if k not in ("project_id", "drawing_id") and v is not None} + return conn.http_call("put", f"{conn.base_url}/projects/{project_id}/drawings/{drawing_id}", json_data=data).json() + + +def delete_drawing_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + drawing_id = params.get("drawing_id") + if not project_id or not drawing_id: + return {"error": "project_id and drawing_id are required"} + conn = _get_connector(gns3_ctx) + conn.http_call("delete", f"{conn.base_url}/projects/{project_id}/drawings/{drawing_id}") + return {"message": f"Drawing {drawing_id} deleted", "drawing_id": drawing_id} diff --git a/gns3server/api/routes/mcp/snapshots.py b/gns3server/api/routes/mcp/snapshots.py new file mode 100644 index 000000000..e5b1685db --- /dev/null +++ b/gns3server/api/routes/mcp/snapshots.py @@ -0,0 +1,79 @@ +# +# Copyright (C) 2026 GNS3 Technologies Inc. +# Author: Yue Guobin +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +""" +MCP tool handlers for GNS3 snapshot management. +""" + +from typing import Any + +import logging + +log = logging.getLogger(__name__) + + +# ── Helper ───────────────────────────────────────────────────────────────── + +def _get_connector(gns3_ctx: dict[str, Any]): + from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector + return Gns3Connector( + url=gns3_ctx["server_url"], + jwt_token=gns3_ctx["jwt_token"], + api_version=3, + verify=False, + ) + + +# ── Tool handlers ────────────────────────────────────────────────────────── + +def get_snapshots_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + if not project_id: + return {"error": "project_id is required"} + conn = _get_connector(gns3_ctx) + snapshots = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/snapshots").json() + return {"snapshots": snapshots, "count": len(snapshots)} + + +def create_snapshot_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + name = params.get("name") + if not project_id or not name: + return {"error": "project_id and name are required"} + conn = _get_connector(gns3_ctx) + result = conn.http_call("post", f"{conn.base_url}/projects/{project_id}/snapshots", json_data={"name": name}).json() + return {"message": f"Snapshot '{name}' created", "snapshot": result} + + +def delete_snapshot_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + snapshot_id = params.get("snapshot_id") + if not project_id or not snapshot_id: + return {"error": "project_id and snapshot_id are required"} + conn = _get_connector(gns3_ctx) + conn.http_call("delete", f"{conn.base_url}/projects/{project_id}/snapshots/{snapshot_id}") + return {"message": f"Snapshot {snapshot_id} deleted", "snapshot_id": snapshot_id} + + +def restore_snapshot_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + snapshot_id = params.get("snapshot_id") + if not project_id or not snapshot_id: + return {"error": "project_id and snapshot_id are required"} + conn = _get_connector(gns3_ctx) + result = conn.http_call("post", f"{conn.base_url}/projects/{project_id}/snapshots/{snapshot_id}/restore").json() + return {"message": f"Snapshot {snapshot_id} restored", "project": result}