mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-08-27 12:30:13 +03:00
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=<jwt> 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
This commit is contained in:
YueGuobin
parent
7086db4226
commit
55b3a7d622
@ -15,161 +15,208 @@
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
"""
|
||||
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=<jwt>",
|
||||
"endpoints": {
|
||||
"sse": "/v3/mcp/transport/sse?token=<jwt>",
|
||||
"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")
|
||||
|
||||
@ -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,
|
||||
},
|
||||
]
|
||||
|
||||
@ -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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user