feat(agent): add gns3-copilot AI assistant integration module

Integrate the gns3-copilot AI assistant module to provide intelligent
   automation and interaction capabilities for GNS3 network emulation.

   Key components:
   - AI agent framework with LLM integration (supports Qwen vision model)
   - GNS3 client library for project topology management
   - Extensive prompt templates for various network operation scenarios
   - Tool library for node creation, linking, configuration, and management
   - Support for English level assessment (A1-C2) and specialized personas
   - Network drawing and topology visualization tools
   - Linux device automation via Nornir/Telnetlib
   - Window controller for UI interaction

   Features:
   - Multi-modal AI agent with vision capabilities
   - Automated network topology deployment and configuration
   - Interactive node and drawing management
   - File-based project operations (read, write, list)
   - Specialized prompts for different scenarios and skill levels
   - Comprehensive tool set for network device management
This commit is contained in:
YueGuobin committed 2026-03-03 23:08:07 +08:00
1 parent 7a2d15cb64
commit 46b262a02c
62 files changed
+18114

No files matched your search

@@ -0,0 +1,78 @@
"""
FlowNet-Lab Tools Package
This package provides various tools for interacting with GNS3 network simulator, including:
- Device configuration command execution
- Display command execution
- Multiple device command execution using Nornir
- VPCS device configuration using telnetlib3
- Node and link management
- Drawing management
- Notes management
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)
- gns3_create_node: GNS3 node creation tool
- gns3_create_link: GNS3 link creation tool
- gns3_start_node: GNS3 node startup tool
- gns3_get_node_temp: GNS3 template retrieval tool
- gns3_get_drawings: GNS3 drawing retrieval tool
- gns3_create_drawing: GNS3 drawing creation tool
- gns3_update_drawing: GNS3 drawing update tool
- gns3_delete_drawing: GNS3 drawing deletion tool
- gns3_create_area_drawing: GNS3 area annotation creation tool (ellipse for 2 nodes)
- gns3_drawing_utils: Drawing utility functions for calculating SVG parameters
- linux_tools_nornir: Linux Telnet batch command execution tool using Nornir
- window_controller: Frontend window control and text input tools (async WebSocket-based)
Note: GNS3TopologyTool is now available from gns3_client package
Author: Guobin Yue
"""
# Import main tool classes
from .config_tools_nornir import ExecuteMultipleDeviceConfigCommands
from .display_tools_nornir import ExecuteMultipleDeviceCommands
from .gns3_create_area_drawing import GNS3CreateAreaDrawingTool
from .gns3_create_link import GNS3LinkTool
from .gns3_create_node import GNS3CreateNodeTool
from .gns3_get_node_temp import GNS3TemplateTool
from .gns3_start_node import GNS3StartNodeTool, GNS3StartNodeQuickTool
from .gns3_update_node_name import GNS3UpdateNodeNameTool
from .linux_tools_nornir import LinuxTelnetBatchTool
from .vpcs_tools_telnetlib3 import VPCSMultiCommands
from .window_controller import WindowControllerTool, TextInputTool
# Dynamic version management
try:
from importlib.metadata import version
__version__ = version("gns3-copilot")
except Exception:
__version__ = "unknown"
__author__ = "Guobin Yue"
__description__ = "AI-powered network automation assistant for GNS3"
__url__ = "https://github.com/yueguobin/gns3-copilot"
# Export main tool classes
__all__ = [
"ExecuteMultipleDeviceConfigCommands",
"ExecuteMultipleDeviceCommands",
"VPCSMultiCommands",
"GNS3CreateNodeTool",
"GNS3LinkTool",
"GNS3StartNodeTool",
"GNS3StartNodeQuickTool",
"GNS3UpdateNodeNameTool",
"GNS3TemplateTool",
"GNS3CreateAreaDrawingTool",
"LinuxTelnetBatchTool",
"WindowControllerTool",
"TextInputTool",
]
# Package initialization message
# print(f"FlowNet-Lab Tools package loaded (version {__version__})")
@@ -0,0 +1,518 @@
"""
This module provides a tool to execute configuration commands on multiple devices
in a GNS3 topology using Nornir.
"""
import json
import logging
from typing import Any
from langchain.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun
from netmiko.exceptions import ReadTimeout
from nornir import InitNornir
from nornir.core import Nornir
from nornir.core.task import AggregatedResult, Result, Task
from nornir_netmiko.tasks import netmiko_send_config
from gns3_copilot.utils import (
get_config,
get_device_ports_from_topology,
)
# config log
logger = logging.getLogger(__name__)
# Local Nornir configuration functions for Cisco IOS Telnet devices
def _get_nornir_defaults() -> dict[str, Any]:
"""Get Nornir default configuration for Cisco IOS."""
return {"data": {"location": "gns3"}}
def _get_nornir_groups_config() -> dict[str, dict[str, Any]]:
"""Get Nornir groups configuration for Cisco IOS Telnet devices."""
return {
"cisco_IOSv_telnet": {
"platform": "cisco_ios",
"hostname": get_config("GNS3_SERVER_HOST"),
"timeout": 120,
"username": get_config("GNS3_SERVER_USERNAME"),
"password": get_config("GNS3_SERVER_PASSWORD"),
"connection_options": {
"netmiko": {"extras": {"device_type": "cisco_ios_telnet"}}
},
},
}
def _get_nornir_group(group_name: str = "cisco_IOSv_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 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.
IMPORTANT SAFETY NOTE:
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').
For viewing information, use the 'execute_multiple_device_commands' tool.
Input should be a JSON object containing project_id and device configurations.
Example input:
{
"project_id": "f32ebf3d-ef8c-4910-b0d6-566ed828cd24",
"device_configs": [
{
"device_name": "R-1",
"config_commands": [
"interface Loopback0",
"ip address 1.1.1.1 255.255.255.255",
"description CONFIG_BY_TOOL"
]
},
{
"device_name": "R-2",
"config_commands": [
"interface Loopback0",
"ip address 2.2.2.2 255.255.255.255",
"description CONFIG_BY_TOOL"
]
}
]
}
Returns a list of dictionaries, each containing the device name and configuration results.
IMPORTANT SAFETY WARNING:
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
- format / erase nvram / delete flash:
- boot system commands
- factory-reset commands
- any commands that require user confirmation prompts
"""
def _run(
self,
tool_input: str, # or Union[str, List[Any], Dict[str, Any]]
run_manager: CallbackManagerForToolRun | None = None,
**kwargs: Any,
) -> list[dict[str, Any]]:
"""
Executes configuration commands on multiple devices in the current 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
configuration results.
"""
# Log received input
logger.info("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]
):
return device_configs_list
# Create a mapping of device names to their configuration 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 all devices concurrently in a single run
try:
task_result = dynamic_nr.run(
task=self._run_all_device_configs_with_single_retry,
device_configs_map=device_configs_map,
)
# Process results for all devices
results = self._process_task_results(
device_configs_list, hosts_data, task_result
)
except Exception as e:
# Overall execution failed
logger.error("Error executing configurations on all devices: %s", e)
return [{"error": f"Execution error: {str(e)}"}]
logger.info(
"Multiple device configuration execution completed. Results: %s",
json.dumps(results, indent=2, ensure_ascii=False),
)
return results
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")
try:
_result = task.run(
task=netmiko_send_config, config_commands=config_commands
)
return Result(host=task.host, result=_result.result)
except ReadTimeout as e:
# Log ReadTimeout exception with full details
logger.error(
"ReadTimeout occurred for device %s: %s",
device_name,
str(e),
)
return Result(
host=task.host,
result=f"Configuration 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.
if "netmiko_send_config (failed)" in str(e):
_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
logger.error(
"Configuration failed for device %s: %s (Exception type: %s)",
device_name,
str(e),
type(e).__name__,
)
return Result(
host=task.host,
result=f"Configuration failed (Unhandled Exception): {str(e)}",
failed=True,
)
def _validate_tool_input(
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.
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
"""
import re
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 configuration commands."""
device_configs_map = {}
for device_config in device_config_list:
device_name = device_config["device_name"]
config_commands = device_config["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)
# 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("cisco_IOSv_telnet")
defaults = _get_nornir_defaults()
# Log nornir account information
gns3_host = get_config("GNS3_SERVER_HOST")
logger.info(
"Initializing Nornir with account: host=%s, platform=%s, timeout=%d",
gns3_host,
groups_data.get("platform"),
groups_data.get("timeout"),
)
return InitNornir(
inventory={
"plugin": "DictInventory",
"options": {
"hosts": hosts_data,
"groups": {"cisco_IOSv_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,
) -> list[dict[str, Any]]:
"""Process the 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["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 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"Configuration 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
results.append(device_result)
return results
if __name__ == "__main__":
# Example usage with new format
# example tool_input with project_id
input_paras = json.dumps(
{
"project_id": "f32ebf3d-ef8c-4910-b0d6-566ed828cd24",
"device_configs": [
{
"device_name": "R-1",
"config_commands": [
"interface Loopback1110",
"ip address 201.201.201.201 255.255.255.255",
"description CONFIG_BY_TOOL",
],
},
{
"device_name": "R-2",
"config_commands": [
"interface Loopback1110",
"ip address 202.202.202.202 255.255.255.255",
"description CONFIG_BY_TOOL",
],
},
{
"device_name": "SW-2",
"config_commands": [
"interface Loopback1110",
"ip address 202.202.202.202 255.255.255.255",
"description CONFIG_BY_TOOL",
],
},
{
"device_name": "SW-1",
"config_commands": [
"interface Loopback1110",
"ip address 202.202.202.202 255.255.255.255",
"description CONFIG_BY_TOOL",
],
},
],
}
)
exe_config = ExecuteMultipleDeviceConfigCommands()
failed_count = 0
for _i in range(0, 5):
exe_results = exe_config._run(tool_input=input_paras)
for result in exe_results:
for result in exe_results:
if result.get("status") == "failed":
failed_count += 1
print(f"Failed Count: {failed_count}")
# print("Execution results:")
# print(json.dumps(result, indent=2, ensure_ascii=False))
@@ -0,0 +1,515 @@
"""
This module provides a tool to execute display commands on multiple devices
in a GNS3 topology using Nornir.
"""
import json
import logging
import re
from typing import Any
from langchain.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun
from netmiko.exceptions import ReadTimeout
from nornir import InitNornir
from nornir.core import Nornir
from nornir.core.task import AggregatedResult, Result, Task
from nornir_netmiko.tasks import netmiko_multiline
from gns3_copilot.utils import (
get_config,
get_device_ports_from_topology,
)
# config log
logger = logging.getLogger(__name__)
# Local Nornir configuration functions for Cisco IOS Telnet devices
def _get_nornir_defaults() -> dict[str, Any]:
"""Get Nornir default configuration for Cisco IOS."""
return {"data": {"location": "gns3"}}
def _get_nornir_groups_config() -> dict[str, dict[str, Any]]:
"""Get Nornir groups configuration for Cisco IOS Telnet devices."""
return {
"cisco_IOSv_telnet": {
"platform": "cisco_ios",
"hostname": get_config("GNS3_SERVER_HOST"),
"timeout": 120,
"username": get_config("GNS3_SERVER_USERNAME"),
"password": get_config("GNS3_SERVER_PASSWORD"),
"connection_options": {
"netmiko": {"extras": {"device_type": "cisco_ios_telnet"}}
},
},
}
def _get_nornir_group(group_name: str = "cisco_IOSv_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 ExecuteMultipleDeviceCommands(BaseTool):
"""
A READ-ONLY diagnostic tool for viewing network device configurations and protocol states.
**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.
**Allowed Command Types:**
- Display commands: show, display (Cisco/Huawei)
- Debug commands: debug ip routing, debug ospf events
- Verification commands: ping, traceroute, telnet
- Status commands: show version, show running-config, show ip route
**Strictly Prohibited:**
- Configuration mode: configure terminal, config t
- Interface configuration: interface, ip address
- Protocol configuration: router ospf, router bgp
- Any command that modifies device state
"""
name: str = "execute_multiple_device_commands"
description: str = """
**READ-ONLY DIAGNOSTIC TOOL** - View network device configurations and protocol states.
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
- 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 commands that modify device state
- If you need to configure devices, provide guidance to the student instead
**Input Format:**
{
"project_id": "f32ebf3d-ef8c-4910-b0d6-566ed828cd24",
"device_configs": [
{
"device_name": "R-1",
"commands": ["show version", "show ip interface brief"]
},
{
"device_name": "R-2",
"commands": ["show version", "show ip ospf neighbor"]
}
]
}
**Returns:** List of device outputs for diagnostic analysis.
"""
def _run(
self,
tool_input: str | bytes | list[Any] | dict[str, Any],
run_manager: CallbackManagerForToolRun | None = None,
**kwargs: Any,
) -> list[dict[str, Any]]:
"""
Executes READ-ONLY diagnostic commands on multiple devices in current GNS3 topology.
This method only permits display/show commands and does not allow any configuration
changes to network devices.
Args:
tool_input (str): A JSON string containing project_id and diagnostic commands to execute.
Returns:
List[Dict[str, Any]]: A list of dictionaries containing device names and command outputs.
"""
# Log received input
logger.info("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]
):
return device_configs_list
# 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 all devices concurrently in a single run
try:
task_result = dynamic_nr.run(
task=self._run_all_device_configs_with_single_retry,
device_configs_map=device_configs_map,
)
# Process results for all devices
results = self._process_task_results(
device_configs_list, hosts_data, task_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 _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."""
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")
try:
_result = task.run(
task=netmiko_multiline,
commands=diagnostic_commands,
enable=True,
read_timeout=60,
)
return Result(host=task.host, result=_result.result)
except ReadTimeout as e:
# Log ReadTimeout exception with full details
logger.error(
"ReadTimeout occurred for device %s: %s",
device_name,
str(e),
)
return Result(
host=task.host,
result=f"diagnostic command execution 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.
if "netmiko_multiline (failed)" in str(e):
_result = task.run(
task=netmiko_multiline,
commands=diagnostic_commands,
enable=True,
read_timeout=60,
)
return Result(host=task.host, result=_result.result)
# Log any other exceptions with full details
logger.error(
"diagnostic command execution failed for device %s: %s (Exception type: %s)",
device_name,
str(e),
type(e).__name__,
)
return Result(
host=task.host,
result=f"diagnostic command execution failed (Unhandled Exception): {str(e)}",
failed=True,
)
def _validate_tool_input(
self, tool_input: str | bytes | list[Any] | dict[str, Any]
) -> tuple[list[dict[str, Any]], str | None]:
"""
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.
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 diagnostic commands."""
device_diagnostic_map = {}
for device_config in device_config_list:
device_name = device_config["device_name"]
diagnostic_commands = device_config["commands"]
device_diagnostic_map[device_name] = diagnostic_commands
return device_diagnostic_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)
# 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("cisco_IOSv_telnet")
defaults = _get_nornir_defaults()
# Log nornir account information
gns3_host = get_config("GNS3_SERVER_HOST")
logger.info(
"Initializing Nornir with account: host=%s, platform=%s, timeout=%d",
gns3_host,
groups_data.get("platform"),
groups_data.get("timeout"),
)
return InitNornir(
inventory={
"plugin": "DictInventory",
"options": {
"hosts": hosts_data,
"groups": {"cisco_IOSv_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,
) -> list[dict[str, Any]]:
"""Process the task results and format them for return."""
results = []
for device_config in device_configs_list:
device_name = device_config["device_name"]
diagnostic_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 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"Diagnostic 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["diagnostic_commands"] = diagnostic_commands
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": "R-2",
"commands": ["show version", "show ip interface brief"],
},
{
"device_name": "R-1",
"commands": ["show version", "show ip interface brief"],
},
{
"device_name": "SW-1",
"commands": ["show version", "show ip interface brief"],
},
{
"device_name": "SW-2",
"commands": ["show version", "show ip interface brief"],
},
],
}
)
exe_cmd = ExecuteMultipleDeviceCommands()
failed_count = 0
for _i in range(0, 1):
exe_results = exe_cmd._run(tool_input=device_commands)
for result in exe_results:
for result in exe_results:
if result.get("status") == "failed":
failed_count += 1
print(f"Failed Count: {failed_count}")
# print("Execution results:")
# print(json.dumps(result, indent=2, ensure_ascii=False))
@@ -0,0 +1,403 @@
"""
GNS3 area annotation drawing tool for creating visual area markers.
This tool creates visual annotations (ellipses) for network devices to represent
groupings such as OSPF areas, EIGRP AS numbers, or other protocol-defined regions.
Currently supports two-node ellipse annotations.
"""
import json
import logging
from typing import Any
from langchain.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun
from gns3_copilot.gns3_client import (
GNS3GetNodesTool,
Project,
get_gns3_connector,
)
from gns3_copilot.utils.gns3_drawing_utils import (
calculate_two_node_ellipse,
calculate_two_node_rectangle,
)
# Configure logging
logger = logging.getLogger(__name__)
class GNS3CreateAreaDrawingTool(BaseTool):
"""
A LangChain tool to create visual area annotations for network devices.
Creates ellipse annotations that connect two network devices,
automatically calculating optimal position, size, and rotation based on node coordinates.
**Input:**
A JSON object containing the project_id, area_name, and node_names.
Example input:
{
"project_id": "uuid-of-project",
"area_name": "Area 0",
"node_names": ["R-1", "R-2"]
}
**Output:**
A dictionary containing the creation results:
{
"project_id": "uuid-of-project",
"area_name": "Area 0",
"node_count": 2,
"shape_type": "ellipse",
"created_drawings": [
{
"drawing_id": "uuid-of-drawing1",
"type": "ellipse",
"status": "success"
},
{
"drawing_id": "uuid-of-drawing2",
"type": "text",
"status": "success"
}
],
"total_drawings": 2,
"successful_drawings": 2,
"failed_drawings": 0
}
**Use Cases:**
- Protocol Domains: OSPF areas, BGP AS, IS-IS levels
- Logical Isolation: VRF, VLAN, MSTP
- High Availability: VRRP, HSRP, Stack, M-LAG
- External Boundaries: Internet, DMZ
- Management Networks: OOB, Management
**Semantic Color Coding:**
The tool automatically applies business-professional colors based on area_name keywords:
| Category | Colors | Keywords in area_name | Border Style |
|------------------------|---------------------|------------------------------|--------------|
| Primary Routing | Blue (#2196F3) | BGP, AS, Area 0, Backbone | Solid |
| Secondary Routing | Light Blue (#64B5F6)| Area, Level | Solid |
| Logical Isolation | Purple (#9C27B0) | VRF, VLAN, MSTP | Dashed |
| High Availability | Amber (#FFC107) | VRRP, HSRP, HA, Stack | Solid |
| External/Boundary | Red (#EF5350) | INET, OUT, External, DMZ | Solid |
| Management | Gray (#757575) | MGMT, OOB | Dashed |
**Implementation Details:**
- Retrieves node coordinates from the GNS3 project topology
- Calculates rotated ellipse parameters (center, radius, rotation angle)
- Generates professional SVG graphics with semantic colors
- Creates two drawings: ellipse shape and text label
- All coordinates are integers as required by GNS3 API
**Example Usage:**
User: "Configure OSPF area 0 on R-1 and R-2"
→ Call: create_gns3_area_drawing(project_id="xxx", area_name="Area 0", node_names=["R-1", "R-2"])
User: "Create VLAN 10 for SW-1 and SW-2"
→ Call: create_gns3_area_drawing(project_id="xxx", area_name="VLAN 10", node_names=["SW-1", "SW-2"])
User: "Configure VRRP group 1 on R-1 and R-2"
→ Call: create_gns3_area_drawing(project_id="xxx", area_name="VRRP Group 1", node_names=["R-1", "R-2"])
"""
name: str = "create_gns3_area_drawing"
description: str = """
Creates a visual annotation (ellipse or rectangle) to mark logical groupings between two network devices.
Use for protocol domains (OSPF areas, BGP AS, IS-IS levels), logical isolation (VRF, VLAN),
or high availability groups (VRRP, HSRP). Automatically calculates optimal position,
size, rotation, and applies semantic colors.
Parameters:
- project_id: GNS3 project UUID
- area_name: Logical group name (e.g., "Area 0", "VLAN 10", "VRRP Group 1")
- node_names: List of exactly 2 node names (e.g., ["R-1", "R-2"])
- shape_type: Shape type, either "ellipse" (default) or "rectangle"
"""
def _run(
self,
tool_input: str,
run_manager: CallbackManagerForToolRun | None = None,
**kwargs: Any,
) -> dict[str, Any]:
"""
Creates a visual area annotation for network devices.
Args:
tool_input: A JSON string containing project_id, area_name, and node_names.
run_manager: LangChain run manager (unused).
Returns:
dict: A dictionary with creation results or an error message.
"""
# Log received input
logger.info("Received input: %s", tool_input)
try:
# Parse input JSON
input_data = json.loads(tool_input)
project_id = input_data.get("project_id")
area_name = input_data.get("area_name")
node_names = input_data.get("node_names", [])
shape_type = input_data.get("shape_type", "ellipse")
# Validate input
if not project_id:
logger.error("Invalid input: Missing project_id.")
return {"error": "Missing project_id."}
if not area_name:
logger.error("Invalid input: Missing area_name.")
return {"error": "Missing area_name."}
if not isinstance(node_names, list) or len(node_names) == 0:
logger.error("Invalid input: node_names must be a non-empty array.")
return {"error": "node_names must be a non-empty array."}
# Validate: requires exactly 2 nodes
if len(node_names) != 2:
logger.error(
"Invalid input: Exactly 2 nodes are required, got %d.",
len(node_names),
)
return {
"error": f"Exactly 2 nodes are required, got {len(node_names)}. "
"Please provide exactly 2 node names."
}
# Initialize Gns3Connector using factory function
logger.info("Connecting to GNS3 server...")
gns3_server = get_gns3_connector()
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
return {
"error": "Failed to connect to GNS3 server. Please check your configuration."
}
# Initialize Project object for drawing creation
logger.info("Initializing project for drawing creation...")
project = Project(project_id=project_id, connector=gns3_server)
project.get() # Load project details
# Get node information using GNS3GetNodesTool to retrieve complete node data
# including height, width, coordinates, etc.
logger.info(
"Retrieving node information for project %s...",
project_id,
)
get_nodes_tool = GNS3GetNodesTool()
nodes_result = get_nodes_tool._run(json.dumps({"project_id": project_id}))
# Check if node retrieval was successful
if "error" in nodes_result:
logger.error("Failed to retrieve nodes: %s", nodes_result["error"])
return {"error": f"Failed to retrieve nodes: {nodes_result['error']}"}
# Build a dictionary mapping node names to their complete information
nodes_dict = {node["name"]: node for node in nodes_result.get("nodes", [])}
# Validate all requested nodes exist
for node_name in node_names:
if node_name not in nodes_dict:
logger.error("Node %s not found in project", node_name)
return {
"error": f"Node '{node_name}' not found in project topology."
}
# Retrieve all requested nodes
nodes = [nodes_dict[node_name] for node_name in node_names]
# Log node information
node_info_list = [
f"{node_name} at ({node['x']}, {node['y']}, size: {node.get('width', 'N/A')}x{node.get('height', 'N/A')})"
for node_name, node in zip(node_names, nodes, strict=True)
]
logger.info("Found nodes: %s", ", ".join(node_info_list))
# Validate shape_type
if shape_type not in ["ellipse", "rectangle"]:
logger.error("Invalid shape_type: %s", shape_type)
return {
"error": f"Invalid shape_type '{shape_type}'. Must be 'ellipse' or 'rectangle'."
}
# Calculate shape parameters for 2 nodes
logger.info("Calculating %s parameters for 2 nodes...", shape_type)
# Use the appropriate calculation function based on shape_type
if shape_type == "ellipse":
shape_result = calculate_two_node_ellipse(nodes[0], nodes[1], area_name)
shape_key = "ellipse"
else: # rectangle
shape_result = calculate_two_node_rectangle(
nodes[0], nodes[1], area_name
)
shape_key = "rectangle"
metadata = shape_result["metadata"]
logger.info(
"Two-node %s: center=(%.2f, %.2f), distance=%.2f, angle=%.2f°",
shape_type,
metadata["center_x"],
metadata["center_y"],
metadata["distance"],
metadata["angle_deg"],
)
# Prepare drawings for creation
# Ensure all coordinates are integers as required by GNS3 API
drawings = [
{
"svg": shape_result[shape_key]["svg"],
"x": int(shape_result[shape_key]["x"]),
"y": int(shape_result[shape_key]["y"]),
"z": shape_result[shape_key]["z"],
"locked": False,
"rotation": int(shape_result[shape_key]["rotation"]),
},
{
"svg": shape_result["text"]["svg"],
"x": int(shape_result["text"]["x"]),
"y": int(shape_result["text"]["y"]),
"z": shape_result["text"]["z"],
"locked": False,
"rotation": int(shape_result["text"]["rotation"]),
},
]
# Create drawings using Project method
logger.info(
"Creating %d drawings in project %s...", len(drawings), project_id
)
results: list[dict[str, Any]] = []
for i, drawing_data in enumerate(drawings):
try:
drawing_type = shape_type if i == 0 else "text"
logger.info(
"Creating drawing %d/%d: %s at (%d, %d) with rotation %d°...",
i + 1,
len(drawings),
drawing_type,
drawing_data["x"],
drawing_data["y"],
drawing_data["rotation"],
)
result = project.create_drawing(
svg=drawing_data["svg"],
x=drawing_data["x"],
y=drawing_data["y"],
z=drawing_data["z"],
locked=drawing_data["locked"],
rotation=drawing_data["rotation"],
)
drawing_info = {
"drawing_id": result.get("drawing_id"),
"type": drawing_type,
"status": "success",
}
results.append(drawing_info)
except Exception as e:
error_info = {
"type": "ellipse" if i == 0 else "text",
"error": f"Drawing {i + 1} creation failed: {str(e)}",
"status": "failed",
}
results.append(error_info)
logger.error("Failed to create drawing %d: %s", i + 1, e)
# Calculate summary statistics
successful_drawings = len(
[r for r in results if r.get("status") == "success"]
)
failed_drawings = len([r for r in results if r.get("status") == "failed"])
# Prepare final result
final_result = {
"project_id": project_id,
"area_name": area_name,
"node_count": len(node_names),
"nodes": node_names,
"shape_type": shape_type,
"created_drawings": results,
"total_drawings": len(drawings),
"successful_drawings": successful_drawings,
"failed_drawings": failed_drawings,
}
# Log the final result
logger.info(
"Area annotation creation completed: %d successful, %d failed out of %d total drawings.",
successful_drawings,
failed_drawings,
len(drawings),
)
# Return JSON-formatted result
return final_result
except json.JSONDecodeError as e:
logger.error("Invalid JSON input: %s", e)
return {"error": f"Invalid JSON input: {e}"}
except Exception as e:
logger.error("Failed to process area annotation request: %s", e)
return {"error": f"Failed to process area annotation request: {str(e)}"}
if __name__ == "__main__":
# Test the tool locally
from pprint import pprint
# Test with ellipse (default)
test_input_ellipse = json.dumps(
{
"project_id": "d7fc094c-685e-4db1-ac11-5e33a1b2e066", # Replace with actual project UUID
"area_name": "Core Area",
"node_names": ["R-6", "R-4"], # Replace with actual node names
"shape_type": "ellipse",
}
)
# Test with rectangle
test_input_rectangle = json.dumps(
{
"project_id": "d7fc094c-685e-4db1-ac11-5e33a1b2e066", # Replace with actual project UUID
"area_name": "VLAN 10",
"node_names": ["SW-1", "SW-2"], # Replace with actual node names
"shape_type": "rectangle",
}
)
tool = GNS3CreateAreaDrawingTool()
result = tool._run(test_input_ellipse)
pprint(result)
"""
example output:
{
'area_name': 'Area 0',
'created_drawings': [
{'drawing_id': 'uuid-of-drawing1', 'status': 'success', 'type': 'ellipse'},
{'drawing_id': 'uuid-of-drawing2', 'status': 'success', 'type': 'text'}
],
'failed_drawings': 0,
'node_count': 2,
'nodes': ['R-1', 'R-2'],
'project_id': '2245149a-71c8-4387-9d1f-441a683ef7e7',
'shape_type': 'ellipse',
'successful_drawings': 2,
'total_drawings': 2
}
"""
@@ -0,0 +1,274 @@
"""
GNS3 link creation tool for connecting network nodes.
Provides functionality to create links between nodes in GNS3 projects
using the GNS3 API connector.
"""
import json
import logging
from pprint import pprint
from typing import Any
from langchain.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun
from gns3_copilot.gns3_client import Link, get_gns3_connector
# Configure logging
logger = logging.getLogger(__name__)
class GNS3LinkTool(BaseTool):
"""
Tool for creating network links between GNS3 nodes.
Creates one or more links between specified nodes in a GNS3 project
by connecting their network ports. Supports batch link creation
with error handling for individual link failures.
"""
name: str = "create_gns3_link"
description: str = """
Creates one or more links between nodes in a GNS3 project.
Input: A JSON string with:
- `project_id` (str): The UUID of the GNS3 project.
- `links` (list): A non-empty array of link definitions, each containing:
- `node_id1` (str): UUID of the first node.
- `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.
Example Input:
{
"project_id": "uuid-of-project",
"links": [
{
"node_id1": "uuid-of-node1",
"port1": "Ethernet0/0",
"node_id2": "uuid-of-node2",
"port2": "Ethernet0/0"
}
]
}
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')
Example Output:
[
{
"link_id": "uuid-of-link",
"node_id1": "uuid-of-node1",
"port1" : "Ethernet0/0",
"node_id2": "uuid-of-node2",
"port2": "Ethernet0/0"
},
{
"error": "Port not found in link 1"
},
]
"""
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.
run_manager: LangChain run manager (unused).
Returns:
list: A list containing dictionaries with created link details or error messages.
"""
# Log received input
logger.info("Received input: %s", tool_input)
try:
# Parse input JSON
input_data = json.loads(tool_input)
project_id = input_data.get("project_id")
links_data = input_data.get("links", [])
# Validate input
if not project_id:
logger.error("Missing required field: project_id")
return [{"error": "Missing required field: project_id"}]
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"}]
# Initialize Gns3Connector using factory function
logger.info("Connecting to GNS3 server...")
gns3_server = get_gns3_connector()
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
return [
{
"error": "Failed to connect to GNS3 server. Please check your configuration."
}
]
created_links = []
# Process each link definition
for i, link_data in enumerate(links_data):
try:
logger.info("Creating link %d/%d", i + 1, len(links_data))
# Extract link parameters
node_id1 = link_data.get("node_id1")
port1 = link_data.get("port1")
node_id2 = link_data.get("node_id2")
port2 = link_data.get("port2")
# Validate link parameters
if not all([node_id1, port1, node_id2, port2]):
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
)
if not node1 or not node2:
error_msg = f"Node not found in link {i}"
logger.error(error_msg)
created_links.append({"error": error_msg})
continue
# Find port information
port1_info = next(
(
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
),
None,
)
if not port1_info or not port2_info:
error_msg = f"Port not found in link {i}"
logger.error(error_msg)
created_links.append({"error": error_msg})
continue
# Create the link
link = Link(
project_id=project_id,
connector=gns3_server,
nodes=[
{
"node_id": node_id1,
"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),
"label": {"text": port2},
},
],
)
link.create()
link.get()
# Collect link details
link_info = {
"link_id": link.link_id,
"node_id1": node_id1,
"port1": port1,
"node_id2": node_id2,
"port2": port2,
}
created_links.append(link_info)
except Exception as e:
error_msg = f"Failed to create link {i}: {str(e)}"
logger.error(error_msg)
created_links.append({"error": error_msg})
# Log final results
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,
len(links_data) - success_count,
)
return created_links
except json.JSONDecodeError as e:
logger.error("Invalid JSON input: %s", e)
return [{"error": f"Invalid JSON input: {e}"}]
except Exception as e:
logger.error("Failed to process link creation: %s", e)
return [{"error": f"Failed to process link creation: {str(e)}"}]
if __name__ == "__main__":
# Test with single link
single_link_input = json.dumps(
{
"project_id": "your-project-uuid",
"links": [
{
"node_id1": "your-node1-uuid",
"port1": "Ethernet0/0",
"node_id2": "your-node2-uuid",
"port2": "Ethernet0/0",
}
],
}
)
# Test with multiple links
multiple_links_input = json.dumps(
{
"project_id": "your-project-uuid",
"links": [
{
"node_id1": "your-node1-uuid",
"port1": "Ethernet0/0",
"node_id2": "your-node2-uuid",
"port2": "Ethernet0/0",
},
{
"node_id1": "your-node1-uuid",
"port1": "Ethernet0/1",
"node_id2": "your-node3-uuid",
"port2": "Ethernet0/0",
},
],
}
)
tool = GNS3LinkTool()
print("=== Testing Single Link Creation ===")
result = tool._run(single_link_input)
pprint(result)
print("\n=== Testing Multiple Links Creation ===")
result = tool._run(multiple_links_input)
pprint(result)
@@ -0,0 +1,268 @@
"""
GNS3 node creation tool for network topology building.
Provides functionality to create multiple nodes in GNS3 projects
using specified templates and coordinates through the GNS3 API.
"""
import json
import logging
from pprint import pprint
from typing import Any
from langchain.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun
from gns3_copilot.gns3_client import Node, get_gns3_connector
# Configure logging
logger = logging.getLogger(__name__)
class GNS3CreateNodeTool(BaseTool):
"""
A LangChain tool to create multiple nodes in a GNS3 project
using specified templates and coordinates.
**Input:**
A JSON object containing the project_id and an array of nodes with template_id,
x and y coordinates.
Example input:
{
"project_id": "uuid-of-project",
"nodes": [
{
"template_id": "uuid-of-template",
"x": 100,
"y": -200
},
{
"template_id": "uuid-of-template2",
"x": -200,
"y": 300
}
]
}
**Output:**
A dictionary containing the creation results for all nodes.
Example output:
{
"project_id": "uuid-of-project",
"created_nodes": [
{
"node_id": "uuid-of-node1",
"name": "NodeName1",
"status": "success"
},
{
"node_id": "uuid-of-node2",
"name": "NodeName2",
"status": "success"
}
],
"total_nodes": 2,
"successful_nodes": 2,
"failed_nodes": 0
}
If an error occurs during input validation, returns a dictionary with an 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.
Example input:
{
"project_id": "uuid-of-project",
"nodes": [
{
"template_id": "uuid-of-template",
"x": 100,
"y": -200
},
{
"template_id": "uuid-of-template2",
"x": -200,
"y": 300
}
]
}
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.
"""
def _run(
self,
tool_input: str,
run_manager: CallbackManagerForToolRun | None = None,
**kwargs: Any,
) -> dict[str, Any]:
"""
Creates multiple nodes in a GNS3 project using the provided templates and coordinates.
Args:
tool_input (str): A JSON string containing 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.
"""
# Log received input
logger.info("Received input: %s", tool_input)
try:
# Parse input JSON
input_data = json.loads(tool_input)
project_id = input_data.get("project_id")
nodes = input_data.get("nodes", [])
# Validate input
if not project_id:
logger.error("Invalid input: Missing project_id.")
return {"error": "Missing project_id."}
if not isinstance(nodes, list) or len(nodes) == 0:
logger.error("Invalid input: nodes must be a non-empty array.")
return {"error": "nodes must be a non-empty array."}
# 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)
return {"error": f"Node {i + 1} must be a dictionary."}
template_id = node_data.get("template_id")
x = node_data.get("x")
y = node_data.get("y")
if not all(
[
template_id,
isinstance(x, (int, float)),
isinstance(y, (int, float)),
]
):
logger.error(
"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."
}
# Initialize Gns3Connector using factory function
logger.info("Connecting to GNS3 server...")
gns3_server = get_gns3_connector()
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
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)
results: list[dict[str, Any]] = []
for i, node_data in enumerate(nodes):
try:
template_id = node_data.get("template_id")
x = node_data.get("x")
y = node_data.get("y")
logger.info(
"Creating node %d/%d with template %s at coordinates (%s, %s)...",
i + 1,
len(nodes),
template_id,
x,
y,
)
# Create node
node = Node(
project_id=project_id,
template_id=template_id,
x=x,
y=y,
connector=gns3_server,
)
node.create()
# Retrieve node details
node.get()
node_info = {
"node_id": node.node_id,
"name": node.name,
"status": "success",
}
results.append(node_info)
except Exception as e:
error_info = {
"error": f"Node {i + 1} creation failed: {str(e)}",
"status": "failed",
}
results.append(error_info)
logger.error("Failed to create node %d: %s", i + 1, e)
# 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"])
# Prepare final result
final_result = {
"project_id": project_id,
"created_nodes": results,
"total_nodes": len(nodes),
"successful_nodes": successful_nodes,
"failed_nodes": failed_nodes,
}
# Log the final result
logger.info(
"Node creation completed: %d successful, %d failed out of %d total nodes.",
successful_nodes,
failed_nodes,
len(nodes),
)
# Return JSON-formatted result
return final_result
except json.JSONDecodeError as e:
logger.error("Invalid JSON input: %s", e)
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)}"}
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
"nodes": [
{
"template_id": "b923a635-b7cc-4cb5-9a86-9357e04c02f7", # Replace with actual template UUID
"x": 100,
"y": -200,
},
{
"template_id": "b923a635-b7cc-4cb5-9a86-9357e04c02f7", # Replace with actual template UUID
"x": 200,
"y": -300,
},
],
}
)
tool = GNS3CreateNodeTool()
result = tool._run(test_input)
pprint(result)
@@ -0,0 +1,117 @@
"""
GNS3 template retrieval tool for device discovery.
Provides functionality to retrieve all available device templates
from a GNS3 server, including template names, IDs, and types.
"""
import json
import logging
from pprint import pprint
from typing import Any
from langchain.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun
from gns3_copilot.gns3_client import get_gns3_connector
# Configure logging
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.
**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.
**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.
"""
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.
"""
def _run(
self,
tool_input: str = "",
run_manager: CallbackManagerForToolRun | None = None,
) -> dict[str, Any]:
"""
Connects to the GNS3 server and retrieves a list of all available device templates.
Args:
tool_input (str): Optional input (not used in this tool).
run_manager: LangChain run manager (unused).
Returns:
dict: A dictionary containing the list of templates or an error message.
"""
try:
# Initialize Gns3Connector using factory function
logger.info("Connecting to GNS3 server...")
gns3_server = get_gns3_connector()
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
return {
"error": "Failed to connect to GNS3 server. Please check your configuration."
}
# Retrieve all available templates
templates = gns3_server.get_templates()
# Extract name, template_id, and template_type
template_info = [
{
"name": template.get("name", "N/A"),
"template_id": template.get("template_id", "N/A"),
"template_type": template.get("template_type", "N/A"),
}
for template in templates
]
# Return JSON-formatted result with full logging
result = {"templates": template_info}
logger.info(
"Template retrieval completed. Total templates: %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
)
return {"error": f"Failed to retrieve templates: {str(e)}"}
if __name__ == "__main__":
# Test's tool locally
tool = GNS3TemplateTool()
result = tool._run("")
pprint(result)
@@ -0,0 +1,381 @@
"""
GNS3 node startup tool for network device activation.
Provides functionality to start one or multiple nodes in GNS3 projects
with progress tracking and status monitoring.
"""
import json
import logging
import time
from pprint import pprint
from typing import Any
from langchain.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun
from gns3_copilot.gns3_client import Node, get_gns3_connector
# Configure logging
logger = logging.getLogger(__name__)
def show_progress_bar(
duration: int = 120, interval: int = 1, node_count: int = 1
) -> None:
"""
Display a simple text progress bar for node startup.
Args:
duration: Total duration of the progress bar in seconds
interval: Update interval in seconds
node_count: Number of nodes being started
"""
print(f"Starting {node_count} node(s), please wait...")
for elapsed in range(duration):
# Calculate progress percentage
progress = (elapsed + 1) / duration * 100
# Create progress bar display
bar_length = 30
filled_length = int(bar_length * elapsed // duration)
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)
time.sleep(interval)
print(f"\n{node_count} node(s) startup completed!")
class GNS3StartNodeTool(BaseTool):
"""
A LangChain tool to start one or multiple nodes in a GNS3 project.
**Input**:
A JSON object with project_id and node_ids (list of node IDs).
Example:
{
"project_id": "uuid-of-project",
"node_ids": ["uuid-of-node-1", "uuid-of-node-2"]
}
**Output**:
A dictionary with all nodes' details:
{
"project_id": "...",
"total_nodes": 2,
"successful": 2,
"failed": 0,
"nodes": [
{"node_id": "...", "name": "...", "status": "..."},
{"node_id": "...", "name": "...", "status": "..."}
]
}
"""
name: str = "start_gns3_node"
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.
"""
def _run(
self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None
) -> dict[str, Any]:
try:
# Parse input JSON
input_data = json.loads(tool_input)
project_id = input_data.get("project_id")
node_ids = input_data.get("node_ids")
# 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."}
if not isinstance(node_ids, list):
logger.error("node_ids must be a list.")
return {"error": "node_ids must be a list."}
# Initialize Gns3Connector using factory function
logger.info("Connecting to GNS3 server...")
gns3_server = get_gns3_connector()
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
return {
"error": "Failed to connect to GNS3 server. Please check your configuration."
}
# First loop: Send start commands for all nodes
logger.info(
"Sending start commands for %d nodes in project %s...",
len(node_ids),
project_id,
)
for node_id in node_ids:
try:
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
)
except Exception as 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
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)
)
# 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.get() # Get latest status
node_info = {
"node_id": node.node_id,
"name": node.name or "N/A",
"status": node.status or "unknown",
}
results.append(node_info)
except Exception as e:
logger.error("Failed to get status for node %s: %s", node_id, e)
results.append(
{
"node_id": node_id,
"name": "N/A",
"status": "error",
"error": str(e),
}
)
# Analyze results
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
response = {
"project_id": project_id,
"total_nodes": len(node_ids),
"successful": len(successful_nodes),
"failed": len(failed_nodes),
"nodes": results,
}
logger.info(
"Start operation completed: %d successful, %d failed",
len(successful_nodes),
len(failed_nodes),
)
return response
except json.JSONDecodeError as e:
logger.error("Invalid JSON input: %s", e)
return {"error": f"Invalid JSON input: {e}"}
except Exception as e:
logger.error("Failed to start nodes: %s", e)
return {"error": f"Failed to start nodes: {str(e)}"}
class GNS3StartNodeQuickTool(BaseTool):
"""
A LangChain tool to start one or multiple 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
workflows where long waits would cause HTTP timeouts.
**Input**:
A JSON object with project_id and node_ids (list of node IDs).
Example:
{
"project_id": "uuid-of-project",
"node_ids": ["uuid-of-node-1", "uuid-of-node-2"]
}
**Output**:
A dictionary with all nodes' details immediately after sending start commands:
{
"project_id": "...",
"total_nodes": 2,
"successful": 2,
"failed": 0,
"nodes": [
{"node_id": "...", "name": "...", "status": "started"},
{"node_id": "...", "name": "...", "status": "started"}
],
"note": "Start commands sent. Nodes are booting in background."
}
"""
name: str = "start_gns3_node_quick"
description: str = """
Starts one or multiple 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.
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]:
try:
# Parse input JSON
input_data = json.loads(tool_input)
project_id = input_data.get("project_id")
node_ids = input_data.get("node_ids")
# 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."}
if not isinstance(node_ids, list):
logger.error("node_ids must be a list.")
return {"error": "node_ids must be a list."}
# Initialize Gns3Connector using factory function
logger.info("Connecting to GNS3 server...")
gns3_server = get_gns3_connector()
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
return {
"error": "Failed to connect to GNS3 server. Please check your configuration."
}
# Send start commands for all nodes and collect initial status
logger.info(
"Sending start commands for %d nodes in project %s...",
len(node_ids),
project_id,
)
results = []
for node_id in node_ids:
try:
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)
results.append(
{
"node_id": node_id,
"name": "N/A",
"status": "error",
"error": "Node not found",
}
)
continue
# Send start command
node.start()
logger.info("Start command sent for node %s (%s)", node_id, node.name)
# Get immediate status (will likely still be 'starting' or 'stopped')
node.get()
node_info = {
"node_id": node.node_id,
"name": node.name or "N/A",
"status": node.status or "unknown",
}
results.append(node_info)
except Exception as e:
logger.error("Failed to start node %s: %s", node_id, e)
results.append(
{
"node_id": node_id,
"name": "N/A",
"status": "error",
"error": str(e),
}
)
# Analyze results (count based on successful command sending)
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
response = {
"project_id": project_id,
"total_nodes": len(node_ids),
"successful": len(successful_nodes),
"failed": len(failed_nodes),
"nodes": results,
"note": "Start commands sent. Nodes are booting in background. Check node status later.",
}
logger.info(
"Quick start commands sent: %d successful, %d failed",
len(successful_nodes),
len(failed_nodes),
)
return response
except json.JSONDecodeError as e:
logger.error("Invalid JSON input: %s", e)
return {"error": f"Invalid JSON input: {e}"}
except Exception as e:
logger.error("Failed to start nodes: %s", e)
return {"error": f"Failed to start nodes: {str(e)}"}
if __name__ == "__main__":
# Test with single node
print("=== Testing single node startup ===")
test_input_single = json.dumps(
{
"project_id": "f32ebf3d-ef8c-4910-b0d6-566ed828cd24", # Replace with actual project UUID
"node_ids": [
"fbeda109-9a74-4d8c-a749-cc3847911a90"
], # Replace with actual node UUID
}
)
tool = GNS3StartNodeTool()
result_single = tool._run(test_input_single)
pprint(result_single)
# Test with multiple nodes
print("\n=== Testing multiple nodes startup ===")
test_input_multiple = json.dumps(
{
"project_id": "f32ebf3d-ef8c-4910-b0d6-566ed828cd24", # Replace with actual project UUID
"node_ids": [
"fbeda109-9a74-4d8c-a749-cc3847911a90", # Replace with actual node UUIDs
"another-node-uuid-here",
"third-node-uuid-here",
],
}
)
result_multiple = tool._run(test_input_multiple)
pprint(result_multiple)
@@ -0,0 +1,216 @@
"""
GNS3 node name update tool for renaming network devices.
Provides functionality to update the name of one or multiple nodes in GNS3 projects.
"""
import json
import logging
from pprint import pprint
from typing import Any
from langchain.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun
from gns3_copilot.gns3_client import Node, get_gns3_connector
# Configure logging
logger = logging.getLogger(__name__)
class GNS3UpdateNodeNameTool(BaseTool):
"""
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.
Example:
{
"project_id": "uuid-of-project",
"nodes": [
{"node_id": "uuid-of-node-1", "new_name": "Router1"},
{"node_id": "uuid-of-node-2", "new_name": "Switch1"}
]
}
**Output**:
A dictionary with all nodes' update results:
{
"project_id": "...",
"total_nodes": 2,
"successful": 2,
"failed": 0,
"nodes": [
{"node_id": "...", "old_name": "...", "new_name": "...", "status": "success"},
{"node_id": "...", "old_name": "...", "new_name": "...", "status": "success"}
]
}
"""
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.
"""
def _run(
self, tool_input: str, run_manager: CallbackManagerForToolRun | None = None
) -> dict[str, Any]:
try:
# Parse input JSON
input_data = json.loads(tool_input)
project_id = input_data.get("project_id")
nodes = input_data.get("nodes", [])
# Validate input
if not project_id:
logger.error("Missing required field: project_id.")
return {"error": "Missing required field: project_id."}
if not isinstance(nodes, list) or len(nodes) == 0:
logger.error("nodes must be a non-empty list.")
return {"error": "nodes must be a non-empty list."}
# Validate each node entry
for i, node_data in enumerate(nodes):
if not isinstance(node_data, dict):
logger.error("Node %d must be a dictionary.", i + 1)
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."}
# Initialize Gns3Connector
logger.info("Connecting to GNS3 server...")
gns3_server = get_gns3_connector()
if gns3_server is None:
logger.error("Failed to create GNS3 connector")
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)
results = []
for i, node_data in enumerate(nodes):
try:
node_id = node_data.get("node_id")
new_name = node_data.get("new_name")
logger.info(
"Updating node %d/%d: %s -> %s",
i + 1,
len(nodes),
node_id,
new_name,
)
# Get node to retrieve current name
node = Node(
project_id=project_id, node_id=node_id, connector=gns3_server
)
node.get()
old_name = node.name
# Update node name
node.update(name=new_name)
# Verify update
node.get()
if node.name == new_name:
node_info = {
"node_id": node_id,
"old_name": old_name,
"new_name": new_name,
"status": "success",
}
results.append(node_info)
logger.info(
"Successfully updated node name: %s -> %s", old_name, new_name
)
else:
error_info = {
"node_id": node_id,
"old_name": old_name,
"new_name": new_name,
"current_name": node.name,
"status": "failed",
"error": "Name verification failed",
}
results.append(error_info)
logger.error("Failed to update node name for %s", node_id)
except Exception as e:
error_info = {
"node_id": node_data.get("node_id"),
"new_name": node_data.get("new_name"),
"status": "failed",
"error": str(e),
}
results.append(error_info)
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"]
failed_nodes = [r for r in results if r.get("status") == "failed"]
# Construct final response
response = {
"project_id": project_id,
"total_nodes": len(nodes),
"successful": len(successful_nodes),
"failed": len(failed_nodes),
"nodes": results,
}
logger.info(
"Name update completed: %d successful, %d failed",
len(successful_nodes),
len(failed_nodes),
)
return response
except json.JSONDecodeError as e:
logger.error("Invalid JSON input: %s", e)
return {"error": f"Invalid JSON input: {e}"}
except Exception as e:
logger.error("Failed to update node names: %s", e)
return {"error": f"Failed to update node names: {str(e)}"}
if __name__ == "__main__":
# Test with single node
print("=== Testing single node name update ===")
test_input_single = json.dumps(
{
"project_id": "your-project-uuid", # Replace with actual project UUID
"nodes": [
{
"node_id": "your-node-uuid", # Replace with actual node UUID
"new_name": "Router1",
}
],
}
)
tool = GNS3UpdateNodeNameTool()
result_single = tool._run(test_input_single)
pprint(result_single)
# Test with multiple nodes
print("\n=== Testing multiple nodes name update ===")
test_input_multiple = json.dumps(
{
"project_id": "your-project-uuid", # Replace with actual project UUID
"nodes": [
{"node_id": "node-uuid-1", "new_name": "Router1"},
{"node_id": "node-uuid-2", "new_name": "Switch1"},
{"node_id": "node-uuid-3", "new_name": "PC1"},
],
}
)
result_multiple = tool._run(test_input_multiple)
pprint(result_multiple)
@@ -0,0 +1,659 @@
"""
This module uses Nornir + Netmiko to batch execute Linux commands on GNS3 topology devices
via Telnet console.
"""
import json
import logging
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_config,
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": get_config("GNS3_SERVER_HOST"),
"timeout": 120,
"username": get_config("LINUX_TELNET_USERNAME"),
"password": get_config("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 <host>, traceroute -n
✅ Processes: ps aux, ps aux --sort=-%mem | head -20
✅ Services: systemctl status <service> --no-pager
✅ Logs: journalctl -u <unit> --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 = get_config("LINUX_TELNET_USERNAME")
linux_password = get_config("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}")
@@ -0,0 +1,439 @@
"""
Multi-device VPCS command execution tool using telnetlib3 with threading.
Supports concurrent execution of multiple command groups across multiple VPCS devices.
"""
import json
import logging
import re
import threading
from time import sleep
from typing import Any
from langchain.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun
from telnetlib3 import Telnet
from gns3_copilot.utils import get_config, get_device_ports_from_topology
logger = logging.getLogger(__name__)
class VPCSMultiCommands(BaseTool):
"""
A tool for VPCS (Virtual PC Simulator) devices to view PC configurations 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.
**Allowed VPCS Commands:**
- IP configuration: ip <address>/<mask> <gateway> (Basic PC IP setup)
- View configuration: ip, show ip
- Connectivity testing: ping <destination>
- Display ARP: arp
- Display version: version
- Save/Load: save, load
**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.
"""
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.
**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
**When to Use This Tool:**
- Configure IP addresses on virtual PCs: ip 10.10.0.12/24 10.10.0.254
- Test connectivity from PCs: ping 10.10.0.254
- View PC IP configuration: ip, show ip
- Display ARP table: arp
- Check PC version: version
**Input Format:**
{
"project_id": "f32ebf3d-ef8c-4910-b0d6-566ed828cd24",
"device_configs": [
{
"device_name": "PC1",
"commands": ["ip", "ping 10.10.0.254"]
},
{
"device_name": "PC2",
"commands": ["show ip"]
}
]
}
**Returns:** PC command outputs for IP configuration and connectivity testing.
**Note:** For network devices (Cisco/Huawei routers), use execute_multiple_device_commands instead.
"""
def _connect_and_execute_commands(
self,
device_name: str,
commands: list[str],
results_list: list[Any],
index: int,
device_ports: dict[str, Any],
gns3_host: str,
) -> None:
"""
Internal method to connect to VPCS device and execute commands.
VPCS devices are lightweight virtual PCs used in GNS3 labs for testing
network connectivity and basic IP configuration.
"""
logger.info(
"Starting connection for device '%s' with %d commands",
device_name,
len(commands),
)
# 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
)
results_list[index] = {
"device_name": device_name,
"status": "error",
"output": f"Device '{device_name}' not found in topology or missing console port",
"commands": commands,
}
return
port = device_ports[device_name]["port"]
host = gns3_host
logger.info("Connecting to device '%s' at %s:%d", device_name, host, port)
tn = Telnet()
try:
tn.open(host=host, port=port, timeout=30)
logger.info(
"Successfully connected to device '%s' at %s:%d",
device_name,
host,
port,
)
# Initialize connection
tn.write(b"\n")
sleep(0.5)
tn.write(b"\n")
sleep(0.5)
tn.write(b"\n")
sleep(0.5)
tn.write(b"\n")
sleep(0.5)
tn.expect([rb"PC\d+>"])
logger.info("Connection initialized for device '%s'", device_name)
# Execute all commands and merge output
combined_output = ""
for i, command in enumerate(commands):
logger.info(
"Executing command %d/%d on device '%s': %s",
i + 1,
len(commands),
device_name,
command,
)
tn.write(command.encode(encoding="ascii") + b"\n")
sleep(5)
tn.expect([rb"PC\d+>"])
output = tn.read_very_eager().decode("utf-8")
combined_output += output
# Add result to list
results_list[index] = {
"device_name": device_name,
"status": "success",
"output": combined_output,
"commands": commands,
}
logger.info(
"Successfully executed all %d commands on device '%s'",
len(commands),
device_name,
)
except Exception as e:
logger.error(
"Error executing commands on device '%s': %s", device_name, str(e)
)
results_list[index] = {
"device_name": device_name,
"status": "error",
"output": str(e),
"commands": commands,
}
finally:
tn.close()
logger.debug("Connection closed for device '%s'", device_name)
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}$"
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)
return is_valid
def _validate_tool_input(
self, tool_input: str | bytes | list[Any] | dict[str, Any]
) -> tuple[list[dict[str, Any]], str]:
"""
Validate device command input and extract project_id and device_configs.
Args:
tool_input: The input received from the LangChain/LangGraph tool call.
Returns:
Tuple containing (device_configs_list, project_id) or (error_list, "")
"""
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 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 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__
)
# 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__}"
)
logger.error(error_msg)
return ([{"error": error_msg}], "")
# Extract and validate project_id
project_id = parsed_input.get("project_id")
if not project_id:
error_msg = "Missing required field 'project_id' in input"
logger.error(error_msg)
return ([{"error": error_msg}], "")
# Validate project_id format
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}], "")
# Extract and validate device_configs
device_configs = parsed_input.get("device_configs")
if device_configs is None:
error_msg = "Missing required field 'device_configs' in input"
logger.error(error_msg)
return ([{"error": error_msg}], "")
# 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__}"
logger.error(error_msg)
return ([{"error": error_msg}], "")
# Handle empty list
if not device_configs:
logger.warning("Device configs list is empty.")
return [], ""
# 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__}"
)
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'"
logger.error(error_msg)
return ([{"error": error_msg}], "")
if "commands" not in item:
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__}"
)
logger.error(error_msg)
return ([{"error": error_msg}], "")
logger.info(
"Input validated successfully. project_id=%s, device_configs_count=%d",
project_id,
len(device_configs),
)
return device_configs, project_id
def _run(
self,
tool_input: str,
run_manager: CallbackManagerForToolRun | None = None,
**kwargs: Any,
) -> list[dict[str, Any]]:
"""
Main method to 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.
Args:
tool_input: JSON string containing project_id and device_configs with VPCS commands
Returns:
List of execution results for each VPCS device
"""
# Log received input
logger.info("Received input: %s", tool_input)
# Validate tool input and extract project_id and device_configs
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]
):
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
)
logger.info(
"Retrieved port mappings for %d devices: %s",
len(device_ports),
list(device_ports.keys()),
)
# Get host IP from database
gns3_host = get_config("GNS3_SERVER_HOST", "127.0.0.1")
logger.info("Using GNS3 server host: %s", gns3_host)
# Initialize results list (pre-allocate space for concurrent writes)
results: list[dict[str, Any]] = [{} for _ in range(len(device_configs))]
threads = []
# Create thread for each command group
logger.info("Starting parallel execution for %d devices", len(device_configs))
for i, cmd_group in enumerate(device_configs):
device_name = cmd_group["device_name"]
thread = threading.Thread(
target=self._connect_and_execute_commands,
args=(
cmd_group["device_name"],
cmd_group["commands"],
results,
i,
device_ports,
gns3_host,
),
)
threads.append(thread)
thread.start()
# Wait for all threads to complete
for thread in threads:
thread.join()
# Count successful and failed executions
success_count = sum(1 for r in results if r.get("status") == "success")
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",
len(results),
success_count,
error_count,
)
return results
if __name__ == "__main__":
# Example usage
command_groups = json.dumps(
{
"project_id": "f32ebf3d-ef8c-4910-b0d6-566ed828cd24",
"device_configs": [
{
"device_name": "PC1",
"commands": [
"ip 10.10.0.12/24 10.10.0.254",
"ping 10.10.0.254",
],
},
{
"device_name": "PC2",
"commands": ["ip 10.10.0.13/24 10.10.0.254"],
},
{
"device_name": "PC3",
"commands": [
"ip 10.20.0.22/24 10.20.0.254",
"ping 10.20.0.254",
],
},
{
"device_name": "PC4",
"commands": ["ip 10.20.0.23/24 10.20.0.254"],
},
],
}
)
exe_cmd = VPCSMultiCommands()
result = exe_cmd._run(tool_input=command_groups)
print("Execution results:")
print(json.dumps(result, indent=2, ensure_ascii=False))
@@ -0,0 +1,726 @@
# 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 <https://www.gnu.org/licenses/>.
"""
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)