diff --git a/gns3server/agent/gns3_copilot/gns3_client/__init__.py b/gns3server/agent/gns3_copilot/gns3_client/__init__.py index 4a0b6e599..e1c6907b6 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/__init__.py +++ b/gns3server/agent/gns3_copilot/gns3_client/__init__.py @@ -27,7 +27,11 @@ Main functions: - get_gns3_connector_with_llm_config: Factory function to create connector AND retrieve LLM config """ -from .connector_factory import get_gns3_connector, get_gns3_connector_with_llm_config +from .connector_factory import ( + get_gns3_connector, + get_gns3_connector_with_llm_config, + get_gns3_server_host, +) from .custom_gns3fy import ( CONSOLE_TYPES, LINK_TYPES, @@ -93,6 +97,7 @@ __all__ = [ "GNS3UpdateDrawingTool", "get_gns3_connector", "get_gns3_connector_with_llm_config", + "get_gns3_server_host", "add_file_to_index", "get_file_list", ] diff --git a/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py b/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py index 606e0ace6..19c4fd607 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py +++ b/gns3server/agent/gns3_copilot/gns3_client/connector_factory.py @@ -298,3 +298,40 @@ def _detect_url_for_api() -> Optional[str]: # Fallback return DEFAULT_GNS3_URL + + +def get_gns3_server_host() -> str: + """ + Get GNS3 server hostname from Controller or Config. + + This is a convenience function for extracting the hostname only, + useful for Nornir tools that need the GNS3 server address. + + Uses the same priority order as get_gns3_connector: + 1. Controller.instance().compute("local") + 2. Config.instance().settings.Server + 3. Fallback to DEFAULT_GNS3_URL host + + Returns: + Hostname/IP address string + + Example: + from gns3_copilot.gns3_client import get_gns3_server_host + + host = get_gns3_server_host() + print(f"GNS3 server host: {host}") + """ + url = _detect_url_for_api() + + # Extract host from URL + # URL format: protocol://host:port + try: + # Remove protocol prefix + host_part = url.split("://")[1] + # Extract host (before the port) + host = host_part.split(":")[0] + logger.debug("Extracted GNS3 server host: %s from URL: %s", host, url) + return host + except Exception as e: + logger.warning("Failed to extract host from URL %s: %s, using fallback", url, e) + return DEFAULT_GNS3_URL.split("://")[1].split(":")[0] 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 7aa6f1b57..762c73b7e 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py +++ b/gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py @@ -8,6 +8,8 @@ import logging import os from typing import Any +from gns3_copilot.gns3_client import get_gns3_server_host + from langchain.tools import BaseTool from langchain_core.callbacks import CallbackManagerForToolRun from netmiko.exceptions import ReadTimeout @@ -33,7 +35,7 @@ def _get_nornir_groups_config() -> dict[str, dict[str, Any]]: return { "cisco_IOSv_telnet": { "platform": "cisco_ios", - "hostname": os.getenv("GNS3_SERVER_HOST", "127.0.0.1"), + "hostname": get_gns3_server_host(), "timeout": 120, "username": "", "password": "", @@ -371,7 +373,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool): defaults = _get_nornir_defaults() # Log nornir account information - gns3_host = os.getenv("GNS3_SERVER_HOST", "127.0.0.1") + gns3_host = get_gns3_server_host() logger.info( "Initializing Nornir with account: host=%s, platform=%s, timeout=%d", 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 66ebc268d..35f4ea10e 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py +++ b/gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py @@ -9,6 +9,8 @@ import os import re from typing import Any +from gns3_copilot.gns3_client import get_gns3_server_host + from langchain.tools import BaseTool from langchain_core.callbacks import CallbackManagerForToolRun from netmiko.exceptions import ReadTimeout @@ -34,7 +36,7 @@ def _get_nornir_groups_config() -> dict[str, dict[str, Any]]: return { "cisco_IOSv_telnet": { "platform": "cisco_ios", - "hostname": os.getenv("GNS3_SERVER_HOST", "127.0.0.1"), + "hostname": get_gns3_server_host(), "timeout": 120, "username": "", "password": "", @@ -385,7 +387,7 @@ class ExecuteMultipleDeviceCommands(BaseTool): defaults = _get_nornir_defaults() # Log nornir account information - gns3_host = os.getenv("GNS3_SERVER_HOST", "127.0.0.1") + gns3_host = get_gns3_server_host() logger.info( "Initializing Nornir with account: host=%s, platform=%s, timeout=%d", diff --git a/gns3server/agent/gns3_copilot/tools_v2/linux_tools_nornir.py b/gns3server/agent/gns3_copilot/tools_v2/linux_tools_nornir.py deleted file mode 100644 index 3743cc048..000000000 --- a/gns3server/agent/gns3_copilot/tools_v2/linux_tools_nornir.py +++ /dev/null @@ -1,657 +0,0 @@ -""" -This module uses Nornir + Netmiko to batch execute Linux commands on GNS3 topology devices -via Telnet console. -""" - -import json -import logging -import os -import re -import time -from typing import Any - -from langchain.tools import BaseTool -from langchain_core.callbacks import CallbackManagerForToolRun -from nornir import InitNornir -from nornir.core import Nornir -from nornir.core.task import AggregatedResult, Result, Task -from nornir_netmiko.tasks import netmiko_send_command - -from gns3_copilot.utils import get_device_ports_from_topology - -# config log -logger = logging.getLogger(__name__) - - -# Local Nornir configuration functions for Linux Telnet devices -def _get_nornir_defaults() -> dict[str, Any]: - """Get Nornir default configuration for Linux Telnet.""" - return {"data": {"location": "gns3"}} - - -def _get_nornir_groups_config() -> dict[str, dict[str, Any]]: - """Get Nornir groups configuration for Linux Telnet devices.""" - return { - "linux_telnet": { - "platform": "linux", - "hostname": os.getenv("GNS3_SERVER_HOST", "127.0.0.1"), - "timeout": 120, - "username": os.getenv("LINUX_TELNET_USERNAME", ""), - "password": os.getenv("LINUX_TELNET_PASSWORD", ""), - "connection_options": { - "netmiko": { - "platform": "linux", - "extras": { - "device_type": "generic_telnet", - "global_delay_factor": 3, - "timeout": 120, - "fast_cli": False, - }, - } - }, - }, - } - - -def _get_nornir_group(group_name: str = "linux_telnet") -> dict[str, Any]: - """Get Nornir group configuration for a specific group.""" - all_groups = _get_nornir_groups_config() - return all_groups.get(group_name, {}) - - -class LinuxTelnetBatchTool(BaseTool): - """ - A tool to execute commands on Linux devices via Telnet console in GNS3 labs. - - **CRITICAL: NON-INTERACTIVE ONLY** - - This tool ONLY supports non-interactive commands that exit immediately. - Interactive commands will cause the tool to hang and fail. - - **Strictly Prohibited (Interactive Commands):** - - Text editors: vi, vim, nano, emacs - - Interactive viewers: less, more - - Continuous monitors: top, htop, iotop (use top -b -n 1 instead) - - Interactive shells: bash, sh, python, REPL environments - - Commands requiring user input: passwd, chsh, interactive installers - - Any command with pagination that waits for user input - - **Required: Non-Interactive Alternatives** - - INSTEAD OF: top → USE: top -b -n 1 (batch mode, single iteration) - - INSTEAD OF: vi file.txt → USE: cat file.txt or head file.txt - - INSTEAD OF: less file.txt → USE: cat file.txt or head -n 50 file.txt - - INSTEAD OF: ping host → USE: ping -c 4 host (limited count) - - INSTEAD OF: tail → USE: tail -n 20 (explicit line count) - - **Allowed Command Types:** - - System info: uname -a, hostnamectl, cat /etc/os-release - - Network diagnostics: ip a, ip route, ping -c 4, traceroute -n - - Process listing: ps aux, ps aux --sort=-%mem | head -20 - - Service status: systemctl status ssh --no-pager - - Log viewing: journalctl -u ssh --no-pager -n 50, cat /var/log/syslog | tail -50 - - File operations: ls -la, cat, head, tail (with explicit limits), find, grep - """ - - name: str = "linux_telnet_batch_commands" - description: str = """ - **LINUX DIAGNOSTIC TOOL** - Execute non-interactive commands on Linux devices via Telnet. - - **CRITICAL: NON-INTERACTIVE COMMANDS ONLY** - - This tool will HANG and FAIL if you use interactive commands. All commands MUST exit immediately without user input. - - **STRICTLY FORBIDDEN:** - ❌ Text editors: vi, vim, nano, emacs - ❌ Interactive pagers: less, more - ❌ Continuous monitors: top, htop, iotop (unless using batch mode) - ❌ Interactive shells: bash, sh, python, REPL - ❌ User input commands: passwd, chsh, SSH with password prompts - ❌ Any command that waits for keyboard input or pagination - - **NON-INTERACTIVE ALTERNATIVES (USE THESE):** - ✅ System info: uname -a, hostnamectl, cat /etc/os-release - ✅ Network: ip a, ip route, ping -c 4 , traceroute -n - ✅ Processes: ps aux, ps aux --sort=-%mem | head -20 - ✅ Services: systemctl status --no-pager - ✅ Logs: journalctl -u --no-pager -n 50, tail -50 /var/log/syslog - ✅ Files: cat, head -n 50, tail -n 20, grep, find - ✅ Batch monitoring: top -b -n 1 (NOT interactive top) - - **Input Format:** - { - "project_id": "f32ebf3d-ef8c-4910-b0d6-566ed828cd24", - "device_configs": [ - { - "device_name": "debian01", - "commands": ["uname -a", "ip a", "ping -c 4 8.8.8.8"] - }, - { - "device_name": "ubuntu01", - "commands": ["hostnamectl", "systemctl status ssh --no-pager"] - } - ] - } - - **Important Notes:** - - ALL commands must use --no-pager flag when available - - Add explicit limits: head -n 50, tail -n 20, ping -c 4 - - For testing: Execute server/client commands on one device at a time, NOT simultaneously - - This tool is for diagnostics and information gathering - """ - - def _run( - self, - tool_input: str | bytes | list[Any] | dict[str, Any], - run_manager: CallbackManagerForToolRun | None = None, - **kwargs: Any, - ) -> list[dict[str, Any]]: - """ - Execute non-interactive Linux commands on multiple devices via Telnet. - - **CRITICAL:** All commands MUST be non-interactive and exit immediately. - Interactive commands (vi, top, less, etc.) will cause execution to hang. - - Args: - tool_input: JSON string containing project_id and device_configs with commands - - Returns: - List of execution results for each Linux device - """ - # Log received input - logger.info("Received input: %s", tool_input) - - # Validate input first (before checking credentials) - 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] - ): - return device_configs_list - - # Check credentials only for valid inputs - linux_username = os.getenv("LINUX_TELNET_USERNAME", "") - linux_password = os.getenv("LINUX_TELNET_PASSWORD", "") - - if not linux_username or not linux_password: - user_message = ( - "Sorry, I can't proceed just yet.\n\n" - "You haven't configured the Linux login credentials (username and password) yet.\n" - "Please go to the **Settings** page and fill in the Linux username and password under the login credentials section.\n\n" - "Once you've saved them, just come back and say anything (like 'Done' or 'Configured'), " - "and I'll immediately continue with the task!\n\n" - "Need help finding the settings page? Let me know — happy to guide you!" - ) - logger.warning( - "Linux login credentials not configured — user prompted to set them up" - ) - return [ - { - "error": user_message, - "action_required": "configure_linux_credentials", - "user_message": user_message, # optional, if your frontend uses a separate field - } - ] - - # Create a mapping of device names to their display commands - device_configs_map = self._configs_map(device_configs_list) - - # Prepare device hosts data - try: - 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)}] - - # Initialize Nornir - try: - dynamic_nr = self._initialize_nornir(hosts_data) - except ValueError as e: - logger.error("Failed to initialize Nornir: %s", e) - return [{"error": str(e)}] - - results = [] - - # Execute login first, then commands - try: - # Step 1: Execute login for all devices - login_result = dynamic_nr.run(task=self._linux_telnet_login) - - # Step 2: Check login results and execute commands for successful logins - successful_logins = [] - failed_logins = [] - - for device_name, result in login_result.items(): - if result.failed: - failed_logins.append(device_name) - logger.error( - "Device %s login failed: %s", device_name, result.result - ) - else: - successful_logins.append(device_name) - logger.info( - "Device %s login successful: %s", device_name, result.result - ) - - task_result: AggregatedResult | dict[str, Any] - - # Step 3: Execute commands only for devices with successful login - if successful_logins: - # Filter device_configs_map to only include successfully logged in devices - filtered_device_configs_map = { - device_name: commands - for device_name, commands in device_configs_map.items() - if device_name in successful_logins - } - - task_result = dynamic_nr.run( - task=self._run_all_device_configs_with_single_retry, - device_configs_map=filtered_device_configs_map, - ) - else: - task_result = AggregatedResult("empty_command_execution") - - # Step 4: Process results for all devices - results = self._process_task_results( - device_configs_list, hosts_data, task_result, login_result - ) - - except Exception as e: - # Overall execution failed - logger.error("Error executing display on all devices: %s", e) - return [{"error": f"Execution error: {str(e)}"}] - - logger.info( - "Multiple device display execution completed. Results: %s", - json.dumps(results, indent=2, ensure_ascii=False), - ) - - return results - - def _linux_telnet_login(self, task: Task) -> Result: - """ - Smart Linux Telnet login: detect login status and only login when needed. - - This method handles authentication for Linux Telnet sessions in GNS3 labs. - It detects whether a device is already logged in before attempting authentication. - """ - try: - net_connect = task.host.get_connection("netmiko", task.nornir.config) - - # Clear the buffer + press Enter several times to wake up the device. - net_connect.clear_buffer() - time.sleep(0.3) - net_connect.write_channel("\n\n") - - # Read the device output (wait up to 10 seconds) - output = net_connect.read_channel_timing(read_timeout=10) - logger.info("Device %s initial output: %s", task.host.name, output) - - # Check if output contains "login:" prompt - if re.search(r"(?i)(^|\n).{0,60}(debian\s+)?login:\s*$", output): - # Need to login, execute login process - logger.info( - "Device %s requires login - detected login prompt", task.host.name - ) - - # Send username - net_connect.write_channel(f"{task.host.username}\n") - time.sleep(1) - output = net_connect.read_until_prompt_or_pattern( - "Password:", read_timeout=10 - ) - - # Send password - net_connect.write_channel(f"{task.host.password}\n") - time.sleep(1) - output += net_connect.read_until_prompt_or_pattern( - r"[$#]", read_timeout=10 - ) - - logger.info("Device %s login successful", task.host.name) - return Result(host=task.host, result="Login successful") - - # Already logged in, return directly - logger.info( - "Device %s already logged in - no login prompt detected", task.host.name - ) - return Result(host=task.host, result="Already logged in") - - except Exception as e: - logger.error("Device %s login failed: %s", task.host.name, e) - return Result(host=task.host, result=f"Login failed: {str(e)}", failed=True) - - def _run_all_device_configs_with_single_retry( - self, task: Task, device_configs_map: dict[str, list[str]] - ) -> Result: - """ - Execute non-interactive commands one-by-one on a single Linux device. - - **WARNING:** Each command MUST exit immediately without user input. - If any command hangs waiting for input, the entire execution will fail. - - Optimized for generic_telnet with $ prompt and passwordless sudo. - """ - device_name = task.host.name - config_commands = device_configs_map.get(device_name, []) - - if not config_commands: - return Result(host=task.host, result="No display commands to execute") - - _outputs = {} - for _cmd in config_commands: - try: - # Use timing mode + increased delay_factor to ensure stability with $ prompt - # and passwordless sudo - _result = task.run( - task=netmiko_send_command, - command_string=_cmd, - use_timing=True, - delay_factor=3, - max_loops=5000, - ) - _outputs[_cmd] = _result.result - except Exception as e: - _outputs[_cmd] = f"Command execution failed: {str(e)}" - - return Result(host=task.host, result=_outputs) - - def _validate_tool_input( - self, tool_input: str | bytes | list[Any] | dict[str, Any] - ) -> tuple[list[dict[str, Any]], str | None]: - """ - Validate device display command input, handling 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. - - Returns: - Tuple containing (device_configs_list, project_id) or (error_list, None) - """ - - parsed_input = None - - # 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. - 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) - 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__ - ) - - # Handle new format: {"project_id": "...", "device_configs": [...]} - if isinstance(parsed_input, dict): - project_id = parsed_input.get("project_id") - device_configs = parsed_input.get("device_configs") - - # Validate project_id - if not project_id: - error_msg = "Missing required 'project_id' field in input" - logger.error(error_msg) - 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." - ) - logger.error(error_msg) - return ([{"error": error_msg}], None) - - # Validate device_configs - if not isinstance(device_configs, list): - error_msg = "'device_configs' must be an array" - logger.error(error_msg) - return ([{"error": error_msg}], None) - - if not device_configs: - logger.warning("Device configs list is empty.") - return [], project_id - - return device_configs, project_id - - # Handle legacy format: [...] - elif isinstance(parsed_input, list): - logger.warning( - "Using legacy input 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__}" - ) - logger.error(error_msg) - return ([{"error": error_msg}], None) - - def _validate_project_id(self, project_id: str) -> bool: - """ - Validate project_id format (UUID). - - Args: - project_id: The project ID to validate - - 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}$" - return bool(re.match(uuid_pattern, project_id, re.IGNORECASE)) - - def _configs_map( - self, device_config_list: list[dict[str, Any]] - ) -> dict[str, list[str]]: - """Create a mapping of device names to their display commands.""" - device_configs_map = {} - for device_config in device_config_list: - device_name = device_config["device_name"] - config_commands = device_config["commands"] - device_configs_map[device_name] = config_commands - - return device_configs_map - - def _prepare_device_hosts_data( - 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 - ] - - # 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}" - ) - raise ValueError(error_msg) - - # Force all devices to use linux_telnet group (compatible with generic_telnet) - for _, _host_info in hosts_data.items(): - _host_info["groups"] = ["linux_telnet"] - - # Check for missing devices - missing_devices = set(device_names) - set(hosts_data.keys()) - if missing_devices: - logger.warning( - "Some devices not found in topology (Project ID: %s): %s", - project_id or "default", - missing_devices, - ) - - return hosts_data - - def _initialize_nornir(self, hosts_data: dict[str, dict[str, Any]]) -> Nornir: - """Initialize Nornir with the provided hosts data.""" - try: - # Get latest environment configuration - groups_data = _get_nornir_group("linux_telnet") - defaults = _get_nornir_defaults() - - return InitNornir( - inventory={ - "plugin": "DictInventory", - "options": { - "hosts": hosts_data, - "groups": {"linux_telnet": groups_data}, - "defaults": defaults, - }, - }, - runner={ - "plugin": "threaded", - "options": {"num_workers": 10}, - }, - logging={"enabled": False}, - ) - except Exception as e: - logger.error("Failed to initialize Nornir: %s", e) - raise ValueError(f"Failed to initialize Nornir: {e}") from e - - def _process_task_results( - self, - device_configs_list: list[dict[str, Any]], - hosts_data: dict[str, dict[str, Any]], - task_result: AggregatedResult, - login_result: AggregatedResult | None = None, - ) -> list[dict[str, Any]]: - """Process task results and format them for return.""" - results = [] - - for device_config in device_configs_list: - device_name = device_config["device_name"] - config_commands = device_config["commands"] - - # Check if device is in topology - if device_name not in hosts_data: - device_result = { - "device_name": device_name, - "status": "failed", - "error": ( - f"Device '{device_name}' not found in topology or missing console_port" - ), - } - results.append(device_result) - continue - - # Check login result first - if login_result and device_name in login_result: - login_status = login_result[device_name] - if login_status.failed: - device_result = { - "device_name": device_name, - "status": "failed", - "error": f"Login failed: {login_status.result}", - "login_status": login_status.result, - } - results.append(device_result) - continue - - # Check if device has execution results - if device_name not in task_result: - device_result = { - "device_name": device_name, - "status": "failed", - "error": (f"Device '{device_name}' not found in task results"), - } - results.append(device_result) - continue - - # Process execution results - multi_result = task_result[device_name] - device_result = {"device_name": device_name} - - if multi_result[0].failed: - # Execution failed - device_result["status"] = "failed" - device_result["error"] = ( - f"Command execution failed: {multi_result[0].result}" - ) - device_result["output"] = multi_result[0].result - else: - # Execution successful - device_result["status"] = "success" - device_result["output"] = multi_result[0].result - device_result["config_commands"] = config_commands - - # Add login status if available - if login_result and device_name in login_result: - device_result["login_status"] = login_result[device_name].result - - results.append(device_result) - - return results - - -if __name__ == "__main__": - # Example usage with new format - device_commands = json.dumps( - { - "project_id": "f32ebf3d-ef8c-4910-b0d6-566ed828cd24", - "device_configs": [ - { - "device_name": "Debian12.6-1", - "commands": [ - "uname -a", - "hostnamectl || hostname", - "cat /etc/os-release", - "whoami && id", - "id", - "pwd", - "top -b -n 1 | head -20", - "ip neigh show | grep -v REACHABLE | grep -v PERMANENT", - "ping -c 3 114.114.114.114", - "ps aux --sort=-%mem | head -15", - "journalctl -u ssh --no-pager -n 20", - 'find /etc -name "*.conf" | head -10', - ], - }, - { - "device_name": "Debian12.6-2", - "commands": [ - "uname -a", - "hostnamectl || hostname", - "cat /etc/os-release", - "whoami && id", - "id", - "pwd", - "top -b -n 1 | head -20", - "ip neigh show | grep -v REACHABLE | grep -v PERMANENT", - "ping -c 3 114.114.114.114", - "ps aux --sort=-%mem | head -15", - "journalctl -u ssh --no-pager -n 20", - 'find /etc -name "*.conf" | head -10', - ], - }, - ], - } - ) - - exe_cmd = LinuxTelnetBatchTool() - - failed_count = 0 - - for _i in range(0, 1): - exe_results = exe_cmd._run(tool_input=device_commands) - for exe_result in exe_results: - for exe_result in exe_results: - if exe_result.get("status") == "failed": - failed_count += 1 - - print("Execution results:") - print(json.dumps(exe_results, indent=2, ensure_ascii=False)) - print(f"Failed Count: {failed_count}") diff --git a/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_telnetlib3.py b/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_telnetlib3.py index 7002d5449..215cea76d 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_telnetlib3.py +++ b/gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_telnetlib3.py @@ -15,6 +15,7 @@ from langchain.tools import BaseTool from langchain_core.callbacks import CallbackManagerForToolRun from telnetlib3 import Telnet +from gns3_copilot.gns3_client import get_gns3_server_host from gns3_copilot.utils import get_device_ports_from_topology logger = logging.getLogger(__name__) @@ -358,8 +359,8 @@ class VPCSMultiCommands(BaseTool): list(device_ports.keys()), ) - # Get host IP from environment variable - gns3_host = os.getenv("GNS3_SERVER_HOST", "127.0.0.1") + # Get GNS3 server host from connector factory + gns3_host = get_gns3_server_host() logger.info("Using GNS3 server host: %s", gns3_host) # Initialize results list (pre-allocate space for concurrent writes) diff --git a/gns3server/agent/gns3_copilot/tools_v2/window_controller.py b/gns3server/agent/gns3_copilot/tools_v2/window_controller.py deleted file mode 100644 index 342cf9806..000000000 --- a/gns3server/agent/gns3_copilot/tools_v2/window_controller.py +++ /dev/null @@ -1,726 +0,0 @@ -# SPDX-License-Identifier: AGPL-3.0-or-later -# -# This file is part of FlowNet-Lab. -# -# FlowNet-Lab is free software: you can redistribute it and/or modify it -# under the terms of the GNU Affero General Public License as published by the -# Free Software Foundation, either version 3 of the License, or (at your -# option) any later version. -# -# FlowNet-Lab 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 Affero General Public License -# for more details. -# -# You should have received a copy of the GNU Affero General Public License along -# with FlowNet-Lab. If not, see . - -""" -Window Controller Tool - Frontend Proxy Tool - -This tool acts as a proxy to frontend window operations. -It sends commands to the frontend via WebSocket and waits for execution results. - -Architecture: - Agent → WindowControllerTool._arun() → WebSocket → Frontend - ← (waits for future) ← execution result -""" - -import json -import uuid -from typing import Optional, Type, Any - -from langchain.tools import BaseTool -from langchain_core.callbacks import ( - AsyncCallbackManagerForToolRun, - CallbackManagerForToolRun, -) -from pydantic import BaseModel, Field - -import logging -from backend.core.session_manager import get_session_manager - -logger = logging.getLogger(__name__) - - -# Window type configurations with minimal required parameters -# Size and position should be dynamically generated by the LLM based on context -WINDOW_CONFIGS = { - "ai_chat": { - "window_type": "ai_chat", - "title": "AI Chat", - "min_size": {"width": 400, "height": 500}, - "description": "AI conversation interface for chatting with the assistant", - }, - "network_topology": { - "window_type": "network_topology", - "title": "Network Topology", - "min_size": {"width": 600, "height": 400}, - "description": "GNS3 network topology visualization view", - }, - "terminal": { - "window_type": "terminal", - "title": "Terminal", - "min_size": {"width": 500, "height": 300}, - "description": "Device terminal emulator for command-line access", - }, - "calibre_books": { - "window_type": "calibre_books", - "title": "Calibre Books", - "min_size": {"width": 500, "height": 400}, - "description": "Calibre ebook library browser", - }, - "notes": { - "window_type": "notes", - "title": "Notes", - "min_size": {"width": 300, "height": 300}, - "description": "Note-taking and management interface", - }, - "pdf_reader": { - "window_type": "pdf_reader", - "title": "PDF Reader", - "min_size": {"width": 500, "height": 400}, - "description": "PDF document viewer and reader", - }, - "gns3_projects": { - "window_type": "gns3_projects", - "title": "GNS3 Projects", - "min_size": {"width": 500, "height": 400}, - "description": "GNS3 project selection and management", - }, - "video_recorder": { - "window_type": "video_recorder", - "title": "Video Recorder", - "min_size": {"width": 400, "height": 300}, - "description": "Screen recording tool", - }, - "settings": { - "window_type": "settings", - "title": "Settings", - "min_size": {"width": 400, "height": 300}, - "description": "System settings and configuration", - }, -} - -VALID_ACTIONS = { - "open": ["window_type"], - "close": ["window_title"], - "focus": ["window_title"], - "minimize": ["window_title"], - "maximize": ["window_title"], - "restore": ["window_title"], - "center": ["window_title"], - "move": ["window_title", "x", "y"], - "resize": ["window_title", "width", "height"], -} - - -class WindowControllerInput(BaseModel): - """Input schema for WindowControllerTool.""" - - action: str = Field( - ..., - description=f"Window action to perform. Valid actions: {list(VALID_ACTIONS.keys())}" - ) - window_type: Optional[str] = Field( - None, - description=f"Type of window to open (required for 'open' action). Valid types: {list(WINDOW_CONFIGS.keys())}" - ) - window_title: Optional[str] = Field( - None, - description="Title of the target window (required for close, focus, minimize, maximize, center actions)" - ) - x: Optional[float] = Field(None, description="X coordinate for move action") - y: Optional[float] = Field(None, description="Y coordinate for move action") - width: Optional[float] = Field(None, description="Width for resize action or when opening a new window") - height: Optional[float] = Field(None, description="Height for resize action or when opening a new window") - size: Optional[dict] = Field(None, description="Complete size object: {\"width\": 800, \"height\": 600} (for open action)") - position: Optional[dict] = Field(None, description="Complete position object: {\"x\": 200, \"y\": 150} (for open action)") - - -class WindowControllerTool(BaseTool): - """ - Proxy tool for frontend window control operations. - - This tool sends window control commands to the frontend via WebSocket - and waits for the frontend to execute and return results. - - The actual window operations (open, close, focus, etc.) are executed - in the browser, not on the backend. - """ - - name: str = "window_controller" - description: str = """ -Control windows in the frontend web interface. - -Supported actions: -- open: Open a new window (requires: window_type) - You SHOULD intelligently generate size and position based on window type and context. - Example: {"action": "open", "window_type": "terminal", "size": {"width": 900, "height": 600}, "position": {"x": 250, "y": 150}} - -- close: Close a window (requires: window_title) - Example: {"action": "close", "window_title": "Notes"} - -- focus: Bring a window to front - increases z-index to make it visible above other windows (requires: window_title) - Use this when a window is visible but behind other windows. - Does NOT change minimized/maximized state. - Example: {"action": "focus", "window_title": "Notes"} - -- minimize: Minimize a window - hide it from view (requires: window_title) - Window becomes hidden and can be restored later. - Example: {"action": "minimize", "window_title": "Notes"} - -- maximize: Maximize a window - expand to fullscreen (requires: window_title) - Window fills the entire screen. Can be restored to previous size. - Example: {"action": "maximize", "window_title": "Notes"} - -- restore: Restore a minimized or maximized window to its normal state (requires: window_title) - Use this when user says "restore", "recover", "show", "unhide", "unminimize", "unmaximize" a window. - This is the CORRECT action when window is minimized or maximized and user wants it back to normal. - Example: {"action": "restore", "window_title": "Notes"} - -- center: Center a window on screen (requires: window_title) - Window occupies 90%% of viewport in center. - Example: {"action": "center", "window_title": "Notes"} - -- move: Move a window (requires: window_title, x, y) - Example: {"action": "move", "window_title": "Notes", "x": 100, "y": 100} - -- resize: Resize a window (requires: window_title, width, height) - Example: {"action": "resize", "window_title": "Notes", "width": 800, "height": 600} - -Available window types and recommended sizes: -- "ai_chat": AI Chat interface (600x800 default, good for right side of screen) -- "network_topology": Network topology visualization (1000x700 default, needs large space) -- "terminal": Command execution terminal (900x600 default, good for bottom or left) -- "calibre_books": E-book library browser (1200x800 default, needs wide space) -- "notes": Note-taking interface (700x600 default, flexible) -- "pdf_reader": PDF document viewer (1000x700 default, needs large space) -- "gns3_projects": GNS3 project selector (600x500 default, medium size) -- "video_recorder": Screen recording tool (800x600 default, medium size) -- "settings": Application configuration (800x600 default, medium size) - -Guidelines for intelligent window placement: -- For productivity tools (notes, pdf_reader, calibre_books): Place on right side (x: 800-1200, y: 100-200) -- For system tools (terminal, settings): Place on left side (x: 100-300, y: 100-200) -- For visualization (network_topology): Place in center or large area (x: 200-400, y: 100-200) -- Avoid overlapping windows by checking their purpose -- Consider typical screen resolution (1920x1080) when positioning - -When opening windows, ALWAYS include both size and position parameters for better UX. -""" - args_schema: Type[BaseModel] = WindowControllerInput - - # Session ID is injected when creating the tool instance - session_id: str = "default" - - def _run( - self, - action: str, - window_type: Optional[str] = None, - window_title: Optional[str] = None, - x: Optional[float] = None, - y: Optional[float] = None, - width: Optional[float] = None, - height: Optional[float] = None, - run_manager: Optional[CallbackManagerForToolRun] = None, - ) -> str: - """ - Synchronous version - not supported. - - Use the async version (_arun) instead. - """ - raise NotImplementedError( - "WindowControllerTool requires async execution. " - "Use the async version (_arun) or ensure your agent supports async tools." - ) - - async def _arun( - self, - action: str, - window_type: Optional[str] = None, - window_title: Optional[str] = None, - x: Optional[float] = None, - y: Optional[float] = None, - width: Optional[float] = None, - height: Optional[float] = None, - size: Optional[dict] = None, - position: Optional[dict] = None, - run_manager: Optional[AsyncCallbackManagerForToolRun] = None, - ) -> str: - """ - Async implementation: Send command to frontend and wait for result. - - Args: - action: Window action to perform - window_type: Type of window (for 'open' action) - window_title: Title of target window (for other actions) - x, y: Position coordinates (for 'move' action) - width, height: Size dimensions (for 'resize' action or 'open' action) - size: Complete size dict for 'open' action - position: Complete position dict for 'open' action - run_manager: LangChain callback manager - - Returns: - JSON string with execution result from frontend - """ - import asyncio - - session_manager = get_session_manager() - session = session_manager.get_session(self.session_id) - - if not session: - logger.error(f"Session not found: {self.session_id}") - return json.dumps({ - "status": "error", - "error": f"Session not found: {self.session_id}" - }) - - # Validate action - if action not in VALID_ACTIONS: - return json.dumps({ - "status": "error", - "error": f"Invalid action '{action}'. Valid actions: {list(VALID_ACTIONS.keys())}" - }) - - # Build input data - input_data = { - "action": action, - } - - # Add optional parameters based on action - if action == "open": - if not window_type: - return json.dumps({ - "status": "error", - "error": "Missing required parameter: window_type" - }) - if window_type not in WINDOW_CONFIGS: - return json.dumps({ - "status": "error", - "error": f"Invalid window_type '{window_type}'. Available: {list(WINDOW_CONFIGS.keys())}" - }) - input_data["window_type"] = window_type - - # Get base configuration - base_config = WINDOW_CONFIGS[window_type] - window_config = base_config.copy() - - # Merge LLM-provided size (supports both formats: dict or width/height) - if size: - window_config["size"] = size - elif width is not None and height is not None: - window_config["size"] = {"width": width, "height": height} - else: - # Use default size based on window type - window_config["size"] = _get_default_size(window_type) - - # Merge LLM-provided position (supports both formats: dict or x/y) - if position: - window_config["position"] = position - elif x is not None and y is not None: - window_config["position"] = {"x": x, "y": y} - else: - # Use default position based on window type - window_config["position"] = _get_default_position(window_type) - - # Add min_size from base config - if "min_size" in base_config: - window_config["min_size"] = base_config["min_size"] - - # Add window_config to input_data - input_data["window_config"] = window_config - - logger.info(f"Opening window: type={window_type}, config={window_config}") - - elif action in ["close", "focus", "minimize", "maximize", "restore", "center"]: - if not window_title: - return json.dumps({ - "status": "error", - "error": f"Missing required parameter: window_title (for action '{action}')" - }) - - # Check if window exists in backend state before sending command - from backend.core.window_state_manager import get_window_state_manager - state_manager = get_window_state_manager() - - logger.info( - f"[WindowController] Looking for window '{window_title}' " - f"in session {self.session_id}..." - ) - - window = state_manager.get_window_by_title(self.session_id, window_title) - - if window: - logger.info( - f"[WindowController] ✅ Found window '{window_title}' " - f"(id={window.id}, type={window.window_type})" - ) - else: - # Log all available windows for debugging - state = state_manager.get_state(self.session_id) - if state: - available_titles = [w.title for w in state.windows] - logger.warning( - f"[WindowController] ❌ Window '{window_title}' not found. " - f"Available windows: {available_titles}" - ) - else: - logger.warning( - f"[WindowController] ❌ Window '{window_title}' not found. " - f"No state exists for session {self.session_id}" - ) - - if not window: - # Window not found - return success (idempotent operation) - logger.info( - f"Window '{window_title}' not found in state, " - f"treating as already {'closed' if action == 'close' else 'not existent'}" - ) - return json.dumps({ - "status": "success", - "action": action, - "window_title": window_title, - "already_closed": True if action == "close" else False, - "not_found": True - }) - - input_data["window_title"] = window_title - - elif action == "move": - if not window_title: - return json.dumps({ - "status": "error", - "error": "Missing required parameter: window_title" - }) - if x is None or y is None: - return json.dumps({ - "status": "error", - "error": "Missing required parameters: x and y (for move action)" - }) - input_data["window_title"] = window_title - input_data["x"] = x - input_data["y"] = y - - elif action == "resize": - if not window_title: - return json.dumps({ - "status": "error", - "error": "Missing required parameter: window_title" - }) - # Support both formats: separate width/height or size dict - if size: - input_data["window_title"] = window_title - input_data["width"] = size.get("width") - input_data["height"] = size.get("height") - elif width is not None and height is not None: - input_data["window_title"] = window_title - input_data["width"] = width - input_data["height"] = height - else: - return json.dumps({ - "status": "error", - "error": "Missing required parameters: width and height (or size dict) for resize action" - }) - - # Generate task_id - task_id = str(uuid.uuid4()) - - # Create Future for waiting on frontend result - future = session.create_future(task_id) - - # Send tool_call to frontend - try: - await session.websocket.send_json({ - "type": "tool_call", - "task_id": task_id, - "tool": "window_controller", - "args": input_data, # Include all parameters including window_config - "session_id": self.session_id - }) - - logger.info(f"WindowController: Sent tool_call to frontend, task_id={task_id}, action={action}") - - except Exception as e: - session.pending_futures.pop(task_id, None) - logger.error(f"WindowController: Failed to send tool_call: {e}") - return json.dumps({ - "status": "error", - "error": f"Failed to send command to frontend: {str(e)}" - }) - - # Wait for frontend to execute and return result (timeout 30 seconds) - try: - result = await asyncio.wait_for(future, timeout=30.0) - - logger.info(f"WindowController: Received result from frontend, task_id={task_id}") - - # Return result to Agent (should already be a dict or JSON string) - if isinstance(result, dict): - return json.dumps(result, ensure_ascii=False) - return result - - except asyncio.TimeoutError: - # Timeout: remove pending future - session.pending_futures.pop(task_id, None) - - logger.warning(f"WindowController: Timeout waiting for frontend, task_id={task_id}, action={action}") - - return json.dumps({ - "status": "error", - "error": "Timeout waiting for frontend to execute window operation", - "action": action, - "timeout_seconds": 30 - }) - - except Exception as e: - # Error: remove pending future - session.pending_futures.pop(task_id, None) - - logger.error(f"WindowController: Error waiting for result: {e}") - - return json.dumps({ - "status": "error", - "error": f"Error waiting for frontend response: {str(e)}" - }) - - -def _get_default_size(window_type: str) -> dict: - """ - Get default size for a window type based on its purpose. - """ - default_sizes = { - "ai_chat": {"width": 600, "height": 800}, - "network_topology": {"width": 1000, "height": 700}, - "terminal": {"width": 900, "height": 600}, - "calibre_books": {"width": 1200, "height": 800}, - "notes": {"width": 700, "height": 600}, - "pdf_reader": {"width": 1000, "height": 700}, - "gns3_projects": {"width": 600, "height": 500}, - "video_recorder": {"width": 800, "height": 600}, - "settings": {"width": 800, "height": 600}, - } - return default_sizes.get(window_type, {"width": 800, "height": 600}) - - -def _get_default_position(window_type: str) -> dict: - """ - Get default position for a window type based on its category. - Strategy: Place windows in different areas of screen to minimize overlap. - """ - # Productivity tools: right side - if window_type in ["notes", "pdf_reader", "calibre_books"]: - return {"x": 900, "y": 150} - - # System tools: left side - if window_type in ["terminal", "settings", "gns3_projects"]: - return {"x": 250, "y": 150} - - # Visualization: center - if window_type == "network_topology": - return {"x": 200, "y": 100} - - # Chat: right edge - if window_type == "ai_chat": - return {"x": 1100, "y": 100} - - # Tools: middle-right - if window_type == "video_recorder": - return {"x": 700, "y": 200} - - # Default fallback - return {"x": 200, "y": 150} - - -def create_window_controller_tool(session_id: str) -> WindowControllerTool: - """ - Factory function to create WindowControllerTool with specific session_id. - - Args: - session_id: Session identifier for WebSocket connection - - Returns: - WindowControllerTool instance configured for this session - """ - return WindowControllerTool(session_id=session_id) - - -# ============================================================================ -# TextInputTool - Smart text input to frontend windows -# ============================================================================ - - -class TextInputInput(BaseModel): - """Input schema for TextInputTool.""" - text: str = Field(..., description="The text content to input") - window_title: Optional[str] = Field( - None, - description="Target window title (strongly recommended for reliable input). Examples: 'AI Chat', 'Notes', 'Terminal'" - ) - selector: Optional[str] = Field( - None, - description="CSS selector for specific input element (advanced use only)" - ) - - -class TextInputTool(BaseTool): - """ - A tool for smart text input to frontend windows. - - This tool sends text input commands to the frontend via WebSocket - and waits for the frontend to execute and return results. - - The actual text input operations are executed in the browser. - """ - - name: str = "text_input" - description: str = """ - Inputs text into frontend windows in FlowNet-Lab. - - IMPORTANT: You MUST specify window_title parameter for reliable text input. - The focused window does NOT necessarily have a focused input field. - - How to use: - 1. Use window_controller to focus/open the target window first - 2. Use text_input with window_title parameter to send text - - Examples: - - {"text": "Hello, how are you?", "window_title": "AI Chat"} - - {"text": "Meeting notes", "window_title": "Notes"} - - {"text": "user@example.com", "selector": "#email"} # Advanced: CSS selector - - Returns a dictionary with operation status and target information. - """ - args_schema: Type[BaseModel] = TextInputInput - - # Session ID is injected when creating the tool instance - session_id: str = "default" - - def _run( - self, - text: str, - window_title: Optional[str] = None, - selector: Optional[str] = None, - run_manager: CallbackManagerForToolRun | None = None, - ) -> str: - """ - Synchronous version - not supported. - - Use the async version (_arun) instead. - """ - raise NotImplementedError( - "TextInputTool requires async execution. " - "Use the async version (_arun) or ensure your agent supports async tools." - ) - - async def _arun( - self, - text: str, - window_title: Optional[str] = None, - selector: Optional[str] = None, - tool_input: Optional[str] = None, # Legacy format support - run_manager: Optional[AsyncCallbackManagerForToolRun] = None, - ) -> str: - """ - Async implementation: Send command to frontend and wait for result. - - Args: - text: The text content to input (from args_schema) - window_title: Target window title (from args_schema) - selector: CSS selector for specific element (from args_schema) - tool_input: Legacy JSON string format (backward compatibility) - run_manager: LangChain callback manager - - Returns: - JSON string with execution result from frontend - """ - import asyncio - - session_manager = get_session_manager() - session = session_manager.get_session(self.session_id) - - if not session: - logger.error(f"Session not found: {self.session_id}") - return json.dumps({ - "status": "error", - "error": f"Session not found: {self.session_id}" - }) - - try: - # Handle legacy tool_input format - if tool_input: - input_data = json.loads(tool_input) - text = input_data.get("text", text) - window_title = input_data.get("window_title", window_title) - selector = input_data.get("selector", selector) - - # Validate text - if not text: - return json.dumps({"error": "Missing text field."}) - - # Build command data - command_data = { - "text": text, - } - if window_title: - command_data["window_title"] = window_title - if selector: - command_data["selector"] = selector - - # Generate task_id - task_id = str(uuid.uuid4()) - - # Create Future for waiting on frontend result - future = session.create_future(task_id) - - # Send tool_call to frontend - await session.websocket.send_json({ - "type": "tool_call", - "task_id": task_id, - "tool": "text_input", - "args": command_data, - "session_id": self.session_id - }) - - logger.info(f"TextInput: Sent tool_call to frontend, task_id={task_id}") - - # Wait for frontend to execute and return result (timeout 30 seconds) - result = await asyncio.wait_for(future, timeout=30.0) - - logger.info(f"TextInput: Received result from frontend, task_id={task_id}") - - # Return result to Agent - if isinstance(result, dict): - return json.dumps(result, ensure_ascii=False) - return result - - except asyncio.TimeoutError: - session.pending_futures.pop(task_id, None) - logger.warning(f"TextInput: Timeout waiting for frontend, task_id={task_id}") - return json.dumps({ - "status": "error", - "error": "Timeout waiting for frontend to execute text input", - "timeout_seconds": 30 - }) - - except json.JSONDecodeError as e: - logger.error(f"Invalid JSON input: {e}") - return json.dumps({"error": f"Invalid JSON input: {e}"}) - - except Exception as e: - session.pending_futures.pop(task_id, None) - logger.error(f"TextInput: Error waiting for result: {e}") - return json.dumps({"error": f"Error waiting for frontend response: {str(e)}"}) - - -def create_text_input_tool(session_id: str) -> TextInputTool: - """ - Factory function to create TextInputTool with specific session_id. - - Args: - session_id: Session identifier for WebSocket connection - - Returns: - TextInputTool instance configured for this session - """ - return TextInputTool(session_id=session_id)