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
This commit is contained in:
YueGuobin 2026-06-04 13:53:05 +08:00
parent 6c7e8fb370
commit 7086db4226
No known key found for this signature in database
3 changed files with 464 additions and 0 deletions

View File

@ -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 <http://www.gnu.org/licenses/>.
"""
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)

View File

@ -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 <http://www.gnu.org/licenses/>.
"""
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()

View File

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