mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-11-03 14:01:15 +02:00
refactor: move MCP service from api/routes to agent package
MCP is an optional AI feature that already depends on agent.gns3_copilot (Gns3Connector, nornir/netmiko tools) and whose MCP_AVAILABLE feature flag lives in gns3server/agent. Moving it there collocates all AI features under one tree and removes AI code from the core REST routes. - git mv gns3server/api/routes/mcp -> gns3server/agent/mcp (no content changes) - api/server.py, core/tasks.py: update import paths - tests: tests/api/routes/mcp -> tests/agent/mcp, rewrite patch BASE and handler imports; fix MCP_DIR depth in test_tool_params.py - agent/__init__.py: probe the SDK via importlib.import_module so the top-level name "mcp" is not bound in the agent namespace (it would shadow the new gns3server.agent.mcp subpackage and break 'from gns3server.agent import mcp') - docs: update source file paths Verified: full suite 1567 passed; 82 MCP tools registered, SSE mounted at /v3/mcp/transport.
This commit is contained in:
21 files changed
+65
-62
No files matched your search
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,89 @@
|
||||
#
|
||||
# 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 appliance management.
|
||||
"""
|
||||
|
||||
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 ──────────────────────────────────────────────────────────
|
||||
|
||||
VALID_APPLIANCE_FIELDS = {
|
||||
"appliance_id", "name", "category", "description", "vendor_name",
|
||||
"vendor_url", "product_name", "product_url", "documentation_url",
|
||||
"status", "availability", "maintainer", "usage", "symbol",
|
||||
"images", "versions", "tags", "builtin",
|
||||
"first_port_name", "port_name_format", "port_segment_size",
|
||||
"linked_clone", "docker", "iou", "dynamips", "qemu",
|
||||
}
|
||||
|
||||
|
||||
def get_appliances_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
appliances = conn.http_call("get", f"{conn.base_url}/appliances").json()
|
||||
fields = params.get("fields")
|
||||
if fields:
|
||||
if not isinstance(fields, list):
|
||||
return {"error": "fields must be a list, e.g. [\"name\", \"category\"]"}
|
||||
invalid = [f for f in fields if f not in VALID_APPLIANCE_FIELDS]
|
||||
if invalid:
|
||||
return {
|
||||
"error": f"Unknown fields: {invalid}",
|
||||
"available_fields": sorted(VALID_APPLIANCE_FIELDS),
|
||||
}
|
||||
appliances = [{k: a[k] for k in fields if k in a} for a in appliances]
|
||||
return {"appliances": appliances, "count": len(appliances)}
|
||||
|
||||
|
||||
def get_appliance_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
appliance_id = params.get("appliance_id")
|
||||
if not appliance_id:
|
||||
return {"error": "appliance_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return conn.http_call("get", f"{conn.base_url}/appliances/{appliance_id}").json()
|
||||
|
||||
|
||||
def install_appliance_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
appliance_id = params.get("appliance_id")
|
||||
if not appliance_id:
|
||||
return {"error": "appliance_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
url = f"{conn.base_url}/appliances/{appliance_id}/install"
|
||||
request_params = {}
|
||||
version = params.get("version")
|
||||
if version:
|
||||
request_params["version"] = version
|
||||
result = conn.http_call("post", url, params=request_params).json()
|
||||
return {"message": f"Appliance {appliance_id} installation requested", "result": result}
|
||||
@@ -0,0 +1,95 @@
|
||||
#
|
||||
# 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 compute management.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
def list_computes_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
computes = conn.http_call("get", f"{conn.base_url}/computes").json()
|
||||
return {"computes": computes, "count": len(computes)}
|
||||
|
||||
|
||||
def get_compute_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
compute_id = params.get("compute_id")
|
||||
if not compute_id:
|
||||
return {"error": "compute_id is required (use compute_list to get the UUID)"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return conn.http_call("get", f"{conn.base_url}/computes/{compute_id}").json()
|
||||
|
||||
|
||||
def get_compute_images_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
emulator = params.get("emulator")
|
||||
compute_id = params.get("compute_id")
|
||||
if not emulator:
|
||||
return {"error": "emulator is required (e.g. qemu, iou, docker)"}
|
||||
if not compute_id:
|
||||
return {"error": "compute_id is required (use compute_list to get the UUID)"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
images = conn.http_call("get", f"{conn.base_url}/computes/{compute_id}/{emulator}/images").json()
|
||||
return {"images": images, "count": len(images)}
|
||||
|
||||
|
||||
COMPUTE_TOOLS = [
|
||||
{
|
||||
"name": "list_computes",
|
||||
"description": "List all compute nodes available to the server",
|
||||
"parameters": {"type": "object", "properties": {}},
|
||||
"handler": list_computes_handler,
|
||||
},
|
||||
{
|
||||
"name": "get_compute",
|
||||
"description": "Get detailed information about a compute node",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"compute_id": {"type": "string", "description": "Compute ID (default: local)"},
|
||||
},
|
||||
},
|
||||
"handler": get_compute_handler,
|
||||
},
|
||||
{
|
||||
"name": "get_compute_images",
|
||||
"description": "List available images for an emulator on a compute node",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"emulator": {"type": "string", "description": "Emulator type (e.g. qemu, iou, docker)"},
|
||||
"compute_id": {"type": "string", "description": "Compute ID (default: local)"},
|
||||
},
|
||||
"required": ["emulator"],
|
||||
},
|
||||
"handler": get_compute_images_handler,
|
||||
},
|
||||
]
|
||||
@@ -0,0 +1,150 @@
|
||||
#
|
||||
# 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 device configuration via Nornir + Netmiko.
|
||||
|
||||
These tools connect to network device consoles via telnet/SSH and execute
|
||||
configuration or diagnostic commands. Device connection info is automatically
|
||||
discovered from the project topology using the device's tags for device_type.
|
||||
|
||||
Prerequisites:
|
||||
- Device must be started (use node_start / node_start_all)
|
||||
- Device must have a 'device_type:<type>' tag set in GNS3
|
||||
(right-click → Configure → Tags → add 'device_type:cisco_ios_telnet')
|
||||
- Device must have a console port assigned
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from jinja2 import Template as JinjaTemplate, TemplateError as JinjaError
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _render_template(template: str, device_configs: list[dict], commands_field: str = "config_commands") -> list[dict]:
|
||||
"""Render a Jinja2 template for each device's vars into the specified commands field.
|
||||
|
||||
Entries with the same device_name are merged into a single entry
|
||||
so they share one Nornir session and avoid output fragmentation.
|
||||
|
||||
Each device in device_configs can have:
|
||||
- "vars": dict of template variables (rendered into commands_field)
|
||||
- commands_field: existing commands merged after rendering if present
|
||||
|
||||
Args:
|
||||
commands_field: field name for the rendered commands, e.g. "config_commands", "commands"
|
||||
"""
|
||||
jinja = JinjaTemplate(template)
|
||||
merged: dict[str, dict] = {}
|
||||
for dev in device_configs:
|
||||
name = dev.get("device_name")
|
||||
if not name:
|
||||
continue
|
||||
vars_data = dev.get("vars", {})
|
||||
if name not in merged:
|
||||
merged[name] = {"device_name": name, commands_field: list(dev.get(commands_field, []))}
|
||||
entry = merged[name]
|
||||
if vars_data:
|
||||
try:
|
||||
output = jinja.render(**vars_data)
|
||||
lines = [l for l in output.splitlines() if l.strip()]
|
||||
entry[commands_field].extend(lines)
|
||||
except JinjaError as e:
|
||||
error_msg = f"Template rendering failed for '{name}': {e}"
|
||||
log.error(error_msg)
|
||||
return [{"error": error_msg}]
|
||||
return list(merged.values())
|
||||
|
||||
|
||||
# ── Tool handlers ──────────────────────────────────────────────────────────
|
||||
|
||||
def device_config_send_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
"""Send configuration commands to network devices via console."""
|
||||
project_id = params.get("project_id")
|
||||
device_configs = params.get("device_configs")
|
||||
template = params.get("template")
|
||||
if not project_id or not device_configs:
|
||||
return [{"error": "project_id and device_configs are required"}]
|
||||
|
||||
if template:
|
||||
device_configs = _render_template(template, device_configs, commands_field="config_commands")
|
||||
if len(device_configs) == 1 and "error" in device_configs[0]:
|
||||
return device_configs
|
||||
|
||||
from gns3server.agent.gns3_copilot.tools_v2.config_tools_nornir import ExecuteMultipleDeviceConfigCommands
|
||||
|
||||
tool = ExecuteMultipleDeviceConfigCommands()
|
||||
input_data = json.dumps({
|
||||
"project_id": project_id,
|
||||
"device_configs": device_configs,
|
||||
})
|
||||
return tool._run(
|
||||
input_data,
|
||||
jwt_token=gns3_ctx["jwt_token"],
|
||||
url=gns3_ctx["server_url"],
|
||||
)
|
||||
|
||||
|
||||
def device_show_run_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
"""Run read-only diagnostic (show) commands on network devices."""
|
||||
project_id = params.get("project_id")
|
||||
device_configs = params.get("device_configs")
|
||||
template = params.get("template")
|
||||
if not project_id or not device_configs:
|
||||
return [{"error": "project_id and device_configs (list of {device_name, commands}) are required"}]
|
||||
|
||||
if template:
|
||||
device_configs = _render_template(template, device_configs, commands_field="commands")
|
||||
if len(device_configs) == 1 and "error" in device_configs[0]:
|
||||
return device_configs
|
||||
|
||||
from gns3server.agent.gns3_copilot.tools_v2.display_tools_nornir import ExecuteMultipleDeviceCommands
|
||||
|
||||
tool = ExecuteMultipleDeviceCommands()
|
||||
input_data = json.dumps({
|
||||
"project_id": project_id,
|
||||
"device_configs": device_configs,
|
||||
})
|
||||
return tool._run(
|
||||
input_data,
|
||||
jwt_token=gns3_ctx["jwt_token"],
|
||||
url=gns3_ctx["server_url"],
|
||||
)
|
||||
|
||||
|
||||
def vpcs_config_set_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
"""Configure VPCS devices (set IP, gateway, etc.)."""
|
||||
project_id = params.get("project_id")
|
||||
device_configs = params.get("device_configs")
|
||||
if not project_id or not device_configs:
|
||||
return [{"error": "project_id and device_configs are required"}]
|
||||
|
||||
from gns3server.agent.gns3_copilot.tools_v2.vpcs_tools_netmiko import VPCSCommands
|
||||
|
||||
tool = VPCSCommands()
|
||||
input_data = json.dumps({
|
||||
"project_id": project_id,
|
||||
"device_configs": device_configs,
|
||||
})
|
||||
return tool._run(
|
||||
input_data,
|
||||
jwt_token=gns3_ctx["jwt_token"],
|
||||
url=gns3_ctx["server_url"],
|
||||
)
|
||||
@@ -0,0 +1,96 @@
|
||||
#
|
||||
# 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 drawing management.
|
||||
"""
|
||||
|
||||
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_drawings_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)
|
||||
drawings = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/drawings").json()
|
||||
return {"drawings": drawings, "count": len(drawings)}
|
||||
|
||||
|
||||
def create_drawing_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
svg = params.get("svg")
|
||||
if not project_id or not svg:
|
||||
return {"error": "project_id and svg are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
data = {
|
||||
"svg": svg,
|
||||
"x": params.get("x", 0),
|
||||
"y": params.get("y", 0),
|
||||
"z": params.get("z", 0),
|
||||
"locked": params.get("locked", False),
|
||||
"rotation": params.get("rotation", 0),
|
||||
}
|
||||
result = conn.http_call("post", f"{conn.base_url}/projects/{project_id}/drawings", json_data=data).json()
|
||||
return {"message": "Drawing created", "drawing": result}
|
||||
|
||||
|
||||
def get_drawing_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
drawing_id = params.get("drawing_id")
|
||||
if not project_id or not drawing_id:
|
||||
return {"error": "project_id and drawing_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return conn.http_call("get", f"{conn.base_url}/projects/{project_id}/drawings/{drawing_id}").json()
|
||||
|
||||
|
||||
def update_drawing_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
drawing_id = params.get("drawing_id")
|
||||
if not project_id or not drawing_id:
|
||||
return {"error": "project_id and drawing_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
data = {k: v for k, v in params.items() if k not in ("project_id", "drawing_id") and v is not None}
|
||||
return conn.http_call("put", f"{conn.base_url}/projects/{project_id}/drawings/{drawing_id}", json_data=data).json()
|
||||
|
||||
|
||||
def delete_drawing_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
drawing_id = params.get("drawing_id")
|
||||
if not project_id or not drawing_id:
|
||||
return {"error": "project_id and drawing_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("delete", f"{conn.base_url}/projects/{project_id}/drawings/{drawing_id}")
|
||||
return {"message": f"Drawing {drawing_id} deleted", "drawing_id": drawing_id}
|
||||
@@ -0,0 +1,77 @@
|
||||
#
|
||||
# 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 image management.
|
||||
"""
|
||||
|
||||
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_images_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
images = conn.http_call("get", f"{conn.base_url}/images").json()
|
||||
return {"images": images, "count": len(images)}
|
||||
|
||||
|
||||
def get_image_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
image_id = params.get("image_id")
|
||||
if not image_id:
|
||||
return {"error": "image_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return conn.http_call("get", f"{conn.base_url}/images/{image_id}").json()
|
||||
|
||||
|
||||
def delete_image_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
image_id = params.get("image_id")
|
||||
if not image_id:
|
||||
return {"error": "image_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("delete", f"{conn.base_url}/images/{image_id}")
|
||||
return {"message": f"Image {image_id} deleted", "image_id": image_id}
|
||||
|
||||
|
||||
def prune_images_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
# Returns 204 No Content on success (empty body, no .json())
|
||||
conn.http_call("delete", f"{conn.base_url}/images/prune")
|
||||
return {"message": "Unused images pruned"}
|
||||
|
||||
|
||||
def install_images_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
# Returns 204 No Content on success (empty body, no .json())
|
||||
conn.http_call("post", f"{conn.base_url}/images/install")
|
||||
return {"message": "Image installation completed"}
|
||||
@@ -0,0 +1,633 @@
|
||||
#
|
||||
# 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
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
import logging
|
||||
|
||||
from gns3server.services import auth_service
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
BATCH_MAX_WORKERS = 100
|
||||
|
||||
|
||||
# ── 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,
|
||||
)
|
||||
|
||||
|
||||
def _normalize_link_nodes(nodes) -> list[dict[str, Any]]:
|
||||
"""
|
||||
Normalize link node entries, accepting both standard object format and
|
||||
compact array format to reduce token usage.
|
||||
|
||||
Standard: [{"node_id": "uuid", "adapter_number": 0, "port_number": 0}]
|
||||
Compact: ["uuid", 0, 0, "uuid", 0, 0]
|
||||
|
||||
Returns the normalized list, or raises ValueError with a clear message
|
||||
on format errors so the AI can self-correct.
|
||||
"""
|
||||
if not nodes:
|
||||
return nodes
|
||||
if not isinstance(nodes, list):
|
||||
raise ValueError(f"nodes must be a list, got {type(nodes).__name__}: {nodes}")
|
||||
# Standard object format: [{"node_id": "...", ...}]
|
||||
if isinstance(nodes[0], dict):
|
||||
return nodes
|
||||
# Compact array format: ["uuid", ad, pt, "uuid", ad, pt]
|
||||
if all(not isinstance(n, dict) for n in nodes):
|
||||
if len(nodes) != 6:
|
||||
raise ValueError(
|
||||
f"Compact link format requires exactly 6 elements "
|
||||
f"[node_id, adapter, port, node_id, adapter, port], "
|
||||
f"but got {len(nodes)} elements: {nodes}"
|
||||
)
|
||||
if not isinstance(nodes[0], str) or not isinstance(nodes[3], str):
|
||||
raise ValueError(
|
||||
f"Compact link format expects node_id (string) at positions 0 and 3, "
|
||||
f"got types {type(nodes[0]).__name__} and {type(nodes[3]).__name__}: {nodes}"
|
||||
)
|
||||
return [
|
||||
{"node_id": nodes[0], "adapter_number": nodes[1], "port_number": nodes[2]},
|
||||
{"node_id": nodes[3], "adapter_number": nodes[4], "port_number": nodes[5]},
|
||||
]
|
||||
raise ValueError(
|
||||
f"Unrecognized link nodes format. "
|
||||
f"Use standard [{{\"node_id\":\"..\",\"adapter_number\":0,\"port_number\":0}},...] "
|
||||
f"or compact [\"id\",0,0,\"id\",0,0], got: {nodes}"
|
||||
)
|
||||
|
||||
|
||||
# ── Tool handlers ──────────────────────────────────────────────────────────
|
||||
|
||||
VALID_LINK_FIELDS = {
|
||||
"link_id", "project_id", "link_type", "nodes", "suspend",
|
||||
"link_style", "filters", "show_filters_icon",
|
||||
"capturing", "capture_file_name", "capture_file_path",
|
||||
"capture_compute_id", "wireshark",
|
||||
}
|
||||
|
||||
|
||||
LINK_DEFAULT_FIELDS = ["link_id", "link_type", "nodes"]
|
||||
|
||||
|
||||
def _filter_link_response(link: dict, fields: list[str] = None) -> dict:
|
||||
"""Filter link response to only include requested fields."""
|
||||
if not fields:
|
||||
fields = LINK_DEFAULT_FIELDS
|
||||
return {k: link[k] for k in fields if k in link}
|
||||
|
||||
|
||||
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.http_call("get", f"{conn.base_url}/projects/{project_id}/links").json()
|
||||
fields = params.get("fields")
|
||||
if fields:
|
||||
if not isinstance(fields, list):
|
||||
return {"error": "fields must be a list, e.g. [\"link_id\", \"nodes\"]"}
|
||||
invalid = [f for f in fields if f not in VALID_LINK_FIELDS]
|
||||
if invalid:
|
||||
return {
|
||||
"error": f"Unknown fields: {invalid}",
|
||||
"available_fields": sorted(VALID_LINK_FIELDS),
|
||||
}
|
||||
links = [{k: l[k] for k in fields if k in l} for l in links]
|
||||
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.http_call("get", f"{conn.base_url}/projects/{project_id}/links/{link_id}").json()
|
||||
|
||||
|
||||
def create_link_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"}
|
||||
|
||||
fields = params.get("fields")
|
||||
if fields is not None and not isinstance(fields, list):
|
||||
return {"error": "fields must be a list, e.g. [\"link_id\", \"nodes\"]"}
|
||||
|
||||
links = params.get("links")
|
||||
# Batch mode: links=[{nodes, link_type?, filters?, suspend?}]
|
||||
if links is not None:
|
||||
if not isinstance(links, list) or not links:
|
||||
return {"error": "links must be a non-empty array"}
|
||||
results = []
|
||||
conn = _get_connector(gns3_ctx)
|
||||
def _create_one(link_data):
|
||||
raw_nodes = link_data.get("nodes")
|
||||
if not raw_nodes:
|
||||
return {"status": "error", "error": "nodes is required for each link"}
|
||||
try:
|
||||
body = {"nodes": _normalize_link_nodes(raw_nodes)}
|
||||
if link_data.get("link_type"):
|
||||
body["link_type"] = link_data["link_type"]
|
||||
if link_data.get("filters"):
|
||||
body["filters"] = link_data["filters"]
|
||||
if link_data.get("suspend"):
|
||||
body["suspend"] = link_data["suspend"]
|
||||
url = f"{conn.base_url}/projects/{project_id}/links"
|
||||
resp = conn.http_call("post", url, json_data=body).json()
|
||||
return {"status": "success", "link": _filter_link_response(resp, fields)}
|
||||
except Exception as e:
|
||||
return {"status": "error", "error": str(e)}
|
||||
with ThreadPoolExecutor(max_workers=min(len(links), BATCH_MAX_WORKERS)) as pool:
|
||||
futures = {pool.submit(_create_one, l): l for l in links}
|
||||
for future in as_completed(futures):
|
||||
results.append(future.result())
|
||||
return results
|
||||
|
||||
# Single mode
|
||||
nodes = params.get("nodes")
|
||||
if not nodes:
|
||||
return {"error": "nodes is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
data = {"nodes": _normalize_link_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"
|
||||
resp = conn.http_call("post", url, json_data=data).json()
|
||||
return _filter_link_response(resp, fields)
|
||||
|
||||
|
||||
def delete_link_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"}
|
||||
link_ids = params.get("link_ids")
|
||||
if link_ids:
|
||||
if not isinstance(link_ids, list):
|
||||
return {"error": "link_ids must be a list"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
def _del(lid):
|
||||
try:
|
||||
conn.http_call("delete", f"{conn.base_url}/projects/{project_id}/links/{lid}")
|
||||
return {"link_id": lid, "status": "deleted"}
|
||||
except Exception as e:
|
||||
return {"link_id": lid, "status": "error", "error": str(e)}
|
||||
with ThreadPoolExecutor(max_workers=min(len(link_ids), BATCH_MAX_WORKERS)) as pool:
|
||||
return list(pool.map(_del, link_ids))
|
||||
link_id = params.get("link_id")
|
||||
if not link_id:
|
||||
return {"error": "link_id or link_ids is 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)
|
||||
|
||||
# Extract update parameters - handle nested kwargs structure from MCP clients
|
||||
if "kwargs" in params and isinstance(params["kwargs"], dict):
|
||||
update_data = params["kwargs"]
|
||||
else:
|
||||
update_data = {k: v for k, v in params.items() if k not in ("project_id", "link_id", "kwargs")}
|
||||
|
||||
url = f"{conn.base_url}/projects/{project_id}/links/{link_id}"
|
||||
return conn.http_call("put", url, json_data=update_data).json()
|
||||
|
||||
|
||||
# ── Link capture / reset handlers ──────────────────────────────────────
|
||||
|
||||
|
||||
def reset_link_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"}
|
||||
link_ids = params.get("link_ids")
|
||||
if link_ids:
|
||||
if not isinstance(link_ids, list):
|
||||
return {"error": "link_ids must be a list"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
def _rst(lid):
|
||||
try:
|
||||
url = f"{conn.base_url}/projects/{project_id}/links/{lid}/reset"
|
||||
r = conn.http_call("post", url).json()
|
||||
return {"link_id": lid, "status": "reset", "link": r}
|
||||
except Exception as e:
|
||||
return {"link_id": lid, "status": "error", "error": str(e)}
|
||||
with ThreadPoolExecutor(max_workers=min(len(link_ids), BATCH_MAX_WORKERS)) as pool:
|
||||
return list(pool.map(_rst, link_ids))
|
||||
link_id = params.get("link_id")
|
||||
if not link_id:
|
||||
return {"error": "link_id or link_ids is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
url = f"{conn.base_url}/projects/{project_id}/links/{link_id}/reset"
|
||||
result = conn.http_call("post", url).json()
|
||||
return {"message": f"Link {link_id} reset", "link": result}
|
||||
|
||||
|
||||
def _batch_capture(project_id, link_ids, action, data_builder, conn):
|
||||
"""Helper for batch capture start/stop."""
|
||||
def _act(lid):
|
||||
try:
|
||||
url = f"{conn.base_url}/projects/{project_id}/links/{lid}/capture/{action}"
|
||||
kwargs = data_builder(lid) if data_builder else {}
|
||||
conn.http_call("post", url, **kwargs)
|
||||
return {"link_id": lid, "status": "success"}
|
||||
except Exception as e:
|
||||
return {"link_id": lid, "status": "error", "error": str(e)}
|
||||
with ThreadPoolExecutor(max_workers=min(len(link_ids), BATCH_MAX_WORKERS)) as pool:
|
||||
return list(pool.map(_act, link_ids))
|
||||
|
||||
|
||||
def start_capture_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"}
|
||||
link_ids = params.get("link_ids")
|
||||
if link_ids:
|
||||
if not isinstance(link_ids, list):
|
||||
return {"error": "link_ids must be a list"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
dlt = params.get("data_link_type", "DLT_EN10MB")
|
||||
ws = params.get("wireshark", False)
|
||||
fname = params.get("capture_file_name")
|
||||
def _build(lid):
|
||||
data = {"data_link_type": dlt, "wireshark": ws}
|
||||
if fname:
|
||||
data["capture_file_name"] = fname
|
||||
return {"json_data": data}
|
||||
return _batch_capture(project_id, link_ids, "start", _build, conn)
|
||||
link_id = params.get("link_id")
|
||||
if not link_id:
|
||||
return {"error": "link_id or link_ids is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
data = {
|
||||
"data_link_type": params.get("data_link_type", "DLT_EN10MB"),
|
||||
"wireshark": params.get("wireshark", False),
|
||||
}
|
||||
if params.get("capture_file_name"):
|
||||
data["capture_file_name"] = params["capture_file_name"]
|
||||
url = f"{conn.base_url}/projects/{project_id}/links/{link_id}/capture/start"
|
||||
result = conn.http_call("post", url, json_data=data).json()
|
||||
return {"message": f"Capture started on link {link_id}", "link": result}
|
||||
|
||||
|
||||
def stop_capture_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"}
|
||||
link_ids = params.get("link_ids")
|
||||
if link_ids:
|
||||
if not isinstance(link_ids, list):
|
||||
return {"error": "link_ids must be a list"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return _batch_capture(project_id, link_ids, "stop", None, conn)
|
||||
link_id = params.get("link_id")
|
||||
if not link_id:
|
||||
return {"error": "link_id or link_ids is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
url = f"{conn.base_url}/projects/{project_id}/links/{link_id}/capture/stop"
|
||||
conn.http_call("post", url)
|
||||
return {"message": f"Capture stopped on link {link_id}", "link_id": link_id}
|
||||
|
||||
|
||||
def download_capture_file_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"}
|
||||
username = gns3_ctx.get("jwt_username")
|
||||
download_token = auth_service.create_access_token(username, token_version=gns3_ctx.get("jwt_token_version", 0), expires_in=10) if username else None
|
||||
|
||||
link_ids = params.get("link_ids")
|
||||
if link_ids:
|
||||
if not isinstance(link_ids, list):
|
||||
return {"error": "link_ids must be a list"}
|
||||
results = []
|
||||
for lid in link_ids:
|
||||
url = f"{gns3_ctx['server_url']}/v3/projects/{project_id}/links/{lid}/capture/file"
|
||||
entry = {"link_id": lid, "download_url": url}
|
||||
if download_token:
|
||||
cmd = f"curl -L -o capture_{lid}.pcap -H 'Authorization: Bearer {download_token}' '{url}'"
|
||||
entry["curl_command"] = cmd
|
||||
results.append(entry)
|
||||
return {"downloads": results, "count": len(results), "note": "Files are in pcap format. Links include a 10-minute token."}
|
||||
|
||||
link_id = params.get("link_id")
|
||||
if not link_id:
|
||||
return {"error": "link_id or link_ids is required"}
|
||||
download_url = f"{gns3_ctx['server_url']}/v3/projects/{project_id}/links/{link_id}/capture/file"
|
||||
result = {
|
||||
"link_id": link_id,
|
||||
"download_url": download_url,
|
||||
"note": "The file is in pcap format and can be analyzed with Wireshark or tcpdump.",
|
||||
}
|
||||
if download_token:
|
||||
result["curl_command"] = f"curl -L -o capture.pcap -H 'Authorization: Bearer {download_token}' '{download_url}'"
|
||||
result["note"] += " The download link includes a 10-minute token."
|
||||
return result
|
||||
|
||||
|
||||
# ── Marker (traffic-insight) handlers ──────────────────────────────────
|
||||
|
||||
|
||||
def link_marker_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
"""
|
||||
Manage traffic-insight markers on a specific link.
|
||||
|
||||
Actions:
|
||||
- create: POST /projects/{pid}/links/{lid}/markers
|
||||
- update: PUT /projects/{pid}/links/{lid}/markers/{name}
|
||||
- delete: DELETE /projects/{pid}/links/{lid}/markers/{name}
|
||||
"""
|
||||
project_id = params.get("project_id")
|
||||
link_id = params.get("link_id")
|
||||
action = params.get("action")
|
||||
if not all([project_id, link_id, action]):
|
||||
return {"error": "project_id, link_id and action are required"}
|
||||
if action not in ("create", "update", "delete"):
|
||||
return {"error": f"Unknown action: {action}. Supported: create, update, delete"}
|
||||
|
||||
conn = _get_connector(gns3_ctx)
|
||||
base = f"{conn.base_url}/projects/{project_id}/links/{link_id}/markers"
|
||||
|
||||
if action == "create":
|
||||
bpf = params.get("bpf")
|
||||
if not bpf:
|
||||
return {"error": "bpf is required for create action"}
|
||||
body: dict[str, Any] = {"bpf": bpf}
|
||||
for opt in ("name", "tag", "capture_node_id", "color", "highlight_duration"):
|
||||
if params.get(opt) is not None:
|
||||
body[opt] = params[opt]
|
||||
# direction: "tx"/"rx" set a one-way filter; "both"/omitted = no filter.
|
||||
if params.get("direction") in ("tx", "rx"):
|
||||
body["direction"] = params["direction"]
|
||||
return conn.http_call("post", base, json_data=body).json()
|
||||
|
||||
marker_name = params.get("marker_name")
|
||||
if not marker_name:
|
||||
return {"error": "marker_name is required for update/delete actions"}
|
||||
|
||||
url = f"{base}/{marker_name}"
|
||||
|
||||
if action == "update":
|
||||
body = {}
|
||||
for opt in ("bpf", "tag", "enabled", "color", "highlight_duration"):
|
||||
if params.get(opt) is not None:
|
||||
body[opt] = params[opt]
|
||||
# direction tri-state: omitted=preserve, "tx"/"rx"=set, "both"=clear (→ null).
|
||||
direction = params.get("direction")
|
||||
if direction == "both":
|
||||
body["direction"] = None
|
||||
elif direction in ("tx", "rx"):
|
||||
body["direction"] = direction
|
||||
if not body:
|
||||
return {"error": "At least one update field is required (bpf, tag, enabled, direction, color, highlight_duration)"}
|
||||
return conn.http_call("put", url, json_data=body).json()
|
||||
|
||||
# action == "delete"
|
||||
conn.http_call("delete", url)
|
||||
return {"message": f"Marker '{marker_name}' deleted from link {link_id}", "link_id": link_id, "marker_name": marker_name}
|
||||
|
||||
|
||||
def marker_definition_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
"""
|
||||
Manage project-level marker definitions (auto-fanout to all links).
|
||||
|
||||
Actions:
|
||||
- create: POST /projects/{pid}/marker-definitions → fans out global-{name} to every link
|
||||
- update: PUT /projects/{pid}/marker-definitions/{name}
|
||||
- delete: DELETE /projects/{pid}/marker-definitions/{name}
|
||||
- list: GET /projects/{pid}/marker-definitions
|
||||
"""
|
||||
project_id = params.get("project_id")
|
||||
action = params.get("action")
|
||||
if not all([project_id, action]):
|
||||
return {"error": "project_id and action are required"}
|
||||
if action not in ("create", "update", "delete", "list"):
|
||||
return {"error": f"Unknown action: {action}. Supported: create, update, delete, list"}
|
||||
|
||||
conn = _get_connector(gns3_ctx)
|
||||
base = f"{conn.base_url}/projects/{project_id}/marker-definitions"
|
||||
|
||||
if action == "list":
|
||||
return conn.http_call("get", base).json()
|
||||
|
||||
if action == "create":
|
||||
bpf = params.get("bpf")
|
||||
if not bpf:
|
||||
return {"error": "bpf is required for create action"}
|
||||
body: dict[str, Any] = {"bpf": bpf}
|
||||
for opt in ("name", "tag", "color", "highlight_duration", "data_link_type"):
|
||||
if params.get(opt) is not None:
|
||||
body[opt] = params[opt]
|
||||
# No direction: a definition fans out to every link and auto-selects its
|
||||
# capture node on each, so tx/rx (which is relative to that node) has no
|
||||
# consistent meaning. Encode direction in the BPF instead.
|
||||
return conn.http_call("post", base, json_data=body).json()
|
||||
|
||||
def_name = params.get("def_name")
|
||||
if not def_name:
|
||||
return {"error": "def_name is required for update/delete actions"}
|
||||
|
||||
url = f"{base}/{def_name}"
|
||||
|
||||
if action == "update":
|
||||
body = {}
|
||||
for opt in ("bpf", "tag", "color", "highlight_duration", "data_link_type"):
|
||||
if params.get(opt) is not None:
|
||||
body[opt] = params[opt]
|
||||
if not body:
|
||||
return {"error": "At least one update field is required (bpf, tag, color, highlight_duration, data_link_type)"}
|
||||
return conn.http_call("put", url, json_data=body).json()
|
||||
|
||||
# action == "delete"
|
||||
conn.http_call("delete", url)
|
||||
return {"message": f"Marker definition '{def_name}' deleted", "project_id": project_id, "def_name": def_name}
|
||||
|
||||
|
||||
# ── 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). Must use array format: frequency_drop: [N], packet_loss: [rate], delay: [ms, jitter], corrupt: [rate], bpf: [expression]"
|
||||
},
|
||||
},
|
||||
"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). Must use array format: frequency_drop: [N], packet_loss: [rate], delay: [ms, jitter], corrupt: [rate], bpf: [expression]. Example: {\"frequency_drop\": [10], \"packet_loss\": [5]}"
|
||||
},
|
||||
},
|
||||
"required": ["project_id", "link_id"],
|
||||
},
|
||||
"handler": update_link_handler,
|
||||
},
|
||||
{
|
||||
"name": "reset_link",
|
||||
"description": "Reset a link, clearing its state (counters, filters, etc.)",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {"type": "string", "description": "Project UUID"},
|
||||
"link_id": {"type": "string", "description": "Link UUID"},
|
||||
},
|
||||
"required": ["project_id", "link_id"],
|
||||
},
|
||||
"handler": reset_link_handler,
|
||||
},
|
||||
{
|
||||
"name": "start_capture",
|
||||
"description": "Start packet capture on a link. The capture file can later be downloaded with download_capture_file.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {"type": "string", "description": "Project UUID"},
|
||||
"link_id": {"type": "string", "description": "Link UUID"},
|
||||
"data_link_type": {"type": "string", "description": "Data link type (optional, default: DLT_EN10MB)"},
|
||||
"capture_file_name": {"type": "string", "description": "Capture file name (optional)"},
|
||||
"wireshark": {"type": "boolean", "description": "Open Wireshark automatically (optional, default: false)"},
|
||||
},
|
||||
"required": ["project_id", "link_id"],
|
||||
},
|
||||
"handler": start_capture_handler,
|
||||
},
|
||||
{
|
||||
"name": "stop_capture",
|
||||
"description": "Stop packet capture on a link. After stopping, the capture file can be downloaded.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {"type": "string", "description": "Project UUID"},
|
||||
"link_id": {"type": "string", "description": "Link UUID"},
|
||||
},
|
||||
"required": ["project_id", "link_id"],
|
||||
},
|
||||
"handler": stop_capture_handler,
|
||||
},
|
||||
{
|
||||
"name": "download_capture_file",
|
||||
"description": "Get the download URL and instructions for a PCAP capture file from a link. "
|
||||
"Use the returned curl command to download the file. "
|
||||
"The PCAP file can be analyzed with Wireshark or tcpdump.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {"type": "string", "description": "Project UUID"},
|
||||
"link_id": {"type": "string", "description": "Link UUID"},
|
||||
},
|
||||
"required": ["project_id", "link_id"],
|
||||
},
|
||||
"handler": download_capture_file_handler,
|
||||
},
|
||||
]
|
||||
@@ -0,0 +1,678 @@
|
||||
#
|
||||
# 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
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
import hashlib
|
||||
import logging
|
||||
|
||||
from gns3server.services import auth_service
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
BATCH_MAX_WORKERS = 100
|
||||
|
||||
# ── Constants ──────────────────────────────────────────────────────────────
|
||||
|
||||
# Maximum bytes to return from get_node_file (safety net).
|
||||
# Larger files are truncated with a truncated=True flag.
|
||||
MAX_NODE_FILE_BYTES = 50 * 1024 # 50 KiB
|
||||
|
||||
|
||||
# ── 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.http_call("get", f"{conn.base_url}/projects/{project_id}/nodes").json()
|
||||
fields = params.get("fields")
|
||||
if fields:
|
||||
if not isinstance(fields, list):
|
||||
return {"error": "fields must be a list of field names, e.g. [\"name\", \"status\"]"}
|
||||
invalid = [f for f in fields if f not in VALID_NODE_FIELDS]
|
||||
if invalid:
|
||||
return {
|
||||
"error": f"Unknown fields: {invalid}",
|
||||
"available_fields": sorted(VALID_NODE_FIELDS),
|
||||
}
|
||||
nodes = [{k: n[k] for k in fields if k in n} for n in nodes]
|
||||
return {"nodes": nodes, "count": len(nodes)}
|
||||
|
||||
|
||||
VALID_NODE_FIELDS = {
|
||||
# NodeBase
|
||||
"compute_id", "name", "node_type", "node_id",
|
||||
"console", "console_type", "console_auto_start",
|
||||
"aux", "aux_type", "properties", "label", "symbol",
|
||||
"x", "y", "z", "locked",
|
||||
"port_name_format", "port_segment_size", "first_port_name",
|
||||
"custom_adapters", "tags",
|
||||
# Node
|
||||
"template_id", "project_id", "node_directory", "status",
|
||||
"command_line", "width", "height", "ports", "console_host",
|
||||
}
|
||||
|
||||
|
||||
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)
|
||||
node = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}").json()
|
||||
|
||||
fields = params.get("fields")
|
||||
if fields:
|
||||
if not isinstance(fields, list):
|
||||
return {"error": "fields must be a list of field names, e.g. [\"name\", \"status\"]"}
|
||||
invalid = [f for f in fields if f not in VALID_NODE_FIELDS]
|
||||
if invalid:
|
||||
return {
|
||||
"error": f"Unknown fields: {invalid}",
|
||||
"available_fields": sorted(VALID_NODE_FIELDS),
|
||||
}
|
||||
return {k: node[k] for k in fields if k in node}
|
||||
|
||||
return node
|
||||
|
||||
|
||||
def _batch_lifecycle(project_id, node_ids, action, conn, action_label):
|
||||
"""Helper to run a lifecycle action on multiple nodes in parallel."""
|
||||
def _act(nid):
|
||||
try:
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/{nid}/{action}")
|
||||
return {"node_id": nid, "status": "success", "message": f"Node {nid} {action_label}"}
|
||||
except Exception as e:
|
||||
return {"node_id": nid, "status": "error", "error": str(e)}
|
||||
with ThreadPoolExecutor(max_workers=min(len(node_ids), BATCH_MAX_WORKERS)) as pool:
|
||||
return list(pool.map(_act, node_ids))
|
||||
|
||||
|
||||
def start_node_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"}
|
||||
node_ids = params.get("node_ids")
|
||||
if node_ids:
|
||||
if not isinstance(node_ids, list):
|
||||
return {"error": "node_ids must be a list"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return _batch_lifecycle(project_id, node_ids, "start", conn, "started")
|
||||
node_id = params.get("node_id")
|
||||
if not node_id:
|
||||
return {"error": "node_id or node_ids is 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")
|
||||
if not project_id:
|
||||
return {"error": "project_id is required"}
|
||||
node_ids = params.get("node_ids")
|
||||
if node_ids:
|
||||
if not isinstance(node_ids, list):
|
||||
return {"error": "node_ids must be a list"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return _batch_lifecycle(project_id, node_ids, "stop", conn, "stopped")
|
||||
node_id = params.get("node_id")
|
||||
if not node_id:
|
||||
return {"error": "node_id or node_ids is 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 suspend_node_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"}
|
||||
node_ids = params.get("node_ids")
|
||||
if node_ids:
|
||||
if not isinstance(node_ids, list):
|
||||
return {"error": "node_ids must be a list"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return _batch_lifecycle(project_id, node_ids, "suspend", conn, "suspended")
|
||||
node_id = params.get("node_id")
|
||||
if not node_id:
|
||||
return {"error": "node_id or node_ids is 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 _filter_node_response(node: dict, fields: list[str] = None) -> dict:
|
||||
"""Filter node response to only include requested fields."""
|
||||
if not fields:
|
||||
fields = ["node_id", "name", "node_type", "status", "console"]
|
||||
return {k: node[k] for k in fields if k in node}
|
||||
|
||||
|
||||
def create_node_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"}
|
||||
|
||||
fields = params.get("fields")
|
||||
if fields is not None and not isinstance(fields, list):
|
||||
return {"error": "fields must be a list, e.g. [\"node_id\", \"name\"]"}
|
||||
|
||||
nodes = params.get("nodes")
|
||||
# Batch mode: nodes=[{template_id?, x, y, name?, compute_id?}]
|
||||
# When top-level template_id is set, it applies to all nodes as a default
|
||||
if nodes is not None:
|
||||
if not isinstance(nodes, list) or not nodes:
|
||||
return {"error": "nodes must be a non-empty array"}
|
||||
default_tid = params.get("template_id")
|
||||
results = []
|
||||
conn = _get_connector(gns3_ctx)
|
||||
def _create_one(node_data):
|
||||
tid = node_data.get("template_id", default_tid)
|
||||
if not tid:
|
||||
return {"template_id": tid, "status": "error", "error": "template_id is required"}
|
||||
try:
|
||||
url = f"{conn.base_url}/projects/{project_id}/templates/{tid}"
|
||||
body = {
|
||||
"x": node_data.get("x", 0),
|
||||
"y": node_data.get("y", 0),
|
||||
"compute_id": node_data.get("compute_id", "local"),
|
||||
}
|
||||
node_name = node_data.get("name")
|
||||
if node_name:
|
||||
body["name"] = node_name
|
||||
resp = conn.http_call("post", url, json_data=body).json()
|
||||
return {"template_id": tid, "status": "success", "node": _filter_node_response(resp, fields)}
|
||||
except Exception as e:
|
||||
return {"template_id": tid, "status": "error", "error": str(e)}
|
||||
with ThreadPoolExecutor(max_workers=min(len(nodes), BATCH_MAX_WORKERS)) as pool:
|
||||
futures = {pool.submit(_create_one, n): n for n in nodes}
|
||||
for future in as_completed(futures):
|
||||
results.append(future.result())
|
||||
return results
|
||||
|
||||
# Single mode
|
||||
template_id = params.get("template_id")
|
||||
if not template_id:
|
||||
return {"error": "template_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
data = {
|
||||
"x": params.get("x", 0),
|
||||
"y": params.get("y", 0),
|
||||
"compute_id": params.get("compute_id", "local"),
|
||||
}
|
||||
node_name = params.get("name")
|
||||
if node_name:
|
||||
data["name"] = node_name
|
||||
url = f"{conn.base_url}/projects/{project_id}/templates/{template_id}"
|
||||
resp = conn.http_call("post", url, json_data=data).json()
|
||||
return _filter_node_response(resp, fields)
|
||||
|
||||
|
||||
def delete_node_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"}
|
||||
node_ids = params.get("node_ids")
|
||||
if node_ids:
|
||||
if not isinstance(node_ids, list):
|
||||
return {"error": "node_ids must be a list"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
def _del(nid):
|
||||
try:
|
||||
conn.http_call("delete", f"{conn.base_url}/projects/{project_id}/nodes/{nid}")
|
||||
return {"node_id": nid, "status": "deleted"}
|
||||
except Exception as e:
|
||||
return {"node_id": nid, "status": "error", "error": str(e)}
|
||||
with ThreadPoolExecutor(max_workers=min(len(node_ids), BATCH_MAX_WORKERS)) as pool:
|
||||
return list(pool.map(_del, node_ids))
|
||||
node_id = params.get("node_id")
|
||||
if not node_id:
|
||||
return {"error": "node_id or node_ids is 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)
|
||||
|
||||
# Extract update parameters - handle nested kwargs structure from MCP clients
|
||||
if "kwargs" in params and isinstance(params["kwargs"], dict):
|
||||
update_data = params["kwargs"]
|
||||
else:
|
||||
update_data = {k: v for k, v in params.items() if k not in ("project_id", "node_id", "kwargs")}
|
||||
|
||||
url = f"{conn.base_url}/projects/{project_id}/nodes/{node_id}"
|
||||
return conn.http_call("put", url, json_data=update_data).json()
|
||||
|
||||
|
||||
def get_node_console_info_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)
|
||||
node = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}").json()
|
||||
|
||||
console_type = node.get("console_type", "unknown")
|
||||
# Short-lived JWT for the WebSocket URL (10 min)
|
||||
username = gns3_ctx.get("jwt_username")
|
||||
ws_token = auth_service.create_access_token(username, token_version=gns3_ctx.get("jwt_token_version", 0), expires_in=10) if username else None
|
||||
raw_url = f"{gns3_ctx['server_url']}/v3/projects/{project_id}/nodes/{node_id}/console/ws"
|
||||
if ws_token:
|
||||
raw_url += f"?token={ws_token}"
|
||||
# Convert http scheme to ws for direct websocat usage
|
||||
ws_url = raw_url.replace("https://", "wss://").replace("http://", "ws://")
|
||||
|
||||
result = {
|
||||
"node_id": node_id,
|
||||
"node_name": node.get("name"),
|
||||
"console_type": console_type,
|
||||
"ws_url": ws_url,
|
||||
"command": f"websocat -t --no-close {ws_url}",
|
||||
}
|
||||
if ws_token:
|
||||
# Fingerprint of the minted token: compare it against what actually reached the
|
||||
# server (logged on WebSocket auth rejection) to detect copy corruption, and
|
||||
# re-request the URL once token_ttl_seconds has elapsed.
|
||||
result["token_sha256_prefix"] = hashlib.sha256(ws_token.encode()).hexdigest()[:8]
|
||||
result["token_ttl_seconds"] = 600
|
||||
if console_type in ("vnc",):
|
||||
result["vnc_url"] = f"/v3/projects/{project_id}/nodes/{node_id}/console/vnc?token={gns3_ctx['jwt_token']}"
|
||||
return result
|
||||
|
||||
|
||||
# ── Node file handlers ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def list_node_files_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)
|
||||
url = f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/files"
|
||||
query = {}
|
||||
if params.get("path"):
|
||||
query["path"] = params["path"]
|
||||
if params.get("recursive"):
|
||||
query["recursive"] = "true"
|
||||
files = conn.http_call("get", url, params=query if query else None).json()
|
||||
return {"files": files, "count": len(files)}
|
||||
|
||||
|
||||
def get_node_file_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")
|
||||
file_path = params.get("file_path")
|
||||
if not project_id or not node_id or not file_path:
|
||||
return {"error": "project_id, node_id and file_path are required"}
|
||||
|
||||
offset = params.get("offset", 0)
|
||||
limit = params.get("limit", 200)
|
||||
|
||||
conn = _get_connector(gns3_ctx)
|
||||
url = f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/files/{file_path}"
|
||||
raw = conn.http_call("get", url).text
|
||||
|
||||
total_bytes = len(raw.encode("utf-8"))
|
||||
truncated = False
|
||||
if total_bytes > MAX_NODE_FILE_BYTES:
|
||||
raw = raw[:MAX_NODE_FILE_BYTES]
|
||||
truncated = True
|
||||
|
||||
lines = raw.splitlines(keepends=False)
|
||||
total_lines = len(lines)
|
||||
|
||||
# Apply offset/limit
|
||||
selected = lines[offset: offset + limit] if offset < total_lines else []
|
||||
has_more = (offset + limit) < total_lines or truncated
|
||||
|
||||
return {
|
||||
"file_path": file_path,
|
||||
"content": "\n".join(selected),
|
||||
"metadata": {
|
||||
"total_lines": total_lines,
|
||||
"total_bytes": total_bytes,
|
||||
"offset": offset,
|
||||
"limit": limit,
|
||||
"returned_lines": len(selected),
|
||||
"returned_bytes": len("\n".join(selected).encode("utf-8")),
|
||||
"truncated": truncated or has_more,
|
||||
"has_more": has_more,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def write_node_file_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")
|
||||
file_path = params.get("file_path")
|
||||
content = params.get("content")
|
||||
if not project_id or not node_id or not file_path or content is None:
|
||||
return {"error": "project_id, node_id, file_path and content are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
url = f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/files/{file_path}"
|
||||
conn.http_call("post", url, data=content, headers={"Content-Type": "text/plain"})
|
||||
return {"message": f"File {file_path} written to node {node_id}", "file_path": file_path, "node_id": node_id}
|
||||
|
||||
|
||||
def delete_node_file_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")
|
||||
file_path = params.get("file_path")
|
||||
if not project_id or not node_id or not file_path:
|
||||
return {"error": "project_id, node_id and file_path are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
url = f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/files/{file_path}"
|
||||
conn.http_call("delete", url)
|
||||
return {"message": f"File {file_path} deleted from node {node_id}", "file_path": file_path, "node_id": node_id}
|
||||
|
||||
|
||||
# ── Node bulk / advanced handlers ────────────────────────────────────
|
||||
|
||||
|
||||
def start_all_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)
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/start")
|
||||
return {"message": "All nodes started", "project_id": project_id}
|
||||
|
||||
|
||||
def stop_all_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)
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/stop")
|
||||
return {"message": "All nodes stopped", "project_id": project_id}
|
||||
|
||||
|
||||
def suspend_all_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)
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/suspend")
|
||||
return {"message": "All nodes suspended", "project_id": project_id}
|
||||
|
||||
|
||||
def duplicate_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)
|
||||
data = {k: v for k, v in params.items() if k not in ("project_id", "node_id") and v is not None}
|
||||
result = conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/duplicate", json_data=data).json()
|
||||
return {"message": f"Node {node_id} duplicated", "node": result}
|
||||
|
||||
|
||||
def isolate_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}/isolate")
|
||||
return {"message": f"Node {node_id} isolated (all links suspended)", "node_id": node_id}
|
||||
|
||||
|
||||
def unisolate_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}/unisolate")
|
||||
return {"message": f"Node {node_id} unisolated (links resumed)", "node_id": node_id}
|
||||
|
||||
|
||||
def get_node_links_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)
|
||||
links = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/links").json()
|
||||
return {"links": links, "count": len(links)}
|
||||
|
||||
|
||||
# ── 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": "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,
|
||||
},
|
||||
{
|
||||
"name": "get_node_console_info",
|
||||
"description": "Get console WebSocket URL for a node (use websocat to connect)",
|
||||
"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_console_info_handler,
|
||||
},
|
||||
{
|
||||
"name": "list_node_files",
|
||||
"description": "List files in a node directory with metadata (name, size, type, modified time). "
|
||||
"Use recursive=true for a full recursive listing. "
|
||||
"Check file sizes before reading large files with get_node_file.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {"type": "string", "description": "Project UUID"},
|
||||
"node_id": {"type": "string", "description": "Node UUID"},
|
||||
"path": {"type": "string", "description": "Subdirectory path within node directory (optional)"},
|
||||
"recursive": {"type": "boolean", "description": "Recursively list all files (optional, default: false)"},
|
||||
},
|
||||
"required": ["project_id", "node_id"],
|
||||
},
|
||||
"handler": list_node_files_handler,
|
||||
},
|
||||
{
|
||||
"name": "get_node_file",
|
||||
"description": "Read a text file from a node directory. Returns file content line-by-line with offset/limit support. "
|
||||
"Best practice: start with offset=0, limit=200 to preview, then increase offset to read more. "
|
||||
"Large files (>50KB) are auto-truncated; check the metadata.truncated flag. "
|
||||
"For binary files, check file type via list_node_files first.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {"type": "string", "description": "Project UUID"},
|
||||
"node_id": {"type": "string", "description": "Node UUID"},
|
||||
"file_path": {"type": "string", "description": "Path to the file within the node directory"},
|
||||
"offset": {"type": "integer", "description": "Line offset to start reading from (optional, default: 0)"},
|
||||
"limit": {"type": "integer", "description": "Maximum number of lines to return (optional, default: 200)"},
|
||||
},
|
||||
"required": ["project_id", "node_id", "file_path"],
|
||||
},
|
||||
"handler": get_node_file_handler,
|
||||
},
|
||||
{
|
||||
"name": "write_node_file",
|
||||
"description": "Write content to a file in a node directory. Creates the file if it doesn't exist. "
|
||||
"Overwrites existing content. Useful for updating configuration files on nodes.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {"type": "string", "description": "Project UUID"},
|
||||
"node_id": {"type": "string", "description": "Node UUID"},
|
||||
"file_path": {"type": "string", "description": "Path to the file within the node directory"},
|
||||
"content": {"type": "string", "description": "Content to write to the file"},
|
||||
},
|
||||
"required": ["project_id", "node_id", "file_path", "content"],
|
||||
},
|
||||
"handler": write_node_file_handler,
|
||||
},
|
||||
{
|
||||
"name": "delete_node_file",
|
||||
"description": "Delete a file from a node directory. Cannot be undone. "
|
||||
"Use list_node_files to confirm the file path before deleting.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {"type": "string", "description": "Project UUID"},
|
||||
"node_id": {"type": "string", "description": "Node UUID"},
|
||||
"file_path": {"type": "string", "description": "Path to the file within the node directory"},
|
||||
},
|
||||
"required": ["project_id", "node_id", "file_path"],
|
||||
},
|
||||
"handler": delete_node_file_handler,
|
||||
},
|
||||
]
|
||||
@@ -0,0 +1,340 @@
|
||||
#
|
||||
# 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 tools for GNS3 project management.
|
||||
|
||||
Tool 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]):
|
||||
"""Create a Gns3Connector from the GNS3 context dict."""
|
||||
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 list_projects_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
projects = conn.http_call("get", f"{conn.base_url}/projects").json()
|
||||
return {"projects": projects, "count": len(projects)}
|
||||
|
||||
|
||||
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(gns3_ctx)
|
||||
project = conn.http_call("get", f"{conn.base_url}/projects/{project_id}").json()
|
||||
if project is None:
|
||||
return {"error": f"Project '{project_id}' not found"}
|
||||
return project
|
||||
|
||||
|
||||
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(gns3_ctx)
|
||||
return conn.http_call("post", f"{conn.base_url}/projects", json_data={"name": name}).json()
|
||||
|
||||
|
||||
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(gns3_ctx)
|
||||
conn.http_call("delete", f"{conn.base_url}/projects/{project_id}")
|
||||
return {"message": f"Project '{project_id}' deleted", "project_id": project_id}
|
||||
|
||||
|
||||
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(gns3_ctx)
|
||||
url = f"{conn.base_url}/projects/{project_id}/open"
|
||||
return conn.http_call("post", url).json()
|
||||
|
||||
|
||||
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(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}
|
||||
|
||||
|
||||
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(gns3_ctx)
|
||||
url = f"{conn.base_url}/projects/{project_id}/stats"
|
||||
return conn.http_call("get", url).json()
|
||||
|
||||
|
||||
def update_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(gns3_ctx)
|
||||
kwargs = {k: v for k, v in params.items() if k != "project_id" and v is not None}
|
||||
return conn.http_call("put", f"{conn.base_url}/projects/{project_id}", json_data=kwargs).json()
|
||||
|
||||
|
||||
def duplicate_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"}
|
||||
name = params.get("name")
|
||||
if not name:
|
||||
return {"error": "name is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
kwargs = {k: v for k, v in params.items() if k not in ("project_id",) and v is not None}
|
||||
return conn.http_call("post", f"{conn.base_url}/projects/{project_id}/duplicate", json_data=kwargs).json()
|
||||
|
||||
|
||||
def get_project_readme_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)
|
||||
try:
|
||||
url = f"{conn.base_url}/projects/{project_id}/files/README.txt"
|
||||
content = conn.http_call("get", url).text
|
||||
return {"project_id": project_id, "file": "README.txt", "content": content}
|
||||
except Exception as e:
|
||||
if "404" in str(e):
|
||||
return {"project_id": project_id, "file": "README.txt", "content": None, "message": "README.txt does not exist yet"}
|
||||
raise
|
||||
|
||||
|
||||
def update_project_readme_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"}
|
||||
content = params.get("content")
|
||||
if content is None:
|
||||
return {"error": "content is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
url = f"{conn.base_url}/projects/{project_id}/files/README.txt"
|
||||
conn.http_call("post", url, data=content, headers={"Content-Type": "text/plain"})
|
||||
return {"message": "README.txt updated", "project_id": project_id}
|
||||
|
||||
|
||||
def lock_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(gns3_ctx)
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/lock")
|
||||
return {"message": f"Project {project_id} locked", "project_id": project_id}
|
||||
|
||||
|
||||
def unlock_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(gns3_ctx)
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/unlock")
|
||||
return {"message": f"Project {project_id} unlocked", "project_id": project_id}
|
||||
|
||||
|
||||
def load_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
path = params.get("path")
|
||||
if not path:
|
||||
return {"error": "path is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
result = conn.http_call("post", f"{conn.base_url}/projects/load", json_data={"path": path}).json()
|
||||
return {"message": f"Project loaded from {path}", "project": result}
|
||||
|
||||
|
||||
def get_locked_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(gns3_ctx)
|
||||
locked = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/locked").json()
|
||||
return {"project_id": project_id, "locked": locked}
|
||||
|
||||
|
||||
# ── Tool definitions (consumed by mcp/__init__.py) ─────────────────────────
|
||||
|
||||
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"},
|
||||
},
|
||||
"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"},
|
||||
},
|
||||
"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"},
|
||||
},
|
||||
"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"},
|
||||
},
|
||||
"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"},
|
||||
},
|
||||
"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"},
|
||||
},
|
||||
"required": ["project_id"],
|
||||
},
|
||||
"handler": get_project_stats_handler,
|
||||
},
|
||||
{
|
||||
"name": "update_project",
|
||||
"description": "Update a project's properties (name, auto_close, auto_open, etc.)",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {"type": "string", "description": "Project UUID"},
|
||||
"name": {"type": "string", "description": "New project name"},
|
||||
"auto_close": {"type": "boolean", "description": "Close project when last client leaves"},
|
||||
"auto_open": {"type": "boolean", "description": "Project opens when GNS3 starts"},
|
||||
"auto_start": {"type": "boolean", "description": "Project starts when opened"},
|
||||
"scene_width": {"type": "integer", "description": "Width of the drawing area"},
|
||||
"scene_height": {"type": "integer", "description": "Height of the drawing area"},
|
||||
"zoom": {"type": "integer", "description": "Zoom of the drawing area"},
|
||||
"show_layers": {"type": "boolean", "description": "Show layers on the drawing area"},
|
||||
"snap_to_grid": {"type": "boolean", "description": "Snap to grid on the drawing area"},
|
||||
"show_grid": {"type": "boolean", "description": "Show the grid on the drawing area"},
|
||||
"grid_size": {"type": "integer", "description": "Grid size for the drawing area for nodes"},
|
||||
"drawing_grid_size": {"type": "integer", "description": "Grid size for the drawing area for drawings"},
|
||||
"show_interface_labels": {"type": "boolean", "description": "Show interface labels on the drawing area"},
|
||||
},
|
||||
"required": ["project_id"],
|
||||
},
|
||||
"handler": update_project_handler,
|
||||
},
|
||||
{
|
||||
"name": "duplicate_project",
|
||||
"description": "Duplicate a project",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {"type": "string", "description": "UUID of the project to duplicate"},
|
||||
"name": {"type": "string", "description": "New project name"},
|
||||
"reset_mac_addresses": {"type": "boolean", "description": "Reset MAC addresses for this project"},
|
||||
},
|
||||
"required": ["project_id", "name"],
|
||||
},
|
||||
"handler": duplicate_project_handler,
|
||||
},
|
||||
{
|
||||
"name": "get_project_readme",
|
||||
"description": "Get the content of a project's README.md file (project documentation, Markdown format)",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {"type": "string", "description": "Project UUID"},
|
||||
},
|
||||
"required": ["project_id"],
|
||||
},
|
||||
"handler": get_project_readme_handler,
|
||||
},
|
||||
{
|
||||
"name": "update_project_readme",
|
||||
"description": "Update or create a project's README.md file (project documentation, Markdown format)",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {"type": "string", "description": "Project UUID"},
|
||||
"content": {"type": "string", "description": "Content to write to README.md (Markdown format)"},
|
||||
},
|
||||
"required": ["project_id", "content"],
|
||||
},
|
||||
"handler": update_project_readme_handler,
|
||||
},
|
||||
]
|
||||
@@ -0,0 +1,50 @@
|
||||
#
|
||||
# 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 server information.
|
||||
"""
|
||||
|
||||
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_version_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return conn.http_call("get", f"{conn.base_url}/version").json()
|
||||
|
||||
|
||||
def get_statistics_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return conn.http_call("get", f"{conn.base_url}/statistics").json()
|
||||
@@ -0,0 +1,79 @@
|
||||
#
|
||||
# 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 snapshot management.
|
||||
"""
|
||||
|
||||
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_snapshots_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)
|
||||
snapshots = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/snapshots").json()
|
||||
return {"snapshots": snapshots, "count": len(snapshots)}
|
||||
|
||||
|
||||
def create_snapshot_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
name = params.get("name")
|
||||
if not project_id or not name:
|
||||
return {"error": "project_id and name are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
result = conn.http_call("post", f"{conn.base_url}/projects/{project_id}/snapshots", json_data={"name": name}).json()
|
||||
return {"message": f"Snapshot '{name}' created", "snapshot": result}
|
||||
|
||||
|
||||
def delete_snapshot_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
snapshot_id = params.get("snapshot_id")
|
||||
if not project_id or not snapshot_id:
|
||||
return {"error": "project_id and snapshot_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("delete", f"{conn.base_url}/projects/{project_id}/snapshots/{snapshot_id}")
|
||||
return {"message": f"Snapshot {snapshot_id} deleted", "snapshot_id": snapshot_id}
|
||||
|
||||
|
||||
def restore_snapshot_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
snapshot_id = params.get("snapshot_id")
|
||||
if not project_id or not snapshot_id:
|
||||
return {"error": "project_id and snapshot_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
result = conn.http_call("post", f"{conn.base_url}/projects/{project_id}/snapshots/{snapshot_id}/restore").json()
|
||||
return {"message": f"Snapshot {snapshot_id} restored", "project": result}
|
||||
@@ -0,0 +1,101 @@
|
||||
#
|
||||
# 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 symbol management.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
import logging
|
||||
|
||||
from gns3server.services import auth_service
|
||||
|
||||
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_symbols_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
symbols = conn.http_call("get", f"{conn.base_url}/symbols").json()
|
||||
return {"symbols": symbols, "count": len(symbols)}
|
||||
|
||||
|
||||
def get_symbol_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
symbol_id = params.get("symbol_id")
|
||||
if not symbol_id:
|
||||
return {"error": "symbol_id is required"}
|
||||
download_url = f"{gns3_ctx['server_url']}/v3/symbols/{symbol_id}/raw"
|
||||
username = gns3_ctx.get("jwt_username")
|
||||
download_token = auth_service.create_access_token(username, token_version=gns3_ctx.get("jwt_token_version", 0), expires_in=10) if username else None
|
||||
result = {
|
||||
"symbol_id": symbol_id,
|
||||
"download_url": download_url,
|
||||
"note": "Symbol files are SVG images.",
|
||||
}
|
||||
if download_token:
|
||||
safe_name = symbol_id.replace(':', '').replace('/', '_')
|
||||
result["curl_command"] = f"curl -L -o '{safe_name}.svg' -H 'Authorization: Bearer {download_token}' '{download_url}'"
|
||||
result["note"] += " Download link includes a 10-minute token."
|
||||
return result
|
||||
|
||||
|
||||
def get_symbol_dimensions_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
symbol_id = params.get("symbol_id")
|
||||
if not symbol_id:
|
||||
return {"error": "symbol_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return conn.http_call("get", f"{conn.base_url}/symbols/{symbol_id}/dimensions").json()
|
||||
|
||||
|
||||
def get_default_symbols_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
symbols = conn.http_call("get", f"{conn.base_url}/symbols/default_symbols").json()
|
||||
return {"default_symbols": symbols}
|
||||
|
||||
|
||||
def upload_symbol_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
symbol_id = params.get("symbol_id")
|
||||
content = params.get("content")
|
||||
if not symbol_id or content is None:
|
||||
return {"error": "symbol_id and content (SVG data) are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
url = f"{conn.base_url}/symbols/{symbol_id}/raw"
|
||||
conn.http_call("post", url, data=content, headers={"Content-Type": "image/svg+xml"})
|
||||
return {"message": f"Symbol {symbol_id} uploaded", "symbol_id": symbol_id}
|
||||
|
||||
|
||||
def delete_symbol_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
symbol_id = params.get("symbol_id")
|
||||
if not symbol_id:
|
||||
return {"error": "symbol_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("delete", f"{conn.base_url}/symbols/{symbol_id}")
|
||||
return {"message": f"Symbol {symbol_id} deleted", "symbol_id": symbol_id}
|
||||
@@ -0,0 +1,229 @@
|
||||
#
|
||||
# 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 template 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 ──────────────────────────────────────────────────────────
|
||||
|
||||
VALID_TEMPLATE_FIELDS = {
|
||||
"template_id", "name", "version", "category", "default_name_format",
|
||||
"symbol", "template_type", "compute_id", "usage", "tags", "builtin",
|
||||
"created_at", "updated_at",
|
||||
}
|
||||
|
||||
TEMPLATE_DEFAULT_FIELDS = ["template_id", "name", "template_type", "category", "default_name_format"]
|
||||
|
||||
|
||||
def _filter_templates(templates, fields):
|
||||
"""Filter each template to only include requested fields."""
|
||||
if not fields:
|
||||
fields = TEMPLATE_DEFAULT_FIELDS
|
||||
if isinstance(templates, dict):
|
||||
return {k: templates[k] for k in fields if k in templates}
|
||||
return [{k: t[k] for k in fields if k in t} for t in templates]
|
||||
|
||||
|
||||
def list_templates_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
templates = conn.http_call("get", f"{conn.base_url}/templates").json()
|
||||
fields = params.get("fields")
|
||||
if fields:
|
||||
if not isinstance(fields, list):
|
||||
return {"error": "fields must be a list, e.g. [\"template_id\", \"name\"]"}
|
||||
invalid = [f for f in fields if f not in VALID_TEMPLATE_FIELDS]
|
||||
if invalid:
|
||||
return {
|
||||
"error": f"Unknown fields: {invalid}",
|
||||
"available_fields": sorted(VALID_TEMPLATE_FIELDS),
|
||||
}
|
||||
templates = _filter_templates(templates, fields)
|
||||
return {"templates": templates, "count": len(templates)}
|
||||
|
||||
|
||||
def get_template_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
template_id = params.get("template_id")
|
||||
name = params.get("name")
|
||||
|
||||
if not template_id and not name:
|
||||
return {"error": "template_id or name is required"}
|
||||
|
||||
conn = _get_connector(gns3_ctx)
|
||||
|
||||
if template_id:
|
||||
template = conn.http_call("get", f"{conn.base_url}/templates/{template_id}").json()
|
||||
else:
|
||||
# Find template by name
|
||||
all_templates = conn.http_call("get", f"{conn.base_url}/templates").json()
|
||||
matches = [t for t in all_templates if t.get("name") == name]
|
||||
if not matches:
|
||||
return {"error": f"Template '{name}' not found"}
|
||||
template = matches[0]
|
||||
|
||||
return template
|
||||
|
||||
|
||||
def create_template_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
name = params.get("name")
|
||||
template_type = params.get("template_type")
|
||||
if not name or not template_type:
|
||||
return {"error": "name and template_type are required"}
|
||||
|
||||
conn = _get_connector(gns3_ctx)
|
||||
data = {
|
||||
"name": name,
|
||||
"template_type": template_type,
|
||||
"compute_id": params.get("compute_id", "local"),
|
||||
}
|
||||
# Pass through optional template-type-specific fields (image, qemu_path, etc.)
|
||||
for key in ("image",):
|
||||
if key in params:
|
||||
data[key] = params[key]
|
||||
return conn.http_call("post", f"{conn.base_url}/templates", json_data=data).json()
|
||||
|
||||
|
||||
def update_template_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
template_id = params.get("template_id")
|
||||
name = params.get("name")
|
||||
|
||||
if not template_id and not name:
|
||||
return {"error": "template_id or name is required"}
|
||||
|
||||
conn = _get_connector(gns3_ctx)
|
||||
|
||||
# Resolve name to ID if needed
|
||||
if not template_id and name:
|
||||
all_templates = conn.http_call("get", f"{conn.base_url}/templates").json()
|
||||
matches = [t for t in all_templates if t.get("name") == name]
|
||||
if not matches:
|
||||
return {"error": f"Template '{name}' not found"}
|
||||
template_id = matches[0]["template_id"]
|
||||
|
||||
update_data = {k: v for k, v in params.items() if k not in ("template_id", "name", "kwargs")}
|
||||
# Support nested kwargs from MCP clients
|
||||
if "kwargs" in params and isinstance(params["kwargs"], dict):
|
||||
update_data = params["kwargs"]
|
||||
|
||||
return conn.http_call("put", f"{conn.base_url}/templates/{template_id}", json_data=update_data).json()
|
||||
|
||||
|
||||
def delete_template_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
template_id = params.get("template_id")
|
||||
name = params.get("name")
|
||||
|
||||
if not template_id and not name:
|
||||
return {"error": "template_id or name is required"}
|
||||
|
||||
conn = _get_connector(gns3_ctx)
|
||||
|
||||
if not template_id and name:
|
||||
all_templates = conn.http_call("get", f"{conn.base_url}/templates").json()
|
||||
matches = [t for t in all_templates if t.get("name") == name]
|
||||
if not matches:
|
||||
return {"error": f"Template '{name}' not found"}
|
||||
template_id = matches[0]["template_id"]
|
||||
|
||||
conn.http_call("delete", f"{conn.base_url}/templates/{template_id}")
|
||||
return {"message": f"Template deleted"}
|
||||
|
||||
|
||||
# ── Tool definitions ───────────────────────────────────────────────────────
|
||||
|
||||
TEMPLATE_TOOLS = [
|
||||
{
|
||||
"name": "list_templates",
|
||||
"description": "List all available templates on the server",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {},
|
||||
},
|
||||
"handler": list_templates_handler,
|
||||
},
|
||||
{
|
||||
"name": "get_template",
|
||||
"description": "Get detailed information about a specific template",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"template_id": {"type": "string", "description": "Template UUID"},
|
||||
"name": {"type": "string", "description": "Template name"},
|
||||
},
|
||||
},
|
||||
"handler": get_template_handler,
|
||||
},
|
||||
{
|
||||
"name": "create_template",
|
||||
"description": "Create a new template",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {"type": "string", "description": "Template name"},
|
||||
"template_type": {"type": "string", "description": "Template type (e.g. qemu, docker, dynamips)"},
|
||||
"compute_id": {"type": "string", "description": "Compute ID (optional, default: local)"},
|
||||
},
|
||||
"required": ["name", "template_type"],
|
||||
},
|
||||
"handler": create_template_handler,
|
||||
},
|
||||
{
|
||||
"name": "update_template",
|
||||
"description": "Update an existing template's properties",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"template_id": {"type": "string", "description": "Template UUID"},
|
||||
"name": {"type": "string", "description": "Template name"},
|
||||
},
|
||||
},
|
||||
"handler": update_template_handler,
|
||||
},
|
||||
{
|
||||
"name": "delete_template",
|
||||
"description": "Delete a template",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"template_id": {"type": "string", "description": "Template UUID"},
|
||||
"name": {"type": "string", "description": "Template name"},
|
||||
},
|
||||
},
|
||||
"handler": delete_template_handler,
|
||||
},
|
||||
]
|
||||
Reference in new issue
Block a user