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