feat: Add snapshot and drawing MCP tools

- Add snapshot tools: get_snapshots, create_snapshot, delete_snapshot, restore_snapshot
- Add drawing tools: get_drawings, create_drawing, get_drawing, update_drawing, delete_drawing
This commit is contained in:
YueGuobin 2026-06-10 13:55:13 +08:00
parent acce79d243
commit e7fbb3ec10
No known key found for this signature in database
3 changed files with 300 additions and 0 deletions

View File

@ -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,
})
# ── Authwrapped SSE app ──────────────────────────────────────────────
def _make_auth_wrapper(inner_app):

View File

@ -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 <http://www.gnu.org/licenses/>.
"""
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}

View File

@ -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 <http://www.gnu.org/licenses/>.
"""
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}