From 4921b989484019df44746bac31f0ba1885eda246 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Sun, 7 Jun 2026 23:14:07 +0800 Subject: [PATCH] Add MCP project tools: update, duplicate, and README operations - Add update_project and duplicate_project MCP tools - Add get_project_readme and update_project_readme tools - Add Gns3Connector methods: update_project, duplicate_project, get_project_file, write_project_file --- .../gns3_copilot/gns3_client/custom_gns3fy.py | 76 +++++++++++- gns3server/api/routes/mcp/__init__.py | 64 +++++++++- gns3server/api/routes/mcp/projects.py | 111 ++++++++++++++++++ 3 files changed, 249 insertions(+), 2 deletions(-) diff --git a/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py b/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py index d565975e7..2c83ddf7a 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py +++ b/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py @@ -55,7 +55,7 @@ from typing import Any from typing import ParamSpec from typing import TypeVar from typing import cast -from urllib.parse import urlparse +from urllib.parse import urlparse, quote import jwt import requests @@ -708,6 +708,80 @@ class Gns3Connector: self.http_call("delete", _url) return None + def update_project(self, project_id: str, **kwargs: Any) -> dict[str, Any]: + """ + Update a project's properties. + + **Required Attributes:** + + - `project_id` + + **Optional Attributes:** + + - `name`, `auto_close`, `auto_open`, `auto_start` + - `scene_height`, `scene_width`, `zoom` + - `show_layers`, `snap_to_grid`, `show_grid`, `grid_size`, `drawing_grid_size` + - `show_interface_labels`, `supplier`, `variables` + + **Returns** + + JSON project information + """ + _url = f"{self.base_url}/projects/{project_id}" + _response = self.http_call("put", _url, json_data=kwargs) + return cast(dict[str, Any], _response.json()) + + def duplicate_project(self, project_id: str, **kwargs: Any) -> dict[str, Any]: + """ + Duplicate a project from a given project_id. + + **Required Attributes:** + + - `project_id` + - `name` (in kwargs) + + **Returns** + + JSON project information + """ + _url = f"{self.base_url}/projects/{project_id}/duplicate" + if "name" not in kwargs: + raise ValueError("Parameter 'name' is mandatory") + _response = self.http_call("post", _url, json_data=kwargs) + return cast(dict[str, Any], _response.json()) + + def get_project_file(self, project_id: str, file_path: str) -> str: + """ + Get the content of a file in a project. + + **Required Attributes:** + + - `project_id` + - `file_path` + + **Returns** + + File content as text string + """ + encoded_path = quote(file_path, safe="/") + _url = f"{self.base_url}/projects/{project_id}/files/{encoded_path}" + _response = self.http_call("get", _url) + return _response.text + + def write_project_file(self, project_id: str, file_path: str, content: str) -> None: + """ + Write content to a file in a project. Creates the file if it doesn't exist. + + **Required Attributes:** + + - `project_id` + - `file_path` + - `content` + """ + encoded_path = quote(file_path, safe="/") + _url = f"{self.base_url}/projects/{project_id}/files/{encoded_path}" + self.http_call("post", _url, data=content, headers={"Content-Type": "text/plain"}) + def get_computes(self) -> list[dict[str, Any]]: """ Returns a list of computes. diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index 39d19a501..b4bbd1ce5 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -48,7 +48,8 @@ from gns3server.utils.request_utils import extract_client_info from .projects import ( list_projects_handler, get_project_handler, create_project_handler, delete_project_handler, open_project_handler, close_project_handler, - get_project_stats_handler, + get_project_stats_handler, update_project_handler, duplicate_project_handler, + get_project_readme_handler, update_project_readme_handler, ) from .nodes import ( get_nodes_handler, get_node_handler, start_node_handler, @@ -255,6 +256,67 @@ async def get_project_stats( return await asyncio.to_thread(_run_handler_sync, get_project_stats_handler, {"project_id": project_id}) +@mcp.tool() +async def update_project( + project_id: Annotated[str, Field(description="UUID of the project to update")], + name: Annotated[str, Field(description="New project name")] = None, + auto_close: Annotated[bool, Field(description="Close project when last client leaves")] = None, + auto_open: Annotated[bool, Field(description="Project opens when GNS3 starts")] = None, + auto_start: Annotated[bool, Field(description="Project starts when opened")] = None, + scene_width: Annotated[int, Field(description="Width of the drawing area")] = None, + scene_height: Annotated[int, Field(description="Height of the drawing area")] = None, + zoom: Annotated[int, Field(description="Zoom of the drawing area")] = None, + show_layers: Annotated[bool, Field(description="Show layers on the drawing area")] = None, + snap_to_grid: Annotated[bool, Field(description="Snap to grid on the drawing area")] = None, + show_grid: Annotated[bool, 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, + drawing_grid_size: Annotated[int, 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, +) -> list[dict[str, Any]]: + """Update a project's properties (name, auto_close, auto_open, etc.).""" + params = {"project_id": project_id} + local_vars = { + "name": name, "auto_close": auto_close, "auto_open": auto_open, "auto_start": auto_start, + "scene_width": scene_width, "scene_height": scene_height, "zoom": zoom, + "show_layers": show_layers, "snap_to_grid": snap_to_grid, "show_grid": show_grid, + "grid_size": grid_size, "drawing_grid_size": drawing_grid_size, "show_interface_labels": show_interface_labels, + } + for key, val in local_vars.items(): + if val is not None: + params[key] = val + return await asyncio.to_thread(_run_handler_sync, update_project_handler, params) + + +@mcp.tool() +async def duplicate_project( + project_id: Annotated[str, Field(description="UUID of the project to duplicate")], + name: Annotated[str, Field(description="New project name")], + reset_mac_addresses: Annotated[bool, Field(description="Reset MAC addresses for this project")] = False, +) -> list[dict[str, Any]]: + """Duplicate a project.""" + params = {"project_id": project_id, "name": name} + if reset_mac_addresses: + params["reset_mac_addresses"] = reset_mac_addresses + return await asyncio.to_thread(_run_handler_sync, duplicate_project_handler, params) + + +@mcp.tool() +async def get_project_readme( + project_id: Annotated[str, Field(description="UUID of the project")], +) -> list[dict[str, Any]]: + """Get the content of a project's README.txt file — the project documentation.""" + return await asyncio.to_thread(_run_handler_sync, get_project_readme_handler, {"project_id": project_id}) + + +@mcp.tool() +async def update_project_readme( + project_id: Annotated[str, Field(description="UUID of the project")], + content: Annotated[str, Field(description="Content to write to README.txt")], +) -> list[dict[str, Any]]: + """Update or create a project's README.txt file — the project documentation.""" + return await asyncio.to_thread(_run_handler_sync, update_project_readme_handler, {"project_id": project_id, "content": content}) + + # ── Node tools ──────────────────────────────────────────────────────── @mcp.tool() diff --git a/gns3server/api/routes/mcp/projects.py b/gns3server/api/routes/mcp/projects.py index daec65ceb..d718a0e1d 100644 --- a/gns3server/api/routes/mcp/projects.py +++ b/gns3server/api/routes/mcp/projects.py @@ -109,6 +109,53 @@ def get_project_stats_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) return conn.http_call("get", url).json() +def update_project_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) + kwargs = {k: v for k, v in params.items() if k != "project_id" and v is not None} + return conn.update_project(project_id=project_id, **kwargs) + + +def duplicate_project_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"} + name = params.get("name") + if not name: + return {"error": "name is required"} + conn = _get_connector(gns3_ctx) + kwargs = {k: v for k, v in params.items() if k not in ("project_id",) and v is not None} + return conn.duplicate_project(project_id=project_id, **kwargs) + + +def get_project_readme_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) + try: + content = conn.get_project_file(project_id=project_id, file_path="README.txt") + return {"project_id": project_id, "file": "README.txt", "content": content} + except Exception as e: + if "404" in str(e): + return {"project_id": project_id, "file": "README.txt", "content": None, "message": "README.txt does not exist yet"} + raise + + +def update_project_readme_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"} + content = params.get("content") + if content is None: + return {"error": "content is required"} + conn = _get_connector(gns3_ctx) + conn.write_project_file(project_id=project_id, file_path="README.txt", content=content) + return {"message": "README.txt updated", "project_id": project_id} + + # ── Tool definitions (consumed by mcp/__init__.py) ───────────────────────── PROJECT_TOOLS = [ @@ -191,4 +238,68 @@ PROJECT_TOOLS = [ }, "handler": get_project_stats_handler, }, + { + "name": "update_project", + "description": "Update a project's properties (name, auto_close, auto_open, etc.)", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "name": {"type": "string", "description": "New project name"}, + "auto_close": {"type": "boolean", "description": "Close project when last client leaves"}, + "auto_open": {"type": "boolean", "description": "Project opens when GNS3 starts"}, + "auto_start": {"type": "boolean", "description": "Project starts when opened"}, + "scene_width": {"type": "integer", "description": "Width of the drawing area"}, + "scene_height": {"type": "integer", "description": "Height of the drawing area"}, + "zoom": {"type": "integer", "description": "Zoom of the drawing area"}, + "show_layers": {"type": "boolean", "description": "Show layers on the drawing area"}, + "snap_to_grid": {"type": "boolean", "description": "Snap to grid on the drawing area"}, + "show_grid": {"type": "boolean", "description": "Show the grid on the drawing area"}, + "grid_size": {"type": "integer", "description": "Grid size for the drawing area for nodes"}, + "drawing_grid_size": {"type": "integer", "description": "Grid size for the drawing area for drawings"}, + "show_interface_labels": {"type": "boolean", "description": "Show interface labels on the drawing area"}, + }, + "required": ["project_id"], + }, + "handler": update_project_handler, + }, + { + "name": "duplicate_project", + "description": "Duplicate a project", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "UUID of the project to duplicate"}, + "name": {"type": "string", "description": "New project name"}, + "reset_mac_addresses": {"type": "boolean", "description": "Reset MAC addresses for this project"}, + }, + "required": ["project_id", "name"], + }, + "handler": duplicate_project_handler, + }, + { + "name": "get_project_readme", + "description": "Get the content of a project's README.txt file (project documentation)", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + }, + "required": ["project_id"], + }, + "handler": get_project_readme_handler, + }, + { + "name": "update_project_readme", + "description": "Update or create a project's README.txt file (project documentation)", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "content": {"type": "string", "description": "Content to write to README.txt"}, + }, + "required": ["project_id", "content"], + }, + "handler": update_project_readme_handler, + }, ]