diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index 907c0195c..dad350e5c 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -51,6 +51,7 @@ from .projects import ( get_project_stats_handler, update_project_handler, duplicate_project_handler, get_project_readme_handler, update_project_readme_handler, lock_project_handler, unlock_project_handler, + load_project_handler, get_locked_project_handler, ) from .server import ( get_version_handler, get_statistics_handler, @@ -58,6 +59,7 @@ from .server import ( from .symbols import ( get_symbols_handler, get_symbol_handler, get_symbol_dimensions_handler, get_default_symbols_handler, + upload_symbol_handler, delete_symbol_handler, ) from .appliances import ( get_appliances_handler, get_appliance_handler, @@ -971,6 +973,26 @@ async def unlock_project( }) +@mcp.tool() +async def get_locked_project( + project_id: Annotated[str, Field(description="UUID of the project")], +) -> list[dict[str, Any]]: + """Check whether a project is locked (preventing edits to drawings and nodes).""" + return await asyncio.to_thread(_run_handler_sync, get_locked_project_handler, { + "project_id": project_id, + }) + + +@mcp.tool() +async def load_project( + path: Annotated[str, Field(description="Filesystem path to the .gns3 project file")], +) -> list[dict[str, Any]]: + """Load a project from a file path on the server's filesystem.""" + return await asyncio.to_thread(_run_handler_sync, load_project_handler, { + "path": path, + }) + + # ── Server info tools ───────────────────────────────────────────────── @@ -1021,6 +1043,26 @@ async def get_default_symbols() -> list[dict[str, Any]]: return await asyncio.to_thread(_run_handler_sync, get_default_symbols_handler, {}) +@mcp.tool() +async def upload_symbol( + symbol_id: Annotated[str, Field(description="Symbol ID to upload (e.g. ':/symbols/my_symbol.svg')")], +) -> list[dict[str, Any]]: + """Upload or update a custom symbol on the server.""" + return await asyncio.to_thread(_run_handler_sync, upload_symbol_handler, { + "symbol_id": symbol_id, + }) + + +@mcp.tool() +async def delete_symbol( + symbol_id: Annotated[str, Field(description="Symbol ID to delete")], +) -> list[dict[str, Any]]: + """Delete a custom symbol from the server.""" + return await asyncio.to_thread(_run_handler_sync, delete_symbol_handler, { + "symbol_id": symbol_id, + }) + + # ── Appliance tools ─────────────────────────────────────────────────── diff --git a/gns3server/api/routes/mcp/projects.py b/gns3server/api/routes/mcp/projects.py index ba1bc1484..bcb348380 100644 --- a/gns3server/api/routes/mcp/projects.py +++ b/gns3server/api/routes/mcp/projects.py @@ -176,6 +176,24 @@ def unlock_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> return {"message": f"Project {project_id} unlocked", "project_id": project_id} +def load_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + path = params.get("path") + if not path: + return {"error": "path is required"} + conn = _get_connector(gns3_ctx) + result = conn.http_call("post", f"{conn.base_url}/projects/load", json_data={"path": path}).json() + return {"message": f"Project loaded from {path}", "project": result} + + +def get_locked_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) + locked = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/locked").json() + return {"project_id": project_id, "locked": locked} + + # ── Tool definitions (consumed by mcp/__init__.py) ───────────────────────── PROJECT_TOOLS = [ diff --git a/gns3server/api/routes/mcp/symbols.py b/gns3server/api/routes/mcp/symbols.py index ca40a6feb..5ff2d56d8 100644 --- a/gns3server/api/routes/mcp/symbols.py +++ b/gns3server/api/routes/mcp/symbols.py @@ -66,3 +66,21 @@ def get_default_symbols_handler(params: dict[str, Any], gns3_ctx: dict[str, Any] conn = _get_connector(gns3_ctx) symbols = conn.http_call("get", f"{conn.base_url}/symbols/default_symbols").json() return {"default_symbols": symbols} + + +def upload_symbol_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + symbol_id = params.get("symbol_id") + if not symbol_id: + return {"error": "symbol_id is required"} + conn = _get_connector(gns3_ctx) + result = conn.http_call("post", f"{conn.base_url}/symbols/{symbol_id}").json() + return {"message": f"Symbol {symbol_id} uploaded", "symbol": result} + + +def delete_symbol_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + symbol_id = params.get("symbol_id") + if not symbol_id: + return {"error": "symbol_id is required"} + conn = _get_connector(gns3_ctx) + conn.http_call("delete", f"{conn.base_url}/symbols/{symbol_id}") + return {"message": f"Symbol {symbol_id} deleted", "symbol_id": symbol_id}