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
This commit is contained in:
YueGuobin 2026-06-07 23:14:07 +08:00
parent 95b1a35300
commit 4921b98948
No known key found for this signature in database
3 changed files with 249 additions and 2 deletions

View File

@ -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.

View File

@ -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()

View File

@ -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,
},
]