diff --git a/gns3server/agent/gns3_copilot/gns3_client/gns3_topology_reader.py b/gns3server/agent/gns3_copilot/gns3_client/gns3_topology_reader.py index 6365e6be4..5ec0f6935 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_topology_reader.py +++ b/gns3server/agent/gns3_copilot/gns3_client/gns3_topology_reader.py @@ -70,6 +70,8 @@ class GNS3TopologyTool(BaseTool): tool_input: Any = None, run_manager: Any = None, project_id: str | None = None, + jwt_token: str | None = None, + url: str | None = None, ) -> dict: """ Synchronous method to retrieve the topology of a specific GNS3 project. @@ -80,6 +82,8 @@ class GNS3TopologyTool(BaseTool): run_manager: Callback manager for tool run. project_id: The UUID of the specific GNS3 project to retrieve topology from. + jwt_token: JWT token for authentication (used by MCP handlers). + url: GNS3 server URL (used by MCP handlers). Returns: dict: A dictionary containing the project ID, name, status, nodes, @@ -102,8 +106,10 @@ class GNS3TopologyTool(BaseTool): } # Initialize Gns3Connector using factory function + # jwt_token/url can be passed explicitly (e.g. from MCP handlers) + # or auto-detected (e.g. from gns3-copilot agent) logger.debug("Connecting to GNS3 server...") - server = get_gns3_connector() + server = get_gns3_connector(jwt_token=jwt_token, url=url) if server is None: logger.error("Failed to create GNS3 connector") diff --git a/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py b/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py index 44af6a703..7bce576ee 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py +++ b/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py @@ -169,6 +169,8 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): self, tool_input: str, # or Union[str, List[Any], Dict[str, Any]] run_manager: CallbackManagerForToolRun | None = None, + jwt_token: str | None = None, + url: str | None = None, **kwargs: Any, ) -> list[dict[str, Any]]: """ @@ -177,6 +179,8 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): Args: tool_input (str): A JSON string containing project_id and device configuration commands to execute. + jwt_token: JWT token for GNS3 API auth (MCP handlers). + url: GNS3 server URL (MCP handlers). Returns: List[Dict[str, Any]]: A list of dicts containing device names and @@ -214,7 +218,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): # Prepare device hosts data try: hosts_data = self._prepare_device_hosts_data( - device_configs_list, project_id + device_configs_list, project_id, jwt_token=jwt_token, url=url ) except ValueError as e: logger.error("Failed to prepare device hosts data: %s", e) @@ -547,6 +551,8 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): self, device_config_list: list[dict[str, Any]], project_id: str | None = None, + jwt_token: str | None = None, + url: str | None = None, ) -> dict[str, dict[str, Any]]: """Prepare device hosts data from topology information.""" # Extract device names list @@ -556,7 +562,9 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): ] # Get device port information with project_id - hosts_data = get_device_ports_from_topology(device_names, project_id) + hosts_data = get_device_ports_from_topology( + device_names, project_id, jwt_token=jwt_token, url=url + ) if not hosts_data: error_msg = ( diff --git a/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py b/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py index abe7eae47..7e0061ceb 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py +++ b/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py @@ -171,6 +171,8 @@ class ExecuteMultipleDeviceCommands(BaseTool): self, tool_input: str | bytes | list[Any] | dict[str, Any], run_manager: CallbackManagerForToolRun | None = None, + jwt_token: str | None = None, + url: str | None = None, **kwargs: Any, ) -> list[dict[str, Any]]: """ @@ -181,6 +183,8 @@ class ExecuteMultipleDeviceCommands(BaseTool): Args: tool_input: JSON string with project_id and diagnostic commands. + jwt_token: JWT token for GNS3 API auth (MCP handlers). + url: GNS3 server URL (MCP handlers). Returns: List[Dict]: A list of dicts with device names and outputs. @@ -210,7 +214,7 @@ class ExecuteMultipleDeviceCommands(BaseTool): # Prepare device hosts data try: hosts_data = self._prepare_device_hosts_data( - device_configs_list, project_id + device_configs_list, project_id, jwt_token=jwt_token, url=url ) except ValueError as e: logger.error("Failed to prepare device hosts data: %s", e) @@ -490,6 +494,8 @@ class ExecuteMultipleDeviceCommands(BaseTool): self, device_config_list: list[dict[str, Any]], project_id: str | None = None, + jwt_token: str | None = None, + url: str | None = None, ) -> dict[str, dict[str, Any]]: """Prepare device hosts data from topology information.""" # Extract device names list @@ -499,7 +505,9 @@ class ExecuteMultipleDeviceCommands(BaseTool): ] # Get device port information with project_id - hosts_data = get_device_ports_from_topology(device_names, project_id) + hosts_data = get_device_ports_from_topology( + device_names, project_id, jwt_token=jwt_token, url=url + ) if not hosts_data: error_msg = ( diff --git a/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py b/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py index e143b700e..b7bb3d661 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py +++ b/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py @@ -154,6 +154,8 @@ class VPCSCommands(BaseTool): self, tool_input: str | bytes | list[Any] | dict[str, Any], run_manager: CallbackManagerForToolRun | None = None, + jwt_token: str | None = None, + url: str | None = None, **kwargs: Any, ) -> list[dict[str, Any]]: """ @@ -161,6 +163,8 @@ class VPCSCommands(BaseTool): Args: tool_input: JSON string with project_id and VPCS commands. + jwt_token: JWT token for GNS3 API auth (MCP handlers). + url: GNS3 server URL (MCP handlers). Returns: List of dicts with device names and command outputs. @@ -183,7 +187,7 @@ class VPCSCommands(BaseTool): # Prepare device hosts data try: hosts_data = self._prepare_device_hosts_data( - device_configs_list, project_id + device_configs_list, project_id, jwt_token=jwt_token, url=url ) except ValueError as e: logger.error("Failed to prepare device hosts data: %s", e) @@ -408,7 +412,11 @@ class VPCSCommands(BaseTool): } def _prepare_device_hosts_data( - self, device_configs_list: list[dict[str, Any]], project_id: str + self, + device_configs_list: list[dict[str, Any]], + project_id: str, + jwt_token: str | None = None, + url: str | None = None, ) -> dict[str, dict[str, Any]]: """ Prepare Nornir inventory hosts data for VPCS devices. @@ -416,6 +424,8 @@ class VPCSCommands(BaseTool): Args: device_configs_list: List of device configurations project_id: GNS3 project ID + jwt_token: JWT token for GNS3 API auth (MCP handlers). + url: GNS3 server URL (MCP handlers). Returns: Dictionary mapping device names to their host data @@ -431,7 +441,7 @@ class VPCSCommands(BaseTool): # Get device port mappings from topology device_ports = get_device_ports_from_topology( - device_names, project_id=project_id + device_names, project_id=project_id, jwt_token=jwt_token, url=url ) # Build Nornir inventory hosts data diff --git a/gns3server/agent/gns3_copilot/utils/get_gns3_device_port.py b/gns3server/agent/gns3_copilot/utils/get_gns3_device_port.py index 3ac1bc992..a5c950150 100644 --- a/gns3server/agent/gns3_copilot/utils/get_gns3_device_port.py +++ b/gns3server/agent/gns3_copilot/utils/get_gns3_device_port.py @@ -36,6 +36,8 @@ logger = logging.getLogger(__name__) def get_device_ports_from_topology( device_names: list[str], project_id: str | None = None, + jwt_token: str | None = None, + url: str | None = None, ) -> dict[str, dict[str, Any]]: """ Get device connection information from GNS3 topology @@ -43,6 +45,8 @@ def get_device_ports_from_topology( Args: device_names: List of device names to look up project_id: UUID of the specific GNS3 project to retrieve topology from + jwt_token: JWT token for authentication (used by MCP handlers). + url: GNS3 server URL (used by MCP handlers). Returns: Dictionary mapping device names to their connection data: @@ -71,7 +75,7 @@ def get_device_ports_from_topology( # Get topology information topo = GNS3TopologyTool() - topology = topo._run(project_id=project_id) + topology = topo._run(project_id=project_id, jwt_token=jwt_token, url=url) # Dynamically build hosts_data from topology hosts_data: dict[str, dict[str, Any]] = {} diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index 19235e69e..c6b3aee57 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -70,6 +70,10 @@ from .images import ( delete_image_handler, prune_images_handler, install_images_handler, ) +from .device_config import ( + device_config_send_handler, device_command_run_handler, + vpcs_config_set_handler, +) from .nodes import ( get_nodes_handler, get_node_handler, start_node_handler, stop_node_handler, reload_node_handler, suspend_node_handler, @@ -1133,6 +1137,71 @@ async def image_install() -> list[dict[str, Any]]: return await asyncio.to_thread(_run_handler_sync, install_images_handler, {}) +# ── Device config tools ─────────────────────────────────────────────── +# These tools connect to network device consoles via telnet/SSH using +# Nornir + Netmiko. Devices must be started and have a device_type tag. +# +# Workflow: +# 1. node_list(project_id) → identify device names +# 2. node_start_all(project_id) → ensure devices are running +# 3. device_config_send(project_id, device_configs=[...]) → push config +# 4. device_command_run(project_id, device_commands=[...]) → verify + + +@mcp.tool() +async def device_config_send( + project_id: Annotated[str, Field(description="UUID of the project")], + device_configs: Annotated[list, Field( + description="List of device configs. Each entry: {\"device_name\": \"R1\", \"config_commands\": [\"int lo0\", \"ip add 1.1.1.1 255.255.255.255\"]}" + )], +) -> list[dict[str, Any]]: + """Send configuration commands to network devices via console (telnet/SSH). + + Devices must be started first (use node_start or node_start_all). + Device type is auto-detected from the 'device_type:' tag on each node. + Common device types: cisco_ios_telnet, cisco_xr_telnet, huawei_telnet, gns3_huawei_telnet_ce + """ + return await asyncio.to_thread(_run_handler_sync, device_config_send_handler, { + "project_id": project_id, "device_configs": device_configs, + }) + + +@mcp.tool() +async def device_command_run( + project_id: Annotated[str, Field(description="UUID of the project")], + device_commands: Annotated[list, Field( + description="List of device show commands. Each entry: {\"device_name\": \"R1\", \"show_commands\": [\"show ip int brief\", \"show running-config\"]}" + )], +) -> list[dict[str, Any]]: + """Run read-only diagnostic (show) commands on network devices via console. + + Use this to inspect device status, view configurations, or verify changes. + Devices must be started first. + """ + return await asyncio.to_thread(_run_handler_sync, device_command_run_handler, { + "project_id": project_id, "device_commands": device_commands, + }) + + +@mcp.tool() +async def vpcs_config_set( + project_id: Annotated[str, Field(description="UUID of the project")], + device_configs: Annotated[list, Field( + description="List of VPCS configs. Each entry: {\"device_name\": \"PC1\", \"commands\": [\"ip 10.0.0.1/24 10.0.0.254\", \"save\"]}" + )], +) -> list[dict[str, Any]]: + """Configure VPCS devices (set IP addresses, gateway, etc.). + + VPCS-specific configuration commands: + - ip
/ Set IP and gateway + - save Save config to startup.vpc + - ping Test connectivity + """ + return await asyncio.to_thread(_run_handler_sync, vpcs_config_set_handler, { + "project_id": project_id, "device_configs": device_configs, + }) + + # ── Auth‑wrapped SSE app ────────────────────────────────────────────── def _make_auth_wrapper(inner_app): diff --git a/gns3server/api/routes/mcp/device_config.py b/gns3server/api/routes/mcp/device_config.py new file mode 100644 index 000000000..790d4b89f --- /dev/null +++ b/gns3server/api/routes/mcp/device_config.py @@ -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 . + +""" +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:' 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 + +log = logging.getLogger(__name__) + + +# ── 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") + if not project_id or not device_configs: + return [{"error": "project_id and device_configs are required"}] + + 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_command_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_commands = params.get("device_commands") + if not project_id or not device_commands: + return [{"error": "project_id and device_commands are required"}] + + from gns3server.agent.gns3_copilot.tools_v2.display_tools_nornir import ExecuteMultipleDeviceCommands + + tool = ExecuteMultipleDeviceCommands() + input_data = json.dumps({ + "project_id": project_id, + "device_commands": device_commands, + }) + 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"], + )