diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index 5ee8e26ee..a24d10dac 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -55,6 +55,14 @@ from .projects import ( from .server import ( get_version_handler, get_statistics_handler, ) +from .symbols import ( + get_symbols_handler, get_symbol_handler, + get_symbol_dimensions_handler, get_default_symbols_handler, +) +from .appliances import ( + get_appliances_handler, get_appliance_handler, + install_appliance_handler, +) from .nodes import ( get_nodes_handler, get_node_handler, start_node_handler, stop_node_handler, reload_node_handler, suspend_node_handler, @@ -973,6 +981,70 @@ async def get_statistics() -> list[dict[str, Any]]: return await asyncio.to_thread(_run_handler_sync, get_statistics_handler, {}) +# ── Symbol tools ────────────────────────────────────────────────────── + + +@mcp.tool() +async def get_symbols() -> list[dict[str, Any]]: + """List all available symbols on the server.""" + return await asyncio.to_thread(_run_handler_sync, get_symbols_handler, {}) + + +@mcp.tool() +async def get_symbol( + symbol_id: Annotated[str, Field(description="Symbol ID (e.g. ':/symbols/router.svg')")], +) -> list[dict[str, Any]]: + """Get details about a specific symbol.""" + return await asyncio.to_thread(_run_handler_sync, get_symbol_handler, { + "symbol_id": symbol_id, + }) + + +@mcp.tool() +async def get_symbol_dimensions( + symbol_id: Annotated[str, Field(description="Symbol ID to get dimensions for")], +) -> list[dict[str, Any]]: + """Get the dimensions (width, height) of a symbol.""" + return await asyncio.to_thread(_run_handler_sync, get_symbol_dimensions_handler, { + "symbol_id": symbol_id, + }) + + +@mcp.tool() +async def get_default_symbols() -> list[dict[str, Any]]: + """Get the default symbol mapping for each node type.""" + return await asyncio.to_thread(_run_handler_sync, get_default_symbols_handler, {}) + + +# ── Appliance tools ─────────────────────────────────────────────────── + + +@mcp.tool() +async def get_appliances() -> list[dict[str, Any]]: + """List all available appliances (template library).""" + return await asyncio.to_thread(_run_handler_sync, get_appliances_handler, {}) + + +@mcp.tool() +async def get_appliance( + appliance_id: Annotated[str, Field(description="UUID of the appliance")], +) -> list[dict[str, Any]]: + """Get detailed information about a specific appliance.""" + return await asyncio.to_thread(_run_handler_sync, get_appliance_handler, { + "appliance_id": appliance_id, + }) + + +@mcp.tool() +async def install_appliance( + appliance_id: Annotated[str, Field(description="UUID of the appliance to install")], +) -> list[dict[str, Any]]: + """Install (download and set up) an appliance from the template library.""" + return await asyncio.to_thread(_run_handler_sync, install_appliance_handler, { + "appliance_id": appliance_id, + }) + + # ── Auth‑wrapped SSE app ────────────────────────────────────────────── def _make_auth_wrapper(inner_app): diff --git a/gns3server/api/routes/mcp/appliances.py b/gns3server/api/routes/mcp/appliances.py new file mode 100644 index 000000000..426538560 --- /dev/null +++ b/gns3server/api/routes/mcp/appliances.py @@ -0,0 +1,63 @@ +# +# Copyright (C) 2026 GNS3 Technologies Inc. +# Author: Yue Guobin +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +""" +MCP tool handlers for GNS3 appliance management. +""" + +from typing import Any + +import logging + +log = logging.getLogger(__name__) + + +# ── Helper ───────────────────────────────────────────────────────────────── + +def _get_connector(gns3_ctx: dict[str, Any]): + from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector + return Gns3Connector( + url=gns3_ctx["server_url"], + jwt_token=gns3_ctx["jwt_token"], + api_version=3, + verify=False, + ) + + +# ── Tool handlers ────────────────────────────────────────────────────────── + +def get_appliances_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + conn = _get_connector(gns3_ctx) + appliances = conn.http_call("get", f"{conn.base_url}/appliances").json() + return {"appliances": appliances, "count": len(appliances)} + + +def get_appliance_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + appliance_id = params.get("appliance_id") + if not appliance_id: + return {"error": "appliance_id is required"} + conn = _get_connector(gns3_ctx) + return conn.http_call("get", f"{conn.base_url}/appliances/{appliance_id}").json() + + +def install_appliance_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + appliance_id = params.get("appliance_id") + if not appliance_id: + return {"error": "appliance_id is required"} + conn = _get_connector(gns3_ctx) + result = conn.http_call("post", f"{conn.base_url}/appliances/{appliance_id}/install").json() + return {"message": f"Appliance {appliance_id} installation requested", "result": result} diff --git a/gns3server/api/routes/mcp/symbols.py b/gns3server/api/routes/mcp/symbols.py new file mode 100644 index 000000000..ca40a6feb --- /dev/null +++ b/gns3server/api/routes/mcp/symbols.py @@ -0,0 +1,68 @@ +# +# Copyright (C) 2026 GNS3 Technologies Inc. +# Author: Yue Guobin +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +""" +MCP tool handlers for GNS3 symbol management. +""" + +from typing import Any + +import logging + +log = logging.getLogger(__name__) + + +# ── Helper ───────────────────────────────────────────────────────────────── + +def _get_connector(gns3_ctx: dict[str, Any]): + from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector + return Gns3Connector( + url=gns3_ctx["server_url"], + jwt_token=gns3_ctx["jwt_token"], + api_version=3, + verify=False, + ) + + +# ── Tool handlers ────────────────────────────────────────────────────────── + +def get_symbols_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + conn = _get_connector(gns3_ctx) + symbols = conn.http_call("get", f"{conn.base_url}/symbols").json() + return {"symbols": symbols, "count": len(symbols)} + + +def get_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) + return conn.http_call("get", f"{conn.base_url}/symbols/{symbol_id}").json() + + +def get_symbol_dimensions_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) + return conn.http_call("get", f"{conn.base_url}/symbols/{symbol_id}/dimensions").json() + + +def get_default_symbols_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> 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}