From b786f0b7ebaa068f8e0e8ae82604f42167666fee Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Wed, 10 Jun 2026 13:58:42 +0800 Subject: [PATCH] feat: Add image management MCP tools - Image tools: get_images, get_image, delete_image, prune_images, install_images --- gns3server/api/routes/mcp/__init__.py | 46 ++++++++++++++++ gns3server/api/routes/mcp/images.py | 75 +++++++++++++++++++++++++++ 2 files changed, 121 insertions(+) create mode 100644 gns3server/api/routes/mcp/images.py diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index a24d10dac..907c0195c 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -63,6 +63,11 @@ from .appliances import ( get_appliances_handler, get_appliance_handler, install_appliance_handler, ) +from .images import ( + get_images_handler, get_image_handler, + delete_image_handler, prune_images_handler, + install_images_handler, +) from .nodes import ( get_nodes_handler, get_node_handler, start_node_handler, stop_node_handler, reload_node_handler, suspend_node_handler, @@ -1045,6 +1050,47 @@ async def install_appliance( }) +# ── Image tools ─────────────────────────────────────────────────────── + + +@mcp.tool() +async def get_images() -> list[dict[str, Any]]: + """List all images available on the server across all emulators.""" + return await asyncio.to_thread(_run_handler_sync, get_images_handler, {}) + + +@mcp.tool() +async def get_image( + image_id: Annotated[str, Field(description="ID or filename of the image")], +) -> list[dict[str, Any]]: + """Get detailed information about a specific image.""" + return await asyncio.to_thread(_run_handler_sync, get_image_handler, { + "image_id": image_id, + }) + + +@mcp.tool() +async def delete_image( + image_id: Annotated[str, Field(description="ID or filename of the image to delete")], +) -> list[dict[str, Any]]: + """Delete an image from the server. Cannot be undone.""" + return await asyncio.to_thread(_run_handler_sync, delete_image_handler, { + "image_id": image_id, + }) + + +@mcp.tool() +async def prune_images() -> list[dict[str, Any]]: + """Remove all unused images from the server to free up disk space.""" + return await asyncio.to_thread(_run_handler_sync, prune_images_handler, {}) + + +@mcp.tool() +async def install_images() -> list[dict[str, Any]]: + """Request the server to install pending images (download from registry).""" + return await asyncio.to_thread(_run_handler_sync, install_images_handler, {}) + + # ── Auth‑wrapped SSE app ────────────────────────────────────────────── def _make_auth_wrapper(inner_app): diff --git a/gns3server/api/routes/mcp/images.py b/gns3server/api/routes/mcp/images.py new file mode 100644 index 000000000..4b00c1a22 --- /dev/null +++ b/gns3server/api/routes/mcp/images.py @@ -0,0 +1,75 @@ +# +# 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 image 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_images_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + conn = _get_connector(gns3_ctx) + images = conn.http_call("get", f"{conn.base_url}/images").json() + return {"images": images, "count": len(images)} + + +def get_image_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + image_id = params.get("image_id") + if not image_id: + return {"error": "image_id is required"} + conn = _get_connector(gns3_ctx) + return conn.http_call("get", f"{conn.base_url}/images/{image_id}").json() + + +def delete_image_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + image_id = params.get("image_id") + if not image_id: + return {"error": "image_id is required"} + conn = _get_connector(gns3_ctx) + conn.http_call("delete", f"{conn.base_url}/images/{image_id}") + return {"message": f"Image {image_id} deleted", "image_id": image_id} + + +def prune_images_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + conn = _get_connector(gns3_ctx) + result = conn.http_call("delete", f"{conn.base_url}/images/prune").json() + return {"message": "Unused images pruned", "result": result} + + +def install_images_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + conn = _get_connector(gns3_ctx) + result = conn.http_call("post", f"{conn.base_url}/images/install").json() + return {"message": "Image installation requested", "result": result}