feat: Add symbol upload/delete, project load, and locked check MCP tools

This commit is contained in:
YueGuobin 2026-06-10 13:59:39 +08:00
parent b786f0b7eb
commit 41594faf43
No known key found for this signature in database
3 changed files with 78 additions and 0 deletions

View File

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

View File

@ -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 = [

View File

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