Merge pull request #2777 from yueguobin/mcp/project-tools

Add MCP project tools: update, duplicate, and README operations

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.

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.md file — the project documentation (Markdown format)."""
+    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.md (Markdown format)")],
+) -> list[dict[str, Any]]:
+    """Update or create a project's README.md file — the project documentation (Markdown format)."""
+    return await asyncio.to_thread(_run_handler_sync, update_project_readme_handler, {"project_id": project_id, "content": content})
+
+
 # ── Node tools ────────────────────────────────────────────────────────
 
 @mcp.tool()

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.md file (project documentation, Markdown format)",
+        "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.md file (project documentation, Markdown format)",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "project_id": {"type": "string", "description": "Project UUID"},
+                "content": {"type": "string", "description": "Content to write to README.md (Markdown format)"},
+            },
+            "required": ["project_id", "content"],
+        },
+        "handler": update_project_readme_handler,
+    },
 ]