feat: add Node and Link MCP tools, update copyright

- Add 9 node tools and 5 link tools
- Update copyright year to 2026, add author
This commit is contained in:
YueGuobin 2026-06-04 23:19:00 +08:00
parent 5b23ac81d0
commit cad216705a
No known key found for this signature in database
4 changed files with 630 additions and 2 deletions

View File

@ -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)
# ── Authwrapped SSE app ──────────────────────────────────────────────
def _make_auth_wrapper(inner_app):

View File

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

View File

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

View File

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