From 7086db42265b1b59c37f7506713396f2b2c0439f Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Thu, 4 Jun 2026 13:53:05 +0800 Subject: [PATCH 01/20] feat: add MCP (Model Context Protocol) service with project tools - Add MCPTool/MCPToolRegistry system for centralized tool registration - Add 7 project-related MCP tools: list_projects, get_project, create_project, delete_project, open_project, close_project, get_project_stats - Tools use Gns3Connector (custom_gns3fy) to call GNS3 REST API via HTTP loopback, keeping the MCP layer decoupled from controller internals - Handlers run in thread pool via asyncio.to_thread() to avoid blocking the event loop on synchronous requests calls - Unified POST /v3/mcp/execute endpoint with JWT authentication --- gns3server/api/routes/mcp/__init__.py | 175 ++++++++++++++++ gns3server/api/routes/mcp/projects.py | 287 ++++++++++++++++++++++++++ gns3server/api/server.py | 2 + 3 files changed, 464 insertions(+) create mode 100644 gns3server/api/routes/mcp/__init__.py create mode 100644 gns3server/api/routes/mcp/projects.py diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py new file mode 100644 index 000000000..5e3771c74 --- /dev/null +++ b/gns3server/api/routes/mcp/__init__.py @@ -0,0 +1,175 @@ +# +# Copyright (C) 2020 GNS3 Technologies Inc. +# +# 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 (Model Context Protocol) service routes for GNS3 server. + +Provides a unified tool execution interface that wraps existing GNS3 API +functionality. Tools are registered via MCPToolRegistry and executed +through a single POST /v3/mcp/execute endpoint. +""" + +from fastapi import APIRouter, Depends, HTTPException, status, Request +from typing import Dict, Any, List, Callable, Optional +from pydantic import BaseModel +import asyncio +import logging + +from gns3server import schemas +from gns3server.api.routes.controller.dependencies.authentication import get_current_active_user +from gns3server.config import Config + +log = logging.getLogger(__name__) + +router = APIRouter(prefix="/mcp", tags=["MCP"]) + + +# ── Tool Registration ────────────────────────────────────────────────────── + +class MCPTool: + """ + An MCP tool binds a name, description, parameter schema, and handler together. + """ + + def __init__( + self, + name: str, + description: str, + parameters_schema: Dict[str, Any], + handler: Callable, + required_permission: Optional[str] = None, + ): + self.name = name + self.description = description + self.parameters_schema = parameters_schema + self.handler = handler + self.required_permission = required_permission + + def as_dict(self) -> Dict[str, Any]: + return { + "name": self.name, + "description": self.description, + "parameters": self.parameters_schema, + } + + +class MCPToolRegistry: + """Central registry — tools are registered once, listed/executed on demand.""" + + def __init__(self): + self._tools: Dict[str, MCPTool] = {} + + def register_tool(self, tool: MCPTool) -> None: + self._tools[tool.name] = tool + log.info(f"Registered MCP tool: {tool.name}") + + def get_tool(self, name: str) -> Optional[MCPTool]: + return self._tools.get(name) + + def list_tools(self) -> List[Dict[str, Any]]: + return [t.as_dict() for t in self._tools.values()] + + async def execute( + self, tool_name: str, parameters: Dict[str, Any], **context + ) -> Dict[str, Any]: + tool = self.get_tool(tool_name) + if tool is None: + return {"status": "error", "error": f"Tool '{tool_name}' not found"} + + try: + # Handlers use synchronous Gns3Connector (requests library), + # so run them in a thread to avoid blocking the event loop. + result = await asyncio.to_thread(tool.handler, parameters, **context) + return {"status": "success", "data": result} + except Exception as e: + log.error(f"Error executing tool '{tool_name}': {e}") + return {"status": "error", "error": str(e)} + + +# Global registry instance +registry = MCPToolRegistry() + +# Import tool modules to trigger registration +from . import projects # noqa: F401 — triggers register_tools() + + +# ── Pydantic request / response models ───────────────────────────────────── + +class ExecuteToolRequest(BaseModel): + tool: str + parameters: Dict[str, Any] = {} + + +class ExecuteToolResponse(BaseModel): + status: str + data: Optional[Dict[str, Any]] = None + error: Optional[str] = None + + +# ── MCP Endpoints ────────────────────────────────────────────────────────── + +@router.get("/") +async def mcp_root(): + """MCP service root — capability discovery.""" + return { + "name": "GNS3 MCP Server", + "version": "1.0.0", + "capabilities": {"tools": True, "resources": False, "prompts": False}, + } + + +@router.get("/tools") +async def list_tools(): + """List every registered MCP tool with its parameter schema.""" + tools = registry.list_tools() + return {"tools": tools, "count": len(tools)} + + +@router.post("/execute", response_model=ExecuteToolResponse) +async def execute_tool( + request: ExecuteToolRequest, + http_request: Request, + current_user: schemas.User = Depends(get_current_active_user), +): + """ + Execute an MCP tool by name. + Authentication is enforced via the existing JWT mechanism. + The tool handler receives a Gns3Connector pre-configured with the + current user's JWT token so it calls GNS3's own REST API (not the + controller internals), keeping the MCP layer fully decoupled. + """ + + # Extract the raw JWT token from the Authorization header + auth_header = http_request.headers.get("Authorization", "") + jwt_token = auth_header.replace("Bearer ", "") if auth_header.startswith("Bearer ") else None + + # Build the local GNS3 API base URL from the server config + config = Config.instance().settings + host = config.Server.host + if host == "0.0.0.0": + host = "127.0.0.1" + port = config.Server.port + scheme = "https" if config.Server.enable_ssl else "http" + server_url = f"{scheme}://{host}:{port}" + + result = await registry.execute( + request.tool, + request.parameters, + current_user=current_user, + jwt_token=jwt_token, + server_url=server_url, + ) + return ExecuteToolResponse(**result) diff --git a/gns3server/api/routes/mcp/projects.py b/gns3server/api/routes/mcp/projects.py new file mode 100644 index 000000000..033a0ca8c --- /dev/null +++ b/gns3server/api/routes/mcp/projects.py @@ -0,0 +1,287 @@ +# +# Copyright (C) 2020 GNS3 Technologies Inc. +# +# 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 tools for GNS3 project management. + +Each handler receives (parameters, current_user, jwt_token, server_url) and +uses a Gns3Connector (from custom_gns3fy) to call GNS3's own REST API. +This keeps the MCP layer decoupled from the controller internals. +""" + +from typing import Dict, Any + +from . import registry, MCPTool + +import logging + +log = logging.getLogger(__name__) + + +# ── Helper ───────────────────────────────────────────────────────────────── + +def _get_connector(server_url: str, jwt_token: str): + """Create a Gns3Connector using the user's JWT token.""" + from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector + return Gns3Connector( + url=server_url, + jwt_token=jwt_token, + api_version=3, + verify=False, + ) + + +# ── Tool: list_projects ──────────────────────────────────────────────────── + +def list_projects_handler( + params: Dict[str, Any], + current_user=None, + jwt_token=None, + server_url=None, +) -> Dict[str, Any]: + """Return all projects via GNS3 REST API.""" + conn = _get_connector(server_url, jwt_token) + projects = conn.get_projects() + return {"projects": projects, "count": len(projects)} + + +# ── Tool: get_project ────────────────────────────────────────────────────── + +def get_project_handler( + params: Dict[str, Any], + current_user=None, + jwt_token=None, + server_url=None, +) -> Dict[str, Any]: + """Return a single project by project_id.""" + project_id = params.get("project_id") + if not project_id: + return {"error": "project_id is required"} + + conn = _get_connector(server_url, jwt_token) + project = conn.get_project(project_id=project_id) + if project is None: + return {"error": f"Project '{project_id}' not found"} + return project + + +# ── Tool: create_project ─────────────────────────────────────────────────── + +def create_project_handler( + params: Dict[str, Any], + current_user=None, + jwt_token=None, + server_url=None, +) -> Dict[str, Any]: + """Create a new project via GNS3 REST API.""" + name = params.get("name") + if not name: + return {"error": "name is required"} + + conn = _get_connector(server_url, jwt_token) + project_data = {"name": name} + if "description" in params: + project_data["description"] = params["description"] + + project = conn.create_project(**project_data) + return project + + +# ── Tool: delete_project ─────────────────────────────────────────────────── + +def delete_project_handler( + params: Dict[str, Any], + current_user=None, + jwt_token=None, + server_url=None, +) -> Dict[str, Any]: + """Delete a project by project_id via GNS3 REST API.""" + project_id = params.get("project_id") + if not project_id: + return {"error": "project_id is required"} + + conn = _get_connector(server_url, jwt_token) + conn.delete_project(project_id=project_id) + return {"message": f"Project '{project_id}' deleted", "project_id": project_id} + + +# ── Tool: open_project ───────────────────────────────────────────────────── + +def open_project_handler( + params: Dict[str, Any], + current_user=None, + jwt_token=None, + server_url=None, +) -> Dict[str, Any]: + """Open a closed project via GNS3 REST API.""" + project_id = params.get("project_id") + if not project_id: + return {"error": "project_id is required"} + + conn = _get_connector(server_url, jwt_token) + url = f"{conn.base_url}/projects/{project_id}/open" + response = conn.http_call("post", url) + return response.json() + + +# ── Tool: close_project ──────────────────────────────────────────────────── + +def close_project_handler( + params: Dict[str, Any], + current_user=None, + jwt_token=None, + server_url=None, +) -> Dict[str, Any]: + """Close an open project via GNS3 REST API.""" + project_id = params.get("project_id") + if not project_id: + return {"error": "project_id is required"} + + conn = _get_connector(server_url, jwt_token) + url = f"{conn.base_url}/projects/{project_id}/close" + conn.http_call("post", url) + return {"message": f"Project '{project_id}' closed", "project_id": project_id} + + +# ── Tool: get_project_stats ──────────────────────────────────────────────── + +def get_project_stats_handler( + params: Dict[str, Any], + current_user=None, + jwt_token=None, + server_url=None, +) -> Dict[str, Any]: + """Return project statistics via GNS3 REST API.""" + project_id = params.get("project_id") + if not project_id: + return {"error": "project_id is required"} + + conn = _get_connector(server_url, jwt_token) + url = f"{conn.base_url}/projects/{project_id}/stats" + response = conn.http_call("get", url) + return response.json() + + +# ── Register all project tools ───────────────────────────────────────────── + +def register_tools(): + """Register every project-related MCP tool into the global registry.""" + tools = [ + MCPTool( + name="list_projects", + description="List all GNS3 projects accessible to the current user", + parameters_schema={"type": "object", "properties": {}}, + handler=list_projects_handler, + ), + MCPTool( + name="get_project", + description="Get detailed information about a specific project", + parameters_schema={ + "type": "object", + "properties": { + "project_id": { + "type": "string", + "description": "Project UUID", + } + }, + "required": ["project_id"], + }, + handler=get_project_handler, + ), + MCPTool( + name="create_project", + description="Create a new GNS3 project", + parameters_schema={ + "type": "object", + "properties": { + "name": {"type": "string", "description": "Project name"}, + "description": { + "type": "string", + "description": "Optional project description", + }, + }, + "required": ["name"], + }, + handler=create_project_handler, + ), + MCPTool( + name="delete_project", + description="Delete a GNS3 project permanently", + parameters_schema={ + "type": "object", + "properties": { + "project_id": { + "type": "string", + "description": "UUID of the project to delete", + } + }, + "required": ["project_id"], + }, + handler=delete_project_handler, + ), + MCPTool( + name="open_project", + description="Open a closed GNS3 project", + parameters_schema={ + "type": "object", + "properties": { + "project_id": { + "type": "string", + "description": "Project UUID", + } + }, + "required": ["project_id"], + }, + handler=open_project_handler, + ), + MCPTool( + name="close_project", + description="Close an open GNS3 project", + parameters_schema={ + "type": "object", + "properties": { + "project_id": { + "type": "string", + "description": "Project UUID", + } + }, + "required": ["project_id"], + }, + handler=close_project_handler, + ), + MCPTool( + name="get_project_stats", + description="Get statistics (nodes, links, snapshots, drawings) for a project", + parameters_schema={ + "type": "object", + "properties": { + "project_id": { + "type": "string", + "description": "Project UUID", + } + }, + "required": ["project_id"], + }, + handler=get_project_stats_handler, + ), + ] + + for tool in tools: + registry.register_tool(tool) + + +# Auto-register on import +register_tools() diff --git a/gns3server/api/server.py b/gns3server/api/server.py index 196ddb504..54e6eb51d 100644 --- a/gns3server/api/server.py +++ b/gns3server/api/server.py @@ -45,6 +45,7 @@ from gns3server.controller.controller_error import ( from gns3server.api.routes import controller, index from gns3server.api.routes.compute import compute_api +from gns3server.api.routes import mcp from gns3server.core import tasks import logging @@ -75,6 +76,7 @@ def get_application() -> FastAPI: application.include_router(controller.router, prefix="/v3") application.mount("/static", StaticFiles(packages=[('gns3server', 'static')], html=True), name="static") application.mount("/v3/compute", compute_api, name="compute") + application.include_router(mcp.router, prefix="/v3", tags=["MCP"]) return application From 55b3a7d622677365d66cc4ca9508d636eee5ef2a Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Thu, 4 Jun 2026 22:19:43 +0800 Subject: [PATCH 02/20] feat: implement standard MCP protocol with SSE transport - Use FastMCP (Anthropic MCP SDK) for tool registration and SSE transport - Mount SSE app under /v3/mcp/transport with JWT token authentication - Token passed via ?token= query parameter on SSE connection - Token validated against GNS3 auth_service and stored in contextvars - Tool handlers create Gns3Connector with JWT token to call GNS3 REST API - 7 project tools: list_projects, get_project, create_project, delete_project, open_project, close_project, get_project_stats - Unauthenticated SSE connections return 401 --- gns3server/api/routes/mcp/__init__.py | 321 +++++++++++++++----------- gns3server/api/routes/mcp/projects.py | 296 ++++++++---------------- gns3server/api/server.py | 3 + 3 files changed, 288 insertions(+), 332 deletions(-) diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index 5e3771c74..e50931872 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -15,161 +15,208 @@ # along with this program. If not, see . """ -MCP (Model Context Protocol) service routes for GNS3 server. +MCP (Model Context Protocol) service for GNS3 server. -Provides a unified tool execution interface that wraps existing GNS3 API -functionality. Tools are registered via MCPToolRegistry and executed -through a single POST /v3/mcp/execute endpoint. +Implements the standard MCP protocol over SSE transport using FastMCP: + + /v3/mcp/sse — SSE stream + /v3/mcp/messages/ — JSON-RPC messages + +Tools are registered via @mcp.tool() decorators. """ -from fastapi import APIRouter, Depends, HTTPException, status, Request -from typing import Dict, Any, List, Callable, Optional -from pydantic import BaseModel +import contextvars +import json import asyncio import logging +from typing import Any +from urllib.parse import parse_qs + +from fastapi import APIRouter +from fastapi.responses import Response + +from mcp.server.fastmcp import FastMCP -from gns3server import schemas -from gns3server.api.routes.controller.dependencies.authentication import get_current_active_user from gns3server.config import Config log = logging.getLogger(__name__) + +# ── Per‑connection JWT token ───────────────────────────────────────── +# Set during SSE authentication, read by tool handlers running in the +# same asyncio task (contextvars propagate through asyncio.to_thread). + +_jwt_token_var: contextvars.ContextVar[str | None] = contextvars.ContextVar( + "mcp_jwt_token", default=None +) + + +# ── Token validation ────────────────────────────────────────────────── + +async def _validate_token(token: str) -> bool: + """Return True if token is a valid GNS3 JWT.""" + from gns3server.services import auth_service + try: + auth_service.get_username_from_token(token) + return True + except Exception: + return False + + +# ── Server URL helper ───────────────────────────────────────────────── + +def _server_url() -> str: + cfg = Config.instance().settings + host = cfg.Server.host + if host == "0.0.0.0": + host = "127.0.0.1" + scheme = "https" if cfg.Server.enable_ssl else "http" + return f"{scheme}://{host}:{cfg.Server.port}" + + +# ── FastMCP Server ──────────────────────────────────────────────────── + +mcp = FastMCP("GNS3 MCP Server") + + +# ── Tool handlers ───────────────────────────────────────────────────── + +def _run_handler_sync(handler, params: dict[str, Any]) -> list[dict[str, Any]]: + """Run a synchronous Gns3Connector handler in a thread.""" + ctx = { + "server_url": _server_url(), + "jwt_token": _jwt_token_var.get(), + } + result = handler(params, ctx) + return [{"type": "text", "text": json.dumps(result, ensure_ascii=False, default=str)}] + + +@mcp.tool() +async def list_projects() -> list[dict[str, Any]]: + """List all GNS3 projects accessible to the current user.""" + from .projects import list_projects_handler + return await asyncio.to_thread(_run_handler_sync, list_projects_handler, {}) + + +@mcp.tool() +async def get_project(project_id: str) -> list[dict[str, Any]]: + """Get detailed information about a specific project. + + Args: + project_id: Project UUID + """ + from .projects import get_project_handler + return await asyncio.to_thread(_run_handler_sync, get_project_handler, {"project_id": project_id}) + + +@mcp.tool() +async def create_project(name: str, description: str = "") -> list[dict[str, Any]]: + """Create a new GNS3 project. + + Args: + name: Project name + description: Optional project description + """ + from .projects import create_project_handler + params = {"name": name} + if description: + params["description"] = description + return await asyncio.to_thread(_run_handler_sync, create_project_handler, params) + + +@mcp.tool() +async def delete_project(project_id: str) -> list[dict[str, Any]]: + """Delete a GNS3 project permanently. + + Args: + project_id: UUID of the project to delete + """ + from .projects import delete_project_handler + return await asyncio.to_thread(_run_handler_sync, delete_project_handler, {"project_id": project_id}) + + +@mcp.tool() +async def open_project(project_id: str) -> list[dict[str, Any]]: + """Open a closed GNS3 project. + + Args: + project_id: Project UUID + """ + from .projects import open_project_handler + return await asyncio.to_thread(_run_handler_sync, open_project_handler, {"project_id": project_id}) + + +@mcp.tool() +async def close_project(project_id: str) -> list[dict[str, Any]]: + """Close an open GNS3 project. + + Args: + project_id: Project UUID + """ + from .projects import close_project_handler + return await asyncio.to_thread(_run_handler_sync, close_project_handler, {"project_id": project_id}) + + +@mcp.tool() +async def get_project_stats(project_id: str) -> list[dict[str, Any]]: + """Get statistics (nodes, links, snapshots, drawings) for a project. + + Args: + project_id: Project UUID + """ + from .projects import get_project_stats_handler + return await asyncio.to_thread(_run_handler_sync, get_project_stats_handler, {"project_id": project_id}) + + +# ── Auth‑wrapped SSE app ────────────────────────────────────────────── + +def _make_auth_wrapper(sse_app): + """Wrap the SSE app with JWT validation from ?token= query parameter. + + The wrapper intercepts GET requests (SSE connections), validates the + JWT token, and stores it in a context variable so tool handlers can + use it to call the GNS3 REST API. POST messages are passed through + unchanged (they are authenticated by their session association). + """ + + async def auth_wrapper(scope, receive, send): + if scope["type"] == "http" and scope["method"] == "GET": + params = parse_qs(scope.get("query_string", b"").decode()) + tokens = params.get("token", []) + if not tokens or not await _validate_token(tokens[0]): + response = Response("Missing or invalid token", status_code=401) + await response(scope, receive, send) + return + _jwt_token_var.set(tokens[0]) + await sse_app(scope, receive, send) + + return auth_wrapper + + +# ── FastAPI router ──────────────────────────────────────────────────── + router = APIRouter(prefix="/mcp", tags=["MCP"]) -# ── Tool Registration ────────────────────────────────────────────────────── - -class MCPTool: - """ - An MCP tool binds a name, description, parameter schema, and handler together. - """ - - def __init__( - self, - name: str, - description: str, - parameters_schema: Dict[str, Any], - handler: Callable, - required_permission: Optional[str] = None, - ): - self.name = name - self.description = description - self.parameters_schema = parameters_schema - self.handler = handler - self.required_permission = required_permission - - def as_dict(self) -> Dict[str, Any]: - return { - "name": self.name, - "description": self.description, - "parameters": self.parameters_schema, - } - - -class MCPToolRegistry: - """Central registry — tools are registered once, listed/executed on demand.""" - - def __init__(self): - self._tools: Dict[str, MCPTool] = {} - - def register_tool(self, tool: MCPTool) -> None: - self._tools[tool.name] = tool - log.info(f"Registered MCP tool: {tool.name}") - - def get_tool(self, name: str) -> Optional[MCPTool]: - return self._tools.get(name) - - def list_tools(self) -> List[Dict[str, Any]]: - return [t.as_dict() for t in self._tools.values()] - - async def execute( - self, tool_name: str, parameters: Dict[str, Any], **context - ) -> Dict[str, Any]: - tool = self.get_tool(tool_name) - if tool is None: - return {"status": "error", "error": f"Tool '{tool_name}' not found"} - - try: - # Handlers use synchronous Gns3Connector (requests library), - # so run them in a thread to avoid blocking the event loop. - result = await asyncio.to_thread(tool.handler, parameters, **context) - return {"status": "success", "data": result} - except Exception as e: - log.error(f"Error executing tool '{tool_name}': {e}") - return {"status": "error", "error": str(e)} - - -# Global registry instance -registry = MCPToolRegistry() - -# Import tool modules to trigger registration -from . import projects # noqa: F401 — triggers register_tools() - - -# ── Pydantic request / response models ───────────────────────────────────── - -class ExecuteToolRequest(BaseModel): - tool: str - parameters: Dict[str, Any] = {} - - -class ExecuteToolResponse(BaseModel): - status: str - data: Optional[Dict[str, Any]] = None - error: Optional[str] = None - - -# ── MCP Endpoints ────────────────────────────────────────────────────────── - @router.get("/") async def mcp_root(): - """MCP service root — capability discovery.""" + """MCP service metadata.""" return { "name": "GNS3 MCP Server", "version": "1.0.0", - "capabilities": {"tools": True, "resources": False, "prompts": False}, + "protocol": "Model Context Protocol", + "transport": "SSE", + "authentication": "?token=", + "endpoints": { + "sse": "/v3/mcp/transport/sse?token=", + "messages": "/v3/mcp/transport/messages/", + }, } -@router.get("/tools") -async def list_tools(): - """List every registered MCP tool with its parameter schema.""" - tools = registry.list_tools() - return {"tools": tools, "count": len(tools)} - - -@router.post("/execute", response_model=ExecuteToolResponse) -async def execute_tool( - request: ExecuteToolRequest, - http_request: Request, - current_user: schemas.User = Depends(get_current_active_user), -): - """ - Execute an MCP tool by name. - Authentication is enforced via the existing JWT mechanism. - The tool handler receives a Gns3Connector pre-configured with the - current user's JWT token so it calls GNS3's own REST API (not the - controller internals), keeping the MCP layer fully decoupled. - """ - - # Extract the raw JWT token from the Authorization header - auth_header = http_request.headers.get("Authorization", "") - jwt_token = auth_header.replace("Bearer ", "") if auth_header.startswith("Bearer ") else None - - # Build the local GNS3 API base URL from the server config - config = Config.instance().settings - host = config.Server.host - if host == "0.0.0.0": - host = "127.0.0.1" - port = config.Server.port - scheme = "https" if config.Server.enable_ssl else "http" - server_url = f"{scheme}://{host}:{port}" - - result = await registry.execute( - request.tool, - request.parameters, - current_user=current_user, - jwt_token=jwt_token, - server_url=server_url, - ) - return ExecuteToolResponse(**result) +def register_starlette_routes(app): + """Mount the authenticated SSE app under /v3/mcp/transport.""" + raw_sse_app = mcp.sse_app(mount_path="") + wrapped = _make_auth_wrapper(raw_sse_app) + app.mount("/v3/mcp/transport", wrapped, name="mcp-sse") + log.info("MCP SSE server mounted at /v3/mcp/transport") diff --git a/gns3server/api/routes/mcp/projects.py b/gns3server/api/routes/mcp/projects.py index 033a0ca8c..c5a6cbb4b 100644 --- a/gns3server/api/routes/mcp/projects.py +++ b/gns3server/api/routes/mcp/projects.py @@ -17,14 +17,11 @@ """ MCP tools for GNS3 project management. -Each handler receives (parameters, current_user, jwt_token, server_url) and -uses a Gns3Connector (from custom_gns3fy) to call GNS3's own REST API. -This keeps the MCP layer decoupled from the controller internals. +Tool handlers receive (params, gns3_ctx) and call GNS3's REST API +via Gns3Connector (from custom_gns3fy). """ -from typing import Dict, Any - -from . import registry, MCPTool +from typing import Any import logging @@ -33,255 +30,164 @@ log = logging.getLogger(__name__) # ── Helper ───────────────────────────────────────────────────────────────── -def _get_connector(server_url: str, jwt_token: str): - """Create a Gns3Connector using the user's JWT token.""" +def _get_connector(gns3_ctx: dict[str, Any]): + """Create a Gns3Connector from the GNS3 context dict.""" from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector return Gns3Connector( - url=server_url, - jwt_token=jwt_token, + url=gns3_ctx["server_url"], + jwt_token=gns3_ctx["jwt_token"], api_version=3, verify=False, ) -# ── Tool: list_projects ──────────────────────────────────────────────────── +# ── Tool handlers ────────────────────────────────────────────────────────── -def list_projects_handler( - params: Dict[str, Any], - current_user=None, - jwt_token=None, - server_url=None, -) -> Dict[str, Any]: - """Return all projects via GNS3 REST API.""" - conn = _get_connector(server_url, jwt_token) +def list_projects_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + conn = _get_connector(gns3_ctx) projects = conn.get_projects() return {"projects": projects, "count": len(projects)} -# ── Tool: get_project ────────────────────────────────────────────────────── - -def get_project_handler( - params: Dict[str, Any], - current_user=None, - jwt_token=None, - server_url=None, -) -> Dict[str, Any]: - """Return a single project by project_id.""" +def get_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(server_url, jwt_token) + conn = _get_connector(gns3_ctx) project = conn.get_project(project_id=project_id) if project is None: return {"error": f"Project '{project_id}' not found"} return project -# ── Tool: create_project ─────────────────────────────────────────────────── - -def create_project_handler( - params: Dict[str, Any], - current_user=None, - jwt_token=None, - server_url=None, -) -> Dict[str, Any]: - """Create a new project via GNS3 REST API.""" +def create_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: name = params.get("name") if not name: return {"error": "name is required"} - - conn = _get_connector(server_url, jwt_token) + conn = _get_connector(gns3_ctx) project_data = {"name": name} if "description" in params: project_data["description"] = params["description"] - - project = conn.create_project(**project_data) - return project + return conn.create_project(**project_data) -# ── Tool: delete_project ─────────────────────────────────────────────────── - -def delete_project_handler( - params: Dict[str, Any], - current_user=None, - jwt_token=None, - server_url=None, -) -> Dict[str, Any]: - """Delete a project by project_id via GNS3 REST API.""" +def delete_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(server_url, jwt_token) + conn = _get_connector(gns3_ctx) conn.delete_project(project_id=project_id) return {"message": f"Project '{project_id}' deleted", "project_id": project_id} -# ── Tool: open_project ───────────────────────────────────────────────────── - -def open_project_handler( - params: Dict[str, Any], - current_user=None, - jwt_token=None, - server_url=None, -) -> Dict[str, Any]: - """Open a closed project via GNS3 REST API.""" +def open_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(server_url, jwt_token) + conn = _get_connector(gns3_ctx) url = f"{conn.base_url}/projects/{project_id}/open" - response = conn.http_call("post", url) - return response.json() + return conn.http_call("post", url).json() -# ── Tool: close_project ──────────────────────────────────────────────────── - -def close_project_handler( - params: Dict[str, Any], - current_user=None, - jwt_token=None, - server_url=None, -) -> Dict[str, Any]: - """Close an open project via GNS3 REST API.""" +def close_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(server_url, jwt_token) + conn = _get_connector(gns3_ctx) url = f"{conn.base_url}/projects/{project_id}/close" conn.http_call("post", url) return {"message": f"Project '{project_id}' closed", "project_id": project_id} -# ── Tool: get_project_stats ──────────────────────────────────────────────── - -def get_project_stats_handler( - params: Dict[str, Any], - current_user=None, - jwt_token=None, - server_url=None, -) -> Dict[str, Any]: - """Return project statistics via GNS3 REST API.""" +def get_project_stats_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(server_url, jwt_token) + conn = _get_connector(gns3_ctx) url = f"{conn.base_url}/projects/{project_id}/stats" - response = conn.http_call("get", url) - return response.json() + return conn.http_call("get", url).json() -# ── Register all project tools ───────────────────────────────────────────── +# ── Tool definitions (consumed by mcp/__init__.py) ───────────────────────── -def register_tools(): - """Register every project-related MCP tool into the global registry.""" - tools = [ - MCPTool( - name="list_projects", - description="List all GNS3 projects accessible to the current user", - parameters_schema={"type": "object", "properties": {}}, - handler=list_projects_handler, - ), - MCPTool( - name="get_project", - description="Get detailed information about a specific project", - parameters_schema={ - "type": "object", - "properties": { - "project_id": { - "type": "string", - "description": "Project UUID", - } - }, - "required": ["project_id"], +PROJECT_TOOLS = [ + { + "name": "list_projects", + "description": "List all GNS3 projects accessible to the current user", + "parameters": {"type": "object", "properties": {}}, + "handler": list_projects_handler, + }, + { + "name": "get_project", + "description": "Get detailed information about a specific project", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, }, - handler=get_project_handler, - ), - MCPTool( - name="create_project", - description="Create a new GNS3 project", - parameters_schema={ - "type": "object", - "properties": { - "name": {"type": "string", "description": "Project name"}, - "description": { - "type": "string", - "description": "Optional project description", - }, - }, - "required": ["name"], + "required": ["project_id"], + }, + "handler": get_project_handler, + }, + { + "name": "create_project", + "description": "Create a new GNS3 project", + "parameters": { + "type": "object", + "properties": { + "name": {"type": "string", "description": "Project name"}, + "description": {"type": "string", "description": "Optional project description"}, }, - handler=create_project_handler, - ), - MCPTool( - name="delete_project", - description="Delete a GNS3 project permanently", - parameters_schema={ - "type": "object", - "properties": { - "project_id": { - "type": "string", - "description": "UUID of the project to delete", - } - }, - "required": ["project_id"], + "required": ["name"], + }, + "handler": create_project_handler, + }, + { + "name": "delete_project", + "description": "Delete a GNS3 project permanently", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "UUID of the project to delete"}, }, - handler=delete_project_handler, - ), - MCPTool( - name="open_project", - description="Open a closed GNS3 project", - parameters_schema={ - "type": "object", - "properties": { - "project_id": { - "type": "string", - "description": "Project UUID", - } - }, - "required": ["project_id"], + "required": ["project_id"], + }, + "handler": delete_project_handler, + }, + { + "name": "open_project", + "description": "Open a closed GNS3 project", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, }, - handler=open_project_handler, - ), - MCPTool( - name="close_project", - description="Close an open GNS3 project", - parameters_schema={ - "type": "object", - "properties": { - "project_id": { - "type": "string", - "description": "Project UUID", - } - }, - "required": ["project_id"], + "required": ["project_id"], + }, + "handler": open_project_handler, + }, + { + "name": "close_project", + "description": "Close an open GNS3 project", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, }, - handler=close_project_handler, - ), - MCPTool( - name="get_project_stats", - description="Get statistics (nodes, links, snapshots, drawings) for a project", - parameters_schema={ - "type": "object", - "properties": { - "project_id": { - "type": "string", - "description": "Project UUID", - } - }, - "required": ["project_id"], + "required": ["project_id"], + }, + "handler": close_project_handler, + }, + { + "name": "get_project_stats", + "description": "Get statistics (nodes, links, snapshots, drawings) for a project", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, }, - handler=get_project_stats_handler, - ), - ] - - for tool in tools: - registry.register_tool(tool) - - -# Auto-register on import -register_tools() + "required": ["project_id"], + }, + "handler": get_project_stats_handler, + }, +] diff --git a/gns3server/api/server.py b/gns3server/api/server.py index 54e6eb51d..5def647cc 100644 --- a/gns3server/api/server.py +++ b/gns3server/api/server.py @@ -83,6 +83,9 @@ def get_application() -> FastAPI: app = get_application() +# Register MCP SSE transport routes (Starlette-level, for raw ASGI access) +mcp.register_starlette_routes(app) + # Monkey Patch uvicorn signal handler to detect the application is shutting down app.state.exiting = False unicorn_exit_handler = UvicornServer.handle_exit From 19e7533cd77a54147f2f5867c328ef872b4b72f8 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Thu, 4 Jun 2026 22:38:49 +0800 Subject: [PATCH 03/20] feat: support Authorization header and query param for MCP token - SSE endpoint supports both Authorization: Bearer header and ?token= query param - Claude Code can use headers (no URL exposure) - Claude Desktop (EventSource) can use ?token= URL param --- gns3server/api/routes/mcp/__init__.py | 36 +++++++++++++++++++-------- 1 file changed, 25 insertions(+), 11 deletions(-) diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index e50931872..6dcfbe56e 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -171,23 +171,37 @@ async def get_project_stats(project_id: str) -> list[dict[str, Any]]: # ── Auth‑wrapped SSE app ────────────────────────────────────────────── def _make_auth_wrapper(sse_app): - """Wrap the SSE app with JWT validation from ?token= query parameter. + """Wrap the SSE app with JWT validation. - The wrapper intercepts GET requests (SSE connections), validates the - JWT token, and stores it in a context variable so tool handlers can - use it to call the GNS3 REST API. POST messages are passed through - unchanged (they are authenticated by their session association). + Supports two ways to pass the token (checked in order): + 1. Authorization: Bearer header + 2. ?token= query parameter + + POST messages are passed through (authenticated by their session). """ async def auth_wrapper(scope, receive, send): if scope["type"] == "http" and scope["method"] == "GET": - params = parse_qs(scope.get("query_string", b"").decode()) - tokens = params.get("token", []) - if not tokens or not await _validate_token(tokens[0]): + token = None + + # 1. Try Authorization header first + headers = dict(scope.get("headers", [])) + auth_header = headers.get(b"authorization", b"").decode() + if auth_header.startswith("Bearer "): + token = auth_header[7:] + + # 2. Fall back to ?token= query param + if not token: + params = parse_qs(scope.get("query_string", b"").decode()) + tokens = params.get("token", []) + if tokens: + token = tokens[0] + + if not token or not await _validate_token(token): response = Response("Missing or invalid token", status_code=401) await response(scope, receive, send) return - _jwt_token_var.set(tokens[0]) + _jwt_token_var.set(token) await sse_app(scope, receive, send) return auth_wrapper @@ -206,9 +220,9 @@ async def mcp_root(): "version": "1.0.0", "protocol": "Model Context Protocol", "transport": "SSE", - "authentication": "?token=", + "authentication": ["Authorization: Bearer ", "?token="], "endpoints": { - "sse": "/v3/mcp/transport/sse?token=", + "sse": "/v3/mcp/transport/sse", "messages": "/v3/mcp/transport/messages/", }, } From 1e9b3d58793a0e218790427d2037b5439ad7e018 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Thu, 4 Jun 2026 23:06:20 +0800 Subject: [PATCH 04/20] feat: complete MCP SSE transport with JWT auth - SSE endpoint at /v3/mcp/transport/sse with token auth - Supports Authorization: Bearer header and ?token= query param - JWT validated via GNS3 auth_service, stored in contextvars - Tool handlers use GNS3 REST API via Gns3Connector with JWT token - 7 project tools: list_projects, get_project, create_project, delete_project, open_project, close_project, get_project_stats - Claude Code: claude mcp add --transport sse ... -H 'Authorization: Bearer ' - Claude Desktop: SSE URL with ?token= --- gns3server/api/routes/mcp/__init__.py | 27 +++++++++------------------ 1 file changed, 9 insertions(+), 18 deletions(-) diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index 6dcfbe56e..99fe310f6 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -170,7 +170,7 @@ async def get_project_stats(project_id: str) -> list[dict[str, Any]]: # ── Auth‑wrapped SSE app ────────────────────────────────────────────── -def _make_auth_wrapper(sse_app): +def _make_auth_wrapper(inner_app): """Wrap the SSE app with JWT validation. Supports two ways to pass the token (checked in order): @@ -183,26 +183,21 @@ def _make_auth_wrapper(sse_app): async def auth_wrapper(scope, receive, send): if scope["type"] == "http" and scope["method"] == "GET": token = None - - # 1. Try Authorization header first headers = dict(scope.get("headers", [])) - auth_header = headers.get(b"authorization", b"").decode() - if auth_header.startswith("Bearer "): - token = auth_header[7:] - - # 2. Fall back to ?token= query param + auth = headers.get(b"authorization", b"").decode() + if auth.startswith("Bearer "): + token = auth[7:] if not token: params = parse_qs(scope.get("query_string", b"").decode()) tokens = params.get("token", []) if tokens: token = tokens[0] - if not token or not await _validate_token(token): response = Response("Missing or invalid token", status_code=401) await response(scope, receive, send) return _jwt_token_var.set(token) - await sse_app(scope, receive, send) + await inner_app(scope, receive, send) return auth_wrapper @@ -218,19 +213,15 @@ async def mcp_root(): return { "name": "GNS3 MCP Server", "version": "1.0.0", - "protocol": "Model Context Protocol", - "transport": "SSE", "authentication": ["Authorization: Bearer ", "?token="], - "endpoints": { + "transports": { "sse": "/v3/mcp/transport/sse", - "messages": "/v3/mcp/transport/messages/", }, } def register_starlette_routes(app): - """Mount the authenticated SSE app under /v3/mcp/transport.""" - raw_sse_app = mcp.sse_app(mount_path="") - wrapped = _make_auth_wrapper(raw_sse_app) - app.mount("/v3/mcp/transport", wrapped, name="mcp-sse") + """Mount MCP transports on the FastAPI app.""" + sse_app = _make_auth_wrapper(mcp.sse_app(mount_path="")) + app.mount("/v3/mcp/transport", sse_app, name="mcp-sse") log.info("MCP SSE server mounted at /v3/mcp/transport") From ccdd307d5404fcd6e72dbe29adba89b1da802fd2 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Thu, 4 Jun 2026 23:09:08 +0800 Subject: [PATCH 05/20] docs: add MCP service feature documentation --- docs/features/mcp-service.md | 136 +++++++++++++++++++++++++++++++++++ 1 file changed, 136 insertions(+) create mode 100644 docs/features/mcp-service.md diff --git a/docs/features/mcp-service.md b/docs/features/mcp-service.md new file mode 100644 index 000000000..9bba8337e --- /dev/null +++ b/docs/features/mcp-service.md @@ -0,0 +1,136 @@ +# MCP (Model Context Protocol) Service + +## Overview + +GNS3 Server provides a standard [Model Context Protocol (MCP)](https://modelcontextprotocol.io/) interface, allowing AI assistants like Claude to interact with GNS3 network simulations through SSE (Server-Sent Events) transport. + +The MCP service exposes GNS3 project management operations as MCP tools that can be discovered and called by MCP clients. + +## Endpoints + +| Path | Method | Description | +|------|--------|-------------| +| `/v3/mcp/` | GET | MCP service metadata | +| `/v3/mcp/transport/sse` | GET | SSE stream (MCP connection) | +| `/v3/mcp/transport/messages/` | POST | JSON-RPC messages | + +## Authentication + +The SSE endpoint requires a valid GNS3 JWT token. It supports two ways to pass the token: + +1. **Authorization header** (recommended for Claude Code): + ``` + Authorization: Bearer + ``` + +2. **Query parameter** (required for Claude Desktop, since EventSource does not support custom headers): + ``` + GET /v3/mcp/transport/sse?token= + ``` + +### Getting a Token + +```bash +curl -X POST http://localhost:3080/v3/access/users/authenticate \ + -H "Content-Type: application/json" \ + -d '{"username": "admin", "password": "admin"}' +``` + +### Token Expiry + +Default JWT token lifetime is **1440 minutes (24 hours)**. This can be configured in `gns3_server.conf`: + +```ini +jwt_access_token_expire_minutes = 1440 ; 24 hours +``` + +## Available Tools + +| Tool | Description | Required Parameters | +|------|-------------|-------------------| +| `list_projects` | List all projects | none | +| `get_project` | Get project details | `project_id` | +| `create_project` | Create a project | `name` | +| `delete_project` | Delete a project | `project_id` | +| `open_project` | Open a project | `project_id` | +| `close_project` | Close a project | `project_id` | +| `get_project_stats` | Get project statistics | `project_id` | + +## Configuration + +### Claude Code (CLI) + +```bash +# Get a JWT token +TOKEN=$(curl -s -X POST http://localhost:3080/v3/access/users/authenticate \ + -H "Content-Type: application/json" \ + -d '{"username": "admin", "password": "admin"}' | python3 -c \ + "import sys,json; print(json.load(sys.stdin)['access_token'])") + +# Add MCP server +claude mcp add --transport sse My_GNS3_Server \ + http://localhost:3080/v3/mcp/transport/sse \ + -H "Authorization: Bearer $TOKEN" +``` + +### Claude Desktop + +Add to `claude_desktop_config.json`: + +```json +{ + "mcpServers": { + "My_GNS3_Server": { + "url": "http://localhost:3080/v3/mcp/transport/sse?token=your_jwt_token" + } + } +} +``` + +## Architecture + +```mermaid +sequenceDiagram + participant Client as Claude Code / Claude Desktop + participant MCP as MCP Service
(/v3/mcp/transport) + participant Auth as JWT Auth + participant GNS3 as GNS3 REST API + participant DB as Controller / Database + + Note over Client: Step 1: Connect with JWT + Client->>MCP: GET /sse?token=<jwt>
or Authorization: Bearer <jwt> + + MCP->>Auth: Validate Token + Auth-->>MCP: Token Valid + MCP-->>Client: event: endpoint
data: /messages/?session_id=xxx + + Note over Client: Step 2: Initialize Protocol + Client->>MCP: POST /messages/?session_id=xxx
{"method":"initialize", ...} + MCP-->>Client: event: message
{"result": {"protocolVersion": "...", ...}} + + Note over Client: Step 3: List & Call Tools + Client->>MCP: POST /messages/
{"method":"tools/list"} + MCP-->>Client: event: message
{"result": {"tools": [...]}} + + Client->>MCP: POST /messages/
{"method":"tools/call",
"params": {"name":"list_projects"}} + + MCP->>GNS3: Gns3Connector (HTTP) + GNS3->>DB: Query Projects + DB-->>GNS3: Project Data + GNS3-->>MCP: JSON Response + MCP-->>Client: event: message
{"result": {"content": [...]}} +``` + +## Internal Implementation + +- **FastMCP** (Anthropic MCP SDK) is used for tool registration and SSE transport +- The SSE app is mounted as a Starlette sub-application under `/v3/mcp/transport` +- JWT tokens are validated using GNS3's existing `auth_service` +- Tool handlers use `Gns3Connector` (from `custom_gns3fy`) to call GNS3's own REST API, keeping the MCP layer decoupled +- The JWT token is stored in a `contextvars.ContextVar` so it is available within tool handler threads (Python ≥ 3.9 propagates contextvars through `asyncio.to_thread`) + +### Source Files + +- `gns3server/api/routes/mcp/__init__.py` — FastMCP server, tool definitions, SSE transport, JWT auth wrapper +- `gns3server/api/routes/mcp/projects.py` — Project tool handlers using Gns3Connector +- `gns3server/api/server.py` — Mounts MCP routes via `register_starlette_routes()` From 5b23ac81d0aab3182f9e1c45fb7d060a4e5b1386 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Thu, 4 Jun 2026 23:09:44 +0800 Subject: [PATCH 06/20] docs: add MCP service feature documentation --- docs/features/mcp-service.md | 33 ++++++++++++++------------------- 1 file changed, 14 insertions(+), 19 deletions(-) diff --git a/docs/features/mcp-service.md b/docs/features/mcp-service.md index 9bba8337e..01bfc26f7 100644 --- a/docs/features/mcp-service.md +++ b/docs/features/mcp-service.md @@ -92,33 +92,28 @@ Add to `claude_desktop_config.json`: ```mermaid sequenceDiagram participant Client as Claude Code / Claude Desktop - participant MCP as MCP Service
(/v3/mcp/transport) + participant MCP as MCP Service participant Auth as JWT Auth participant GNS3 as GNS3 REST API - participant DB as Controller / Database - - Note over Client: Step 1: Connect with JWT - Client->>MCP: GET /sse?token=<jwt>
or Authorization: Bearer <jwt> + Note over Client: 1. Connect with JWT + Client->>MCP: GET /sse (token in header or query) MCP->>Auth: Validate Token Auth-->>MCP: Token Valid - MCP-->>Client: event: endpoint
data: /messages/?session_id=xxx + MCP-->>Client: event: endpoint /messages/?session_id=xxx - Note over Client: Step 2: Initialize Protocol - Client->>MCP: POST /messages/?session_id=xxx
{"method":"initialize", ...} - MCP-->>Client: event: message
{"result": {"protocolVersion": "...", ...}} + Note over Client: 2. Initialize + Client->>MCP: POST /messages/ (initialize) + MCP-->>Client: event: message (protocolVersion, capabilities) - Note over Client: Step 3: List & Call Tools - Client->>MCP: POST /messages/
{"method":"tools/list"} - MCP-->>Client: event: message
{"result": {"tools": [...]}} + Note over Client: 3. List & Call Tools + Client->>MCP: POST /messages/ (tools/list) + MCP-->>Client: event: message (tools list) - Client->>MCP: POST /messages/
{"method":"tools/call",
"params": {"name":"list_projects"}} - - MCP->>GNS3: Gns3Connector (HTTP) - GNS3->>DB: Query Projects - DB-->>GNS3: Project Data - GNS3-->>MCP: JSON Response - MCP-->>Client: event: message
{"result": {"content": [...]}} + Client->>MCP: POST /messages/ (tools/call list_projects) + MCP->>GNS3: Gns3Connector HTTP request + GNS3-->>MCP: Projects data + MCP-->>Client: event: message (tool result) ``` ## Internal Implementation From cad216705a852c226c5c6e96a7964c8af990aa97 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Thu, 4 Jun 2026 23:19:00 +0800 Subject: [PATCH 07/20] feat: add Node and Link MCP tools, update copyright - Add 9 node tools and 5 link tools - Update copyright year to 2026, add author --- gns3server/api/routes/mcp/__init__.py | 178 ++++++++++++++++- gns3server/api/routes/mcp/links.py | 184 ++++++++++++++++++ gns3server/api/routes/mcp/nodes.py | 267 ++++++++++++++++++++++++++ gns3server/api/routes/mcp/projects.py | 3 +- 4 files changed, 630 insertions(+), 2 deletions(-) create mode 100644 gns3server/api/routes/mcp/links.py create mode 100644 gns3server/api/routes/mcp/nodes.py diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index 99fe310f6..d728a348c 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -1,5 +1,6 @@ # -# Copyright (C) 2020 GNS3 Technologies Inc. +# 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 @@ -168,6 +169,181 @@ async def get_project_stats(project_id: str) -> list[dict[str, Any]]: return await asyncio.to_thread(_run_handler_sync, get_project_stats_handler, {"project_id": project_id}) +# ── Node tools ──────────────────────────────────────────────────────── + +@mcp.tool() +async def get_nodes(project_id: str) -> list[dict[str, Any]]: + """List all nodes in a project.""" + from .nodes import get_nodes_handler + return await asyncio.to_thread(_run_handler_sync, get_nodes_handler, {"project_id": project_id}) + + +@mcp.tool() +async def get_node(project_id: str, node_id: str) -> list[dict[str, Any]]: + """Get detailed information about a specific node. + + Args: + project_id: Project UUID + node_id: Node UUID + """ + from .nodes import get_node_handler + return await asyncio.to_thread(_run_handler_sync, get_node_handler, {"project_id": project_id, "node_id": node_id}) + + +@mcp.tool() +async def start_node(project_id: str, node_id: str) -> list[dict[str, Any]]: + """Start a node in a project. + + Args: + project_id: Project UUID + node_id: Node UUID + """ + from .nodes import start_node_handler + return await asyncio.to_thread(_run_handler_sync, start_node_handler, {"project_id": project_id, "node_id": node_id}) + + +@mcp.tool() +async def stop_node(project_id: str, node_id: str) -> list[dict[str, Any]]: + """Stop a node in a project. + + Args: + project_id: Project UUID + node_id: Node UUID + """ + from .nodes import stop_node_handler + return await asyncio.to_thread(_run_handler_sync, stop_node_handler, {"project_id": project_id, "node_id": node_id}) + + +@mcp.tool() +async def reload_node(project_id: str, node_id: str) -> list[dict[str, Any]]: + """Reload (restart) a node in a project. + + Args: + project_id: Project UUID + node_id: Node UUID + """ + from .nodes import reload_node_handler + return await asyncio.to_thread(_run_handler_sync, reload_node_handler, {"project_id": project_id, "node_id": node_id}) + + +@mcp.tool() +async def suspend_node(project_id: str, node_id: str) -> list[dict[str, Any]]: + """Suspend a node in a project. + + Args: + project_id: Project UUID + node_id: Node UUID + """ + from .nodes import suspend_node_handler + return await asyncio.to_thread(_run_handler_sync, suspend_node_handler, {"project_id": project_id, "node_id": node_id}) + + +@mcp.tool() +async def create_node(project_id: str, template_id: str, x: int = 0, y: int = 0, compute_id: str = "local") -> list[dict[str, Any]]: + """Create a new node from a template in a project. + + Args: + project_id: Project UUID + template_id: Template UUID + x: X coordinate (optional) + y: Y coordinate (optional) + compute_id: Compute ID (optional, default: local) + """ + from .nodes import create_node_handler + return await asyncio.to_thread(_run_handler_sync, create_node_handler, { + "project_id": project_id, "template_id": template_id, + "x": x, "y": y, "compute_id": compute_id, + }) + + +@mcp.tool() +async def delete_node(project_id: str, node_id: str) -> list[dict[str, Any]]: + """Delete a node from a project. + + Args: + project_id: Project UUID + node_id: Node UUID + """ + from .nodes import delete_node_handler + return await asyncio.to_thread(_run_handler_sync, delete_node_handler, {"project_id": project_id, "node_id": node_id}) + + +@mcp.tool() +async def update_node(project_id: str, node_id: str, **kwargs: Any) -> list[dict[str, Any]]: + """Update a node's properties (name, position, etc.). + + Args: + project_id: Project UUID + node_id: Node UUID + """ + from .nodes import update_node_handler + params = {"project_id": project_id, "node_id": node_id, **kwargs} + return await asyncio.to_thread(_run_handler_sync, update_node_handler, params) + + +# ── Link tools ──────────────────────────────────────────────────────── + +@mcp.tool() +async def get_links(project_id: str) -> list[dict[str, Any]]: + """List all links in a project.""" + from .links import get_links_handler + return await asyncio.to_thread(_run_handler_sync, get_links_handler, {"project_id": project_id}) + + +@mcp.tool() +async def get_link(project_id: str, link_id: str) -> list[dict[str, Any]]: + """Get detailed information about a specific link. + + Args: + project_id: Project UUID + link_id: Link UUID + """ + from .links import get_link_handler + return await asyncio.to_thread(_run_handler_sync, get_link_handler, {"project_id": project_id, "link_id": link_id}) + + +@mcp.tool() +async def create_link(project_id: str, nodes: list, link_type: str = "ethernet", filters: dict = None) -> list[dict[str, Any]]: + """Create a link between two nodes in a project. + + Args: + project_id: Project UUID + nodes: List of node connections, e.g. [{"node_id": "...", "adapter_number": 0, "port_number": 0}, ...] + link_type: Link type - ethernet or serial (optional) + filters: Packet filters (optional) + """ + from .links import create_link_handler + params = {"project_id": project_id, "nodes": nodes, "link_type": link_type} + if filters: + params["filters"] = filters + return await asyncio.to_thread(_run_handler_sync, create_link_handler, params) + + +@mcp.tool() +async def delete_link(project_id: str, link_id: str) -> list[dict[str, Any]]: + """Delete a link from a project. + + Args: + project_id: Project UUID + link_id: Link UUID + """ + from .links import delete_link_handler + return await asyncio.to_thread(_run_handler_sync, delete_link_handler, {"project_id": project_id, "link_id": link_id}) + + +@mcp.tool() +async def update_link(project_id: str, link_id: str, **kwargs: Any) -> list[dict[str, Any]]: + """Update a link's properties (suspend, filters, etc.). + + Args: + project_id: Project UUID + link_id: Link UUID + """ + from .links import update_link_handler + params = {"project_id": project_id, "link_id": link_id, **kwargs} + return await asyncio.to_thread(_run_handler_sync, update_link_handler, params) + + # ── Auth‑wrapped SSE app ────────────────────────────────────────────── def _make_auth_wrapper(inner_app): diff --git a/gns3server/api/routes/mcp/links.py b/gns3server/api/routes/mcp/links.py new file mode 100644 index 000000000..2b9278ba0 --- /dev/null +++ b/gns3server/api/routes/mcp/links.py @@ -0,0 +1,184 @@ +# +# 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 link management. + +Handlers receive (params, gns3_ctx) and call GNS3's REST API +via Gns3Connector (from custom_gns3fy). +""" + +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_links_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) + links = conn.get_links(project_id=project_id) + return {"links": links, "count": len(links)} + + +def get_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + link_id = params.get("link_id") + if not project_id or not link_id: + return {"error": "project_id and link_id are required"} + conn = _get_connector(gns3_ctx) + return conn.get_link(project_id=project_id, link_id=link_id) + + +def create_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + nodes = params.get("nodes") + if not project_id or not nodes: + return {"error": "project_id and nodes are required"} + conn = _get_connector(gns3_ctx) + data = {"nodes": nodes} + if "link_type" in params: + data["link_type"] = params["link_type"] + if "filters" in params: + data["filters"] = params["filters"] + if "suspend" in params: + data["suspend"] = params["suspend"] + url = f"{conn.base_url}/projects/{project_id}/links" + return conn.http_call("post", url, json_data=data).json() + + +def delete_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + link_id = params.get("link_id") + if not project_id or not link_id: + return {"error": "project_id and link_id are required"} + conn = _get_connector(gns3_ctx) + conn.http_call("delete", f"{conn.base_url}/projects/{project_id}/links/{link_id}") + return {"message": f"Link {link_id} deleted", "link_id": link_id} + + +def update_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + link_id = params.get("link_id") + if not project_id or not link_id: + return {"error": "project_id and link_id are required"} + conn = _get_connector(gns3_ctx) + update_data = {k: v for k, v in params.items() if k not in ("project_id", "link_id")} + url = f"{conn.base_url}/projects/{project_id}/links/{link_id}" + return conn.http_call("put", url, json_data=update_data).json() + + +# ── Tool definitions ─────────────────────────────────────────────────────── + +LINK_TOOLS = [ + { + "name": "get_links", + "description": "List all links in a project", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + }, + "required": ["project_id"], + }, + "handler": get_links_handler, + }, + { + "name": "get_link", + "description": "Get detailed information about a specific link", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "link_id": {"type": "string", "description": "Link UUID"}, + }, + "required": ["project_id", "link_id"], + }, + "handler": get_link_handler, + }, + { + "name": "create_link", + "description": "Create a link between two nodes in a project", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "nodes": { + "type": "array", + "description": "List of node connections, each with node_id, adapter_number, port_number", + "items": { + "type": "object", + "properties": { + "node_id": {"type": "string"}, + "adapter_number": {"type": "integer"}, + "port_number": {"type": "integer"}, + }, + }, + }, + "link_type": {"type": "string", "description": "Link type: ethernet or serial (optional)"}, + "filters": {"type": "object", "description": "Packet filters (optional)"}, + }, + "required": ["project_id", "nodes"], + }, + "handler": create_link_handler, + }, + { + "name": "delete_link", + "description": "Delete a link from a project", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "link_id": {"type": "string", "description": "Link UUID"}, + }, + "required": ["project_id", "link_id"], + }, + "handler": delete_link_handler, + }, + { + "name": "update_link", + "description": "Update a link's properties (suspend, filters, etc.)", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "link_id": {"type": "string", "description": "Link UUID"}, + "suspend": {"type": "boolean", "description": "Suspend the link (optional)"}, + "filters": {"type": "object", "description": "Packet filters (optional)"}, + }, + "required": ["project_id", "link_id"], + }, + "handler": update_link_handler, + }, +] diff --git a/gns3server/api/routes/mcp/nodes.py b/gns3server/api/routes/mcp/nodes.py new file mode 100644 index 000000000..87378f6c3 --- /dev/null +++ b/gns3server/api/routes/mcp/nodes.py @@ -0,0 +1,267 @@ +# +# 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 node management. + +Handlers receive (params, gns3_ctx) and call GNS3's REST API +via Gns3Connector (from custom_gns3fy). +""" + +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_nodes_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) + nodes = conn.get_nodes(project_id=project_id) + return {"nodes": nodes, "count": len(nodes)} + + +def get_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + node_id = params.get("node_id") + if not project_id or not node_id: + return {"error": "project_id and node_id are required"} + conn = _get_connector(gns3_ctx) + return conn.get_node(project_id=project_id, node_id=node_id) + + +def start_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + node_id = params.get("node_id") + if not project_id or not node_id: + return {"error": "project_id and node_id are required"} + conn = _get_connector(gns3_ctx) + conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/start", json_data={}) + return {"message": f"Node {node_id} started", "node_id": node_id} + + +def stop_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + node_id = params.get("node_id") + if not project_id or not node_id: + return {"error": "project_id and node_id are required"} + conn = _get_connector(gns3_ctx) + conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/stop", json_data={}) + return {"message": f"Node {node_id} stopped", "node_id": node_id} + + +def reload_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + node_id = params.get("node_id") + if not project_id or not node_id: + return {"error": "project_id and node_id are required"} + conn = _get_connector(gns3_ctx) + conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/reload") + return {"message": f"Node {node_id} reloaded", "node_id": node_id} + + +def suspend_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + node_id = params.get("node_id") + if not project_id or not node_id: + return {"error": "project_id and node_id are required"} + conn = _get_connector(gns3_ctx) + conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/suspend") + return {"message": f"Node {node_id} suspended", "node_id": node_id} + + +def create_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + template_id = params.get("template_id") + if not project_id or not template_id: + return {"error": "project_id and template_id are required"} + conn = _get_connector(gns3_ctx) + data = { + "x": params.get("x", 0), + "y": params.get("y", 0), + "compute_id": params.get("compute_id", "local"), + } + url = f"{conn.base_url}/projects/{project_id}/templates/{template_id}" + return conn.http_call("post", url, json_data=data).json() + + +def delete_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + node_id = params.get("node_id") + if not project_id or not node_id: + return {"error": "project_id and node_id are required"} + conn = _get_connector(gns3_ctx) + conn.http_call("delete", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}") + return {"message": f"Node {node_id} deleted", "node_id": node_id} + + +def update_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + node_id = params.get("node_id") + if not project_id or not node_id: + return {"error": "project_id and node_id are required"} + conn = _get_connector(gns3_ctx) + # Pass all params except project_id/node_id as update fields + update_data = {k: v for k, v in params.items() if k not in ("project_id", "node_id")} + url = f"{conn.base_url}/projects/{project_id}/nodes/{node_id}" + return conn.http_call("put", url, json_data=update_data).json() + + +# ── Tool definitions ─────────────────────────────────────────────────────── + +NODE_TOOLS = [ + { + "name": "get_nodes", + "description": "List all nodes in a project", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + }, + "required": ["project_id"], + }, + "handler": get_nodes_handler, + }, + { + "name": "get_node", + "description": "Get detailed information about a specific node", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "node_id": {"type": "string", "description": "Node UUID"}, + }, + "required": ["project_id", "node_id"], + }, + "handler": get_node_handler, + }, + { + "name": "start_node", + "description": "Start a node in a project", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "node_id": {"type": "string", "description": "Node UUID"}, + }, + "required": ["project_id", "node_id"], + }, + "handler": start_node_handler, + }, + { + "name": "stop_node", + "description": "Stop a node in a project", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "node_id": {"type": "string", "description": "Node UUID"}, + }, + "required": ["project_id", "node_id"], + }, + "handler": stop_node_handler, + }, + { + "name": "reload_node", + "description": "Reload (restart) a node in a project", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "node_id": {"type": "string", "description": "Node UUID"}, + }, + "required": ["project_id", "node_id"], + }, + "handler": reload_node_handler, + }, + { + "name": "suspend_node", + "description": "Suspend a node in a project", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "node_id": {"type": "string", "description": "Node UUID"}, + }, + "required": ["project_id", "node_id"], + }, + "handler": suspend_node_handler, + }, + { + "name": "create_node", + "description": "Create a new node from a template in a project", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "template_id": {"type": "string", "description": "Template UUID"}, + "x": {"type": "integer", "description": "X coordinate (optional)"}, + "y": {"type": "integer", "description": "Y coordinate (optional)"}, + "compute_id": {"type": "string", "description": "Compute ID (optional, default: local)"}, + }, + "required": ["project_id", "template_id"], + }, + "handler": create_node_handler, + }, + { + "name": "delete_node", + "description": "Delete a node from a project", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "node_id": {"type": "string", "description": "Node UUID"}, + }, + "required": ["project_id", "node_id"], + }, + "handler": delete_node_handler, + }, + { + "name": "update_node", + "description": "Update a node's properties (name, position, etc.)", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "node_id": {"type": "string", "description": "Node UUID"}, + "name": {"type": "string", "description": "New node name (optional)"}, + "x": {"type": "integer", "description": "New X position (optional)"}, + "y": {"type": "integer", "description": "New Y position (optional)"}, + "compute_id": {"type": "string", "description": "Compute ID (optional)"}, + }, + "required": ["project_id", "node_id"], + }, + "handler": update_node_handler, + }, +] diff --git a/gns3server/api/routes/mcp/projects.py b/gns3server/api/routes/mcp/projects.py index c5a6cbb4b..daec65ceb 100644 --- a/gns3server/api/routes/mcp/projects.py +++ b/gns3server/api/routes/mcp/projects.py @@ -1,5 +1,6 @@ # -# Copyright (C) 2020 GNS3 Technologies Inc. +# 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 From 6889f5173765cdf4137f6bbec60fedc2cbdda7b7 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Thu, 4 Jun 2026 23:21:01 +0800 Subject: [PATCH 08/20] feat: add 5 Template MCP tools Add list_templates, get_template, create_template, update_template, delete_template. Total MCP tools: 26. --- gns3server/api/routes/mcp/__init__.py | 65 +++++++++++ gns3server/api/routes/mcp/templates.py | 156 +++++++++++++++++++++++++ 2 files changed, 221 insertions(+) create mode 100644 gns3server/api/routes/mcp/templates.py diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index d728a348c..75c67a77b 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -344,6 +344,71 @@ async def update_link(project_id: str, link_id: str, **kwargs: Any) -> list[dict return await asyncio.to_thread(_run_handler_sync, update_link_handler, params) +# ── Template tools ──────────────────────────────────────────────────── + +@mcp.tool() +async def list_templates() -> list[dict[str, Any]]: + """List all available templates on the server.""" + from .templates import list_templates_handler + return await asyncio.to_thread(_run_handler_sync, list_templates_handler, {}) + + +@mcp.tool() +async def get_template(template_id: str = None, name: str = None) -> list[dict[str, Any]]: + """Get detailed information about a specific template. + + Args: + template_id: Template UUID (optional if name is provided) + name: Template name (optional if template_id is provided) + """ + from .templates import get_template_handler + return await asyncio.to_thread(_run_handler_sync, get_template_handler, { + "template_id": template_id, "name": name, + }) + + +@mcp.tool() +async def create_template(name: str, template_type: str, compute_id: str = "local") -> list[dict[str, Any]]: + """Create a new template. + + Args: + name: Template name + template_type: Template type (e.g. qemu, docker, dynamips) + compute_id: Compute ID (optional, default: local) + """ + from .templates import create_template_handler + return await asyncio.to_thread(_run_handler_sync, create_template_handler, { + "name": name, "template_type": template_type, "compute_id": compute_id, + }) + + +@mcp.tool() +async def update_template(template_id: str = None, name: str = None, **kwargs: Any) -> list[dict[str, Any]]: + """Update an existing template's properties. + + Args: + template_id: Template UUID (optional if name is provided) + name: Template name (optional if template_id is provided) + """ + from .templates import update_template_handler + params = {"template_id": template_id, "name": name, **kwargs} + return await asyncio.to_thread(_run_handler_sync, update_template_handler, params) + + +@mcp.tool() +async def delete_template(template_id: str = None, name: str = None) -> list[dict[str, Any]]: + """Delete a template. + + Args: + template_id: Template UUID (optional if name is provided) + name: Template name (optional if template_id is provided) + """ + from .templates import delete_template_handler + return await asyncio.to_thread(_run_handler_sync, delete_template_handler, { + "template_id": template_id, "name": name, + }) + + # ── Auth‑wrapped SSE app ────────────────────────────────────────────── def _make_auth_wrapper(inner_app): diff --git a/gns3server/api/routes/mcp/templates.py b/gns3server/api/routes/mcp/templates.py new file mode 100644 index 000000000..b5df86faf --- /dev/null +++ b/gns3server/api/routes/mcp/templates.py @@ -0,0 +1,156 @@ +# +# 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 template management. + +Handlers receive (params, gns3_ctx) and call GNS3's REST API +via Gns3Connector (from custom_gns3fy). +""" + +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 list_templates_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + conn = _get_connector(gns3_ctx) + templates = conn.get_templates() + return {"templates": templates, "count": len(templates)} + + +def get_template_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + template_id = params.get("template_id") + name = params.get("name") + if not template_id and not name: + return {"error": "template_id or name is required"} + conn = _get_connector(gns3_ctx) + template = conn.get_template(name=name, template_id=template_id) + if template is None: + return {"error": "Template not found"} + return template + + +def create_template_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + name = params.get("name") + template_type = params.get("template_type") + if not name or not template_type: + return {"error": "name and template_type are required"} + conn = _get_connector(gns3_ctx) + return conn.create_template(**params) + + +def update_template_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + template_id = params.get("template_id") + name = params.get("name") + if not template_id and not name: + return {"error": "template_id or name is required"} + conn = _get_connector(gns3_ctx) + return conn.update_template(name=name, template_id=template_id, **{ + k: v for k, v in params.items() if k not in ("template_id", "name") + }) + + +def delete_template_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + template_id = params.get("template_id") + name = params.get("name") + if not template_id and not name: + return {"error": "template_id or name is required"} + conn = _get_connector(gns3_ctx) + conn.delete_template(name=name, template_id=template_id) + return {"message": f"Template deleted"} + + +# ── Tool definitions ─────────────────────────────────────────────────────── + +TEMPLATE_TOOLS = [ + { + "name": "list_templates", + "description": "List all available templates on the server", + "parameters": { + "type": "object", + "properties": {}, + }, + "handler": list_templates_handler, + }, + { + "name": "get_template", + "description": "Get detailed information about a specific template", + "parameters": { + "type": "object", + "properties": { + "template_id": {"type": "string", "description": "Template UUID"}, + "name": {"type": "string", "description": "Template name"}, + }, + }, + "handler": get_template_handler, + }, + { + "name": "create_template", + "description": "Create a new template", + "parameters": { + "type": "object", + "properties": { + "name": {"type": "string", "description": "Template name"}, + "template_type": {"type": "string", "description": "Template type (e.g. qemu, docker, dynamips)"}, + "compute_id": {"type": "string", "description": "Compute ID (optional, default: local)"}, + }, + "required": ["name", "template_type"], + }, + "handler": create_template_handler, + }, + { + "name": "update_template", + "description": "Update an existing template's properties", + "parameters": { + "type": "object", + "properties": { + "template_id": {"type": "string", "description": "Template UUID"}, + "name": {"type": "string", "description": "Template name"}, + }, + }, + "handler": update_template_handler, + }, + { + "name": "delete_template", + "description": "Delete a template", + "parameters": { + "type": "object", + "properties": { + "template_id": {"type": "string", "description": "Template UUID"}, + "name": {"type": "string", "description": "Template name"}, + }, + }, + "handler": delete_template_handler, + }, +] From e416ef8d5e90c92ce79a46dedffb1286edba96e4 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Thu, 4 Jun 2026 23:22:14 +0800 Subject: [PATCH 09/20] feat: add 3 Compute MCP tools - Add list_computes, get_compute, get_compute_images - Total MCP tools: 29 --- gns3server/api/routes/mcp/__init__.py | 34 ++++++++++ gns3server/api/routes/mcp/computes.py | 91 +++++++++++++++++++++++++++ 2 files changed, 125 insertions(+) create mode 100644 gns3server/api/routes/mcp/computes.py diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index 75c67a77b..0f1e113a6 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -409,6 +409,40 @@ async def delete_template(template_id: str = None, name: str = None) -> list[dic }) +# ── Compute tools ───────────────────────────────────────────────────── + +@mcp.tool() +async def list_computes() -> list[dict[str, Any]]: + """List all compute nodes available to the server.""" + from .computes import list_computes_handler + return await asyncio.to_thread(_run_handler_sync, list_computes_handler, {}) + + +@mcp.tool() +async def get_compute(compute_id: str = "local") -> list[dict[str, Any]]: + """Get detailed information about a compute node. + + Args: + compute_id: Compute ID (default: local) + """ + from .computes import get_compute_handler + return await asyncio.to_thread(_run_handler_sync, get_compute_handler, {"compute_id": compute_id}) + + +@mcp.tool() +async def get_compute_images(emulator: str, compute_id: str = "local") -> list[dict[str, Any]]: + """List available images for an emulator on a compute node. + + Args: + emulator: Emulator type (e.g. qemu, iou, docker) + compute_id: Compute ID (default: local) + """ + from .computes import get_compute_images_handler + return await asyncio.to_thread(_run_handler_sync, get_compute_images_handler, { + "emulator": emulator, "compute_id": compute_id, + }) + + # ── Auth‑wrapped SSE app ────────────────────────────────────────────── def _make_auth_wrapper(inner_app): diff --git a/gns3server/api/routes/mcp/computes.py b/gns3server/api/routes/mcp/computes.py new file mode 100644 index 000000000..fa6c70674 --- /dev/null +++ b/gns3server/api/routes/mcp/computes.py @@ -0,0 +1,91 @@ +# +# 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 compute management. +""" + +from typing import Any +import logging + +log = logging.getLogger(__name__) + + +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, + ) + + +def list_computes_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + conn = _get_connector(gns3_ctx) + computes = conn.get_computes() + return {"computes": computes, "count": len(computes)} + + +def get_compute_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + compute_id = params.get("compute_id", "local") + conn = _get_connector(gns3_ctx) + return conn.get_compute(compute_id=compute_id) + + +def get_compute_images_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + emulator = params.get("emulator") + compute_id = params.get("compute_id", "local") + if not emulator: + return {"error": "emulator is required (e.g. qemu, iou, docker)"} + conn = _get_connector(gns3_ctx) + images = conn.get_compute_images(emulator=emulator, compute_id=compute_id) + return {"images": images, "count": len(images)} + + +COMPUTE_TOOLS = [ + { + "name": "list_computes", + "description": "List all compute nodes available to the server", + "parameters": {"type": "object", "properties": {}}, + "handler": list_computes_handler, + }, + { + "name": "get_compute", + "description": "Get detailed information about a compute node", + "parameters": { + "type": "object", + "properties": { + "compute_id": {"type": "string", "description": "Compute ID (default: local)"}, + }, + }, + "handler": get_compute_handler, + }, + { + "name": "get_compute_images", + "description": "List available images for an emulator on a compute node", + "parameters": { + "type": "object", + "properties": { + "emulator": {"type": "string", "description": "Emulator type (e.g. qemu, iou, docker)"}, + "compute_id": {"type": "string", "description": "Compute ID (default: local)"}, + }, + "required": ["emulator"], + }, + "handler": get_compute_images_handler, + }, +] From 0c572f2e9c328b5ac497aba7ee296390757fd0c3 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Thu, 4 Jun 2026 23:25:22 +0800 Subject: [PATCH 10/20] docs: add MCP service design memory record --- .claude/memory/MEMORY.md | 3 + .claude/memory/mcp-service-design.md | 91 ++++++++++++++++++++++++++++ 2 files changed, 94 insertions(+) create mode 100644 .claude/memory/mcp-service-design.md diff --git a/.claude/memory/MEMORY.md b/.claude/memory/MEMORY.md index 9050f7d54..e02b12fbc 100644 --- a/.claude/memory/MEMORY.md +++ b/.claude/memory/MEMORY.md @@ -27,3 +27,6 @@ ### Docker Container Stop Delay - **[Docker Container Stop Delay](./docker-container-stop-delay.md)** - Some containers take ~5s to stop because they don't handle SIGTERM (AlpiNet, OstinatoWireshark) + +### MCP Service +- **[MCP Service Design](./mcp-service-design.md)** - MCP (Model Context Protocol) service architecture using FastMCP with SSE transport, JWT auth, 29 tools across 5 domains diff --git a/.claude/memory/mcp-service-design.md b/.claude/memory/mcp-service-design.md new file mode 100644 index 000000000..b514eeb7f --- /dev/null +++ b/.claude/memory/mcp-service-design.md @@ -0,0 +1,91 @@ +--- +name: mcp-service-design +description: MCP (Model Context Protocol) service architecture and tool design for GNS3 server +metadata: + type: project +--- + +# MCP (Model Context Protocol) Service Design + +## Background + +Provide a standard MCP interface for GNS3 Server, allowing AI assistants (Claude Code, Claude Desktop) to interact with GNS3 network simulations through the Model Context Protocol. + +## Decision/Implementation + +### Transport +- **SSE (Server-Sent Events)** with JWT token authentication +- Endpoint: `/v3/mcp/transport/sse` +- Message endpoint: `/v3/mcp/transport/messages/` + +### Authentication +- JWT token obtained via `/v3/access/users/authenticate` +- Two ways to pass token: + - `Authorization: Bearer ` header (Claude Code via `-H`) + - `?token=` query param (Claude Desktop, EventSource limitation) +- Token validated using GNS3's existing `auth_service` +- Token stored in `contextvars.ContextVar` for per-session isolation +- Python ≥ 3.9 `asyncio.to_thread` propagates contextvars to threads + +### Architecture +``` +Claude Code / Desktop → SSE → Auth Wrapper → FastMCP Server → Tool Handler → Gns3Connector → GNS3 REST API +``` + +### Tool Organization +Tools are separated by domain into individual files under `gns3server/api/routes/mcp/`: + +| File | Domain | Tool Count | +|------|--------|:----------:| +| `projects.py` | Project CRUD, open/close/stats | 7 | +| `nodes.py` | Node CRUD, start/stop/reload/suspend | 9 | +| `links.py` | Link CRUD | 5 | +| `templates.py` | Template CRUD | 5 | +| `computes.py` | Compute list/get/images | 3 | + +**Total: 29 tools** + +### Handler Pattern +- Synchronous functions receiving `(params: dict, gns3_ctx: dict)` +- Run via `asyncio.to_thread()` to avoid blocking the event loop +- `gns3_ctx` contains `server_url` and `jwt_token` +- `Gns3Connector` is created per-handler from `custom_gns3fy` + +### Token Lifetime +- Default: 1440 minutes (24 hours) +- Configurable via `jwt_access_token_expire_minutes` in `gns3_server.conf` + +## Rationale +- **Why not Direct Controller calls**: MCP layer calls GNS3's own REST API through Gns3Connector, keeping full decoupling and supporting future multi-user/multi-instance scenarios +- **Why not Streamable HTTP**: Claude Code supports SSE natively via `--transport sse` with custom headers; Streamable HTTP session manager lifecycle conflicts with FastAPI mount +- **Why not stdio**: stdio is local-only; SSE supports both local and remote deployments + +## Related Files +- `gns3server/api/routes/mcp/__init__.py` — FastMCP server, tool decorators, auth wrapper +- `gns3server/api/routes/mcp/projects.py` — Project tool handlers +- `gns3server/api/routes/mcp/nodes.py` — Node tool handlers +- `gns3server/api/routes/mcp/links.py` — Link tool handlers +- `gns3server/api/routes/mcp/templates.py` — Template tool handlers +- `gns3server/api/routes/mcp/computes.py` — Compute tool handlers +- `gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py` — Gns3Connector client +- `gns3server/api/server.py:87` — MCP route registration + +## Configuration + +### Claude Code +```bash +claude mcp add --transport sse My_GNS3_Server \ + http://host:3080/v3/mcp/transport/sse \ + -H "Authorization: Bearer " +``` + +### Claude Desktop +```json +{ + "mcpServers": { + "My_GNS3_Server": { + "url": "http://host:3080/v3/mcp/transport/sse?token=" + } + } +} +``` From 8775583b836c5624f69d5fffd7246c5b4c41aba0 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Thu, 4 Jun 2026 23:45:34 +0800 Subject: [PATCH 11/20] feat: add get_node_console_info tool Returns console type, host, port and a suggested command (e.g. telnet, vncviewer) for connecting to a node's console. Total: 30 tools. --- gns3server/api/routes/mcp/__init__.py | 14 ++++++++++ gns3server/api/routes/mcp/nodes.py | 38 ++++++++++++++++++++++++++- 2 files changed, 51 insertions(+), 1 deletion(-) diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index 0f1e113a6..313aafcb4 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -281,6 +281,20 @@ async def update_node(project_id: str, node_id: str, **kwargs: Any) -> list[dict return await asyncio.to_thread(_run_handler_sync, update_node_handler, params) +@mcp.tool() +async def get_node_console_info(project_id: str, node_id: str) -> list[dict[str, Any]]: + """Get console connection info for a node (host, port, type, and suggested command). + + Args: + project_id: Project UUID + node_id: Node UUID + """ + from .nodes import get_node_console_info_handler + return await asyncio.to_thread(_run_handler_sync, get_node_console_info_handler, { + "project_id": project_id, "node_id": node_id, + }) + + # ── Link tools ──────────────────────────────────────────────────────── @mcp.tool() diff --git a/gns3server/api/routes/mcp/nodes.py b/gns3server/api/routes/mcp/nodes.py index 87378f6c3..b2839a90e 100644 --- a/gns3server/api/routes/mcp/nodes.py +++ b/gns3server/api/routes/mcp/nodes.py @@ -132,12 +132,35 @@ def update_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dic if not project_id or not node_id: return {"error": "project_id and node_id are required"} conn = _get_connector(gns3_ctx) - # Pass all params except project_id/node_id as update fields update_data = {k: v for k, v in params.items() if k not in ("project_id", "node_id")} url = f"{conn.base_url}/projects/{project_id}/nodes/{node_id}" return conn.http_call("put", url, json_data=update_data).json() +def get_node_console_info_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + project_id = params.get("project_id") + node_id = params.get("node_id") + if not project_id or not node_id: + return {"error": "project_id and node_id are required"} + conn = _get_connector(gns3_ctx) + node = conn.get_node(project_id=project_id, node_id=node_id) + console_type = node.get("console_type", "unknown") + result = { + "node_id": node_id, + "node_name": node.get("name"), + "console_type": console_type, + "console_host": node.get("console_host"), + "console_port": node.get("console"), + } + if console_type == "telnet": + result["command"] = f"telnet {node.get('console_host')} {node.get('console')}" + elif console_type in ("vnc",): + result["command"] = f"vncviewer {node.get('console_host')}::{node.get('console')}" + elif console_type in ("http", "https"): + result["url"] = f"{console_type}://{node.get('console_host')}:{node.get('console')}" + return result + + # ── Tool definitions ─────────────────────────────────────────────────────── NODE_TOOLS = [ @@ -264,4 +287,17 @@ NODE_TOOLS = [ }, "handler": update_node_handler, }, + { + "name": "get_node_console_info", + "description": "Get console connection info for a node (host, port, type, and suggested command)", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "node_id": {"type": "string", "description": "Node UUID"}, + }, + "required": ["project_id", "node_id"], + }, + "handler": get_node_console_info_handler, + }, ] From f34de4c075f9e5b74ec1c1d1d3d05912cf09278c Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Thu, 4 Jun 2026 23:53:03 +0800 Subject: [PATCH 12/20] fix: remove console_host/port from get_node_console_info Return only ws_url + websocat command to avoid LLM misinterpreting direct telnet connection. --- gns3server/api/routes/mcp/__init__.py | 2 +- gns3server/api/routes/mcp/nodes.py | 17 ++++++++--------- 2 files changed, 9 insertions(+), 10 deletions(-) diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index 313aafcb4..92eb79fe2 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -283,7 +283,7 @@ async def update_node(project_id: str, node_id: str, **kwargs: Any) -> list[dict @mcp.tool() async def get_node_console_info(project_id: str, node_id: str) -> list[dict[str, Any]]: - """Get console connection info for a node (host, port, type, and suggested command). + """Get console WebSocket URL for a node (use websocat to connect). Args: project_id: Project UUID diff --git a/gns3server/api/routes/mcp/nodes.py b/gns3server/api/routes/mcp/nodes.py index b2839a90e..7fef57021 100644 --- a/gns3server/api/routes/mcp/nodes.py +++ b/gns3server/api/routes/mcp/nodes.py @@ -144,20 +144,19 @@ def get_node_console_info_handler(params: dict[str, Any], gns3_ctx: dict[str, An return {"error": "project_id and node_id are required"} conn = _get_connector(gns3_ctx) node = conn.get_node(project_id=project_id, node_id=node_id) + console_type = node.get("console_type", "unknown") + ws_url = f"{gns3_ctx['server_url']}/v3/projects/{project_id}/nodes/{node_id}/console/ws?token={gns3_ctx['jwt_token']}" + result = { "node_id": node_id, "node_name": node.get("name"), "console_type": console_type, - "console_host": node.get("console_host"), - "console_port": node.get("console"), + "ws_url": ws_url, + "command": f"websocat {ws_url}", } - if console_type == "telnet": - result["command"] = f"telnet {node.get('console_host')} {node.get('console')}" - elif console_type in ("vnc",): - result["command"] = f"vncviewer {node.get('console_host')}::{node.get('console')}" - elif console_type in ("http", "https"): - result["url"] = f"{console_type}://{node.get('console_host')}:{node.get('console')}" + if console_type in ("vnc",): + result["vnc_url"] = f"/v3/projects/{project_id}/nodes/{node_id}/console/vnc?token={gns3_ctx['jwt_token']}" return result @@ -289,7 +288,7 @@ NODE_TOOLS = [ }, { "name": "get_node_console_info", - "description": "Get console connection info for a node (host, port, type, and suggested command)", + "description": "Get console WebSocket URL for a node (use websocat to connect)", "parameters": { "type": "object", "properties": { From 25dc1e90091a0053816c5ce148c1e96953cc2b50 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Fri, 5 Jun 2026 00:20:13 +0800 Subject: [PATCH 13/20] docs: update MCP feature doc to cover all 30 tools and console WS --- docs/features/mcp-service.md | 69 ++++++++++++++++++++++++++++++++++-- 1 file changed, 66 insertions(+), 3 deletions(-) diff --git a/docs/features/mcp-service.md b/docs/features/mcp-service.md index 01bfc26f7..96ec1f10c 100644 --- a/docs/features/mcp-service.md +++ b/docs/features/mcp-service.md @@ -46,6 +46,10 @@ jwt_access_token_expire_minutes = 1440 ; 24 hours ## Available Tools +**30 tools** across 5 categories: + +### Project (7) + | Tool | Description | Required Parameters | |------|-------------|-------------------| | `list_projects` | List all projects | none | @@ -56,6 +60,49 @@ jwt_access_token_expire_minutes = 1440 ; 24 hours | `close_project` | Close a project | `project_id` | | `get_project_stats` | Get project statistics | `project_id` | +### Node (10) + +| Tool | Description | Required Parameters | +|------|-------------|-------------------| +| `get_nodes` | List all nodes in a project | `project_id` | +| `get_node` | Get node details | `project_id`, `node_id` | +| `start_node` | Start a node | `project_id`, `node_id` | +| `stop_node` | Stop a node | `project_id`, `node_id` | +| `reload_node` | Reload a node | `project_id`, `node_id` | +| `suspend_node` | Suspend a node | `project_id`, `node_id` | +| `create_node` | Create a node from template | `project_id`, `template_id` | +| `delete_node` | Delete a node | `project_id`, `node_id` | +| `update_node` | Update node properties | `project_id`, `node_id` | +| `get_node_console_info` | Get WebSocket console URL | `project_id`, `node_id` | + +### Link (5) + +| Tool | Description | Required Parameters | +|------|-------------|-------------------| +| `get_links` | List all links in a project | `project_id` | +| `get_link` | Get link details | `project_id`, `link_id` | +| `create_link` | Create a link between nodes | `project_id`, `nodes` | +| `delete_link` | Delete a link | `project_id`, `link_id` | +| `update_link` | Update link properties | `project_id`, `link_id` | + +### Template (5) + +| Tool | Description | Required Parameters | +|------|-------------|-------------------| +| `list_templates` | List all templates | none | +| `get_template` | Get template details | `template_id` or `name` | +| `create_template` | Create a template | `name`, `template_type` | +| `update_template` | Update a template | `template_id` or `name` | +| `delete_template` | Delete a template | `template_id` or `name` | + +### Compute (3) + +| Tool | Description | Required Parameters | +|------|-------------|-------------------| +| `list_computes` | List all compute nodes | none | +| `get_compute` | Get compute details | `compute_id` | +| `get_compute_images` | List available images | `emulator` | + ## Configuration ### Claude Code (CLI) @@ -124,8 +171,24 @@ sequenceDiagram - Tool handlers use `Gns3Connector` (from `custom_gns3fy`) to call GNS3's own REST API, keeping the MCP layer decoupled - The JWT token is stored in a `contextvars.ContextVar` so it is available within tool handler threads (Python ≥ 3.9 propagates contextvars through `asyncio.to_thread`) +### Console WebSocket + +The `get_node_console_info` tool returns a WebSocket URL for connecting to a node's console. This endpoint is protocol-agnostic — it works for **telnet**, **ssh**, and **vnc** console types alike. The WebSocket simply proxies raw byte streams between the client and the compute node; protocol negotiation (e.g. SSH key exchange) happens on the compute side. + +Use `websocat` to connect from the command line: + +```bash +websocat wss://host:3080/v3/projects/{project_id}/nodes/{node_id}/console/ws?token= +``` + ### Source Files -- `gns3server/api/routes/mcp/__init__.py` — FastMCP server, tool definitions, SSE transport, JWT auth wrapper -- `gns3server/api/routes/mcp/projects.py` — Project tool handlers using Gns3Connector -- `gns3server/api/server.py` — Mounts MCP routes via `register_starlette_routes()` +| File | Purpose | +|------|---------| +| `gns3server/api/routes/mcp/__init__.py` | FastMCP server, tool decorators, SSE transport, JWT auth wrapper | +| `gns3server/api/routes/mcp/projects.py` | Project tool handlers | +| `gns3server/api/routes/mcp/nodes.py` | Node tool handlers | +| `gns3server/api/routes/mcp/links.py` | Link tool handlers | +| `gns3server/api/routes/mcp/templates.py` | Template tool handlers | +| `gns3server/api/routes/mcp/computes.py` | Compute tool handlers | +| `gns3server/api/server.py` | Mounts MCP routes via `register_starlette_routes()` | From f9f7ba7a5311ab9e77c0304b25256788b526ede6 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Fri, 5 Jun 2026 00:20:34 +0800 Subject: [PATCH 14/20] docs: update memory record to 30 tools --- .claude/memory/mcp-service-design.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.claude/memory/mcp-service-design.md b/.claude/memory/mcp-service-design.md index b514eeb7f..d6bbcc9de 100644 --- a/.claude/memory/mcp-service-design.md +++ b/.claude/memory/mcp-service-design.md @@ -38,12 +38,12 @@ Tools are separated by domain into individual files under `gns3server/api/routes | File | Domain | Tool Count | |------|--------|:----------:| | `projects.py` | Project CRUD, open/close/stats | 7 | -| `nodes.py` | Node CRUD, start/stop/reload/suspend | 9 | +| `nodes.py` | Node CRUD, start/stop/reload/suspend, console WS | 10 | | `links.py` | Link CRUD | 5 | | `templates.py` | Template CRUD | 5 | | `computes.py` | Compute list/get/images | 3 | -**Total: 29 tools** +**Total: 30 tools** ### Handler Pattern - Synchronous functions receiving `(params: dict, gns3_ctx: dict)` From f1078a97e28da58e6d75c5c4c5262075511c0671 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Fri, 5 Jun 2026 01:07:20 +0800 Subject: [PATCH 15/20] test: add /v3/mcp/ to allowed public endpoints --- tests/api/routes/test_routes.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/api/routes/test_routes.py b/tests/api/routes/test_routes.py index dc69cfca6..98c4dcdd6 100644 --- a/tests/api/routes/test_routes.py +++ b/tests/api/routes/test_routes.py @@ -41,7 +41,8 @@ ALLOWED_CONTROLLER_ENDPOINTS = [ ("/v3/symbols", "GET"), ("/v3/symbols/{symbol_id:path}/raw", "GET"), ("/v3/symbols/{symbol_id:path}/dimensions", "GET"), - ("/v3/symbols/default_symbols", "GET") + ("/v3/symbols/default_symbols", "GET"), + ("/v3/mcp/", "GET"), ] class TestRoutes: From a056fa3450a8f7b17aa951a16cf69b31c2d68938 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Fri, 5 Jun 2026 01:16:17 +0800 Subject: [PATCH 16/20] refactor: move all imports to top of __init__.py --- gns3server/api/routes/mcp/__init__.py | 54 ++++++++++++--------------- 1 file changed, 23 insertions(+), 31 deletions(-) diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index 92eb79fe2..a581c4e4e 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -39,6 +39,29 @@ from fastapi.responses import Response from mcp.server.fastmcp import FastMCP from gns3server.config import Config +from gns3server.services import auth_service +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, +) +from .nodes import ( + get_nodes_handler, get_node_handler, start_node_handler, + stop_node_handler, reload_node_handler, suspend_node_handler, + create_node_handler, delete_node_handler, update_node_handler, + get_node_console_info_handler, +) +from .links import ( + get_links_handler, get_link_handler, create_link_handler, + delete_link_handler, update_link_handler, +) +from .templates import ( + list_templates_handler, get_template_handler, create_template_handler, + update_template_handler, delete_template_handler, +) +from .computes import ( + list_computes_handler, get_compute_handler, get_compute_images_handler, +) log = logging.getLogger(__name__) @@ -56,7 +79,6 @@ _jwt_token_var: contextvars.ContextVar[str | None] = contextvars.ContextVar( async def _validate_token(token: str) -> bool: """Return True if token is a valid GNS3 JWT.""" - from gns3server.services import auth_service try: auth_service.get_username_from_token(token) return True @@ -95,7 +117,6 @@ def _run_handler_sync(handler, params: dict[str, Any]) -> list[dict[str, Any]]: @mcp.tool() async def list_projects() -> list[dict[str, Any]]: """List all GNS3 projects accessible to the current user.""" - from .projects import list_projects_handler return await asyncio.to_thread(_run_handler_sync, list_projects_handler, {}) @@ -106,7 +127,6 @@ async def get_project(project_id: str) -> list[dict[str, Any]]: Args: project_id: Project UUID """ - from .projects import get_project_handler return await asyncio.to_thread(_run_handler_sync, get_project_handler, {"project_id": project_id}) @@ -118,7 +138,6 @@ async def create_project(name: str, description: str = "") -> list[dict[str, Any name: Project name description: Optional project description """ - from .projects import create_project_handler params = {"name": name} if description: params["description"] = description @@ -132,7 +151,6 @@ async def delete_project(project_id: str) -> list[dict[str, Any]]: Args: project_id: UUID of the project to delete """ - from .projects import delete_project_handler return await asyncio.to_thread(_run_handler_sync, delete_project_handler, {"project_id": project_id}) @@ -143,7 +161,6 @@ async def open_project(project_id: str) -> list[dict[str, Any]]: Args: project_id: Project UUID """ - from .projects import open_project_handler return await asyncio.to_thread(_run_handler_sync, open_project_handler, {"project_id": project_id}) @@ -154,7 +171,6 @@ async def close_project(project_id: str) -> list[dict[str, Any]]: Args: project_id: Project UUID """ - from .projects import close_project_handler return await asyncio.to_thread(_run_handler_sync, close_project_handler, {"project_id": project_id}) @@ -165,7 +181,6 @@ async def get_project_stats(project_id: str) -> list[dict[str, Any]]: Args: project_id: Project UUID """ - from .projects import get_project_stats_handler return await asyncio.to_thread(_run_handler_sync, get_project_stats_handler, {"project_id": project_id}) @@ -174,7 +189,6 @@ async def get_project_stats(project_id: str) -> list[dict[str, Any]]: @mcp.tool() async def get_nodes(project_id: str) -> list[dict[str, Any]]: """List all nodes in a project.""" - from .nodes import get_nodes_handler return await asyncio.to_thread(_run_handler_sync, get_nodes_handler, {"project_id": project_id}) @@ -186,7 +200,6 @@ async def get_node(project_id: str, node_id: str) -> list[dict[str, Any]]: project_id: Project UUID node_id: Node UUID """ - from .nodes import get_node_handler return await asyncio.to_thread(_run_handler_sync, get_node_handler, {"project_id": project_id, "node_id": node_id}) @@ -198,7 +211,6 @@ async def start_node(project_id: str, node_id: str) -> list[dict[str, Any]]: project_id: Project UUID node_id: Node UUID """ - from .nodes import start_node_handler return await asyncio.to_thread(_run_handler_sync, start_node_handler, {"project_id": project_id, "node_id": node_id}) @@ -210,7 +222,6 @@ async def stop_node(project_id: str, node_id: str) -> list[dict[str, Any]]: project_id: Project UUID node_id: Node UUID """ - from .nodes import stop_node_handler return await asyncio.to_thread(_run_handler_sync, stop_node_handler, {"project_id": project_id, "node_id": node_id}) @@ -222,7 +233,6 @@ async def reload_node(project_id: str, node_id: str) -> list[dict[str, Any]]: project_id: Project UUID node_id: Node UUID """ - from .nodes import reload_node_handler return await asyncio.to_thread(_run_handler_sync, reload_node_handler, {"project_id": project_id, "node_id": node_id}) @@ -234,7 +244,6 @@ async def suspend_node(project_id: str, node_id: str) -> list[dict[str, Any]]: project_id: Project UUID node_id: Node UUID """ - from .nodes import suspend_node_handler return await asyncio.to_thread(_run_handler_sync, suspend_node_handler, {"project_id": project_id, "node_id": node_id}) @@ -249,7 +258,6 @@ async def create_node(project_id: str, template_id: str, x: int = 0, y: int = 0, y: Y coordinate (optional) compute_id: Compute ID (optional, default: local) """ - from .nodes import create_node_handler return await asyncio.to_thread(_run_handler_sync, create_node_handler, { "project_id": project_id, "template_id": template_id, "x": x, "y": y, "compute_id": compute_id, @@ -264,7 +272,6 @@ async def delete_node(project_id: str, node_id: str) -> list[dict[str, Any]]: project_id: Project UUID node_id: Node UUID """ - from .nodes import delete_node_handler return await asyncio.to_thread(_run_handler_sync, delete_node_handler, {"project_id": project_id, "node_id": node_id}) @@ -276,7 +283,6 @@ async def update_node(project_id: str, node_id: str, **kwargs: Any) -> list[dict project_id: Project UUID node_id: Node UUID """ - from .nodes import update_node_handler params = {"project_id": project_id, "node_id": node_id, **kwargs} return await asyncio.to_thread(_run_handler_sync, update_node_handler, params) @@ -289,7 +295,6 @@ async def get_node_console_info(project_id: str, node_id: str) -> list[dict[str, project_id: Project UUID node_id: Node UUID """ - from .nodes import get_node_console_info_handler return await asyncio.to_thread(_run_handler_sync, get_node_console_info_handler, { "project_id": project_id, "node_id": node_id, }) @@ -300,7 +305,6 @@ async def get_node_console_info(project_id: str, node_id: str) -> list[dict[str, @mcp.tool() async def get_links(project_id: str) -> list[dict[str, Any]]: """List all links in a project.""" - from .links import get_links_handler return await asyncio.to_thread(_run_handler_sync, get_links_handler, {"project_id": project_id}) @@ -312,7 +316,6 @@ async def get_link(project_id: str, link_id: str) -> list[dict[str, Any]]: project_id: Project UUID link_id: Link UUID """ - from .links import get_link_handler return await asyncio.to_thread(_run_handler_sync, get_link_handler, {"project_id": project_id, "link_id": link_id}) @@ -326,7 +329,6 @@ async def create_link(project_id: str, nodes: list, link_type: str = "ethernet", link_type: Link type - ethernet or serial (optional) filters: Packet filters (optional) """ - from .links import create_link_handler params = {"project_id": project_id, "nodes": nodes, "link_type": link_type} if filters: params["filters"] = filters @@ -341,7 +343,6 @@ async def delete_link(project_id: str, link_id: str) -> list[dict[str, Any]]: project_id: Project UUID link_id: Link UUID """ - from .links import delete_link_handler return await asyncio.to_thread(_run_handler_sync, delete_link_handler, {"project_id": project_id, "link_id": link_id}) @@ -353,7 +354,6 @@ async def update_link(project_id: str, link_id: str, **kwargs: Any) -> list[dict project_id: Project UUID link_id: Link UUID """ - from .links import update_link_handler params = {"project_id": project_id, "link_id": link_id, **kwargs} return await asyncio.to_thread(_run_handler_sync, update_link_handler, params) @@ -363,7 +363,6 @@ async def update_link(project_id: str, link_id: str, **kwargs: Any) -> list[dict @mcp.tool() async def list_templates() -> list[dict[str, Any]]: """List all available templates on the server.""" - from .templates import list_templates_handler return await asyncio.to_thread(_run_handler_sync, list_templates_handler, {}) @@ -375,7 +374,6 @@ async def get_template(template_id: str = None, name: str = None) -> list[dict[s template_id: Template UUID (optional if name is provided) name: Template name (optional if template_id is provided) """ - from .templates import get_template_handler return await asyncio.to_thread(_run_handler_sync, get_template_handler, { "template_id": template_id, "name": name, }) @@ -390,7 +388,6 @@ async def create_template(name: str, template_type: str, compute_id: str = "loca template_type: Template type (e.g. qemu, docker, dynamips) compute_id: Compute ID (optional, default: local) """ - from .templates import create_template_handler return await asyncio.to_thread(_run_handler_sync, create_template_handler, { "name": name, "template_type": template_type, "compute_id": compute_id, }) @@ -404,7 +401,6 @@ async def update_template(template_id: str = None, name: str = None, **kwargs: A template_id: Template UUID (optional if name is provided) name: Template name (optional if template_id is provided) """ - from .templates import update_template_handler params = {"template_id": template_id, "name": name, **kwargs} return await asyncio.to_thread(_run_handler_sync, update_template_handler, params) @@ -417,7 +413,6 @@ async def delete_template(template_id: str = None, name: str = None) -> list[dic template_id: Template UUID (optional if name is provided) name: Template name (optional if template_id is provided) """ - from .templates import delete_template_handler return await asyncio.to_thread(_run_handler_sync, delete_template_handler, { "template_id": template_id, "name": name, }) @@ -428,7 +423,6 @@ async def delete_template(template_id: str = None, name: str = None) -> list[dic @mcp.tool() async def list_computes() -> list[dict[str, Any]]: """List all compute nodes available to the server.""" - from .computes import list_computes_handler return await asyncio.to_thread(_run_handler_sync, list_computes_handler, {}) @@ -439,7 +433,6 @@ async def get_compute(compute_id: str = "local") -> list[dict[str, Any]]: Args: compute_id: Compute ID (default: local) """ - from .computes import get_compute_handler return await asyncio.to_thread(_run_handler_sync, get_compute_handler, {"compute_id": compute_id}) @@ -451,7 +444,6 @@ async def get_compute_images(emulator: str, compute_id: str = "local") -> list[d emulator: Emulator type (e.g. qemu, iou, docker) compute_id: Compute ID (default: local) """ - from .computes import get_compute_images_handler return await asyncio.to_thread(_run_handler_sync, get_compute_images_handler, { "emulator": emulator, "compute_id": compute_id, }) From b415ae0af6ebb9c42f76b63bf8e2998b14daeba4 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Fri, 5 Jun 2026 12:32:27 +0800 Subject: [PATCH 17/20] fix: add missing fastmcp dependency to resolve CI test failures - Add fastmcp>=3.4.0 to requirements.txt - Fixes ModuleNotFoundError: No module named 'mcp' - Resolves CI test failures in compute routes tests --- requirements.txt | 3 +++ 1 file changed, 3 insertions(+) diff --git a/requirements.txt b/requirements.txt index 76b3f333d..472bbff19 100644 --- a/requirements.txt +++ b/requirements.txt @@ -31,6 +31,9 @@ typing-extensions>=4.15.0 requests>=2.34.2 urllib3>=2.7.0 +# MCP (Model Context Protocol) dependencies +fastmcp>=3.4.0 + # ============================================================================== # AI Copilot Optional Dependencies # ============================================================================== From 4c44a32db7b5391b27634678fba98dbd9fbcc5ac Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Fri, 5 Jun 2026 13:17:23 +0800 Subject: [PATCH 18/20] docs: update get_node_console_info description with websocat connection workflow --- gns3server/api/routes/mcp/__init__.py | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index a581c4e4e..429e7b877 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -289,7 +289,22 @@ async def update_node(project_id: str, node_id: str, **kwargs: Any) -> list[dict @mcp.tool() async def get_node_console_info(project_id: str, node_id: str) -> list[dict[str, Any]]: - """Get console WebSocket URL for a node (use websocat to connect). + """Get console WebSocket URL and send configuration commands via websocat. + + Complete workflow: + 1. Preparation: Call this tool with project_id and node_id + 2. Connection: Use websocat in text mode (-t) to connect + > websocat -t "ws://:3080/v3/projects/{project_id}/nodes/{node_id}/console/ws?token={jwt_token}" + 3. Send commands: Use heredoc (<<<) with \\r\\n as line endings + > websocat -t "ws://..." <<< $'\\r\\nenable\\r\\nshow version\\r\\nexit\\r\\n' + 4. Receive response: websocat receives and displays device output + Use 'timeout' to avoid connection hanging: + > timeout 10 websocat -t "ws://..." <<< $'commands\\r\\n' + + Key points: + - Use \\r\\n (not \\n) to match Telnet/Console protocol + - $'...' format supports escape sequences + - Set timeout to prevent hanging Args: project_id: Project UUID From 62db5caf6a7693857e0de35490af620dd0f9c92e Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Fri, 5 Jun 2026 13:38:25 +0800 Subject: [PATCH 19/20] fix: remove platformdirs upper bound to resolve fastmcp-slim dependency conflict - Remove platformdirs<3 constraint (was for Debian packaging only) - fastmcp-slim>=3.4.0 requires platformdirs>=4.0.0 - Debian packaging is no longer a concern --- requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index 472bbff19..d9f9cbf29 100644 --- a/requirements.txt +++ b/requirements.txt @@ -21,7 +21,7 @@ joserfc==1.7.0 email-validator==2.3.0 watchdog==6.0.0 zstandard==0.25.0 -platformdirs>=2.4.0,<3 # platformdirs >=3 conflicts when building Debian packages +platformdirs>=2.4.0 # fastmcp-slim >=3.4 requires >=4.0.0; upper bound removed for compatibility truststore>=0.10.4; python_version >= '3.10' # Shared dependencies (also used by AI Copilot) From 24953c1712a416a3c566e11df70ccd0dd1abb26a Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Fri, 5 Jun 2026 13:53:02 +0800 Subject: [PATCH 20/20] refactor: convert all MCP tool parameter descriptions to Annotated+Field - Replace Args: docstring blocks with Annotated[str, Field(description=...)] so parameter descriptions appear in inputSchema.properties.*.description - mcp.server.fastmcp does not parse Args: blocks from docstrings; only Annotated with pydantic Field injects descriptions into the structured JSON Schema visible to AI clients via tools/list - Remove redundant Args: blocks from docstrings (info moved to Field) - Restore full 4-step websocat workflow in get_node_console_info docstring with connection, command sending, response receiving, and timeout --- gns3server/api/routes/mcp/__init__.py | 327 +++++++++++--------------- 1 file changed, 140 insertions(+), 187 deletions(-) diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index 429e7b877..0190452a4 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -30,12 +30,14 @@ import contextvars import json import asyncio import logging -from typing import Any +from typing import Any, Annotated from urllib.parse import parse_qs from fastapi import APIRouter from fastapi.responses import Response +from pydantic import Field + from mcp.server.fastmcp import FastMCP from gns3server.config import Config @@ -121,23 +123,19 @@ async def list_projects() -> list[dict[str, Any]]: @mcp.tool() -async def get_project(project_id: str) -> list[dict[str, Any]]: - """Get detailed information about a specific project. - - Args: - project_id: Project UUID - """ +async def get_project( + project_id: Annotated[str, Field(description="UUID of the project")], +) -> list[dict[str, Any]]: + """Get detailed information about a specific project.""" return await asyncio.to_thread(_run_handler_sync, get_project_handler, {"project_id": project_id}) @mcp.tool() -async def create_project(name: str, description: str = "") -> list[dict[str, Any]]: - """Create a new GNS3 project. - - Args: - name: Project name - description: Optional project description - """ +async def create_project( + name: Annotated[str, Field(description="Project name")], + description: Annotated[str, Field(description="Optional project description")] = "", +) -> list[dict[str, Any]]: + """Create a new GNS3 project.""" params = {"name": name} if description: params["description"] = description @@ -145,42 +143,32 @@ async def create_project(name: str, description: str = "") -> list[dict[str, Any @mcp.tool() -async def delete_project(project_id: str) -> list[dict[str, Any]]: - """Delete a GNS3 project permanently. - - Args: - project_id: UUID of the project to delete - """ +async def delete_project( + project_id: Annotated[str, Field(description="UUID of the project to delete")], +) -> list[dict[str, Any]]: + """Delete a GNS3 project permanently.""" return await asyncio.to_thread(_run_handler_sync, delete_project_handler, {"project_id": project_id}) @mcp.tool() -async def open_project(project_id: str) -> list[dict[str, Any]]: - """Open a closed GNS3 project. - - Args: - project_id: Project UUID - """ +async def open_project( + project_id: Annotated[str, Field(description="UUID of the project to open")], +) -> list[dict[str, Any]]: + """Open a closed GNS3 project.""" return await asyncio.to_thread(_run_handler_sync, open_project_handler, {"project_id": project_id}) - @mcp.tool() -async def close_project(project_id: str) -> list[dict[str, Any]]: - """Close an open GNS3 project. - - Args: - project_id: Project UUID - """ +async def close_project( + project_id: Annotated[str, Field(description="UUID of the project to close")], +) -> list[dict[str, Any]]: + """Close an open GNS3 project.""" return await asyncio.to_thread(_run_handler_sync, close_project_handler, {"project_id": project_id}) - @mcp.tool() -async def get_project_stats(project_id: str) -> list[dict[str, Any]]: - """Get statistics (nodes, links, snapshots, drawings) for a project. - - Args: - project_id: Project UUID - """ +async def get_project_stats( + project_id: Annotated[str, Field(description="UUID of the project to get statistics for")], +) -> list[dict[str, Any]]: + """Get statistics (nodes, links, snapshots, drawings) for a project.""" return await asyncio.to_thread(_run_handler_sync, get_project_stats_handler, {"project_id": project_id}) @@ -193,71 +181,55 @@ async def get_nodes(project_id: str) -> list[dict[str, Any]]: @mcp.tool() -async def get_node(project_id: str, node_id: str) -> list[dict[str, Any]]: - """Get detailed information about a specific node. - - Args: - project_id: Project UUID - node_id: Node UUID - """ +async def get_node( + project_id: Annotated[str, Field(description="UUID of the project")], + node_id: Annotated[str, Field(description="UUID of the node")], +) -> list[dict[str, Any]]: + """Get detailed information about a specific node.""" return await asyncio.to_thread(_run_handler_sync, get_node_handler, {"project_id": project_id, "node_id": node_id}) - @mcp.tool() -async def start_node(project_id: str, node_id: str) -> list[dict[str, Any]]: - """Start a node in a project. - - Args: - project_id: Project UUID - node_id: Node UUID - """ +async def start_node( + project_id: Annotated[str, Field(description="UUID of the project")], + node_id: Annotated[str, Field(description="UUID of the node to start")], +) -> list[dict[str, Any]]: + """Start a node in a project.""" return await asyncio.to_thread(_run_handler_sync, start_node_handler, {"project_id": project_id, "node_id": node_id}) - @mcp.tool() -async def stop_node(project_id: str, node_id: str) -> list[dict[str, Any]]: - """Stop a node in a project. - - Args: - project_id: Project UUID - node_id: Node UUID - """ +async def stop_node( + project_id: Annotated[str, Field(description="UUID of the project")], + node_id: Annotated[str, Field(description="UUID of the node to stop")], +) -> list[dict[str, Any]]: + """Stop a node in a project.""" return await asyncio.to_thread(_run_handler_sync, stop_node_handler, {"project_id": project_id, "node_id": node_id}) - @mcp.tool() -async def reload_node(project_id: str, node_id: str) -> list[dict[str, Any]]: - """Reload (restart) a node in a project. - - Args: - project_id: Project UUID - node_id: Node UUID - """ +async def reload_node( + project_id: Annotated[str, Field(description="UUID of the project")], + node_id: Annotated[str, Field(description="UUID of the node to reload")], +) -> list[dict[str, Any]]: + """Reload (restart) a node in a project.""" return await asyncio.to_thread(_run_handler_sync, reload_node_handler, {"project_id": project_id, "node_id": node_id}) - @mcp.tool() -async def suspend_node(project_id: str, node_id: str) -> list[dict[str, Any]]: - """Suspend a node in a project. - - Args: - project_id: Project UUID - node_id: Node UUID - """ +async def suspend_node( + project_id: Annotated[str, Field(description="UUID of the project")], + node_id: Annotated[str, Field(description="UUID of the node to suspend")], +) -> list[dict[str, Any]]: + """Suspend a node in a project.""" return await asyncio.to_thread(_run_handler_sync, suspend_node_handler, {"project_id": project_id, "node_id": node_id}) @mcp.tool() -async def create_node(project_id: str, template_id: str, x: int = 0, y: int = 0, compute_id: str = "local") -> list[dict[str, Any]]: - """Create a new node from a template in a project. - - Args: - project_id: Project UUID - template_id: Template UUID - x: X coordinate (optional) - y: Y coordinate (optional) - compute_id: Compute ID (optional, default: local) - """ +async def create_node( + project_id: Annotated[str, Field(description="UUID of the project")], + template_id: Annotated[str, Field(description="UUID of the template to create the node from")], + x: Annotated[int, Field(description="X coordinate on the project canvas")] = 0, + y: Annotated[int, Field(description="Y coordinate on the project canvas")] = 0, + compute_id: Annotated[str, Field(description="Compute ID (default: local)")] = "local", +) -> list[dict[str, Any]]: + """Create a new node from a template in a project.""" return await asyncio.to_thread(_run_handler_sync, create_node_handler, { "project_id": project_id, "template_id": template_id, "x": x, "y": y, "compute_id": compute_id, @@ -265,50 +237,49 @@ async def create_node(project_id: str, template_id: str, x: int = 0, y: int = 0, @mcp.tool() -async def delete_node(project_id: str, node_id: str) -> list[dict[str, Any]]: - """Delete a node from a project. - - Args: - project_id: Project UUID - node_id: Node UUID - """ +async def delete_node( + project_id: Annotated[str, Field(description="UUID of the project")], + node_id: Annotated[str, Field(description="UUID of the node to delete")], +) -> list[dict[str, Any]]: + """Delete a node from a project.""" return await asyncio.to_thread(_run_handler_sync, delete_node_handler, {"project_id": project_id, "node_id": node_id}) @mcp.tool() -async def update_node(project_id: str, node_id: str, **kwargs: Any) -> list[dict[str, Any]]: - """Update a node's properties (name, position, etc.). - - Args: - project_id: Project UUID - node_id: Node UUID - """ +async def update_node( + project_id: Annotated[str, Field(description="UUID of the project")], + node_id: Annotated[str, Field(description="UUID of the node to update")], + **kwargs: Any, +) -> list[dict[str, Any]]: + """Update a node's properties (name, position, etc.).""" params = {"project_id": project_id, "node_id": node_id, **kwargs} return await asyncio.to_thread(_run_handler_sync, update_node_handler, params) @mcp.tool() -async def get_node_console_info(project_id: str, node_id: str) -> list[dict[str, Any]]: - """Get console WebSocket URL and send configuration commands via websocat. +async def get_node_console_info( + project_id: Annotated[str, Field(description="UUID of the project containing the node")], + node_id: Annotated[str, Field(description="UUID of the node to get console info for")], +) -> list[dict[str, Any]]: + """Get WebSocket console connection info for a node. + + Returns the WebSocket URL, console type (telnet/ssh/vnc), and other + connection details needed to interact with a node's console via WebSocket. Complete workflow: - 1. Preparation: Call this tool with project_id and node_id - 2. Connection: Use websocat in text mode (-t) to connect + 1. Call this tool with project_id and node_id to get the WebSocket URL + 2. Connect to the returned URL using websocat in text mode (-t): > websocat -t "ws://:3080/v3/projects/{project_id}/nodes/{node_id}/console/ws?token={jwt_token}" - 3. Send commands: Use heredoc (<<<) with \\r\\n as line endings + 3. Send device commands with \\r\\n line endings via heredoc: > websocat -t "ws://..." <<< $'\\r\\nenable\\r\\nshow version\\r\\nexit\\r\\n' 4. Receive response: websocat receives and displays device output Use 'timeout' to avoid connection hanging: > timeout 10 websocat -t "ws://..." <<< $'commands\\r\\n' Key points: - - Use \\r\\n (not \\n) to match Telnet/Console protocol - - $'...' format supports escape sequences - - Set timeout to prevent hanging - - Args: - project_id: Project UUID - node_id: Node UUID + - Use \\r\\n (not \\n) to match console protocol line endings + - Use $'...' format for escape sequences in bash + - Set a timeout to prevent hanging connections """ return await asyncio.to_thread(_run_handler_sync, get_node_console_info_handler, { "project_id": project_id, "node_id": node_id, @@ -324,26 +295,22 @@ async def get_links(project_id: str) -> list[dict[str, Any]]: @mcp.tool() -async def get_link(project_id: str, link_id: str) -> list[dict[str, Any]]: - """Get detailed information about a specific link. - - Args: - project_id: Project UUID - link_id: Link UUID - """ +async def get_link( + project_id: Annotated[str, Field(description="UUID of the project")], + link_id: Annotated[str, Field(description="UUID of the link")], +) -> list[dict[str, Any]]: + """Get detailed information about a specific link.""" return await asyncio.to_thread(_run_handler_sync, get_link_handler, {"project_id": project_id, "link_id": link_id}) @mcp.tool() -async def create_link(project_id: str, nodes: list, link_type: str = "ethernet", filters: dict = None) -> list[dict[str, Any]]: - """Create a link between two nodes in a project. - - Args: - project_id: Project UUID - nodes: List of node connections, e.g. [{"node_id": "...", "adapter_number": 0, "port_number": 0}, ...] - link_type: Link type - ethernet or serial (optional) - filters: Packet filters (optional) - """ +async def create_link( + project_id: Annotated[str, Field(description="UUID of the project")], + nodes: Annotated[list, Field(description="List of node connections, e.g. [{\"node_id\": \"...\", \"adapter_number\": 0, \"port_number\": 0}]")], + link_type: Annotated[str, Field(description="Link type - ethernet or serial")] = "ethernet", + filters: Annotated[dict, Field(description="Optional packet filters")] = None, +) -> list[dict[str, Any]]: + """Create a link between two nodes in a project.""" params = {"project_id": project_id, "nodes": nodes, "link_type": link_type} if filters: params["filters"] = filters @@ -351,24 +318,21 @@ async def create_link(project_id: str, nodes: list, link_type: str = "ethernet", @mcp.tool() -async def delete_link(project_id: str, link_id: str) -> list[dict[str, Any]]: - """Delete a link from a project. - - Args: - project_id: Project UUID - link_id: Link UUID - """ +async def delete_link( + project_id: Annotated[str, Field(description="UUID of the project")], + link_id: Annotated[str, Field(description="UUID of the link to delete")], +) -> list[dict[str, Any]]: + """Delete a link from a project.""" return await asyncio.to_thread(_run_handler_sync, delete_link_handler, {"project_id": project_id, "link_id": link_id}) @mcp.tool() -async def update_link(project_id: str, link_id: str, **kwargs: Any) -> list[dict[str, Any]]: - """Update a link's properties (suspend, filters, etc.). - - Args: - project_id: Project UUID - link_id: Link UUID - """ +async def update_link( + project_id: Annotated[str, Field(description="UUID of the project")], + link_id: Annotated[str, Field(description="UUID of the link to update")], + **kwargs: Any, +) -> list[dict[str, Any]]: + """Update a link's properties (suspend, filters, etc.).""" params = {"project_id": project_id, "link_id": link_id, **kwargs} return await asyncio.to_thread(_run_handler_sync, update_link_handler, params) @@ -382,52 +346,45 @@ async def list_templates() -> list[dict[str, Any]]: @mcp.tool() -async def get_template(template_id: str = None, name: str = None) -> list[dict[str, Any]]: - """Get detailed information about a specific template. - - Args: - template_id: Template UUID (optional if name is provided) - name: Template name (optional if template_id is provided) - """ +async def get_template( + template_id: Annotated[str | None, Field(description="Template UUID (optional if name is provided)")] = None, + name: Annotated[str | None, Field(description="Template name (optional if template_id is provided)")] = None, +) -> list[dict[str, Any]]: + """Get detailed information about a specific template.""" return await asyncio.to_thread(_run_handler_sync, get_template_handler, { "template_id": template_id, "name": name, }) @mcp.tool() -async def create_template(name: str, template_type: str, compute_id: str = "local") -> list[dict[str, Any]]: - """Create a new template. - - Args: - name: Template name - template_type: Template type (e.g. qemu, docker, dynamips) - compute_id: Compute ID (optional, default: local) - """ +async def create_template( + name: Annotated[str, Field(description="Template name")], + template_type: Annotated[str, Field(description="Template type (e.g. qemu, docker, dynamips)")], + compute_id: Annotated[str, Field(description="Compute ID (default: local)")] = "local", +) -> list[dict[str, Any]]: + """Create a new template.""" return await asyncio.to_thread(_run_handler_sync, create_template_handler, { "name": name, "template_type": template_type, "compute_id": compute_id, }) @mcp.tool() -async def update_template(template_id: str = None, name: str = None, **kwargs: Any) -> list[dict[str, Any]]: - """Update an existing template's properties. - - Args: - template_id: Template UUID (optional if name is provided) - name: Template name (optional if template_id is provided) - """ +async def update_template( + template_id: Annotated[str | None, Field(description="Template UUID (optional if name is provided)")] = None, + name: Annotated[str | None, Field(description="Template name (optional if template_id is provided)")] = None, + **kwargs: Any, +) -> list[dict[str, Any]]: + """Update an existing template's properties.""" params = {"template_id": template_id, "name": name, **kwargs} return await asyncio.to_thread(_run_handler_sync, update_template_handler, params) @mcp.tool() -async def delete_template(template_id: str = None, name: str = None) -> list[dict[str, Any]]: - """Delete a template. - - Args: - template_id: Template UUID (optional if name is provided) - name: Template name (optional if template_id is provided) - """ +async def delete_template( + template_id: Annotated[str | None, Field(description="Template UUID (optional if name is provided)")] = None, + name: Annotated[str | None, Field(description="Template name (optional if template_id is provided)")] = None, +) -> list[dict[str, Any]]: + """Delete a template.""" return await asyncio.to_thread(_run_handler_sync, delete_template_handler, { "template_id": template_id, "name": name, }) @@ -442,23 +399,19 @@ async def list_computes() -> list[dict[str, Any]]: @mcp.tool() -async def get_compute(compute_id: str = "local") -> list[dict[str, Any]]: - """Get detailed information about a compute node. - - Args: - compute_id: Compute ID (default: local) - """ +async def get_compute( + compute_id: Annotated[str, Field(description="Compute ID (default: local)")] = "local", +) -> list[dict[str, Any]]: + """Get detailed information about a compute node.""" return await asyncio.to_thread(_run_handler_sync, get_compute_handler, {"compute_id": compute_id}) @mcp.tool() -async def get_compute_images(emulator: str, compute_id: str = "local") -> list[dict[str, Any]]: - """List available images for an emulator on a compute node. - - Args: - emulator: Emulator type (e.g. qemu, iou, docker) - compute_id: Compute ID (default: local) - """ +async def get_compute_images( + emulator: Annotated[str, Field(description="Emulator type (e.g. qemu, iou, docker)")], + compute_id: Annotated[str, Field(description="Compute ID (default: local)")] = "local", +) -> list[dict[str, Any]]: + """List available images for an emulator on a compute node.""" return await asyncio.to_thread(_run_handler_sync, get_compute_images_handler, { "emulator": emulator, "compute_id": compute_id, })