NgTech-LTD/gns3-server
2
0 Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2026-11-06 07:11:13 +02:00
Code Issues Packages Projects Releases Wiki Activity

style: fix E501 line-too-long errors in gns3_copilot

Browse Source 
Fix all 423 E501 line length violations across 26 files to comply with
   PEP 8 88-character line limit.

   Changes:
   - Split long f-strings across multiple lines
   - Break long docstring descriptions and parameter lists
   - Split markdown table rows and list examples
   - Break long URL construction f-strings
   - Split long logger messages and comments
   - Add noqa: E501 for SVG strings (cannot be split)

   Modified files:
   - agent/: context_manager.py, gns3_copilot.py, model_factory.py
   - gns3_client/: connector_factory.py, context_helpers.py, custom_gns3fy.py,
                   gns3_project_info.py, gns3_topology_reader.py
   - prompts/: __init__.py, lab_automation_assistant_prompt.py,
               prompt_loader.py, teaching_assistant_prompt.py
   - tools_v2/: __init__.py, config_tools_nornir.py, display_tools_nornir.py,
                gns3_create_link.py, gns3_create_node.py, gns3_get_node_temp.py,
                gns3_start_node.py, gns3_update_node_name.py,
                vpcs_tools_telnetlib3.py
   - utils/: __init__.py, command_filter.py, get_gns3_device_port.py,
             gns3_drawing_utils.py, llm_config_helper.py, message_converters.py,
             parse_tool_content.py, tool_call_stream.py

   All files now pass ruff E501 checks.

   Co-Authored-By: Yue Guobin <yueguobin@outlook.com>
This commit is contained in:
YueGuobin committed 2026-03-10 01:05:17 +08:00
1 parent 25f2d1b56b
commit 7c3b832bca
32 files changed
+1967 -762
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

