feat: Add device configuration MCP tools (config_send, command_run, vpcs_config_set)

- Add jwt_token/url parameters to get_device_ports_from_topology() and GNS3TopologyTool
  for MCP handler compatibility (backward compatible, auto-detection fallback)
- Add jwt_token/url pass-through to ExecuteMultipleDeviceConfigCommands,
  ExecuteMultipleDeviceCommands, and VPCSCommands _run() methods
- Create MCP handler device_config.py wrapping the 3 device config tools
- Register as device_config_send, device_command_run, vpcs_config_set
This commit is contained in:
YueGuobin committed 2026-06-10 14:34:35 +08:00
1 parent 87e14cc49e
commit e7038f1ae3
7 files changed
+215 -9

No files matched your search

@@ -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")
@@ -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 = (
@@ -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 = (
@@ -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
@@ -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]] = {}