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