gns3server/agent/gns3_copilot/tools_v2/__init__.py
+2 -2
View File
@@ -26,7 +26,7 @@
GNS3-Copilot Tools Package
This package provides various tools for interacting with GNS3 network simulator, including:
This package provides various tools for interacting with GNS3 network simulator:
- Device configuration command execution
- Display command execution
- Multiple device command execution using Nornir
@@ -36,7 +36,7 @@ This package provides various tools for interacting with GNS3 network simulator,
Main modules:
- config_tools_nornir: Multiple device configuration command execution tool using Nornir
- display_tools_nornir: Multiple device command execution tool using Nornir
- vpcs_tools_telnetlib3: VPCS device configuration tool using telnetlib3 (concurrent execution)
- vpcs_tools_telnetlib3: VPCS device configuration tool using telnetlib3
- gns3_create_node: GNS3 node creation tool
- gns3_create_link: GNS3 link creation tool
- gns3_start_node: GNS3 node startup tool
gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py
+154 -68
View File
@@ -24,8 +24,8 @@
#
"""
This module provides a tool to execute configuration commands on multiple devices
in a GNS3 topology using Nornir.
This module provides a tool to execute configuration commands on multiple
devices in a GNS3 topology using Nornir.
"""
import json
@@ -45,14 +45,17 @@ from nornir_netmiko.tasks import netmiko_send_config
from gns3server.agent.gns3_copilot.gns3_client import get_gns3_server_host
from gns3server.agent.gns3_copilot.utils import get_device_ports_from_topology
from gns3server.agent.gns3_copilot.utils.command_filter import filter_forbidden_commands
from gns3server.agent.gns3_copilot.utils.command_filter import (
filter_forbidden_commands,
)
# config log
logger = logging.getLogger(__name__)
# Suppress nornir INFO logs in console (reduce verbosity)
# The logging={"enabled": False} in InitNornir only disables plugin internal logs,
# but nornir.core still logs task execution at INFO level. Set to WARNING to suppress these.
# The logging={"enabled": False} in InitNornir only disables plugin
# internal logs, but nornir.core still logs task execution at INFO level.
# Set to WARNING to suppress these.
logging.getLogger("nornir.core").setLevel(logging.WARNING)
logging.getLogger("nornir").setLevel(logging.WARNING)
@@ -63,7 +66,9 @@ def _get_nornir_defaults() -> dict[str, Any]:
return {"data": {"location": "gns3"}}
def _get_nornir_groups_config(device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios") -> dict[str, Any]:
def _get_nornir_groups_config(
device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios"
) -> dict[str, Any]:
"""
Get Nornir group configuration for Cisco IOS Telnet devices.
@@ -80,12 +85,16 @@ def _get_nornir_groups_config(device_type: str = "cisco_ios_telnet", platform: s
"timeout": 120,
"username": "",
"password": "",
"connection_options": {"netmiko": {"extras": {"device_type": device_type}}},
"connection_options": {
"netmiko": {"extras": {"device_type": device_type}}
},
}
def _get_nornir_group(
group_name: str = "cisco_IOSv_telnet", device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios"
group_name: str = "cisco_IOSv_telnet",
device_type: str = "cisco_ios_telnet",
platform: str = "cisco_ios",
) -> dict[str, Any]:
"""
Get Nornir group configuration for a specific group.
@@ -99,26 +108,31 @@ def _get_nornir_group(
Dictionary containing group configuration
"""
# _get_nornir_groups_config now returns the group config directly
return _get_nornir_groups_config(device_type=device_type, platform=platform)
return _get_nornir_groups_config(
device_type=device_type, platform=platform
)
class ExecuteMultipleDeviceConfigCommands(BaseTool):
"""
A tool to execute configuration commands on multiple devices in a GNS3 topology using Nornir.
This class uses Nornir to manage connections and execute configuration commands
on multiple devices concurrently.
A tool to execute configuration commands on multiple devices in a GNS3
topology using Nornir.
This class uses Nornir to manage connections and execute configuration
commands on multiple devices concurrently.
IMPORTANT SAFETY NOTE:
This tool is intended for configuration changes only. Use extreme caution when
executing configuration commands.
This tool is intended for configuration changes only. Use extreme caution
when executing configuration commands.
"""
name: str = "execute_multiple_device_config_commands"
description: str = """
Executes CONFIGURATION commands on multiple devices in the current GNS3 topology.
Use this tool ONLY for changing device settings (e.g., 'configure', 'interface', 'ip address', 'router ospf').
Executes CONFIGURATION commands on multiple devices in the current GNS3
topology.
Use this tool ONLY for changing device settings (e.g., 'configure',
'interface', 'ip address', 'router ospf').
For viewing information, use the 'execute_multiple_device_commands' tool.
Input should be a JSON object containing project_id and device configurations.
Input should be a JSON object containing project_id and device configs.
Example input:
{
"project_id": "<PROJECT_UUID>",
@@ -141,10 +155,11 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
}
]
}
Returns a list of dictionaries, each containing the device name and configuration results.
Returns a list of dicts with device name and config results.
IMPORTANT SAFETY WARNING:
Do NOT use this tool for dangerous operations that could reboot, erase, or factory-reset devices.
Do NOT use this tool for dangerous operations that could reboot, erase,
or factory-reset devices.
Forbidden operations include but are not limited to:
- reload / reboot commands
- write erase / erase startup-config
@@ -161,14 +176,14 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
**kwargs: Any,
) -> list[dict[str, Any]]:
"""
Executes configuration commands on multiple devices in the current GNS3 topology.
Executes configuration commands on multiple devices in GNS3 topology.
Args:
tool_input (str): A JSON string containing project_id and device
configuration commands to execute.
Returns:
List[Dict[str, Any]]: A list of dictionaries containing device names and
List[Dict[str, Any]]: A list of dicts containing device names and
configuration results.
"""
# Log received input
@@ -176,12 +191,18 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
# Validate input
device_configs_list, project_id = self._validate_tool_input(tool_input)
if isinstance(device_configs_list, list) and len(device_configs_list) > 0 and "error" in device_configs_list[0]:
if (
isinstance(device_configs_list, list)
and len(device_configs_list) > 0
and "error" in device_configs_list[0]
):
return device_configs_list
# Filter forbidden commands and store blocked commands info
device_configs_list, blocked_commands_map = self._filter_forbidden_commands_from_device_configs(
device_configs_list
device_configs_list, blocked_commands_map = (
self._filter_forbidden_commands_from_device_configs(
device_configs_list
)
)
# Create a mapping of device names to their configuration commands
@@ -189,7 +210,9 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
# Prepare device hosts data
try:
hosts_data = self._prepare_device_hosts_data(device_configs_list, project_id)
hosts_data = self._prepare_device_hosts_data(
device_configs_list, project_id
)
except ValueError as e:
logger.error("Failed to prepare device hosts data: %s", e)
return [{"error": str(e)}]
@@ -211,11 +234,18 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
)
# Process results for all devices
results = self._process_task_results(device_configs_list, hosts_data, task_result, blocked_commands_map)
results = self._process_task_results(
device_configs_list,
hosts_data,
task_result,
blocked_commands_map,
)
except Exception as e:
# Overall execution failed
logger.error("Error executing configurations on all devices: %s", e)
logger.error(
"Error executing configurations on all devices: %s", e
)
return [{"error": f"Execution error: {str(e)}"}]
logger.info(
@@ -225,16 +255,22 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
return results
def _run_all_device_configs_with_single_retry(self, task: Task, device_configs_map: dict[str, list[str]]) -> Result:
def _run_all_device_configs_with_single_retry(
self, task: Task, device_configs_map: dict[str, list[str]]
) -> Result:
"""Execute configuration commands with single retry mechanism."""
device_name = task.host.name
config_commands = device_configs_map.get(device_name, [])
if not config_commands:
return Result(host=task.host, result="No configuration commands to execute")
return Result(
host=task.host, result="No configuration commands to execute"
)
try:
_result = task.run(task=netmiko_send_config, config_commands=config_commands)
_result = task.run(
task=netmiko_send_config, config_commands=config_commands
)
return Result(host=task.host, result=_result.result)
except ReadTimeout as e:
@@ -251,10 +287,12 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
)
except Exception as e:
# Handle prompt detection issues with Cisco IOSv L2 images where the '#' prompt character
# may be delayed, causing Netmiko prompt detection failures. Implements retry logic.
# Handle Cisco IOSv L2 where '#' prompt char may be delayed,
# causing Netmiko failures. Implements retry logic.
if "netmiko_send_config (failed)" in str(e):
_result = task.run(task=netmiko_send_config, config_commands=config_commands)
_result = task.run(
task=netmiko_send_config, config_commands=config_commands
)
return Result(host=task.host, result=_result.result)
# Log any other exceptions with full details
@@ -274,14 +312,14 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
self, tool_input: str | bytes | list[Any] | dict[str, Any]
) -> tuple[list[dict[str, Any]], str | None]:
"""
Validate device configuration command input, handling both new and legacy input formats.
Supports new format with project_id and device_configs, as well as legacy array format.
Validate device config command input, handling new and legacy formats.
Supports new format with project_id and device_configs.
Args:
tool_input: The input received from the LangChain/LangGraph tool call.
tool_input: Input from the LangChain/LangGraph tool call.
Returns:
Tuple containing (device_configs_list, project_id) or (error_list, None)
Tuple of (device_configs_list, project_id) or (error_list, None)
"""
parsed_input = None
@@ -289,18 +327,26 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
# Compatibility Check and Parsing ---
# Check if the input is a string (or bytes) which needs to be parsed.
if isinstance(tool_input, (str, bytes, bytearray)):
# Handle models (like potentially DeepSeek) that return a raw JSON string.
# Handle models (like DeepSeek) that return a raw JSON string.
try:
parsed_input = json.loads(tool_input)
logger.info("Successfully parsed tool input from JSON string.")
except json.JSONDecodeError as e:
logger.error("Invalid JSON string received as tool input: %s", e)
return ([{"error": f"Invalid JSON string input from model: {e}"}], None)
logger.error(
"Invalid JSON string received as tool input: %s", e
)
return (
[{"error": f"Invalid JSON string input from model: {e}"}],
None,
)
else:
# Handle standard models (like GPT/OpenAI) where the framework
# has already parsed the JSON into a Python object (dict or list).
parsed_input = tool_input
logger.info("Using tool input directly as type: %s", type(parsed_input).__name__)
logger.info(
"Using tool input directly as type: %s",
type(parsed_input).__name__,
)
# Handle new format: {"project_id": "...", "device_configs": [...]}
if isinstance(parsed_input, dict):
@@ -314,7 +360,10 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
return ([{"error": error_msg}], None)
if not self._validate_project_id(project_id):
error_msg = f"Invalid project_id format: {project_id}. Expected UUID format."
error_msg = (
f"Invalid project_id format: {project_id}. "
"Expected UUID format."
)
logger.error(error_msg)
return ([{"error": error_msg}], None)
@@ -332,13 +381,17 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
# Handle legacy format: [...]
elif isinstance(parsed_input, list):
logger.warning("Using legacy input format without project_id. Please use new format with project_id.")
logger.warning(
"Using legacy format without project_id. "
"Please use new format with project_id."
)
return parsed_input, None
else:
error_msg = (
"Tool input must be a JSON object with 'project_id' and 'device_configs' fields, "
f"or a legacy JSON array, but got {type(parsed_input).__name__}"
"Tool input must be a JSON object with 'project_id' and "
f"'device_configs' fields, or a legacy JSON array, but got "
f"{type(parsed_input).__name__}"
)
logger.error(error_msg)
return ([{"error": error_msg}], None)
@@ -353,7 +406,9 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
Returns:
True if valid UUID format, False otherwise
"""
uuid_pattern = r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"
uuid_pattern = (
r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"
)
return bool(re.match(uuid_pattern, project_id, re.IGNORECASE))
def _filter_forbidden_commands_from_device_configs(
@@ -363,12 +418,12 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
Filter out forbidden commands from device configurations.
Args:
device_configs_list: List of device configurations with config_commands.
device_configs_list: List of device configs with config_commands.
Returns:
A tuple of (filtered_device_configs_list, blocked_commands_map):
- filtered_device_configs_list: Device configs with forbidden commands removed.
- blocked_commands_map: Dict mapping device names to their blocked commands info.
Tuple of (filtered_configs, blocked_map):
- filtered_configs: Device configs with forbidden removed.
- blocked_map: Dict mapping device names to blocked info.
"""
filtered_list = []
blocked_commands_map: dict[str, dict[str, str]] = {}
@@ -378,7 +433,9 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
commands = device_config["config_commands"]
# Filter commands
allowed_commands, blocked_info = filter_forbidden_commands(commands)
allowed_commands, blocked_info = filter_forbidden_commands(
commands
)
# Update device config with allowed commands only
filtered_config = device_config.copy()
@@ -397,7 +454,9 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
return filtered_list, blocked_commands_map
def _configs_map(self, device_config_list: list[dict[str, Any]]) -> dict[str, list[str]]:
def _configs_map(
self, device_config_list: list[dict[str, Any]]
) -> dict[str, list[str]]:
"""Create a mapping of device names to their configuration commands."""
device_configs_map = {}
for device_config in device_config_list:
@@ -408,18 +467,23 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
return device_configs_map
def _prepare_device_hosts_data(
self, device_config_list: list[dict[str, Any]], project_id: str | None = None
self,
device_config_list: list[dict[str, Any]],
project_id: str | None = None,
) -> dict[str, dict[str, Any]]:
"""Prepare device hosts data from topology information."""
# Extract device names list
device_names = [device_config["device_name"] for device_config in device_config_list]
device_names = [
device_config["device_name"]
for device_config in device_config_list
]
# Get device port information with project_id
hosts_data = get_device_ports_from_topology(device_names, project_id)
if not hosts_data:
error_msg = (
f"Failed to get device information from topology or no valid devices found. "
"Failed to get device info from topology or no valid devices. "
f"Project ID: {project_id}, Devices: {device_names}"
)
raise ValueError(error_msg)
@@ -435,7 +499,9 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
return hosts_data
def _initialize_nornir(self, hosts_data: dict[str, dict[str, Any]]) -> Nornir:
def _initialize_nornir(
self, hosts_data: dict[str, dict[str, Any]]
) -> Nornir:
"""Initialize Nornir with the provided hosts data."""
try:
# Extract device_type and platform from hosts_data
@@ -445,14 +511,21 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
if hosts_data:
first_device_data = next(iter(hosts_data.values()), {})
device_type = first_device_data.get("device_type", "cisco_ios_telnet")
device_type = first_device_data.get(
"device_type", "cisco_ios_telnet"
)
platform = first_device_data.get("platform", "cisco_ios")
logger.info("Extracted from tags: device_type=%s, platform=%s", device_type, platform)
logger.info(
"Extracted from tags: device_type=%s, platform=%s",
device_type,
platform,
)
# Get latest environment configuration with dynamic device_type and platform
# Get environment config with dynamic device_type and platform
groups_data = _get_nornir_groups_config(
device_type=device_type or "cisco_ios_telnet", platform=platform or "cisco_ios"
device_type=device_type or "cisco_ios_telnet",
platform=platform or "cisco_ios",
)
defaults = _get_nornir_defaults()
@@ -460,7 +533,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
gns3_host = get_gns3_server_host()
logger.info(
"Initializing Nornir with account: host=%s, platform=%s, timeout=%d",
"Initializing Nornir: host=%s, platform=%s, timeout=%d",
gns3_host,
groups_data.get("platform"),
groups_data.get("timeout"),
@@ -505,11 +578,16 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
device_result = {
"device_name": device_name,
"status": "failed",
"error": (f"Device '{device_name}' not found in topology or missing console_port"),
"error": (
f"Device '{device_name}' not found in topology or "
"missing console_port"
),
}
# Add blocked commands info if any
if blocked_commands_info:
device_result["blocked_commands"] = list(blocked_commands_info.keys())
device_result["blocked_commands"] = list(
blocked_commands_info.keys()
)
device_result["blocked_info"] = blocked_commands_info
results.append(device_result)
continue
@@ -519,11 +597,15 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
device_result = {
"device_name": device_name,
"status": "failed",
"error": (f"Device '{device_name}' not found in task results"),
"error": (
f"Device '{device_name}' not found in task results"
),
}
# Add blocked commands info if any
if blocked_commands_info:
device_result["blocked_commands"] = list(blocked_commands_info.keys())
device_result["blocked_commands"] = list(
blocked_commands_info.keys()
)
device_result["blocked_info"] = blocked_commands_info
results.append(device_result)
continue
@@ -535,7 +617,9 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
if multi_result[0].failed:
# Execution failed
device_result["status"] = "failed"
device_result["error"] = f"Configuration execution failed: {multi_result[0].result}"
device_result["error"] = (
f"Configuration execution failed: {multi_result[0].result}"
)
device_result["output"] = multi_result[0].result
else:
# Execution successful
@@ -545,9 +629,11 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
# Add blocked commands info if any
if blocked_commands_info:
device_result["blocked_commands"] = list(blocked_commands_info.keys())
device_result["blocked_commands"] = list(
blocked_commands_info.keys()
)
device_result["blocked_info"] = blocked_commands_info
# Update status if some commands were blocked but execution succeeded
# Update status if some commands were blocked but succeeded
if device_result["status"] == "success":
device_result["status"] = "partial_success"
gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py
+138 -67
View File
@@ -45,14 +45,17 @@ from nornir_netmiko.tasks import netmiko_multiline
from gns3server.agent.gns3_copilot.gns3_client import get_gns3_server_host
from gns3server.agent.gns3_copilot.utils import get_device_ports_from_topology
from gns3server.agent.gns3_copilot.utils.command_filter import filter_forbidden_commands
from gns3server.agent.gns3_copilot.utils.command_filter import (
filter_forbidden_commands,
)
# config log
logger = logging.getLogger(__name__)
# Suppress nornir INFO logs in console (reduce verbosity)
# The logging={"enabled": False} in InitNornir only disables plugin internal logs,
# but nornir.core still logs task execution at INFO level. Set to WARNING to suppress these.
# The logging={"enabled": False} in InitNornir only disables plugin
# internal logs, but nornir.core still logs task execution at INFO level.
# Set to WARNING to suppress these.
logging.getLogger("nornir.core").setLevel(logging.WARNING)
logging.getLogger("nornir").setLevel(logging.WARNING)
@@ -63,7 +66,9 @@ def _get_nornir_defaults() -> dict[str, Any]:
return {"data": {"location": "gns3"}}
def _get_nornir_groups_config(device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios") -> dict[str, Any]:
def _get_nornir_groups_config(
device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios"
) -> dict[str, Any]:
"""
Get Nornir group configuration for Cisco IOS Telnet devices.
@@ -80,12 +85,16 @@ def _get_nornir_groups_config(device_type: str = "cisco_ios_telnet", platform: s
"timeout": 120,
"username": "",
"password": "",
"connection_options": {"netmiko": {"extras": {"device_type": device_type}}},
"connection_options": {
"netmiko": {"extras": {"device_type": device_type}}
},
}
def _get_nornir_group(
group_name: str = "cisco_IOSv_telnet", device_type: str = "cisco_ios_telnet", platform: str = "cisco_ios"
group_name: str = "cisco_IOSv_telnet",
device_type: str = "cisco_ios_telnet",
platform: str = "cisco_ios",
) -> dict[str, Any]:
"""
Get Nornir group configuration for a specific group.
@@ -99,17 +108,19 @@ def _get_nornir_group(
Dictionary containing group configuration
"""
# _get_nornir_groups_config now returns the group config directly
return _get_nornir_groups_config(device_type=device_type, platform=platform)
return _get_nornir_groups_config(
device_type=device_type, platform=platform
)
class ExecuteMultipleDeviceCommands(BaseTool):
"""
A READ-ONLY diagnostic tool for viewing network device configurations and protocol states.
A READ-ONLY diagnostic tool for viewing network device configurations.
**CRITICAL: DIAGNOSIS ONLY - NO CONFIGURATION PERMISSIONS**
This tool is exclusively designed for read-only operations to inspect and diagnose
network devices. It CANNOT and MUST NOT be used for any configuration changes.
This tool is exclusively designed for read-only operations to inspect and
diagnose network devices. It CANNOT and MUST NOT be used for configuration.
**Allowed Command Types:**
- Display commands: show, display (Cisco/Huawei)
@@ -126,21 +137,21 @@ class ExecuteMultipleDeviceCommands(BaseTool):
name: str = "execute_multiple_device_commands"
description: str = """
**READ-ONLY DIAGNOSTIC TOOL** - View network device configurations and protocol states.
**READ-ONLY DIAGNOSTIC TOOL** - View network device configurations.
Use this tool to inspect device information without making any changes.
**PERMITTED USE CASES:**
- View device status: show version, show running-config, show startup-config
- View device status: show version, show running-config
- Check routing: show ip route, show ip ospf neighbor, show bgp summary
- Interface status: show ip interface brief, show interfaces
- Protocol diagnostics: show ospf database, show bgp routes, debug commands
- Connectivity testing: ping, traceroute
**STRICTLY FORBIDDEN:**
- NO configuration commands (configure terminal, interface, router, ip address, etc.)
- NO configuration commands (configure terminal, interface, router, etc.)
- NO commands that modify device state
- If you need to configure devices, provide guidance to the student instead
- If you need to configure, provide guidance to the student instead
**Input Format:**
{
@@ -167,28 +178,34 @@ class ExecuteMultipleDeviceCommands(BaseTool):
**kwargs: Any,
) -> list[dict[str, Any]]:
"""
Executes READ-ONLY diagnostic commands on multiple devices in current GNS3 topology.
Executes READ-ONLY diagnostic commands on multiple devices in GNS3.
This method only permits display/show commands and does not allow any configuration
changes to network devices.
This method only permits display/show commands and does not allow
configuration changes to network devices.
Args:
tool_input (str): A JSON string containing project_id and diagnostic commands to execute.
tool_input: JSON string with project_id and diagnostic commands.
Returns:
List[Dict[str, Any]]: A list of dictionaries containing device names and command outputs.
List[Dict]: A list of dicts with device names and outputs.
"""
# Log received input
logger.debug("Received input: %s", tool_input)
# Validate input
device_configs_list, project_id = self._validate_tool_input(tool_input)
if isinstance(device_configs_list, list) and len(device_configs_list) > 0 and "error" in device_configs_list[0]:
if (
isinstance(device_configs_list, list)
and len(device_configs_list) > 0
and "error" in device_configs_list[0]
):
return device_configs_list
# Filter forbidden commands and store blocked commands info
device_configs_list, blocked_commands_map = self._filter_forbidden_commands_from_device_configs(
device_configs_list
device_configs_list, blocked_commands_map = (
self._filter_forbidden_commands_from_device_configs(
device_configs_list
)
)
# Create a mapping of device names to their display commands
@@ -196,7 +213,9 @@ class ExecuteMultipleDeviceCommands(BaseTool):
# Prepare device hosts data
try:
hosts_data = self._prepare_device_hosts_data(device_configs_list, project_id)
hosts_data = self._prepare_device_hosts_data(
device_configs_list, project_id
)
except ValueError as e:
logger.error("Failed to prepare device hosts data: %s", e)
return [{"error": str(e)}]
@@ -218,7 +237,12 @@ class ExecuteMultipleDeviceCommands(BaseTool):
)
# Process results for all devices
results = self._process_task_results(device_configs_list, hosts_data, task_result, blocked_commands_map)
results = self._process_task_results(
device_configs_list,
hosts_data,
task_result,
blocked_commands_map,
)
except Exception as e:
# Overall execution failed
@@ -232,13 +256,17 @@ class ExecuteMultipleDeviceCommands(BaseTool):
return results
def _run_all_device_configs_with_single_retry(self, task: Task, device_configs_map: dict[str, list[str]]) -> Result:
"""Execute READ-ONLY diagnostic commands with single retry mechanism."""
def _run_all_device_configs_with_single_retry(
self, task: Task, device_configs_map: dict[str, list[str]]
) -> Result:
"""Execute READ-ONLY diagnostic commands with single retry."""
device_name = task.host.name
diagnostic_commands = device_configs_map.get(device_name, [])
if not diagnostic_commands:
return Result(host=task.host, result="No diagnostic commands to execute")
return Result(
host=task.host, result="No diagnostic commands to execute"
)
try:
_result = task.run(
@@ -258,13 +286,13 @@ class ExecuteMultipleDeviceCommands(BaseTool):
)
return Result(
host=task.host,
result=f"diagnostic command execution failed (ReadTimeout): {str(e)}",
result=f"diagnostic command failed (ReadTimeout): {str(e)}",
failed=True,
)
except Exception as e:
# Handle prompt detection issues with Cisco IOSv L2 images where the '#' prompt character
# may be delayed, causing Netmiko prompt detection failures. Implements retry logic.
# Handle Cisco IOSv L2 where '#' prompt char may be delayed,
# causing Netmiko failures. Implements retry logic.
if "netmiko_multiline (failed)" in str(e):
_result = task.run(
task=netmiko_multiline,
@@ -276,14 +304,14 @@ class ExecuteMultipleDeviceCommands(BaseTool):
# Log any other exceptions with full details
logger.error(
"diagnostic command execution failed for device %s: %s (Exception type: %s)",
"diagnostic command failed for device %s: %s (Exception: %s)",
device_name,
str(e),
type(e).__name__,
)
return Result(
host=task.host,
result=f"diagnostic command execution failed (Unhandled Exception): {str(e)}",
result=f"diagnostic command failed (Unhandled): {str(e)}",
failed=True,
)
@@ -293,14 +321,14 @@ class ExecuteMultipleDeviceCommands(BaseTool):
"""
Validate diagnostic command input for read-only device inspection.
Handles both new and legacy input formats. Supports new format with project_id
and device_configs, as well as legacy array format.
Handles both new and legacy input formats. Supports new format with
project_id and device_configs, as well as legacy array format.
Args:
tool_input: The input received from the LangChain/LangGraph tool call.
tool_input: Input from LangChain/LangGraph tool call.
Returns:
Tuple containing (device_configs_list, project_id) or (error_list, None)
Tuple of (device_configs_list, project_id) or (error_list, None)
"""
parsed_input = None
@@ -308,18 +336,26 @@ class ExecuteMultipleDeviceCommands(BaseTool):
# Compatibility Check and Parsing ---
# Check if the input is a string (or bytes) which needs to be parsed.
if isinstance(tool_input, (str, bytes, bytearray)):
# Handle models (like potentially DeepSeek) that return a raw JSON string.
# Handle models (like DeepSeek) that return a raw JSON string.
try:
parsed_input = json.loads(tool_input)
logger.info("Successfully parsed tool input from JSON string.")
except json.JSONDecodeError as e:
logger.error("Invalid JSON string received as tool input: %s", e)
return ([{"error": f"Invalid JSON string input from model: {e}"}], None)
logger.error(
"Invalid JSON string received as tool input: %s", e
)
return (
[{"error": f"Invalid JSON string input from model: {e}"}],
None,
)
else:
# Handle standard models (like GPT/OpenAI) where the framework
# has already parsed the JSON into a Python object (dict or list).
parsed_input = tool_input
logger.info("Using tool input directly as type: %s", type(parsed_input).__name__)
logger.info(
"Using tool input directly as type: %s",
type(parsed_input).__name__,
)
# Handle new format: {"project_id": "...", "device_configs": [...]}
if isinstance(parsed_input, dict):
@@ -333,7 +369,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
return ([{"error": error_msg}], None)
if not self._validate_project_id(project_id):
error_msg = f"Invalid project_id format: {project_id}. Expected UUID format."
error_msg = f"Invalid project_id: {project_id}. Expected UUID."
logger.error(error_msg)
return ([{"error": error_msg}], None)
@@ -351,13 +387,15 @@ class ExecuteMultipleDeviceCommands(BaseTool):
# Handle legacy format: [...]
elif isinstance(parsed_input, list):
logger.warning("Using legacy input format without project_id. Please use new format with project_id.")
logger.warning(
"Legacy input format without project_id. Use new format."
)
return parsed_input, None
else:
error_msg = (
"Tool input must be a JSON object with 'project_id' and 'device_configs' fields, "
f"or a legacy JSON array, but got {type(parsed_input).__name__}"
"Tool input must be JSON with project_id and device_configs, "
f"or legacy JSON array, got {type(parsed_input).__name__}"
)
logger.error(error_msg)
return ([{"error": error_msg}], None)
@@ -372,7 +410,9 @@ class ExecuteMultipleDeviceCommands(BaseTool):
Returns:
True if valid UUID format, False otherwise
"""
uuid_pattern = r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"
uuid_pattern = (
r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"
)
return bool(re.match(uuid_pattern, project_id, re.IGNORECASE))
def _filter_forbidden_commands_from_device_configs(
@@ -385,9 +425,9 @@ class ExecuteMultipleDeviceCommands(BaseTool):
device_configs_list: List of device configurations with commands.
Returns:
A tuple of (filtered_device_configs_list, blocked_commands_map):
- filtered_device_configs_list: Device configs with forbidden commands removed.
- blocked_commands_map: Dict mapping device names to their blocked commands info.
Tuple of (filtered_configs, blocked_map):
- filtered_configs: Device configs with forbidden removed.
- blocked_map: Dict mapping device names to blocked commands.
"""
filtered_list = []
blocked_commands_map: dict[str, dict[str, str]] = {}
@@ -397,7 +437,9 @@ class ExecuteMultipleDeviceCommands(BaseTool):
commands = device_config["commands"]
# Filter commands
allowed_commands, blocked_info = filter_forbidden_commands(commands)
allowed_commands, blocked_info = filter_forbidden_commands(
commands
)
# Update device config with allowed commands only
filtered_config = device_config.copy()
@@ -416,7 +458,9 @@ class ExecuteMultipleDeviceCommands(BaseTool):
return filtered_list, blocked_commands_map
def _configs_map(self, device_config_list: list[dict[str, Any]]) -> dict[str, list[str]]:
def _configs_map(
self, device_config_list: list[dict[str, Any]]
) -> dict[str, list[str]]:
"""Create a mapping of device names to their diagnostic commands."""
device_diagnostic_map = {}
for device_config in device_config_list:
@@ -427,19 +471,24 @@ class ExecuteMultipleDeviceCommands(BaseTool):
return device_diagnostic_map
def _prepare_device_hosts_data(
self, device_config_list: list[dict[str, Any]], project_id: str | None = None
self,
device_config_list: list[dict[str, Any]],
project_id: str | None = None,
) -> dict[str, dict[str, Any]]:
"""Prepare device hosts data from topology information."""
# Extract device names list
device_names = [device_config["device_name"] for device_config in device_config_list]
device_names = [
device_config["device_name"]
for device_config in device_config_list
]
# Get device port information with project_id
hosts_data = get_device_ports_from_topology(device_names, project_id)
if not hosts_data:
error_msg = (
f"Failed to get device information from topology or no valid devices found. "
f"Project ID: {project_id}, Devices: {device_names}"
f"Failed to get device info from topology. "
f"Project: {project_id}, Devices: {device_names}"
)
raise ValueError(error_msg)
@@ -454,7 +503,9 @@ class ExecuteMultipleDeviceCommands(BaseTool):
return hosts_data
def _initialize_nornir(self, hosts_data: dict[str, dict[str, Any]]) -> Nornir:
def _initialize_nornir(
self, hosts_data: dict[str, dict[str, Any]]
) -> Nornir:
"""Initialize Nornir with the provided hosts data."""
try:
# Extract device_type and platform from hosts_data
@@ -464,14 +515,21 @@ class ExecuteMultipleDeviceCommands(BaseTool):
if hosts_data:
first_device_data = next(iter(hosts_data.values()), {})
device_type = first_device_data.get("device_type", "cisco_ios_telnet")
device_type = first_device_data.get(
"device_type", "cisco_ios_telnet"
)
platform = first_device_data.get("platform", "cisco_ios")
logger.info("Extracted from tags: device_type=%s, platform=%s", device_type, platform)
logger.info(
"Extracted from tags: device_type=%s, platform=%s",
device_type,
platform,
)
# Get latest environment configuration with dynamic device_type and platform
# Get environment config with dynamic device_type and platform
groups_data = _get_nornir_groups_config(
device_type=device_type or "cisco_ios_telnet", platform=platform or "cisco_ios"
device_type=device_type or "cisco_ios_telnet",
platform=platform or "cisco_ios",
)
defaults = _get_nornir_defaults()
@@ -479,7 +537,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
gns3_host = get_gns3_server_host()
logger.info(
"Initializing Nornir with account: host=%s, platform=%s, timeout=%d",
"Initializing Nornir: host=%s, platform=%s, timeout=%d",
gns3_host,
groups_data.get("platform"),
groups_data.get("timeout"),
@@ -524,11 +582,16 @@ class ExecuteMultipleDeviceCommands(BaseTool):
device_result = {
"device_name": device_name,
"status": "failed",
"error": (f"Device '{device_name}' not found in topology or missing console_port"),
"error": (
f"Device '{device_name}' not found in topology "
"or missing console_port"
),
}
# Add blocked commands info if any
if blocked_commands_info:
device_result["blocked_commands"] = list(blocked_commands_info.keys())
device_result["blocked_commands"] = list(
blocked_commands_info.keys()
)
device_result["blocked_info"] = blocked_commands_info
results.append(device_result)
continue
@@ -538,11 +601,15 @@ class ExecuteMultipleDeviceCommands(BaseTool):
device_result = {
"device_name": device_name,
"status": "failed",
"error": (f"Device '{device_name}' not found in task results"),
"error": (
f"Device '{device_name}' not found in task results"
),
}
# Add blocked commands info if any
if blocked_commands_info:
device_result["blocked_commands"] = list(blocked_commands_info.keys())
device_result["blocked_commands"] = list(
blocked_commands_info.keys()
)
device_result["blocked_info"] = blocked_commands_info
results.append(device_result)
continue
@@ -554,7 +621,9 @@ class ExecuteMultipleDeviceCommands(BaseTool):
if multi_result[0].failed:
# Execution failed
device_result["status"] = "failed"
device_result["error"] = f"Diagnostic command execution failed: {multi_result[0].result}"
device_result["error"] = (
f"Diagnostic command failed: {multi_result[0].result}"
)
device_result["output"] = multi_result[0].result
else:
# Execution successful
@@ -564,9 +633,11 @@ class ExecuteMultipleDeviceCommands(BaseTool):
# Add blocked commands info if any
if blocked_commands_info:
device_result["blocked_commands"] = list(blocked_commands_info.keys())
device_result["blocked_commands"] = list(
blocked_commands_info.keys()
)
device_result["blocked_info"] = blocked_commands_info
# Update status if some commands were blocked but execution succeeded
# Update status if some commands were blocked but succeeded
if device_result["status"] == "success":
device_result["status"] = "partial_success"
gns3server/agent/gns3_copilot/tools_v2/gns3_create_link.py
+56 -19
View File
@@ -65,7 +65,7 @@ class GNS3LinkTool(BaseTool):
- `port1` (str): Port name of the first node (e.g., 'Ethernet0/0').
- `node_id2` (str): UUID of the second node.
- `port2` (str): Port name of the second node (e.g., 'Ethernet0/0').
Note: Port names must match those retrieved from the `gns3_topology_reader` tool.
Note: Port names must match those from `gns3_topology_reader` tool.
Example Input:
{
@@ -79,9 +79,9 @@ class GNS3LinkTool(BaseTool):
}
]
}
Output: A list of dictionaries, each containing:
- For successfule links: link_id(str),node_id1(str),port1(str),port1(str),node_id(str),port2(str).
- For failed links: error(str) with an error message(e.g., 'Missing required field: project_id')
Output: A list of dicts, each containing:
- For successful links: link_id, node_id1, port1, node_id2, port2.
- For failed links: error(str) with error message.
Example Output:
[
@@ -98,16 +98,20 @@ class GNS3LinkTool(BaseTool):
]
"""
def _run(self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None) -> list[dict[str, Any]]:
def _run(
self,
tool_input: str,
run_manager: CallbackManagerForToolRun | None = None,
) -> list[dict[str, Any]]:
"""
Creates one or multiple links between nodes in a GNS3 project.
Args:
tool_input (str): A JSON string containing project_id and links array.
tool_input: JSON string with project_id and links array.
run_manager: LangChain run manager (unused).
Returns:
list: A list containing dictionaries with created link details or error messages.
list: A list with created link details or error messages.
"""
# Log received input
logger.info("Received input: %s", tool_input)
@@ -125,7 +129,9 @@ class GNS3LinkTool(BaseTool):
if not isinstance(links_data, list) or len(links_data) == 0:
logger.error("Invalid links data: must be a non-empty array")
return [{"error": "Invalid links data: must be a non-empty array"}]
return [
{"error": "Invalid links data: must be a non-empty array"}
]
# Initialize Gns3Connector using factory function
logger.info("Connecting to GNS3 server...")
@@ -133,7 +139,14 @@ class GNS3LinkTool(BaseTool):
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
return [{"error": "Failed to connect to GNS3 server. Please check your configuration."}]
return [
{
"error": (
"Failed to connect to GNS3 server. "
"Please check your configuration."
)
}
]
created_links = []
@@ -150,14 +163,20 @@ class GNS3LinkTool(BaseTool):
# Validate link parameters
if not all([node_id1, port1, node_id2, port2]):
error_msg = f"Missing required fields in link definition {i}"
error_msg = (
f"Missing required fields in link definition {i}"
)
logger.error(error_msg)
created_links.append({"error": error_msg})
continue
# Get node details
node1 = gns3_server.get_node(project_id=project_id, node_id=node_id1)
node2 = gns3_server.get_node(project_id=project_id, node_id=node_id2)
node1 = gns3_server.get_node(
project_id=project_id, node_id=node_id1
)
node2 = gns3_server.get_node(
project_id=project_id, node_id=node_id2
)
if not node1 or not node2:
error_msg = f"Node not found in link {i}"
logger.error(error_msg)
@@ -166,11 +185,19 @@ class GNS3LinkTool(BaseTool):
# Find port information
port1_info = next(
(port for port in node1.get("ports", []) if port.get("name") == port1),
(
port
for port in node1.get("ports", [])
if port.get("name") == port1
),
None,
)
port2_info = next(
(port for port in node2.get("ports", []) if port.get("name") == port2),
(
port
for port in node2.get("ports", [])
if port.get("name") == port2
),
None,
)
if not port1_info or not port2_info:
@@ -186,14 +213,22 @@ class GNS3LinkTool(BaseTool):
nodes=[
{
"node_id": node_id1,
"adapter_number": port1_info.get("adapter_number", 0),
"port_number": port1_info.get("port_number", 0),
"adapter_number": port1_info.get(
"adapter_number", 0
),
"port_number": port1_info.get(
"port_number", 0
),
"label": {"text": port1},
},
{
"node_id": node_id2,
"adapter_number": port2_info.get("adapter_number", 0),
"port_number": port2_info.get("port_number", 0),
"adapter_number": port2_info.get(
"adapter_number", 0
),
"port_number": port2_info.get(
"port_number", 0
),
"label": {"text": port2},
},
],
@@ -217,7 +252,9 @@ class GNS3LinkTool(BaseTool):
created_links.append({"error": error_msg})
# Log final results
success_count = len([link for link in created_links if "error" not in link])
success_count = len(
[link for link in created_links if "error" not in link]
)
logger.info(
"Link creation completed: %d successful, %d failed",
success_count,
gns3server/agent/gns3_copilot/tools_v2/gns3_create_node.py
+48 -24
View File
@@ -51,7 +51,7 @@ class GNS3CreateNodeTool(BaseTool):
using specified templates and coordinates.
**Input:**
A JSON object containing the project_id and an array of nodes with template_id,
A JSON object with project_id and array of nodes with template_id,
x and y coordinates.
Example input:
@@ -92,13 +92,13 @@ class GNS3CreateNodeTool(BaseTool):
"successful_nodes": 2,
"failed_nodes": 0
}
If an error occurs during input validation, returns a dictionary with an error message.
If error occurs during validation, returns dict with error message.
"""
name: str = "create_gns3_node"
description: str = """
Creates multiple nodes in a GNS3 project using specified templates and coordinates.
Input is a JSON object with project_id and an array of nodes, each containing template_id, x, and y coordinates.
Creates multiple nodes in a GNS3 project using templates and coordinates.
Input is a JSON object with project_id and array of nodes.
Example input:
{
"project_id": "uuid-of-project",
@@ -115,10 +115,13 @@ class GNS3CreateNodeTool(BaseTool):
}
]
}
IMPORTANT: Ensure the distance between any two nodes is greater than 250 pixels.
This spacing is necessary to display interface numbers clearly for better topology visualization.
Returns a dictionary with creation results for all nodes, including success/failure status.
If the operation fails during input validation, returns a dictionary with an error message.
IMPORTANT: Ensure distance between any two nodes is greater than 250 px.
This spacing is necessary to display interface numbers clearly for better
topology visualization.
Returns a dictionary with creation results for all nodes, including
success/failure status.
If the operation fails during input validation, returns a dictionary with
an error message.
"""
def _run(
@@ -128,14 +131,15 @@ class GNS3CreateNodeTool(BaseTool):
**kwargs: Any,
) -> dict[str, Any]:
"""
Creates multiple nodes in a GNS3 project using the provided templates and coordinates.
Creates nodes in a GNS3 project with templates and coordinates.
Args:
tool_input (str): A JSON string containing project_id and an array of nodes.
tool_input: A JSON string with project_id and an array of nodes.
run_manager: LangChain run manager (unused).
Returns:
dict: A dictionary with creation results for all nodes or an error message.
dict: A dictionary with creation results for all nodes or an error
message.
"""
# Log received input
logger.info("Received input: %s", tool_input)
@@ -158,7 +162,9 @@ class GNS3CreateNodeTool(BaseTool):
# Validate each node in the array
for i, node_data in enumerate(nodes):
if not isinstance(node_data, dict):
logger.error("Invalid input: Node %d must be a dictionary.", i + 1)
logger.error(
"Invalid input: Node %d must be a dictionary.", i + 1
)
return {"error": f"Node {i + 1} must be a dictionary."}
template_id = node_data.get("template_id")
@@ -173,10 +179,14 @@ class GNS3CreateNodeTool(BaseTool):
]
):
logger.error(
"Invalid input: Node %d missing or invalid template_id, x, or y.",
"Invalid input: Node %d missing or invalid "
"template_id, x, or y.",
i + 1,
)
return {"error": f"Node {i + 1} missing or invalid template_id, x, or y."}
return {
"error": f"Node {i + 1} missing or invalid "
f"template_id, x, or y."
}
# Initialize Gns3Connector using factory function
logger.info("Connecting to GNS3 server...")
@@ -184,10 +194,15 @@ class GNS3CreateNodeTool(BaseTool):
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
return {"error": "Failed to connect to GNS3 server. Please check your configuration."}
return {
"error": "Failed to connect to GNS3 server. "
"Please check your configuration."
}
# Create nodes
logger.info("Creating %d nodes in project %s...", len(nodes), project_id)
logger.info(
"Creating %d nodes in project %s...", len(nodes), project_id
)
results: list[dict[str, Any]] = []
for i, node_data in enumerate(nodes):
@@ -197,7 +212,7 @@ class GNS3CreateNodeTool(BaseTool):
y = node_data.get("y")
logger.info(
"Creating node %d/%d with template %s at coordinates (%s, %s)...",
"Creating node %d/%d with template %s at (%s, %s)...",
i + 1,
len(nodes),
template_id,
@@ -235,8 +250,12 @@ class GNS3CreateNodeTool(BaseTool):
# Continue with next node even if one fails
# Calculate summary statistics
successful_nodes = len([r for r in results if r.get("status") == "success"])
failed_nodes = len([r for r in results if r.get("status") == "failed"])
successful_nodes = len(
[r for r in results if r.get("status") == "success"]
)
failed_nodes = len(
[r for r in results if r.get("status") == "failed"]
)
# Prepare final result
final_result = {
@@ -249,7 +268,7 @@ class GNS3CreateNodeTool(BaseTool):
# Log the final result
logger.info(
"Node creation completed: %d successful, %d failed out of %d total nodes.",
"Node creation completed: %d successful, %d failed, %d total.",
successful_nodes,
failed_nodes,
len(nodes),
@@ -263,22 +282,27 @@ class GNS3CreateNodeTool(BaseTool):
return {"error": f"Invalid JSON input: {e}"}
except Exception as e:
logger.error("Failed to process node creation request: %s", e)
return {"error": f"Failed to process node creation request: {str(e)}"}
return {
"error": f"Failed to process node creation request: {str(e)}"
}
if __name__ == "__main__":
# Test the tool locally with multiple nodes
test_input = json.dumps(
{
"project_id": "d7fc094c-685e-4db1-ac11-5e33a1b2e066", # Replace with actual project UUID
# TODO: Replace with actual project UUID
"project_id": "d7fc094c-685e-4db1-ac11-5e33a1b2e066",
"nodes": [
{
"template_id": "b923a635-b7cc-4cb5-9a86-9357e04c02f7", # Replace with actual template UUID
# TODO: Replace with actual template UUID
"template_id": "b923a635-b7cc-4cb5-9a86-9357e04c02f7",
"x": 100,
"y": -200,
},
{
"template_id": "b923a635-b7cc-4cb5-9a86-9357e04c02f7", # Replace with actual template UUID
# TODO: Replace with actual template UUID
"template_id": "b923a635-b7cc-4cb5-9a86-9357e04c02f7",
"x": 200,
"y": -300,
},
gns3server/agent/gns3_copilot/tools_v2/gns3_get_node_temp.py
+22 -31
View File
@@ -46,39 +46,23 @@ logger = logging.getLogger(__name__)
class GNS3TemplateTool(BaseTool):
"""
A LangChain tool to retrieve all available device templates from a GNS3 server.
The tool connects to the GNS3 server and extracts the name, template_id, and template_type
for each template.
LangChain tool to retrieve available device templates from GNS3 server.
Connects to GNS3 server and extracts name, template_id, and template_type.
**Input:**
No input is required for this tool. It connects to the GNS3 server at the default URL
(http://localhost:3080) and retrieves all templates.
No input required. Connects to GNS3 server at default URL.
**Output:**
A dictionary containing a list of dictionaries, each with the name, template_id, and
template_type of a template. Example output:
{
"templates": [
{"name": "Router1", "template_id": "uuid1", "template_type": "qemu"},
{"name": "Switch1", "template_id": "uuid2", "template_type": "ethernet_switch"}
]
}
If an error occurs, returns a dictionary with an error message.
Dict with list of dicts (name, template_id, template_type).
If error, returns dict with error message.
"""
name: str = "get_gns3_templates"
description: str = """
Retrieves all available device templates from a GNS3 server.
Returns a dictionary containing a list of dictionaries, each with the name, template_id,
and template_type of a template. No input is required.
Example output:
{
"templates": [
{"name": "Router1", "template_id": "uuid1", "template_type": "qemu"},
{"name": "Switch1", "template_id": "uuid2", "template_type": "ethernet_switch"}
]
}
If the connection fails, returns a dictionary with an error message.
Retrieves available device templates from GNS3 server.
Returns dict with list of dicts (name, template_id, template_type).
No input required.
If connection fails, returns dict with error message.
"""
def _run(
@@ -87,14 +71,14 @@ class GNS3TemplateTool(BaseTool):
run_manager: CallbackManagerForToolRun | None = None,
) -> dict[str, Any]:
"""
Connects to the GNS3 server and retrieves a list of all available device templates.
Connects to GNS3 server and retrieves available device templates.
Args:
tool_input (str): Optional input (not used in this tool).
tool_input: Optional input (not used).
run_manager: LangChain run manager (unused).
Returns:
dict: A dictionary containing the list of templates or an error message.
dict: Dict with templates list or error message.
"""
try:
# Initialize Gns3Connector using factory function
@@ -103,7 +87,12 @@ class GNS3TemplateTool(BaseTool):
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
return {"error": "Failed to connect to GNS3 server. Please check your configuration."}
return {
"error": (
"Failed to connect to GNS3 server. "
"Please check your configuration."
)
}
# Retrieve all available templates
templates = gns3_server.get_templates()
@@ -120,14 +109,16 @@ class GNS3TemplateTool(BaseTool):
# Return JSON-formatted result with full logging
result = {"templates": template_info}
logger.info(
"Template retrieval completed. Total templates: %d. Result: %s",
"Template retrieval completed. Total: %d. Result: %s",
len(template_info),
json.dumps(result, indent=2, ensure_ascii=False),
)
return result
except Exception as e:
logger.error("Failed to connect to GNS3 server or retrieve templates: %s", e)
logger.error(
"Failed to connect to GNS3 server or retrieve templates: %s", e
)
return {"error": f"Failed to retrieve templates: {str(e)}"}
gns3server/agent/gns3_copilot/tools_v2/gns3_start_node.py
+101 -31
View File
@@ -46,7 +46,9 @@ from gns3server.agent.gns3_copilot.gns3_client import get_gns3_connector
logger = logging.getLogger(__name__)
def show_progress_bar(duration: int = 120, interval: int = 1, node_count: int = 1) -> None:
def show_progress_bar(
duration: int = 120, interval: int = 1, node_count: int = 1
) -> None:
"""
Display a simple text progress bar for node startup.
@@ -63,7 +65,9 @@ def show_progress_bar(duration: int = 120, interval: int = 1, node_count: int =
# Create progress bar display
bar_length = 30
filled_length = int(bar_length * elapsed // duration)
progress_string = "=" * filled_length + ">" + " " * (bar_length - filled_length - 1)
progress_string = (
"=" * filled_length + ">" + " " * (bar_length - filled_length - 1)
)
# Print progress bar with node count
print(f"\r[{progress_string}] {progress:.1f}%", end="", flush=True)
@@ -102,10 +106,14 @@ class GNS3StartNodeTool(BaseTool):
description: str = """
Starts one or multiple nodes in a GNS3 project.
Input: JSON with project_id and node_ids (list of node IDs).
Returns: A dictionary with all nodes' details including success/failure status.
Returns: A dict with all nodes' details (success/failure status).
"""
def _run(self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None) -> dict[str, Any]:
def _run(
self,
tool_input: str,
run_manager: CallbackManagerForToolRun | None = None,
) -> dict[str, Any]:
try:
# Parse input JSON
input_data = json.loads(tool_input)
@@ -114,8 +122,13 @@ class GNS3StartNodeTool(BaseTool):
# Validate input
if not project_id or not node_ids:
logger.error("Missing required fields: project_id or node_ids.")
return {"error": "Missing required fields: project_id and node_ids."}
logger.error(
"Missing required fields: project_id or node_ids."
)
return {
"error": "Missing required fields: "
"project_id and node_ids."
}
if not isinstance(node_ids, list):
logger.error("node_ids must be a list.")
@@ -127,7 +140,10 @@ class GNS3StartNodeTool(BaseTool):
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
return {"error": "Failed to connect to GNS3 server. Please check your configuration."}
return {
"error": "Failed to connect to GNS3 server. "
"Please check your configuration."
}
# First loop: Send start commands for all nodes
logger.info(
@@ -137,31 +153,49 @@ class GNS3StartNodeTool(BaseTool):
)
for node_id in node_ids:
try:
node = Node(project_id=project_id, node_id=node_id, connector=gns3_server)
node = Node(
project_id=project_id,
node_id=node_id,
connector=gns3_server,
)
# Verify node exists
node.get()
if node.node_id:
node.start()
logger.info("Start command sent for node %s", node_id)
else:
logger.error("Node %s not found in project %s", node_id, project_id)
logger.error(
"Node %s not found in project %s",
node_id,
project_id,
)
except Exception as e:
logger.error("Failed to send start command for node %s: %s", node_id, e)
logger.error(
"Failed to send start command for node %s: %s",
node_id,
e,
)
# Calculate progress bar duration: 140s base + 10s per additional node
# Progress bar duration: 140s base + 10s per extra node
base_duration = 140
extra_duration = max(0, len(node_ids) - 1) * 10
total_duration = base_duration + extra_duration
# Show progress bar
show_progress_bar(duration=total_duration, interval=1, node_count=len(node_ids))
show_progress_bar(
duration=total_duration, interval=1, node_count=len(node_ids)
)
# Second loop: Get status for all nodes
results = []
logger.info("Retrieving status for %d nodes...", len(node_ids))
for node_id in node_ids:
try:
node = Node(project_id=project_id, node_id=node_id, connector=gns3_server)
node = Node(
project_id=project_id,
node_id=node_id,
connector=gns3_server,
)
node.get() # Get latest status
node_info = {
"node_id": node.node_id,
@@ -170,7 +204,9 @@ class GNS3StartNodeTool(BaseTool):
}
results.append(node_info)
except Exception as e:
logger.error("Failed to get status for node %s: %s", node_id, e)
logger.error(
"Failed to get status for node %s: %s", node_id, e
)
results.append(
{
"node_id": node_id,
@@ -181,7 +217,9 @@ class GNS3StartNodeTool(BaseTool):
)
# Analyze results
successful_nodes = [r for r in results if r.get("status") != "error"]
successful_nodes = [
r for r in results if r.get("status") != "error"
]
failed_nodes = [r for r in results if r.get("status") == "error"]
# Construct final response
@@ -211,7 +249,7 @@ class GNS3StartNodeTool(BaseTool):
class GNS3StartNodeQuickTool(BaseTool):
"""
A LangChain tool to start one or multiple nodes in a GNS3 project WITHOUT waiting.
A LangChain tool to start nodes in a GNS3 project WITHOUT waiting.
This tool sends start commands to all nodes and immediately returns status,
without blocking for startup completion. Suitable for automated deployment
@@ -226,7 +264,7 @@ class GNS3StartNodeQuickTool(BaseTool):
}
**Output**:
A dictionary with all nodes' details immediately after sending start commands:
A dict with nodes' details immediately after sending start commands:
{
"project_id": "...",
"total_nodes": 2,
@@ -242,14 +280,18 @@ class GNS3StartNodeQuickTool(BaseTool):
name: str = "start_gns3_node_quick"
description: str = """
Starts one or multiple nodes in a GNS3 project WITHOUT waiting for startup completion.
Starts nodes in a GNS3 project WITHOUT waiting for startup completion.
Use this for automated deployments to avoid HTTP timeouts.
Input: JSON with project_id and node_ids (list of node IDs).
Returns: Dictionary with nodes' details after start commands are sent.
Returns: Dict with nodes' details after start commands are sent.
NOTE: Nodes will continue booting in background after this tool returns.
"""
def _run(self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None) -> dict[str, Any]:
def _run(
self,
tool_input: str,
run_manager: CallbackManagerForToolRun | None = None,
) -> dict[str, Any]:
try:
# Parse input JSON
input_data = json.loads(tool_input)
@@ -258,8 +300,13 @@ class GNS3StartNodeQuickTool(BaseTool):
# Validate input
if not project_id or not node_ids:
logger.error("Missing required fields: project_id or node_ids.")
return {"error": "Missing required fields: project_id and node_ids."}
logger.error(
"Missing required fields: project_id or node_ids."
)
return {
"error": "Missing required fields: "
"project_id and node_ids."
}
if not isinstance(node_ids, list):
logger.error("node_ids must be a list.")
@@ -271,7 +318,10 @@ class GNS3StartNodeQuickTool(BaseTool):
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
return {"error": "Failed to connect to GNS3 server. Please check your configuration."}
return {
"error": "Failed to connect to GNS3 server. "
"Please check your configuration."
}
# Send start commands for all nodes and collect initial status
logger.info(
@@ -283,11 +333,19 @@ class GNS3StartNodeQuickTool(BaseTool):
for node_id in node_ids:
try:
node = Node(project_id=project_id, node_id=node_id, connector=gns3_server)
node = Node(
project_id=project_id,
node_id=node_id,
connector=gns3_server,
)
# Verify node exists and get current info
node.get()
if not node.node_id:
logger.error("Node %s not found in project %s", node_id, project_id)
logger.error(
"Node %s not found in project %s",
node_id,
project_id,
)
results.append(
{
"node_id": node_id,
@@ -300,9 +358,13 @@ class GNS3StartNodeQuickTool(BaseTool):
# Send start command
node.start()
logger.info("Start command sent for node %s (%s)", node_id, node.name)
logger.info(
"Start command sent for node %s (%s)",
node_id,
node.name,
)
# Get immediate status (will likely still be 'starting' or 'stopped')
# Get immediate status (likely 'starting' or 'stopped')
node.get()
node_info = {
"node_id": node.node_id,
@@ -323,7 +385,9 @@ class GNS3StartNodeQuickTool(BaseTool):
)
# Analyze results (count based on successful command sending)
successful_nodes = [r for r in results if r.get("status") != "error"]
successful_nodes = [
r for r in results if r.get("status") != "error"
]
failed_nodes = [r for r in results if r.get("status") == "error"]
# Construct final response
@@ -333,7 +397,10 @@ class GNS3StartNodeQuickTool(BaseTool):
"successful": len(successful_nodes),
"failed": len(failed_nodes),
"nodes": results,
"note": "Start commands sent. Nodes are booting in background. Check node status later.",
"note": (
"Start commands sent. Nodes are booting in background. "
"Check node status later."
),
}
logger.info(
@@ -358,7 +425,9 @@ if __name__ == "__main__":
test_input_single = json.dumps(
{
"project_id": "<PROJECT_UUID>", # Replace with actual project UUID
"node_ids": ["fbeda109-9a74-4d8c-a749-cc3847911a90"], # Replace with actual node UUID
"node_ids": [
"fbeda109-9a74-4d8c-a749-cc3847911a90"
], # Replace with actual node UUID
}
)
tool = GNS3StartNodeTool()
@@ -371,7 +440,8 @@ if __name__ == "__main__":
{
"project_id": "<PROJECT_UUID>", # Replace with actual project UUID
"node_ids": [
"fbeda109-9a74-4d8c-a749-cc3847911a90", # Replace with actual node UUIDs
"fbeda109-9a74-4d8c-a749-cc3847911a90",
# Replace with actual node UUIDs
"another-node-uuid-here",
"third-node-uuid-here",
],
gns3server/agent/gns3_copilot/tools_v2/gns3_update_node_name.py
+61 -18
View File
@@ -26,7 +26,8 @@
GNS3 node name update tool for renaming network devices.
Provides functionality to update the name of one or multiple nodes in GNS3 projects.
Provides functionality to update the name of one or multiple nodes
in GNS3 projects.
"""
import json
@@ -46,10 +47,12 @@ logger = logging.getLogger(__name__)
class GNS3UpdateNodeNameTool(BaseTool):
"""
A LangChain tool to update the name of one or multiple nodes in a GNS3 project.
A LangChain tool to update the name of one or multiple nodes
in a GNS3 project.
**Input**:
A JSON object with project_id and nodes array containing node_id and new_name.
A JSON object with project_id and nodes array containing
node_id and new_name.
Example:
{
"project_id": "uuid-of-project",
@@ -67,8 +70,18 @@ class GNS3UpdateNodeNameTool(BaseTool):
"successful": 2,
"failed": 0,
"nodes": [
{"node_id": "...", "old_name": "...", "new_name": "...", "status": "success"},
{"node_id": "...", "old_name": "...", "new_name": "...", "status": "success"}
{
"node_id": "...",
"old_name": "...",
"new_name": "...",
"status": "success"
},
{
"node_id": "...",
"old_name": "...",
"new_name": "...",
"status": "success"
}
]
}
"""
@@ -76,11 +89,17 @@ class GNS3UpdateNodeNameTool(BaseTool):
name: str = "update_gns3_node_name"
description: str = """
Updates the name of one or multiple nodes in a GNS3 project.
Input: JSON with project_id and nodes array. Each node must have node_id and new_name.
Returns: A dictionary with all nodes' update results including success/failure status.
Input: JSON with project_id and nodes array.
Each node must have node_id and new_name.
Returns: A dictionary with all nodes' update results
including success/failure status.
"""
def _run(self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None) -> dict[str, Any]:
def _run(
self,
tool_input: str,
run_manager: CallbackManagerForToolRun | None = None,
) -> dict[str, Any]:
try:
# Parse input JSON
input_data = json.loads(tool_input)
@@ -103,7 +122,9 @@ class GNS3UpdateNodeNameTool(BaseTool):
return {"error": f"Node {i + 1} must be a dictionary."}
if "node_id" not in node_data or "new_name" not in node_data:
logger.error("Node %d missing node_id or new_name.", i + 1)
return {"error": f"Node {i + 1} missing node_id or new_name."}
return {
"error": f"Node {i + 1} missing node_id or new_name."
}
# Initialize Gns3Connector
logger.info("Connecting to GNS3 server...")
@@ -111,10 +132,17 @@ class GNS3UpdateNodeNameTool(BaseTool):
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
return {"error": "Failed to connect to GNS3 server. Please check your configuration."}
return {
"error": "Failed to connect to GNS3 server. "
"Please check your configuration."
}
# Update node names
logger.info("Updating names for %d nodes in project %s...", len(nodes), project_id)
logger.info(
"Updating names for %d nodes in project %s...",
len(nodes),
project_id,
)
results = []
for i, node_data in enumerate(nodes):
@@ -131,7 +159,11 @@ class GNS3UpdateNodeNameTool(BaseTool):
)
# Get node to retrieve current name
node = Node(project_id=project_id, node_id=node_id, connector=gns3_server)
node = Node(
project_id=project_id,
node_id=node_id,
connector=gns3_server,
)
node.get()
old_name = node.name
@@ -148,7 +180,11 @@ class GNS3UpdateNodeNameTool(BaseTool):
"status": "success",
}
results.append(node_info)
logger.info("Successfully updated node name: %s -> %s", old_name, new_name)
logger.info(
"Successfully updated node name: %s -> %s",
old_name,
new_name,
)
else:
error_info = {
"node_id": node_id,
@@ -159,7 +195,9 @@ class GNS3UpdateNodeNameTool(BaseTool):
"error": "Name verification failed",
}
results.append(error_info)
logger.error("Failed to update node name for %s", node_id)
logger.error(
"Failed to update node name for %s", node_id
)
except Exception as e:
error_info = {
@@ -172,7 +210,9 @@ class GNS3UpdateNodeNameTool(BaseTool):
logger.error("Failed to update node %d: %s", i + 1, e)
# Analyze results
successful_nodes = [r for r in results if r.get("status") == "success"]
successful_nodes = [
r for r in results if r.get("status") == "success"
]
failed_nodes = [r for r in results if r.get("status") == "failed"]
# Construct final response
@@ -205,10 +245,12 @@ if __name__ == "__main__":
print("=== Testing single node name update ===")
test_input_single = json.dumps(
{
"project_id": "your-project-uuid", # Replace with actual project UUID
# Replace with actual project UUID
"project_id": "your-project-uuid",
"nodes": [
{
"node_id": "your-node-uuid", # Replace with actual node UUID
# Replace with actual node UUID
"node_id": "your-node-uuid",
"new_name": "Router1",
}
],
@@ -222,7 +264,8 @@ if __name__ == "__main__":
print("\n=== Testing multiple nodes name update ===")
test_input_multiple = json.dumps(
{
"project_id": "your-project-uuid", # Replace with actual project UUID
# Replace with actual project UUID
"project_id": "your-project-uuid",
"nodes": [
{"node_id": "node-uuid-1", "new_name": "Router1"},
{"node_id": "node-uuid-2", "new_name": "Switch1"},
gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_telnetlib3.py
+79 -37
View File
@@ -25,7 +25,7 @@
"""
Multi-device VPCS command execution tool using telnetlib3 with threading.
Supports concurrent execution of multiple command groups across multiple VPCS devices.
Supports concurrent command execution across VPCS devices.
"""
import json
@@ -47,14 +47,13 @@ logger = logging.getLogger(__name__)
class VPCSMultiCommands(BaseTool):
"""
A tool for VPCS (Virtual PC Simulator) devices to view PC configurations and test connectivity.
A tool for VPCS devices to view PC configs and test connectivity.
**VPCS-SPECIFIC TOOL** - This tool ONLY works with VPCS virtual PC devices.
**IMPORTANT DISTINCTION:**
Unlike network devices (routers/switches), VPCS devices are lightweight virtual PCs that
simulate basic network functionality. Commands like 'ip' are basic PC IP configuration,
NOT network device configuration.
Unlike network devices (routers/switches), VPCS devices are lightweight virtual PCs.
Commands like 'ip' are basic PC IP config, NOT network device config.
**Allowed VPCS Commands:**
- IP configuration: ip <address>/<mask> <gateway> (Basic PC IP setup)
@@ -66,19 +65,19 @@ class VPCSMultiCommands(BaseTool):
**Usage Context:**
This tool is used in lab environments where students need to configure virtual PC IP
addresses and test network connectivity. It does NOT configure network infrastructure.
addresses and test network connectivity. It does NOT configure network infra.
"""
name: str = "execute_vpcs_multi_commands"
description: str = """
**VPCS VIRTUAL PC TOOL** - Configure and test Virtual PC Simulator devices.
This tool ONLY works with VPCS (Virtual PC Simulator) devices, NOT routers/switches.
This tool ONLY works with VPCS devices, NOT routers/switches.
**IMPORTANT: VPCS vs Network Devices**
- VPCS = Lightweight virtual PCs for lab testing (NOT network infrastructure)
- 'ip' command on VPCS = Basic PC IP configuration (like 'ipconfig' on Windows)
- This is NOT the same as configuring router interfaces or routing protocols
- VPCS = Lightweight virtual PCs for lab testing (NOT network infra)
- 'ip' command on VPCS = Basic PC IP config (like 'ipconfig' on Windows)
- NOT the same as configuring router interfaces or routing protocols
**When to Use This Tool:**
- Configure IP addresses on virtual PCs: ip 10.10.0.12/24 10.10.0.254
@@ -102,9 +101,9 @@ class VPCSMultiCommands(BaseTool):
]
}
**Returns:** PC command outputs for IP configuration and connectivity testing.
**Returns:** PC command outputs for IP config and connectivity testing.
**Note:** For network devices (Cisco/Huawei routers), use execute_multiple_device_commands instead.
**Note:** For network devices, use execute_multiple_device_commands.
"""
def _connect_and_execute_commands(
@@ -131,11 +130,17 @@ class VPCSMultiCommands(BaseTool):
# Check if device has port information
if device_name not in device_ports:
logger.warning("Device '%s' not found in topology or missing console port", device_name)
logger.warning(
"Device '%s' not found in topology or missing console port",
device_name,
)
results_list[index] = {
"device_name": device_name,
"status": "error",
"output": f"Device '{device_name}' not found in topology or missing console port",
"output": (
f"Device '{device_name}' not found in topology "
"or missing console port"
),
"commands": commands,
}
return
@@ -143,7 +148,12 @@ class VPCSMultiCommands(BaseTool):
port = device_ports[device_name]["port"]
host = gns3_host
logger.info("Connecting to device '%s' at %s:%d", device_name, host, port)
logger.info(
"Connecting to device '%s' at %s:%d",
device_name,
host,
port,
)
tn = Telnet()
try:
@@ -197,7 +207,11 @@ class VPCSMultiCommands(BaseTool):
)
except Exception as e:
logger.error("Error executing commands on device '%s': %s", device_name, str(e))
logger.error(
"Error executing commands on device '%s': %s",
device_name,
str(e),
)
results_list[index] = {
"device_name": device_name,
"status": "error",
@@ -218,10 +232,14 @@ class VPCSMultiCommands(BaseTool):
Returns:
True if valid UUID format, False otherwise
"""
uuid_pattern = r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"
uuid_pattern = (
r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"
)
is_valid = bool(re.match(uuid_pattern, project_id, re.IGNORECASE))
if not is_valid:
logger.warning("project_id '%s' is not a valid UUID format", project_id)
logger.warning(
"project_id '%s' is not a valid UUID format", project_id
)
return is_valid
def _validate_tool_input(
@@ -231,10 +249,10 @@ class VPCSMultiCommands(BaseTool):
Validate device command input and extract project_id and device_configs.
Args:
tool_input: The input received from the LangChain/LangGraph tool call.
tool_input: Input from the LangChain/LangGraph tool call.
Returns:
Tuple containing (device_configs_list, project_id) or (error_list, "")
Tuple of (device_configs_list, project_id) or (error_list, "")
"""
parsed_input = None
@@ -247,18 +265,23 @@ class VPCSMultiCommands(BaseTool):
parsed_input = json.loads(tool_input)
logger.info("Successfully parsed tool input from JSON string.")
except json.JSONDecodeError as e:
logger.error("Invalid JSON string received as tool input: %s", e)
logger.error(
"Invalid JSON string received as tool input: %s", e
)
return ([{"error": f"Invalid JSON input: {e}"}], "")
else:
# Handle standard models where the framework has already parsed the JSON.
parsed_input = tool_input
logger.info("Using tool input directly as type: %s", type(parsed_input).__name__)
logger.info(
"Using tool input directly as type: %s",
type(parsed_input).__name__,
)
# Validate input is a dictionary
if not isinstance(parsed_input, dict):
error_msg = (
"Tool input must be a JSON object containing 'project_id' and 'device_configs', "
f"but got {type(parsed_input).__name__}"
"Tool input must be a JSON object containing 'project_id' "
f"and 'device_configs', but got {type(parsed_input).__name__}"
)
logger.error(error_msg)
return ([{"error": error_msg}], "")
@@ -272,7 +295,7 @@ class VPCSMultiCommands(BaseTool):
# Validate project_id format
if not self._validate_project_id(project_id):
error_msg = f"Invalid project_id format: {project_id}. Expected UUID format."
error_msg = f"Invalid project_id format: {project_id}. Expected UUID."
logger.error(error_msg)
return ([{"error": error_msg}], "")
@@ -285,7 +308,10 @@ class VPCSMultiCommands(BaseTool):
# Validate device_configs is a list
if not isinstance(device_configs, list):
error_msg = f"'device_configs' must be a list, but got {type(device_configs).__name__}"
error_msg = (
f"'device_configs' must be a list, "
f"but got {type(device_configs).__name__}"
)
logger.error(error_msg)
return ([{"error": error_msg}], "")
@@ -297,24 +323,32 @@ class VPCSMultiCommands(BaseTool):
# Validate each item in device_configs
for i, item in enumerate(device_configs):
if not isinstance(item, dict):
error_msg = f"Item at index {i} must be a dictionary, got {type(item).__name__}"
error_msg = (
f"Item at index {i} must be a dictionary, "
f"got {type(item).__name__}"
)
logger.error(error_msg)
return ([{"error": error_msg}], "")
# Validate required fields in each device config
if "device_name" not in item:
error_msg = f"Item at index {i} missing required field 'device_name'"
error_msg = (
f"Item at index {i} missing required field 'device_name'"
)
logger.error(error_msg)
return ([{"error": error_msg}], "")
if "commands" not in item:
error_msg = f"Item at index {i} missing required field 'commands'"
error_msg = (
f"Item at index {i} missing required field 'commands'"
)
logger.error(error_msg)
return ([{"error": error_msg}], "")
if not isinstance(item["commands"], list):
error_msg = (
f"'commands' in item at index {i} must be a list, " f"but got {type(item['commands']).__name__}"
f"'commands' in item at index {i} must be a list, "
f"but got {type(item['commands']).__name__}"
)
logger.error(error_msg)
return ([{"error": error_msg}], "")
@@ -333,13 +367,14 @@ class VPCSMultiCommands(BaseTool):
**kwargs: Any,
) -> list[dict[str, Any]]:
"""
Main method to execute commands on multiple VPCS virtual PC devices concurrently.
Execute commands on multiple VPCS virtual PC devices concurrently.
VPCS (Virtual PC Simulator) devices are lightweight virtual machines that simulate
basic PC network functionality for lab testing. This is NOT network device configuration.
VPCS devices are lightweight virtual machines that simulate basic PC network
functionality for lab testing. This is NOT network device configuration.
Args:
tool_input: JSON string containing project_id and device_configs with VPCS commands
tool_input: JSON string containing project_id and device_configs with
VPCS commands
Returns:
List of execution results for each VPCS device
@@ -352,14 +387,20 @@ class VPCSMultiCommands(BaseTool):
device_configs, project_id = self._validate_tool_input(tool_input)
# Check if validation returned an error
if isinstance(device_configs, list) and len(device_configs) > 0 and "error" in device_configs[0]:
if (
isinstance(device_configs, list)
and len(device_configs) > 0
and "error" in device_configs[0]
):
return device_configs
# Extract all device names from input using set comprehension
device_names = {config["device_name"] for config in device_configs}
# Get device port mapping with project_id
device_ports = get_device_ports_from_topology(list(device_names), project_id=project_id)
device_ports = get_device_ports_from_topology(
list(device_names), project_id=project_id
)
logger.info(
"Retrieved port mappings for %d devices: %s",
len(device_ports),
@@ -400,7 +441,8 @@ class VPCSMultiCommands(BaseTool):
error_count = sum(1 for r in results if r.get("status") == "error")
logger.info(
"Multi-device command execution completed. Total: %d, Success: %d, Error: %d",
"Multi-device command execution completed. Total: %d, Success: %d, "
"Error: %d",
len(results),
success_count,
error_count,