diff --git a/gns3server/agent/gns3_copilot/__init__.py b/gns3server/agent/gns3_copilot/__init__.py index 5bca7bfe7..ae4b67126 100644 --- a/gns3server/agent/gns3_copilot/__init__.py +++ b/gns3server/agent/gns3_copilot/__init__.py @@ -1,7 +1,7 @@ """ -FlowNet-Lab - AI-powered network automation assistant for GNS3. +GNS3-Copilot - AI-powered network automation assistant for GNS3. -This package provides a command-line interface for launching the FlowNet-Lab +This package provides a command-line interface for launching the GNS3-Copilot Streamlit application with support for Streamlit parameter passthrough. """ diff --git a/gns3server/agent/gns3_copilot/agent/__init__.py b/gns3server/agent/gns3_copilot/agent/__init__.py index 6527381fd..87bcccea3 100644 --- a/gns3server/agent/gns3_copilot/agent/__init__.py +++ b/gns3server/agent/gns3_copilot/agent/__init__.py @@ -1,7 +1,7 @@ """ -FlowNet-Lab Agent Package +GNS3-Copilot Agent Package -This package contains the main FlowNet-Lab agent implementation for network automation tasks. +This package contains the main GNS3-Copilot agent implementation for network automation tasks. """ from .gns3_copilot import agent_builder diff --git a/gns3server/agent/gns3_copilot/agent/gns3_copilot.py b/gns3server/agent/gns3_copilot/agent/gns3_copilot.py index ae4a3dc4d..e378aebde 100644 --- a/gns3server/agent/gns3_copilot/agent/gns3_copilot.py +++ b/gns3server/agent/gns3_copilot/agent/gns3_copilot.py @@ -1,19 +1,19 @@ # SPDX-License-Identifier: AGPL-3.0-or-later # -# This file is part of FlowNet-Lab. +# This file is part of GNS3-Copilot. # -# FlowNet-Lab is free software: you can redistribute it and/or modify it +# GNS3-Copilot 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 +# GNS3-Copilot 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 . +# with GNS3-Copilot. If not, see . # mypy: ignore-errors @@ -55,16 +55,14 @@ from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "backend")) from gns3_copilot.tools_v2 import ( ExecuteMultipleDeviceCommands, - GNS3CreateAreaDrawingTool, GNS3CreateNodeTool, GNS3LinkTool, GNS3StartNodeTool, GNS3TemplateTool, - LinuxTelnetBatchTool, VPCSMultiCommands, ) -# Set up logger for FlowNet-Lab +# Set up logger for GNS3-Copilot logger = logging.getLogger(__name__) # Note: LLM model configuration is now managed by the new llm_model_configs system. @@ -79,15 +77,13 @@ tools = [ GNS3StartNodeTool(), # Start GNS3 nodes ExecuteMultipleDeviceCommands(), # Execute show/display/debug commands on multiple devices (READ-ONLY) VPCSMultiCommands(), # Execute VPCS commands on multiple devices - LinuxTelnetBatchTool(), # Execute Linux commands via Telnet on multiple devices - GNS3CreateAreaDrawingTool(), # Create area drawings in GNS3 topologies ] # Augment the LLM with tools tools_by_name = {tool.name: tool for tool in tools} # Model with tools will be created dynamically by the factory when needed # Log application startup -logger.info("FlowNet-Lab application starting up") +logger.info("GNS3-Copilot application starting up") # Constants for conversation title management DEFAULT_CONVERSATION_TITLE = "New Conversation" @@ -98,7 +94,7 @@ TITLE_MAX_LENGTH = 40 # Define state class MessagesState(TypedDict): """ - FlowNet-Lab conversation state management class. + GNS3-Copilot conversation state management class. Maintains the conversation state for the LangGraph workflow, including message history, call counters, and session titles for comprehensive dialogue management. diff --git a/gns3server/agent/gns3_copilot/agent/model_factory.py b/gns3server/agent/gns3_copilot/agent/model_factory.py index 5ab619433..6467a055e 100644 --- a/gns3server/agent/gns3_copilot/agent/model_factory.py +++ b/gns3server/agent/gns3_copilot/agent/model_factory.py @@ -1,5 +1,5 @@ """ -Model Factory for FlowNet-Lab Agent +Model Factory for GNS3-Copilot Agent This module provides factory functions to create fresh LLM model instances. Configuration is loaded from the llm_model_configs system via connector_factory. diff --git a/gns3server/agent/gns3_copilot/gns3_client/__init__.py b/gns3server/agent/gns3_copilot/gns3_client/__init__.py index aeac912ab..f554287c6 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/__init__.py +++ b/gns3server/agent/gns3_copilot/gns3_client/__init__.py @@ -11,20 +11,13 @@ Main classes: - Node: GNS3 Node management - Link: GNS3 Link management - GNS3TopologyTool: GNS3 topology reading tool -- GNS3ProjectReadFileTool: LangChain tool for reading project files -- GNS3ProjectWriteFileTool: LangChain tool for writing project files -- GNS3ProjectListFilesTool: LangChain tool for listing project files -- GNS3ProjectLock: LangChain tool for locking/unlocking GNS3 projects - -File Manager Modules: -- gns3_project_read_file: GNS3ProjectReadFileTool implementation -- gns3_project_write_file: GNS3ProjectWriteFileTool implementation -- gns3_project_list_files: GNS3ProjectListFilesTool implementation -- gns3_file_index: File index management utilities +- GNS3UpdateDrawingTool: GNS3 drawing update tool Main functions: - get_gns3_connector: Factory function to create Gns3Connector from environment - get_gns3_connector_with_llm_config: Factory function to create connector AND retrieve LLM config +- get_gns3_server_host: Get GNS3 server hostname from Controller or Config +- get_llm_config: Get LLM model configuration for a user """ from .connector_factory import ( @@ -42,21 +35,6 @@ from .custom_gns3fy import ( Node, Project, ) -from .gns3_create_drawing import GNS3CreateDrawingTool -from .gns3_delete_drawing import GNS3DeleteDrawingTool -from .gns3_file_index import add_file_to_index, get_file_list -from .gns3_get_drawings import GNS3GetDrawingsTool -from .gns3_get_nodes import GNS3GetNodesTool -from .gns3_project_create import GNS3ProjectCreate -from .gns3_project_delete import GNS3ProjectDelete -from .gns3_project_list_files import GNS3ProjectListFilesTool -from .gns3_project_lock import GNS3ProjectLock -from .gns3_project_open import GNS3ProjectOpen -from .gns3_project_path import GNS3ProjectPath -from .gns3_project_read_file import GNS3ProjectReadFileTool -from .gns3_project_update import GNS3ProjectUpdate -from .gns3_project_write_file import GNS3ProjectWriteFileTool -from .gns3_projects_list import GNS3ProjectList from .gns3_topology_reader import GNS3TopologyTool from .gns3_update_drawing import GNS3UpdateDrawingTool @@ -81,25 +59,9 @@ __all__ = [ "CONSOLE_TYPES", "LINK_TYPES", "GNS3TopologyTool", - "GNS3ProjectList", - "GNS3ProjectOpen", - "GNS3ProjectPath", - "GNS3ProjectCreate", - "GNS3ProjectDelete", - "GNS3ProjectLock", - "GNS3ProjectUpdate", - "GNS3ProjectReadFileTool", - "GNS3ProjectWriteFileTool", - "GNS3ProjectListFilesTool", - "GNS3CreateDrawingTool", - "GNS3DeleteDrawingTool", - "GNS3GetDrawingsTool", - "GNS3GetNodesTool", "GNS3UpdateDrawingTool", "get_gns3_connector", "get_gns3_connector_with_llm_config", "get_gns3_server_host", "get_llm_config", - "add_file_to_index", - "get_file_list", ] diff --git a/gns3server/agent/gns3_copilot/gns3_client/gns3_create_drawing.py b/gns3server/agent/gns3_copilot/gns3_client/gns3_create_drawing.py deleted file mode 100644 index b0f85edec..000000000 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_create_drawing.py +++ /dev/null @@ -1,292 +0,0 @@ -""" -GNS3 drawing creation tool for adding graphical elements. - -Provides functionality to create multiple drawings in a GNS3 project -using specified SVG content 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 Project, get_gns3_connector - -# Configure logging -logger = logging.getLogger(__name__) - - -class GNS3CreateDrawingTool(BaseTool): - """ - A LangChain tool to create multiple drawings in a GNS3 project - using specified SVG content and coordinates. - - **Input:** - A JSON object containing the project_id and an array of drawings with SVG content, - coordinates, and optional properties. - - Example input: - { - "project_id": "uuid-of-project", - "drawings": [ - { - "svg": "...", - "x": 100, - "y": -200, - "z": 0, - "locked": false, - "rotation": 0 - }, - { - "svg": "...", - "x": -200, - "y": 300, - "z": 1, - "locked": true, - "rotation": 90 - } - ] - } - - **Output:** - A dictionary containing the creation results for all drawings. - Example output: - { - "project_id": "uuid-of-project", - "created_drawings": [ - { - "drawing_id": "uuid-of-drawing1", - "status": "success" - }, - { - "drawing_id": "uuid-of-drawing2", - "status": "success" - } - ], - "total_drawings": 2, - "successful_drawings": 2, - "failed_drawings": 0 - } - If an error occurs during input validation, returns a dictionary with an error message. - """ - - name: str = "create_gns3_drawing" - description: str = """ - Creates multiple drawings in a GNS3 project using specified SVG content and coordinates. - Input is a JSON object with project_id and an array of drawings, each containing svg, x, y, - and optional z, locked, and rotation parameters. - Example input: - { - "project_id": "uuid-of-project", - "drawings": [ - { - "svg": "...", - "x": 100, - "y": -200, - "z": 0, - "locked": false, - "rotation": 0 - } - ] - } - Returns a dictionary with creation results for all drawings, 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 drawings in a GNS3 project using the provided SVG content and coordinates. - - Args: - tool_input (str): A JSON string containing project_id and an array of drawings. - run_manager: LangChain run manager (unused). - - Returns: - dict: A dictionary with creation results for all drawings 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") - drawings = input_data.get("drawings", []) - - # Validate input - if not project_id: - logger.error("Invalid input: Missing project_id.") - return {"error": "Missing project_id."} - - if not isinstance(drawings, list) or len(drawings) == 0: - logger.error("Invalid input: drawings must be a non-empty array.") - return {"error": "drawings must be a non-empty array."} - - # Validate each drawing in the array - for i, drawing_data in enumerate(drawings): - if not isinstance(drawing_data, dict): - logger.error( - "Invalid input: Drawing %d must be a dictionary.", i + 1 - ) - return {"error": f"Drawing {i + 1} must be a dictionary."} - - svg = drawing_data.get("svg") - x = drawing_data.get("x") - y = drawing_data.get("y") - - if not all( - [svg, isinstance(x, (int, float)), isinstance(y, (int, float))] - ): - logger.error( - "Invalid input: Drawing %d missing or invalid svg, x, or y.", - i + 1, - ) - return { - "error": f"Drawing {i + 1} missing or invalid svg, 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 project instance - logger.info( - "Creating %d drawings in project %s...", len(drawings), project_id - ) - project = Project(project_id=project_id, connector=gns3_server) - - # Create drawings - results: list[dict[str, Any]] = [] - - for i, drawing_data in enumerate(drawings): - try: - svg = drawing_data.get("svg") - x = drawing_data.get("x") - y = drawing_data.get("y") - z = drawing_data.get("z", 0) - locked = drawing_data.get("locked", False) - rotation = drawing_data.get("rotation", 0) - - logger.info( - "Creating drawing %d/%d at coordinates (%s, %s)...", - i + 1, - len(drawings), - x, - y, - ) - - # Create drawing using Project method - result = project.create_drawing( - svg=svg, - x=int(x), - y=int(y), - z=int(z), - locked=bool(locked), - rotation=int(rotation), - ) - - drawing_info = { - "drawing_id": result.get("drawing_id"), - "status": "success", - } - - results.append(drawing_info) - - except Exception as e: - error_info = { - "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) - # Continue with next drawing even if one fails - - # 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, - "created_drawings": results, - "total_drawings": len(drawings), - "successful_drawings": successful_drawings, - "failed_drawings": failed_drawings, - } - - # Log the final result - logger.info( - "Drawing 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 drawing creation request: %s", e) - return {"error": f"Failed to process drawing creation request: {str(e)}"} - - -if __name__ == "__main__": - # Test the tool locally with multiple drawings - test_input = json.dumps( - { - "project_id": "0c0fde25-6ead-4413-a283-ea8fd2324291", # Replace with actual project UUID - "drawings": [ - { - "svg": 'Label 1', - "x": 100, - "y": -200, - "z": 0, - "locked": False, - "rotation": 0, - }, - { - "svg": 'Label 2', - "x": 200, - "y": -300, - "z": 1, - "locked": True, - "rotation": 90, - }, - ], - } - ) - tool = GNS3CreateDrawingTool() - result = tool._run(test_input) - pprint(result) - - -""" -example output: -{'created_drawings': [{'drawing_id': '52045be2-d6d9-46f8-85af-c33bf7074b6a', - 'status': 'success'}, - {'drawing_id': '8f1838a7-1aa4-4613-acaa-b300b23e60d5', - 'status': 'success'}], - 'failed_drawings': 0, - 'project_id': '2245149a-71c8-4387-9d1f-441a683ef7e7', - 'successful_drawings': 2, - 'total_drawings': 2} -""" diff --git a/gns3server/agent/gns3_copilot/gns3_client/gns3_delete_drawing.py b/gns3server/agent/gns3_copilot/gns3_client/gns3_delete_drawing.py deleted file mode 100644 index aafb7132d..000000000 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_delete_drawing.py +++ /dev/null @@ -1,160 +0,0 @@ -""" -GNS3 drawing deletion tool for removing graphical elements. - -Provides functionality to delete drawings from a GNS3 project. -""" - -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 Project, get_gns3_connector - -# Configure logging -logger = logging.getLogger(__name__) - - -class GNS3DeleteDrawingTool(BaseTool): - """ - A LangChain tool to delete a drawing from a GNS3 project. - - **Input:** - A JSON object containing the project_id and drawing_id. - - Example input: - { - "project_id": "uuid-of-project", - "drawing_id": "uuid-of-drawing" - } - - **Output:** - A dictionary containing the deletion result. - Example output: - { - "project_id": "uuid-of-project", - "drawing_id": "uuid-of-drawing", - "status": "success" - } - If an error occurs during input validation, returns a dictionary with an error message. - """ - - name: str = "delete_gns3_drawing" - description: str = """ - Deletes a drawing from a GNS3 project. - Input is a JSON object with project_id and drawing_id. - Example input: - { - "project_id": "uuid-of-project", - "drawing_id": "uuid-of-drawing" - } - Returns a dictionary with deletion result and status. - If the operation fails, returns a dictionary with an error message. - """ - - def _run( - self, - tool_input: str, - run_manager: CallbackManagerForToolRun | None = None, - **kwargs: Any, - ) -> dict[str, Any]: - """ - Deletes a drawing from a GNS3 project. - - Args: - tool_input (str): A JSON string containing project_id and drawing_id. - run_manager: LangChain run manager (unused). - - Returns: - dict: A dictionary with the deletion result and status 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") - drawing_id = input_data.get("drawing_id") - - # Validate input - if not project_id: - logger.error("Invalid input: Missing project_id.") - return {"error": "Missing project_id."} - - if not drawing_id: - logger.error("Invalid input: Missing drawing_id.") - return {"error": "Missing drawing_id."} - - # 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 project instance - logger.info( - "Deleting drawing %s from project %s...", drawing_id, project_id - ) - project = Project(project_id=project_id, connector=gns3_server) - - # Delete the drawing - project.delete_drawing(drawing_id=drawing_id) - - # Prepare final result - final_result = { - "project_id": project_id, - "drawing_id": drawing_id, - "status": "success", - } - - # Log the final result - logger.info("Drawing deletion completed successfully.") - - # 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 ValueError as e: - logger.error("Value error: %s", e) - return {"error": str(e)} - except Exception as e: - logger.error("Failed to delete drawing: %s", e) - return {"error": f"Failed to delete drawing: {str(e)}"} - - -if __name__ == "__main__": - # Test the tool locally - test_input = json.dumps( - { - "project_id": "0c0fde25-6ead-4413-a283-ea8fd2324291", # Replace with actual project UUID - "drawing_id": "0728feaf-defd-40e3-ae02-1f97734810e2", # Replace with actual drawing UUID - } - ) - tool = GNS3DeleteDrawingTool() - result = tool._run(test_input) - pprint(result) - - -""" -example output: - -error output when drawing is locked: -{'error': 'Failed to delete drawing: Unknown Status: Drawing ID ' - 'daf3385a-86f0-458d-8563-6e1fbd87af77 cannot be deleted because it ' - 'is locked (Original 409 Error)'} - -successful output: -{'drawing_id': 'daf3385a-86f0-458d-8563-6e1fbd87af77', - 'project_id': '2245149a-71c8-4387-9d1f-441a683ef7e7', - 'status': 'success'} -""" diff --git a/gns3server/agent/gns3_copilot/gns3_client/gns3_file_index.py b/gns3server/agent/gns3_copilot/gns3_client/gns3_file_index.py deleted file mode 100644 index a6c972e26..000000000 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_file_index.py +++ /dev/null @@ -1,257 +0,0 @@ -""" -GNS3 File Index Manager - -This module provides functionality to manage an index of files in GNS3 projects. -Since GNS3 API doesn't provide a way to list files, we maintain an index -file to track files written to each project. -""" - -import json -import logging -from datetime import datetime -from typing import Any - -from gns3_copilot.gns3_client import Project, get_gns3_connector - -# Configure logging -logger = logging.getLogger(__name__) - -# Load environment variables - -# Index file name (stored in project directory) -INDEX_FILE_NAME = ".gns3_copilot_file_index.json" - - -def _get_index_path(project_id: str) -> str: - """ - Get the path to the index file for a project. - - Args: - project_id: The project ID - - Returns: - The path to the index file (relative to project directory) - """ - return INDEX_FILE_NAME - - -def load_file_index(project: Project) -> dict[str, Any]: - """ - Load the file index for a project. - - Args: - project: GNS3 Project instance - - Returns: - A dictionary containing the index data, or an empty index if not found - """ - if project.project_id is None: - raise ValueError("Project ID must be set") - - index_path = _get_index_path(project.project_id) - - try: - content = project.get_file(path=index_path) - index_data = json.loads(content) - - # Validate index structure - if not isinstance(index_data, dict): - logger.warning("Invalid index structure, creating new index") - return _create_empty_index(project.project_id) - - if "files" not in index_data: - index_data["files"] = [] - - logger.info( - "Loaded file index for project %s with %d files", - project.project_id, - len(index_data["files"]), - ) - return index_data - - except Exception as e: - logger.info( - "No existing file index found for project %s, creating new one: %s", - project.project_id, - e, - ) - return _create_empty_index(project.project_id) - - -def save_file_index(project: Project, index_data: dict[str, Any]) -> None: - """ - Save the file index for a project. - - Args: - project: GNS3 Project instance - index_data: The index data to save - - Raises: - ValueError: If failed to save the index - """ - if project.project_id is None: - raise ValueError("Project ID must be set") - - index_path = _get_index_path(project.project_id) - - try: - # Ensure project_id is in the index - index_data["project_id"] = project.project_id - - # Save as JSON with indentation for readability - content = json.dumps(index_data, indent=2, ensure_ascii=False) - - project.write_file(path=index_path, data=content) - logger.info("Saved file index for project %s", project.project_id) - - except Exception as e: - logger.error( - "Failed to save file index for project %s: %s", project.project_id, e - ) - raise ValueError(f"Failed to save file index: {str(e)}") from e - - -def _create_empty_index(project_id: str) -> dict[str, Any]: - """ - Create an empty file index structure. - - Args: - project_id: The project ID - - Returns: - A dictionary with empty index structure - """ - return { - "project_id": project_id, - "files": [], - "created_at": datetime.now().isoformat(), - "updated_at": datetime.now().isoformat(), - } - - -def add_file_to_index( - project: Project, - path: str, - size: int | None = None, -) -> dict[str, Any]: - """ - Add or update a file in the project index. - - Args: - project: GNS3 Project instance - path: The file path - size: Optional file size in bytes - - Returns: - The updated index data - - Raises: - ValueError: If failed to update the index - """ - try: - # Load existing index - index_data = load_file_index(project) - - # Check if file already exists - file_entry = next((f for f in index_data["files"] if f["path"] == path), None) - - now = datetime.now().isoformat() - - if file_entry: - # Update existing file - file_entry["updated_at"] = now - if size is not None: - file_entry["size"] = size - logger.info("Updated file in index: %s", path) - else: - # Add new file - new_entry: dict[str, Any] = { - "path": path, - "created_at": now, - "updated_at": None, - } - if size is not None: - new_entry["size"] = size - - index_data["files"].append(new_entry) - logger.info("Added file to index: %s", path) - - # Update index timestamp - index_data["updated_at"] = now - - # Save index - save_file_index(project, index_data) - - return index_data - - except Exception as e: - logger.error("Failed to add file to index: %s", e) - raise ValueError(f"Failed to add file to index: {str(e)}") from e - - -def get_file_list(project: Project) -> list[dict[str, Any]]: - """ - Get the list of files from the project index. - - Args: - project: GNS3 Project instance - - Returns: - A list of file entries - """ - try: - index_data = load_file_index(project) - files = index_data.get("files", []) - return files if isinstance(files, list) else [] - except Exception as e: - logger.error("Failed to get file list: %s", e) - return [] - - -if __name__ == "__main__": - # Test the file index manager - print("=" * 80) - print("Testing GNS3 File Index Manager") - print("=" * 80) - - try: - # Create connector and project - connector = get_gns3_connector() - if connector is None: - print("ERROR: Failed to create GNS3 connector") - exit(1) - - # Replace with actual project ID for testing - test_project_id = "your-project-uuid" - print(f"\nTesting with project: {test_project_id}") - print("NOTE: Replace 'your-project-uuid' with actual project ID") - - # Create project instance - project = Project(project_id=test_project_id, connector=connector) - - # Load index - print("\n1. Loading file index...") - index_data = load_file_index(project) - print(f" Index loaded: {len(index_data.get('files', []))} files") - - # Add file to index - print("\n2. Adding file to index...") - updated_index = add_file_to_index(project, "test_file.txt", size=1024) - print(f" File added. Total files: {len(updated_index['files'])}") - - # Get file list - print("\n3. Getting file list...") - file_list = get_file_list(project) - print(" Files in index:") - for file_entry in file_list: - print(f" - {file_entry['path']}") - - print("\n" + "=" * 80) - print("Test completed successfully!") - print("=" * 80) - - except Exception as e: - print(f"\nERROR: {e}") - import traceback - - traceback.print_exc() diff --git a/gns3server/agent/gns3_copilot/gns3_client/gns3_get_drawings.py b/gns3server/agent/gns3_copilot/gns3_client/gns3_get_drawings.py deleted file mode 100644 index fd51c46bc..000000000 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_get_drawings.py +++ /dev/null @@ -1,236 +0,0 @@ -""" -GNS3 drawings retrieval tool for managing graphical elements. - -Provides functionality to retrieve all drawings from a GNS3 project, -including their coordinates, SVG content, and other properties. -""" - -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 Project, get_gns3_connector - -# Configure logging -logger = logging.getLogger(__name__) - - -class GNS3GetDrawingsTool(BaseTool): - """ - A LangChain tool to retrieve all drawings from a GNS3 project. - - **Input:** - A JSON object containing the project_id. - - Example input: - { - "project_id": "uuid-of-project" - } - - **Output:** - A dictionary containing all drawings in the project. - Example output: - { - "project_id": "uuid-of-project", - "drawings": [ - { - "drawing_id": "uuid-of-drawing1", - "svg": "...", - "x": 100, - "y": 200, - "z": 0, - "locked": false, - "rotation": 0 - }, - { - "drawing_id": "uuid-of-drawing2", - "svg": "...", - "x": 300, - "y": 400, - "z": 1, - "locked": true, - "rotation": 90 - } - ], - "total_drawings": 2 - } - If an error occurs during input validation, returns a dictionary with an error message. - """ - - name: str = "get_gns3_drawings" - description: str = """ - Retrieves all drawings from a GNS3 project. - Input is a JSON object with project_id. - Example input: - { - "project_id": "uuid-of-project" - } - Returns a dictionary with all drawings in the project, including their drawing_id, - svg content, coordinates (x, y), z-index, locked status, and rotation. - If the operation fails, returns a dictionary with an error message. - """ - - def _run( - self, - tool_input: str, - run_manager: CallbackManagerForToolRun | None = None, - **kwargs: Any, - ) -> dict[str, Any]: - """ - Retrieves all drawings from a GNS3 project. - - Args: - tool_input (str): A JSON string containing project_id. - run_manager: LangChain run manager (unused). - - Returns: - dict: A dictionary with all drawings in the project 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") - - # Validate input - if not project_id: - logger.error("Invalid input: Missing project_id.") - return {"error": "Missing project_id."} - - # 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 project instance - logger.info("Retrieving drawings from project %s...", project_id) - project = Project(project_id=project_id, connector=gns3_server) - - # Get project drawings - project.get_drawings() - - # Prepare drawing list - drawings_list = [] - if project.drawings: - for drawing in project.drawings: - drawing_info = { - "drawing_id": drawing.get("drawing_id"), - "svg": drawing.get("svg"), - "x": drawing.get("x"), - "y": drawing.get("y"), - "z": drawing.get("z"), - "locked": drawing.get("locked"), - "rotation": drawing.get("rotation", 0), - } - drawings_list.append(drawing_info) - - # Prepare final result - final_result = { - "project_id": project_id, - "drawings": drawings_list, - "total_drawings": len(drawings_list), - } - - # Log the final result - logger.info( - "Retrieved %d drawings from project %s.", - len(drawings_list), - project_id, - ) - - # 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 retrieve drawings: %s", e) - return {"error": f"Failed to retrieve drawings: {str(e)}"} - - -if __name__ == "__main__": - # Test the tool locally - test_input = json.dumps( - { - "project_id": "d7fc094c-685e-4db1-ac11-5e33a1b2e066" - } # Replace with actual project UUID - ) - tool = GNS3GetDrawingsTool() - result = tool._run(test_input) - pprint(result) - - -""" -example output: -{'drawings': [{'drawing_id': 'e8202c5e-bd0a-447c-848c-15db2c9af2f0', - 'locked': False, - 'rotation': 0, - 'svg': '', - 'x': -376, - 'y': -381, - 'z': 1}, - {'drawing_id': '7d2f2411-efe8-4f3a-9882-cc0ad7798ee0', - 'locked': False, - 'rotation': 0, - 'svg': '', - 'x': -394, - 'y': 4, - 'z': 1}, - {'drawing_id': '264be3ab-002c-4b1d-886d-b4c323de845a', - 'locked': False, - 'rotation': 0, - 'svg': '哈哈哈\n' - '哈和', - 'x': -596, - 'y': -43, - 'z': 1}, - {'drawing_id': '44210a41-0100-46a0-8962-a73c72171c47', - 'locked': False, - 'rotation': 0, - 'svg': 'Area 0', - 'x': -573, - 'y': -272, - 'z': 1}, - {'drawing_id': 'efd1add9-a798-489d-af26-58b2d73c89dc', - 'locked': False, - 'rotation': 0, - 'svg': '', - 'x': -636, - 'y': -152, - 'z': 1}, - {'drawing_id': '89913ef3-1041-4b11-94e1-c5d65c3a52d4', - 'locked': False, - 'rotation': 0, - 'svg': '', - 'x': -891, - 'y': 66, - 'z': 1}], - 'project_id': '2245149a-71c8-4387-9d1f-441a683ef7e7', - 'total_drawings': 8} -""" diff --git a/gns3server/agent/gns3_copilot/gns3_client/gns3_get_nodes.py b/gns3server/agent/gns3_copilot/gns3_client/gns3_get_nodes.py deleted file mode 100644 index cd9024c68..000000000 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_get_nodes.py +++ /dev/null @@ -1,182 +0,0 @@ -""" -GNS3 node retrieval tool for device information. - -Provides functionality to retrieve all node instances from a project, -including node dimensions, symbol, and other attributes. -""" - -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 GNS3GetNodesTool(BaseTool): - """ - A LangChain tool to retrieve all node instances from a GNS3 project. - The tool connects to the GNS3 server and extracts comprehensive node information - including dimensions, symbol, coordinates, and other attributes. - - **Input:** - A JSON string with project_id parameter. - Example: '{"project_id": "uuid-of-project"}' - - **Output:** - A dictionary containing a list of dictionaries, each with comprehensive node information. - Example output: - { - "nodes": [ - { - "name": "R-1", - "node_id": "uuid1", - "node_type": "iou", - "height": 50, - "width": 50, - "symbol": ":/symbols/router.svg", - "x": -180, - "y": -30, - "z": 1, - "status": "started", - "ports": [ - {"name": "Ethernet0/0", "short_name": "e0/0", "adapter_number": 0, "port_number": 0}, - {"name": "Ethernet0/1", "short_name": "e0/1", "adapter_number": 0, "port_number": 1} - ] - } - ] - } - If an error occurs, returns a dictionary with an error message. - """ - - name: str = "get_gns3_nodes" - description: str = """ - Retrieves all node instances from a GNS3 project. - Requires a JSON string input with 'project_id' parameter. - Returns comprehensive node information including name, node_id, node_type, - height, width, symbol, coordinates (x, y, z), status, and simplified port information. - Example input: '{"project_id": "uuid-of-project"}' - Example output: - { - "nodes": [ - { - "name": "R-1", - "node_id": "uuid1", - "node_type": "iou", - "height": 50, - "width": 50, - "symbol": ":/symbols/router.svg", - "x": -180, - "y": -30, - "z": 1, - "status": "started", - "ports": [ - {"name": "Ethernet0/0", "short_name": "e0/0", "adapter_number": 0, "port_number": 0} - ] - } - ] - } - """ - - def _run( - self, - tool_input: str, - run_manager: CallbackManagerForToolRun | None = None, - ) -> dict[str, Any]: - """ - Connects to the GNS3 server and retrieves all nodes from the specified project. - - Args: - tool_input (str): JSON string containing project_id parameter. - run_manager: LangChain run manager (unused). - - Returns: - dict: A dictionary containing the list of nodes or an error message. - """ - try: - # Parse input JSON - input_data = json.loads(tool_input) - project_id = input_data.get("project_id") - - if not project_id: - logger.error("Missing project_id in input") - return {"error": "Missing required parameter: project_id"} - - # 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 nodes from the project - logger.info(f"Retrieving nodes from project: {project_id}") - nodes = gns3_server.get_nodes(project_id) - - # Simplify and organize node information - node_info = [] - for node in nodes: - # Simplify ports - keep essential information only - simplified_ports = [] - if node.get("ports"): - for port in node["ports"]: - simplified_ports.append( - { - "name": port.get("name", "N/A"), - "short_name": port.get("short_name", "N/A"), - "adapter_number": port.get("adapter_number", 0), - "port_number": port.get("port_number", 0), - } - ) - - # Extract node information - node_data = { - "name": node.get("name", "N/A"), - "node_id": node.get("node_id", "N/A"), - "node_type": node.get("node_type", "N/A"), - "height": node.get("height"), # Node height (read-only) - "width": node.get("width"), # Node width (read-only) - "symbol": node.get("symbol", "N/A"), - "x": node.get("x"), - "y": node.get("y"), - "z": node.get("z"), - "status": node.get("status", "N/A"), - "console": node.get("console"), - "console_type": node.get("console_type", "N/A"), - "ports": simplified_ports, - } - node_info.append(node_data) - - # Return JSON-formatted result - result = {"nodes": node_info} - logger.info( - "Node retrieval completed. Total nodes: %d.", - len(node_info), - ) - return result - - except json.JSONDecodeError as e: - logger.error("Failed to parse input JSON: %s", e) - return {"error": f"Invalid JSON input: {str(e)}"} - except Exception as e: - logger.error("Failed to connect to GNS3 server or retrieve nodes: %s", e) - return {"error": f"Failed to retrieve nodes: {str(e)}"} - - -if __name__ == "__main__": - # Test the tool locally - tool = GNS3GetNodesTool() - - # Example: Replace with your actual project_id - test_project_id = "d7fc094c-685e-4db1-ac11-5e33a1b2e066" - result = tool._run(f'{{"project_id": "{test_project_id}"}}') - pprint(result) diff --git a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_create.py b/gns3server/agent/gns3_copilot/gns3_client/gns3_project_create.py deleted file mode 100644 index bc530aee7..000000000 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_create.py +++ /dev/null @@ -1,167 +0,0 @@ -import logging -from typing import Any - -from langchain.tools import BaseTool - -from gns3_copilot.gns3_client import Project, get_gns3_connector - -# Configure logging -logger = logging.getLogger(__name__) - - -class GNS3ProjectCreate(BaseTool): - """ - Tool to create a new GNS3 project. - - This tool connects to GNS3 server and creates a new project with the specified - name and optional configuration parameters. - """ - - name: str = "create_gns3_project" - description: str = """ - Creates a new GNS3 project with the specified name and optional parameters. - - Input: A dictionary with project parameters. - - name: The name of the project to create (required) - - auto_start: Automatically start the project when opened (optional, default: False) - - auto_close: Automatically close the project when client disconnects (optional, default: False) - - auto_open: Automatically open the project when GNS3 starts (optional, default: False) - - scene_width: Width of the drawing area in pixels (optional) - - scene_height: Height of the drawing area in pixels (optional) - - Example input: - { - "name": "my_new_project", - "auto_start": false, - "auto_close": false, - "auto_open": false - } - - Returns: A dictionary with project creation status and details: - - success: Whether the operation succeeded - - project: Project details (name, project_id, status, etc.) - - message: Status message - - Example output: - { - "success": true, - "project": { - "project_id": "ff8e059c-c33d-47f4-bc11-c7dda8a1d500", - "name": "my_new_project", - "status": "opened" - }, - "message": "Project 'my_new_project' created successfully" - } - """ - - def _run(self, tool_input: Any = None, run_manager: Any = None) -> dict: - """ - Execute the project creation operation. - - Args: - tool_input: Dictionary containing project parameters - run_manager: Run manager for tool execution (optional) - - Returns: - Dictionary with operation result and project details - """ - # Log received input - logger.info("Received input: %s", tool_input) - - try: - # Validate input - if not tool_input or "name" not in tool_input: - return { - "success": False, - "error": "Missing required parameter: name", - } - - name = tool_input["name"] - - # Validate project name is not empty - if not name or not isinstance(name, str) or not name.strip(): - return { - "success": False, - "error": "Project name must be a non-empty string", - } - - # Get optional parameters - auto_start = tool_input.get("auto_start", False) - auto_close = tool_input.get("auto_close", False) - auto_open = tool_input.get("auto_open", False) - scene_width = tool_input.get("scene_width") - scene_height = tool_input.get("scene_height") - - # Initialize Gns3Connector using factory function - logger.info("Connecting to GNS3 server...") - server = get_gns3_connector() - - if server is None: - logger.error("Failed to create GNS3 connector") - return { - "success": False, - "error": "Failed to connect to GNS3 server. Please check your configuration.", - } - - # Create project instance with specified parameters - project_params = { - "name": name, - "auto_start": auto_start, - "auto_close": auto_close, - "auto_open": auto_open, - } - - # Add optional scene parameters if provided - if scene_width is not None: - project_params["scene_width"] = scene_width - if scene_height is not None: - project_params["scene_height"] = scene_height - - project = Project(connector=server, **project_params) - - # Create the project - project.create() - - # Verify project was created successfully - if not project.project_id: - return { - "success": False, - "error": "Failed to create project: project_id not returned", - } - - logger.info( - "Project created successfully: %s (ID: %s)", - project.name, - project.project_id, - ) - - # Prepare result - result = { - "success": True, - "project": { - "project_id": project.project_id, - "name": project.name, - "status": project.status, - "path": project.path, - }, - "message": f"Project '{project.name}' created successfully", - } - - # Log result - logger.info("Project creation result: %s", result) - - # Return success with project details - return result - - except ValueError as e: - logger.error("Validation error creating GNS3 project: %s", str(e)) - return { - "success": False, - "error": f"Validation error: {str(e)}", - } - except Exception as e: - logger.error("Error creating GNS3 project: %s", str(e)) - return { - "success": False, - "error": f"Failed to create GNS3 project: {str(e)}", - } diff --git a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_delete.py b/gns3server/agent/gns3_copilot/gns3_client/gns3_project_delete.py deleted file mode 100644 index f8b186992..000000000 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_delete.py +++ /dev/null @@ -1,152 +0,0 @@ -import logging -from typing import Any - -from langchain.tools import BaseTool - -from gns3_copilot.gns3_client import Project, get_gns3_connector - -# Configure logging -logger = logging.getLogger(__name__) - - -class GNS3ProjectDelete(BaseTool): - """ - Tool to delete a GNS3 project. - - This tool connects to GNS3 server and deletes an existing project, - optionally retrieving project details before deletion. - """ - - name: str = "delete_gns3_project" - description: str = """ - Deletes an existing GNS3 project. - - Input parameters: - Required: - - project_id: The UUID of the project to delete OR - - name: The name of the project to delete (one must be provided) - - Returns: Project deletion status and detailed information including: - - success: Whether the operation succeeded - - project: Deleted project details (name, project_id, status, etc.) - - message: Status message - - Example output: - { - "success": true, - "project": { - "project_id": "ff8e059c-c33d-47f4-bc11-c7dda8a1d500", - "name": "my_project", - "status": "closed" - }, - "message": "Project 'my_project' deleted successfully" - } - """ - - def _run(self, tool_input: Any = None, run_manager: Any = None) -> dict: - """ - Execute the project deletion operation. - - Args: - tool_input: Dictionary containing project identifier - run_manager: Run manager for tool execution (optional) - - Returns: - Dictionary with operation result and deleted project details - """ - # Log received input - logger.info("Received input: %s", tool_input) - - try: - # Validate input - if not tool_input: - return { - "success": False, - "error": "No input provided", - } - - # Check for project identifier - project_id = tool_input.get("project_id") - project_name = tool_input.get("name") - - if not project_id and not project_name: - return { - "success": False, - "error": "Missing required parameter: either 'project_id' or 'name' must be provided", - } - - # Initialize Gns3Connector using factory function - logger.info("Connecting to GNS3 server...") - server = get_gns3_connector() - - if server is None: - logger.error("Failed to create GNS3 connector") - return { - "success": False, - "error": "Failed to connect to GNS3 server. Please check your configuration.", - } - - # Create project instance - if project_id: - project = Project(project_id=project_id, connector=server) - else: - project = Project(name=project_name, connector=server) - - # Get project information before deletion - project.get(get_nodes=False, get_links=False, get_stats=False) - - # Verify project was found - if not project.project_id: - if project_name: - return { - "success": False, - "error": f"Project with name '{project_name}' not found", - } - else: - return { - "success": False, - "error": f"Project with ID '{project_id}' not found", - } - - # Store project details for response - project_details = { - "project_id": project.project_id, - "name": project.name, - "status": project.status, - "path": project.path, - } - - # Delete the project - project.delete() - - logger.info( - "Project deleted successfully: %s (ID: %s)", - project.name, - project.project_id, - ) - - # Prepare result - result = { - "success": True, - "project": project_details, - "message": f"Project '{project.name}' deleted successfully", - } - - # Log result - logger.info("Project deletion result: %s", result) - - # Return success with project details - return result - - except ValueError as e: - logger.error("Validation error deleting GNS3 project: %s", str(e)) - return { - "success": False, - "error": f"Validation error: {str(e)}", - } - except Exception as e: - logger.error("Error deleting GNS3 project: %s", str(e)) - return { - "success": False, - "error": f"Failed to delete GNS3 project: {str(e)}", - } diff --git a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_list_files.py b/gns3server/agent/gns3_copilot/gns3_client/gns3_project_list_files.py deleted file mode 100644 index 5c23ae44e..000000000 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_list_files.py +++ /dev/null @@ -1,239 +0,0 @@ -""" -GNS3 Project List Files Tool - -This module provides a LangChain tool to list files in a GNS3 project. -It uses a file index stored in the project directory to track files. -""" - -import json -import logging -import re - -import requests -from langchain.tools import BaseTool - -from gns3_copilot.gns3_client import Project, get_gns3_connector -from gns3_copilot.gns3_client.gns3_file_index import load_file_index - -# Configure logging -logger = logging.getLogger(__name__) - -# Load environment variables - - -class GNS3ProjectListFilesTool(BaseTool): - """ - Tool to list files in a GNS3 project. - - This tool reads a file index maintained in the project directory - and returns a list of all tracked files. - - Input format (JSON string): - { - "project_id": "uuid-of-project", - "pattern": "*.cfg" # Optional: filter by file pattern - } - - Returns: - JSON string with file list: - { - "project_id": "uuid-of-project", - "files": [ - { - "path": "README.md", - "created_at": "2026-01-01T10:00:00", - "updated_at": "2026-01-01T10:05:00", - "size": 1024 - } - ], - "total_count": 1, - "status": "success" - } - """ - - name: str = "gns3_project_list_files" - description: str = """List files in a GNS3 project using a file index. - - This tool reads a file index (.gns3_copilot_file_index.json) that tracks - all files written to the project. It returns a list of file paths and metadata. - - Args: - project_id (required): The UUID of the GNS3 project - pattern (optional): A glob pattern to filter files (e.g., "*.cfg", "configs/*") - - Returns: - A list of files with their metadata - """ - - def _run(self, tool_input: str) -> str: - """ - Execute the tool to list files in a project. - - Args: - tool_input: JSON string containing project_id and optional pattern - - Returns: - JSON string containing the file list - """ - try: - # Parse input JSON - input_data = json.loads(tool_input) - project_id = input_data.get("project_id") - pattern = input_data.get("pattern") - - # Validate input - if not project_id: - logger.error("Invalid input: Missing project_id.") - return json.dumps({"error": "Missing project_id."}) - - # Validate UUID format (basic check) - uuid_pattern = r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$" - if not re.match(uuid_pattern, project_id): - logger.error("Invalid project_id format: %s", project_id) - return json.dumps({"error": f"Invalid project_id format: {project_id}"}) - - # Create connector - connector = get_gns3_connector() - if connector is None: - logger.error("Failed to create GNS3 connector") - return json.dumps( - { - "project_id": project_id, - "status": "failed", - "error": "Failed to create GNS3 connector. Check server configuration.", - } - ) - - # Create project instance - project = Project(project_id=project_id, connector=connector) - - # Load file index - logger.info("Loading file index for project: %s", project_id) - index_data = load_file_index(project) - - # Get file list - files = index_data.get("files", []) - - # Apply pattern filter if provided - if pattern: - filtered_files = [] - for file_entry in files: - if self._matches_pattern(file_entry["path"], pattern): - filtered_files.append(file_entry) - files = filtered_files - logger.info( - "Filtered %d files matching pattern: %s", len(files), pattern - ) - - # Prepare result - result = { - "project_id": project_id, - "files": files, - "total_count": len(files), - "status": "success", - } - - logger.info( - "Successfully listed %d files for project: %s", len(files), project_id - ) - - return json.dumps(result, indent=2) - - except requests.HTTPError as e: - logger.error("HTTP error listing files: %s", e) - return json.dumps( - { - "project_id": project_id if project_id else "unknown", - "status": "failed", - "error": f"HTTP error: {str(e)}", - } - ) - except json.JSONDecodeError as e: - logger.error("Invalid JSON input: %s", e) - return json.dumps({"error": f"Invalid JSON input: {str(e)}"}) - except Exception as e: - logger.error("Unexpected error listing files: %s", e) - return json.dumps( - { - "project_id": project_id if project_id else "unknown", - "status": "failed", - "error": f"Failed to list files: {str(e)}", - } - ) - - def _matches_pattern(self, path: str, pattern: str) -> bool: - """ - Check if a file path matches a glob pattern. - - Args: - path: The file path to check - pattern: The glob pattern (e.g., "*.cfg", "configs/*") - - Returns: - True if the path matches the pattern - """ - # Convert glob pattern to regex - regex_pattern = pattern.replace(".", r"\.").replace("*", ".*").replace("?", ".") - return bool(re.match(regex_pattern, path)) - - -if __name__ == "__main__": - # Test the tool - print("=" * 80) - print("Testing GNS3ProjectListFilesTool") - print("=" * 80) - - try: - tool = GNS3ProjectListFilesTool() - - # Replace with actual project ID for testing - test_project_id = ( - "1445a4ba-4635-430b-a332-bef438f65932" # Replace with actual project UUID - ) - print(f"\nTesting with project: {test_project_id}") - print("NOTE: Replace 'your-project-uuid' with actual project ID") - - # Test 1: List all files - print("\n1. Listing all files...") - result1 = tool._run( - json.dumps( - { - "project_id": test_project_id, - } - ) - ) - result_data1 = json.loads(result1) - print(f" Status: {result_data1.get('status')}") - print(f" Total files: {result_data1.get('total_count', 0)}") - if result_data1.get("files"): - print(" Files:") - for file_entry in result_data1["files"]: - print(f" - {file_entry['path']}") - - # Test 2: List files with pattern - print("\n2. Listing files matching pattern '*.cfg'...") - result2 = tool._run( - json.dumps({"project_id": test_project_id, "pattern": "*.cfg"}) - ) - result_data2 = json.loads(result2) - print(f" Status: {result_data2.get('status')}") - print(f" Matching files: {result_data2.get('total_count', 0)}") - if result_data2.get("files"): - print(" Files:") - for file_entry in result_data2["files"]: - print(f" - {file_entry['path']}") - - # Test 3: Missing project_id - print("\n3. Testing with missing project_id...") - result3 = tool._run(json.dumps({})) - print(f" Result: {result3}") - - print("\n" + "=" * 80) - print("Test completed successfully!") - print("=" * 80) - - except Exception as e: - print(f"\nERROR: {e}") - import traceback - - traceback.print_exc() diff --git a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_lock.py b/gns3server/agent/gns3_copilot/gns3_client/gns3_project_lock.py deleted file mode 100644 index 5ee4e46b5..000000000 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_lock.py +++ /dev/null @@ -1,305 +0,0 @@ -""" -GNS3 project lock/unlock tool for project management. - -Provides functionality to lock, unlock, or check the lock status of GNS3 projects. -Locking a project prevents accidental modifications to drawings and nodes. -""" - -import logging -from typing import Any - -from langchain.tools import BaseTool - -from gns3_copilot.gns3_client import Project, get_gns3_connector - -# Configure logging -logger = logging.getLogger(__name__) - - -class GNS3ProjectLock(BaseTool): - """ - Tool to lock, unlock, or check the lock status of a GNS3 project. - - This tool connects to GNS3 server and performs lock/unlock operations - on the specified project, or retrieves the current lock status. - - IMPORTANT: Project lock/unlock operations are only supported in GNS3 API v3. - Using this tool with GNS3 API v2 will result in an error. - - Supported operations: - - "locked": Check if the project is currently locked - - "lock": Lock all drawings and nodes in the project - - "unlock": Unlock all drawings and nodes in the project - """ - - name: str = "gns3_project_lock" - description: str = """ - Lock, unlock, or check the lock status of a GNS3 project. - - IMPORTANT: This tool only works with GNS3 API v3. It is not supported in API v2. - - Input parameters: - - project_id: The unique UUID identifier of the project (required) - - operation: The operation to perform - "locked", "lock", or "unlock" (required) - - Operations: - - "locked": Returns the current lock status of the project - - "lock": Locks all drawings and nodes in the project - - "unlock": Unlocks all drawings and nodes in the project - - Returns: Dictionary with operation result including: - - success: Whether the operation succeeded - - operation: The operation performed - - project_id: The project ID - - locked_status: Current lock status (for "locked" operation) - - message: Status message - - error: Error message if operation failed - - Example output (check locked status): - { - "success": true, - "operation": "locked", - "project_id": "ff8e059c-c33d-47f4-bc11-c7dda8a1d500", - "locked_status": false, - "message": "Project is currently unlocked" - } - - Example output (lock project): - { - "success": true, - "operation": "lock", - "project_id": "ff8e059c-c33d-47f4-bc11-c7dda8a1d500", - "message": "Project locked successfully" - } - - Example output (unlock project): - { - "success": true, - "operation": "unlock", - "project_id": "ff8e059c-c33d-47f4-bc11-c7dda8a1d500", - "message": "Project unlocked successfully" - } - - Example output (API v2 not supported): - { - "success": false, - "operation": "lock", - "project_id": "ff8e059c-c33d-47f4-bc11-c7dda8a1d500", - "error": "Project lock/unlock operations are only supported in GNS3 API v3. Current API version: v2" - } - """ - - def _run(self, tool_input: Any = None, run_manager: Any = None) -> dict: - """ - Execute the project lock/unlock operation. - - Args: - tool_input: Dictionary containing project_id and operation - run_manager: Run manager for tool execution (optional) - - Returns: - Dictionary with operation result - """ - # Log received input - logger.info("Received input: %s", tool_input) - - try: - # Validate input - if not tool_input or "project_id" not in tool_input: - return { - "success": False, - "error": "Missing required parameter: project_id", - } - - if "operation" not in tool_input: - return { - "success": False, - "error": "Missing required parameter: operation", - } - - project_id = tool_input["project_id"] - operation = tool_input["operation"] - - # Validate operation type - valid_operations = ["locked", "lock", "unlock"] - if operation not in valid_operations: - return { - "success": False, - "error": f"Invalid operation '{operation}'. " - f"Must be one of: {', '.join(valid_operations)}", - } - - # Initialize Gns3Connector using factory function - logger.info("Connecting to GNS3 server...") - server = get_gns3_connector() - - if server is None: - logger.error("Failed to create GNS3 connector") - return { - "success": False, - "error": "Failed to connect to GNS3 server. " - "Please check your configuration.", - } - - # Create project instance and retrieve project details - project = Project(project_id=project_id, connector=server) - project.get(get_nodes=False, get_links=False, get_stats=False) - - # Check if project was found - if not project.name: - return { - "success": False, - "error": f"Project with ID '{project_id}' not found", - } - - # Perform the requested operation - result: dict[str, Any] = { - "success": False, - "operation": operation, - "project_id": project_id, - } - - if operation == "locked": - # Get current lock status - logger.info("Getting lock status for project %s", project_id) - try: - locked_status = project.get_locked() - result.update( - { - "success": True, - "locked_status": locked_status, - "message": f"Project is currently {'locked' if locked_status else 'unlocked'}", - } - ) - logger.info("Project lock status: %s", locked_status) - except ValueError as e: - # Handle API version incompatibility - error_msg = str(e) - if "only supported in GNS3 API v3" in error_msg: - logger.error("API v2 not supported: %s", error_msg) - result.update( - { - "error": error_msg, - "message": "Project lock/unlock operations require GNS3 API v3. " - "Please upgrade your GNS3 server or use v3 API.", - } - ) - else: - logger.error("Error getting lock status: %s", error_msg) - result.update( - { - "error": error_msg, - "message": f"Failed to get lock status: {error_msg}", - } - ) - - elif operation == "lock": - # Lock the project - logger.info("Locking project %s", project_id) - try: - project.lock_project() - result.update( - { - "success": True, - "message": f"Project '{project.name}' locked successfully", - } - ) - logger.info("Project locked successfully") - except ValueError as e: - # Handle API version incompatibility - error_msg = str(e) - if "only supported in GNS3 API v3" in error_msg: - logger.error("API v2 not supported: %s", error_msg) - result.update( - { - "error": error_msg, - "message": "Project lock/unlock operations require GNS3 API v3. " - "Please upgrade your GNS3 server or use v3 API.", - } - ) - else: - logger.error("Error locking project: %s", error_msg) - result.update( - { - "error": error_msg, - "message": f"Failed to lock project: {error_msg}", - } - ) - - elif operation == "unlock": - # Unlock the project - logger.info("Unlocking project %s", project_id) - try: - project.unlock_project() - result.update( - { - "success": True, - "message": f"Project '{project.name}' unlocked successfully", - } - ) - logger.info("Project unlocked successfully") - except ValueError as e: - # Handle API version incompatibility - error_msg = str(e) - if "only supported in GNS3 API v3" in error_msg: - logger.error("API v2 not supported: %s", error_msg) - result.update( - { - "error": error_msg, - "message": "Project lock/unlock operations require GNS3 API v3. " - "Please upgrade your GNS3 server or use v3 API.", - } - ) - else: - logger.error("Error unlocking project: %s", error_msg) - result.update( - { - "error": error_msg, - "message": f"Failed to unlock project: {error_msg}", - } - ) - - # Log result - logger.info("Project lock operation result: %s", result) - - return result - - except Exception as e: - logger.error("Error performing lock operation on GNS3 project: %s", str(e)) - return { - "success": False, - "error": f"Failed to perform lock operation on GNS3 project: {str(e)}", - } - - -if __name__ == "__main__": - # Test the tool locally - print("=== Testing GNS3ProjectLock Tool ===") - - # Test 1: Check lock status - print("\n--- Test 1: Check lock status ---") - test_input_locked = { - "project_id": "0c0fde25-6ead-4413-a283-ea8fd2324291", # Replace with actual project UUID - "operation": "locked", - } - tool = GNS3ProjectLock() - result_locked = tool._run(test_input_locked) - print(f"Result: {result_locked}") - - # Test 2: Lock project - print("\n--- Test 2: Lock project ---") - test_input_lock = { - "project_id": "0c0fde25-6ead-4413-a283-ea8fd2324291", # Replace with actual project UUID - "operation": "lock", - } - result_lock = tool._run(test_input_lock) - print(f"Result: {result_lock}") - - # Test 3: Unlock project - print("\n--- Test 3: Unlock project ---") - test_input_unlock = { - "project_id": "0c0fde25-6ead-4413-a283-ea8fd2324291", # Replace with actual project UUID - "operation": "unlock", - } - result_unlock = tool._run(test_input_unlock) - print(f"Result: {result_unlock}") diff --git a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_open.py b/gns3server/agent/gns3_copilot/gns3_client/gns3_project_open.py deleted file mode 100644 index 0a5c2777f..000000000 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_open.py +++ /dev/null @@ -1,158 +0,0 @@ -import logging -from typing import Any - -from langchain.tools import BaseTool - -from gns3_copilot.gns3_client import Project, get_gns3_connector - -# Configure logging -logger = logging.getLogger(__name__) - - -class GNS3ProjectOpen(BaseTool): - """ - Tool to open or close a GNS3 project by project_id. - - This tool connects to GNS3 server and opens or closes the specified project. - It returns the project status and details after the operation. - """ - - name: str = "open_gns3_project" - description: str = """ - Opens or closes a GNS3 project by its project_id. - - Input parameters: - - project_id: The unique UUID identifier of project (required) - - open: Set to True to open the project (default: False) - - close: Set to True to close the project (default: False) - - Note: Exactly one of 'open' or 'close' must be set to True. - - Returns: Project operation status and detailed information including: - - success: Whether the operation succeeded - - operation: "open" or "close" - - project: Project details (name, project_id, status, etc.) - - message: Status message - - Example output (open): - { - "success": true, - "operation": "open", - "project": { - "project_id": "ff8e059c-c33d-47f4-bc11-c7dda8a1d500", - "name": "mylab", - "status": "opened" - }, - "message": "Project 'mylab' opened successfully" - } - - Example output (close): - { - "success": true, - "operation": "close", - "project": { - "project_id": "ff8e059c-c33d-47f4-bc11-c7dda8a1d500", - "name": "mylab", - "status": "closed" - }, - "message": "Project 'mylab' closed successfully" - } - """ - - def _run(self, tool_input: Any = None, run_manager: Any = None) -> dict: - """ - Execute project open or close operation. - - Args: - tool_input: Dictionary containing project_id, open, or close - run_manager: Run manager for tool execution (optional) - - Returns: - Dictionary with operation result and project details - """ - # Log received input - logger.info("Received input: %s", tool_input) - - try: - # Validate input - if not tool_input or "project_id" not in tool_input: - return { - "success": False, - "error": "Missing required parameter: project_id", - } - - project_id = tool_input["project_id"] - should_open = tool_input.get("open", False) - should_close = tool_input.get("close", False) - - # Validate that exactly one of open or close is True - if should_open and should_close: - return { - "success": False, - "error": "Cannot set both 'open' and 'close' to True. " - "Please specify only one operation.", - } - - if not should_open and not should_close: - return { - "success": False, - "error": "Either 'open' or 'close' must be set to True.", - } - - # Initialize Gns3Connector using factory function - logger.info("Connecting to GNS3 server...") - server = get_gns3_connector() - - if server is None: - logger.error("Failed to create GNS3 connector") - return { - "success": False, - "error": "Failed to connect to GNS3 server. Please check your configuration.", - } - - # Create project instance and retrieve project details - project = Project(project_id=project_id, connector=server) - project.get(get_nodes=False, get_links=False, get_stats=False) - - # Check if project was found - if not project.name: - return { - "success": False, - "error": f"Project with ID '{project_id}' not found", - } - - # Perform the requested operation - operation = None - if should_open: - project.open() - operation = "open" - action_message = "opened" - else: # should_close - project.close() - operation = "close" - action_message = "closed" - - # Prepare result - result = { - "success": True, - "operation": operation, - "project": { - "project_id": project.project_id, - "name": project.name, - "status": project.status, - }, - "message": f"Project '{project.name}' {action_message} successfully", - } - - # Log result - logger.info("Project operation result: %s", result) - - # Return success with project details - return result - - except Exception as e: - logger.error("Error operating on GNS3 project: %s", str(e)) - return { - "success": False, - "error": f"Failed to operate on GNS3 project: {str(e)}", - } diff --git a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_path.py b/gns3server/agent/gns3_copilot/gns3_client/gns3_project_path.py deleted file mode 100644 index 2102d4df1..000000000 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_path.py +++ /dev/null @@ -1,160 +0,0 @@ -""" -GNS3 Project Path Tool. - -This module provides a LangChain tool for retrieving GNS3 project paths. -These paths are needed for features like Notes to store markdown files -in the project directory. - -Usage: - from gns3_copilot.gns3_client.gns3_project_path import GNS3ProjectPath - tool = GNS3ProjectPath() - result = tool._run({"project_name": "mylab", "project_id": "uuid"}) -""" - -import logging -from typing import Any - -from langchain.tools import BaseTool - -from gns3_copilot.gns3_client import Project, get_gns3_connector - -# Configure logging -logger = logging.getLogger(__name__) - - -class GNS3ProjectPath(BaseTool): - """ - Tool to retrieve the local filesystem path of a GNS3 project. - - This tool connects to GNS3 server and retrieves the local path where - project files are stored on the server. Useful for features that need to - access project-specific files such as Notes. - - Input parameters: - - project_name: Name of the GNS3 project (required) - - project_id: The unique UUID identifier of project (required) - - Returns: Project path information including: - - success: Whether the operation succeeded - - project_path: Full path to the project directory on the server - - project_name: Name of the project - - project_id: UUID of the project - - message: Status message - - Example output (success): - { - "success": true, - "project_path": "/home/user/GNS3/projects/mylab", - "project_name": "mylab", - "project_id": "ff8e059c-c33d-47f4-bc11-c7dda8a1d500", - "message": "Successfully retrieved project path" - } - - Example output (error): - { - "success": false, - "error": "Project with ID 'xxx' not found" - } - """ - - name: str = "get_gns3_project_path" - description: str = """ - Retrieves the local filesystem path for a GNS3 project. - - Input parameters: - - project_name: Name of the GNS3 project (required) - - project_id: The unique UUID identifier of project (required) - - This tool is useful for accessing project-specific files and directories - on the GNS3 server, such as storing notes or configuration files. - - Returns: Project path information including success status, path, - project details, and status message. - """ - - def _run(self, tool_input: Any = None, run_manager: Any = None) -> dict: - """ - Execute the project path retrieval operation. - - Args: - tool_input: Dictionary containing project_name and project_id - run_manager: Run manager for tool execution (optional) - - Returns: - Dictionary with operation result and project path details - """ - # Log received input - logger.info("Received input: %s", tool_input) - - try: - # Validate input - if not tool_input: - return { - "success": False, - "error": "Missing required parameters: project_name and project_id", - } - - project_name = tool_input.get("project_name") - project_id = tool_input.get("project_id") - - if not project_name or not project_id: - return { - "success": False, - "error": "Both project_name and project_id are required parameters", - } - - # Initialize Gns3Connector using factory function - logger.info("Connecting to GNS3 server...") - server = get_gns3_connector() - - if server is None: - logger.error("Failed to create GNS3 connector") - return { - "success": False, - "error": "Failed to connect to GNS3 server. Please check your configuration.", - } - - # Create project instance and retrieve project details - project = Project(project_id=project_id, connector=server) - project.get(get_nodes=False, get_links=False, get_stats=False) - - # Check if project was found - if not project.name: - return { - "success": False, - "error": f"Project with ID '{project_id}' not found", - } - - # Verify project name matches - if project.name != project_name: - logger.warning( - "Project name mismatch: expected '%s', got '%s'", - project_name, - project.name, - ) - # Continue anyway as project_id is more authoritative - - # Return project path if available - if project.path: - result = { - "success": True, - "project_path": project.path, - "project_name": project.name, - "project_id": project.project_id, - "message": f"Successfully retrieved project path for '{project.name}'", - } - # Log result - logger.info("Project path retrieval result: %s", result) - return result - else: - return { - "success": False, - "error": f"Project '{project_name}' exists but has no path attribute", - } - - except Exception as e: - logger.error("Error retrieving project path: %s", str(e)) - return { - "success": False, - "error": f"Failed to retrieve project path: {str(e)}", - } diff --git a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_read_file.py b/gns3server/agent/gns3_copilot/gns3_client/gns3_project_read_file.py deleted file mode 100644 index 64c876976..000000000 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_read_file.py +++ /dev/null @@ -1,191 +0,0 @@ -""" -GNS3 Project Read File Tool - -Provides a LangChain tool for reading files in GNS3 projects. -This tool directly uses the Project.get_file() method from custom_gns3fy. -""" - -import json -import logging -import re -from typing import Any - -from langchain.tools import BaseTool -from langchain_core.callbacks import CallbackManagerForToolRun - -from gns3_copilot.gns3_client import Project, get_gns3_connector - -# Configure logging -logger = logging.getLogger(__name__) - - -class GNS3ProjectReadFileTool(BaseTool): - """ - A LangChain tool to read files from a GNS3 project. - - This tool reads the content of a file located in a GNS3 project directory. - It uses the Project.get_file() method from custom_gns3fy. - - **Input:** - A JSON object containing the project_id and the file path. - - Example input: - { - "project_id": "uuid-of-project", - "path": "README.md" - } - - **Output:** - A dictionary containing the project_id, file path, content, and status. - - Example output: - { - "project_id": "uuid-of-project", - "path": "README.md", - "content": "This is the file content...", - "status": "success" - } - - If an error occurs, returns a dictionary with an error message. - - Example error output: - { - "project_id": "uuid-of-project", - "path": "README.md", - "status": "failed", - "error": "File not found" - } - """ - - name: str = "gns3_project_read_file" - description: str = """ - Reads a file from a GNS3 project directory. - Input is a JSON object with project_id and the file path. - Example input: - { - "project_id": "uuid-of-project", - "path": "README.md" - } - Returns a dictionary with the file content and status. - If the operation fails, returns a dictionary with an error message. - """ - - def _run( - self, - tool_input: str, - run_manager: CallbackManagerForToolRun | None = None, - **kwargs: Any, - ) -> dict[str, Any]: - """ - Reads a file from a GNS3 project. - - Args: - tool_input (str): A JSON string containing project_id and file path. - run_manager: LangChain run manager (unused). - - Returns: - dict: A dictionary with the file content and status, 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") - path = input_data.get("path") - - # Validate input - if not project_id: - logger.error("Invalid input: Missing project_id.") - return {"error": "Missing project_id."} - - if not path: - logger.error("Invalid input: Missing path.") - return {"error": "Missing path."} - - # Validate project_id format (UUID) - 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} - - # Get connector using factory function - connector = get_gns3_connector() - if connector is None: - error_msg = "Failed to create GNS3 connector. Check configuration." - logger.error(error_msg) - return {"error": error_msg} - - logger.info("Connecting to GNS3 server...") - - # Create Project instance - project = Project(project_id=project_id, connector=connector) - - # Read file content - logger.info("Reading file '%s' from project '%s'...", path, project_id) - content = project.get_file(path=path) - - # Prepare successful result - result = { - "project_id": project_id, - "path": path, - "content": content, - "status": "success", - } - - logger.info( - "Successfully read file '%s' from project '%s'", path, project_id - ) - - return result - - except json.JSONDecodeError as e: - logger.error("Invalid JSON input: %s", e) - return {"error": f"Invalid JSON input: {e}"} - except ValueError as e: - logger.error("Value error: %s", e) - return {"error": f"Value error: {e}"} - except Exception as e: - logger.error("Failed to read file from project: %s", e) - return { - "project_id": project_id if "project_id" in locals() else None, - "path": path if "path" in locals() else None, - "status": "failed", - "error": f"Failed to read file: {str(e)}", - } - - 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)) - - -if __name__ == "__main__": - import pprint - - print("=" * 80) - print("Testing GNS3ProjectReadFileTool") - print("=" * 80) - - read_test_input = json.dumps( - { - "project_id": "1445a4ba-4635-430b-a332-bef438f65932", # Replace with actual project UUID - "path": "test_write.txt", - } - ) - - read_tool = GNS3ProjectReadFileTool() - read_result = read_tool._run(read_test_input) - print("\nRead Result:") - pprint.pprint(read_result) diff --git a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_update.py b/gns3server/agent/gns3_copilot/gns3_client/gns3_project_update.py deleted file mode 100644 index 9deaf3218..000000000 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_update.py +++ /dev/null @@ -1,224 +0,0 @@ -import logging -from typing import Any - -from langchain.tools import BaseTool - -from gns3_copilot.gns3_client import Project, get_gns3_connector - -# Configure logging -logger = logging.getLogger(__name__) - - -class GNS3ProjectUpdate(BaseTool): - """ - Tool to update an existing GNS3 project configuration. - - This tool connects to GNS3 server and updates project settings including - auto-start options, scene dimensions, display options, and other parameters. - """ - - name: str = "update_gns3_project" - description: str = """ - Updates an existing GNS3 project configuration. - - Input parameters: - Required: - - project_id: The UUID of the project to update OR - - name: The name of the project to update (one must be provided) - - Optional - Auto control options: - - auto_start: Automatically start the project when opened (default: keep current) - - auto_close: Automatically close the project when client disconnects (default: keep current) - - auto_open: Automatically open the project when GNS3 starts (default: keep current) - - Optional - Scene settings: - - scene_width: Width of the drawing area in pixels (default: keep current) - - scene_height: Height of the drawing area in pixels (default: keep current) - - grid_size: Grid size for the drawing area for nodes (default: keep current) - - drawing_grid_size: Grid size for the drawing area for drawings (default: keep current) - - Optional - Display options: - - show_grid: Show the grid on the drawing area (default: keep current) - - show_interface_labels: Show interface labels on the drawing area (default: keep current) - - show_layers: Show layers on the drawing area (default: keep current) - - snap_to_grid: Snap to grid on the drawing area (default: keep current) - - zoom: Zoom of the drawing area (default: keep current) - - Returns: Project update status and detailed information including: - - success: Whether the operation succeeded - - project: Updated project details (name, project_id, status, settings, etc.) - - updated_fields: List of fields that were updated - - message: Status message - - Example output: - { - "success": true, - "project": { - "project_id": "ff8e059c-c33d-47f4-bc11-c7dda8a1d500", - "name": "my_project", - "status": "opened", - "auto_start": true, - "auto_close": false, - "auto_open": false - }, - "updated_fields": ["auto_start"], - "message": "Project 'my_project' updated successfully" - } - """ - - def _run(self, tool_input: Any = None, run_manager: Any = None) -> dict: - """ - Execute project update operation. - - Args: - tool_input: Dictionary containing project identifier and update parameters - run_manager: Run manager for tool execution (optional) - - Returns: - Dictionary with operation result and updated project details - """ - # Log received input - logger.info("Received input: %s", tool_input) - - try: - # Validate input - if not tool_input: - return { - "success": False, - "error": "No input provided", - } - - # Check for project identifier - project_id = tool_input.get("project_id") - project_name = tool_input.get("name") - - if not project_id and not project_name: - return { - "success": False, - "error": "Missing required parameter: either 'project_id' or 'name' must be provided", - } - - # Remove project_id and name from update parameters - update_params = { - k: v - for k, v in tool_input.items() - if k not in ("project_id", "name") and v is not None - } - - # Check if there's anything to update - if not update_params: - return { - "success": False, - "error": "No update parameters provided", - } - - # Initialize Gns3Connector using factory function - logger.info("Connecting to GNS3 server...") - server = get_gns3_connector() - - if server is None: - logger.error("Failed to create GNS3 connector") - return { - "success": False, - "error": "Failed to connect to GNS3 server. Please check your configuration.", - } - - # Create project instance - if project_id: - project = Project(project_id=project_id, connector=server) - else: - project = Project(name=project_name, connector=server) - - # Get current project information - project.get(get_nodes=False, get_links=False, get_stats=False) - - # Verify project was found - if not project.project_id: - if project_name: - return { - "success": False, - "error": f"Project with name '{project_name}' not found", - } - else: - return { - "success": False, - "error": f"Project with ID '{project_id}' not found", - } - - # Store current values for comparison - old_values = {} - for field in update_params: - old_values[field] = getattr(project, field, None) - - # Update the project - project.update(**update_params) - - # Collect updated fields - updated_fields = [] - for field in update_params: - new_value = getattr(project, field, None) - if old_values[field] != new_value: - updated_fields.append(field) - - logger.info( - "Project updated successfully: %s (ID: %s), updated fields: %s", - project.name, - project.project_id, - ", ".join(updated_fields), - ) - - # Prepare project details for response - project_details = { - "project_id": project.project_id, - "name": project.name, - "status": project.status, - "path": project.path, - } - - # Add optional fields if they were updated or exist - optional_fields = [ - "auto_start", - "auto_close", - "auto_open", - "scene_width", - "scene_height", - "grid_size", - "drawing_grid_size", - "show_grid", - "show_interface_labels", - "show_layers", - "snap_to_grid", - "zoom", - ] - - for field in optional_fields: - value = getattr(project, field, None) - if value is not None: - project_details[field] = value - - # Prepare result - result = { - "success": True, - "project": project_details, - "updated_fields": updated_fields, - "message": f"Project '{project.name}' updated successfully", - } - - # Log result - logger.info("Project update result: %s", result) - - # Return success with project details - return result - - except ValueError as e: - logger.error("Validation error updating GNS3 project: %s", str(e)) - return { - "success": False, - "error": f"Validation error: {str(e)}", - } - except Exception as e: - logger.error("Error updating GNS3 project: %s", str(e)) - return { - "success": False, - "error": f"Failed to update GNS3 project: {str(e)}", - } diff --git a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_write_file.py b/gns3server/agent/gns3_copilot/gns3_client/gns3_project_write_file.py deleted file mode 100644 index ed19ba974..000000000 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_project_write_file.py +++ /dev/null @@ -1,207 +0,0 @@ -""" -GNS3 Project Write File Tool - -Provides a LangChain tool for writing files to GNS3 projects. -This tool directly uses the Project.write_file() method from custom_gns3fy. -""" - -import json -import logging -import re -from typing import Any - -from langchain.tools import BaseTool -from langchain_core.callbacks import CallbackManagerForToolRun - -from gns3_copilot.gns3_client import Project, get_gns3_connector -from gns3_copilot.gns3_client.gns3_file_index import add_file_to_index - -# Configure logging -logger = logging.getLogger(__name__) - - -class GNS3ProjectWriteFileTool(BaseTool): - """ - A LangChain tool to write files to a GNS3 project. - - This tool writes content to a file in a GNS3 project directory. - It uses Project.write_file() method from custom_gns3fy. - - **Input:** - A JSON object containing the project_id, file path, and data to write. - - Example input: - { - "project_id": "uuid-of-project", - "path": "README.md", - "data": "This is the file content..." - } - - **Output:** - A dictionary containing the project_id, file path, and status. - - Example output: - { - "project_id": "uuid-of-project", - "path": "README.md", - "status": "success" - } - - If an error occurs, returns a dictionary with an error message. - - Example error output: - { - "project_id": "uuid-of-project", - "path": "README.md", - "status": "failed", - "error": "Failed to write file" - } - """ - - name: str = "gns3_project_write_file" - description: str = """ - Writes content to a file in a GNS3 project directory. - Input is a JSON object with project_id, file path, and data to write. - Example input: - { - "project_id": "uuid-of-project", - "path": "README.md", - "data": "This is the file content..." - } - Returns a dictionary with the write operation status. - If the operation fails, returns a dictionary with an error message. - """ - - def _run( - self, - tool_input: str, - run_manager: CallbackManagerForToolRun | None = None, - **kwargs: Any, - ) -> str: - """ - Writes a file to a GNS3 project. - - Args: - tool_input (str): A JSON string containing project_id, file path, and data. - run_manager: LangChain run manager (unused). - - Returns: - str: A JSON string with the write operation status, 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") - path = input_data.get("path") - data = input_data.get("data") - - # Validate input - if not project_id: - logger.error("Invalid input: Missing project_id.") - return json.dumps({"error": "Missing project_id."}) - - if not path: - logger.error("Invalid input: Missing path.") - return json.dumps({"error": "Missing path."}) - - if data is None: - logger.error("Invalid input: Missing data.") - return json.dumps({"error": "Missing data."}) - - # Validate project_id format (UUID) - 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 json.dumps({"error": error_msg}) - - # Get connector using factory function - connector = get_gns3_connector() - if connector is None: - error_msg = "Failed to create GNS3 connector. Check configuration." - logger.error(error_msg) - return json.dumps({"error": error_msg}) - - logger.info("Connecting to GNS3 server...") - - # Create Project instance - project = Project(project_id=project_id, connector=connector) - - # Write file content - logger.info("Writing file '%s' to project '%s'...", path, project_id) - project.write_file(path=path, data=data) - - # Update file index - logger.info("Updating file index for project '%s'...", project_id) - file_size = len(str(data)) - add_file_to_index(project, path, size=file_size) - logger.info("Successfully updated file index") - - # Prepare successful result - result = { - "project_id": project_id, - "path": path, - "status": "success", - } - - logger.info( - "Successfully wrote file '%s' to project '%s'", path, project_id - ) - - return json.dumps(result) - - except json.JSONDecodeError as e: - logger.error("Invalid JSON input: %s", e) - return json.dumps({"error": f"Invalid JSON input: {e}"}) - except ValueError as e: - logger.error("Value error: %s", e) - return json.dumps({"error": f"Value error: {e}"}) - except Exception as e: - logger.error("Failed to write file to project: %s", e) - return json.dumps( - { - "project_id": project_id if "project_id" in locals() else None, - "path": path if "path" in locals() else None, - "status": "failed", - "error": f"Failed to write file: {str(e)}", - } - ) - - 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-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$" - return bool(re.match(uuid_pattern, project_id)) - - -if __name__ == "__main__": - import pprint - - print("=" * 80) - print("Testing GNS3ProjectWriteFileTool") - print("=" * 80) - - write_test_input = json.dumps( - { - "project_id": "1445a4ba-4635-430b-a332-bef438f65932", # Replace with actual project UUID - "path": "test_write.txt", - "data": "This is a test file created by GNS3ProjectWriteFileTool\n" - "Created on: 2026-01-01", - } - ) - - write_tool = GNS3ProjectWriteFileTool() - write_result = write_tool._run(write_test_input) - print("\nWrite Result:") - pprint.pprint(write_result) diff --git a/gns3server/agent/gns3_copilot/gns3_client/gns3_projects_list.py b/gns3server/agent/gns3_copilot/gns3_client/gns3_projects_list.py deleted file mode 100644 index 06283677c..000000000 --- a/gns3server/agent/gns3_copilot/gns3_client/gns3_projects_list.py +++ /dev/null @@ -1,77 +0,0 @@ -import logging -from typing import Any - -from langchain.tools import BaseTool - -from gns3_copilot.gns3_client import get_gns3_connector - -# Configure logging -logger = logging.getLogger(__name__) - -""" -example output: - -[('mylab', 'ff8e059c-c33d-47f4-bc11-c7dda8a1d500', 0, 0, 'closed'), - ('q-learning-traffic-management', '69d49a6a-ff7f-45dd-af1e-dc14aff600cc', 0, 0, 'closed'), - ('network_ai', 'f2f7ed27-7aa3-4b11-a64c-da947a2c7210', 6, 8, 'opened'), - ('test', '365dd3ff-cda9-447a-94da-3a6cef75fe77', 0, 0, 'closed'), - ('Soft-RoCE learning', 'd1e4509e-64bd-4109-b954-266223959ee9', 0, 0, 'closed')] - -""" - - -class GNS3ProjectList(BaseTool): - name: str = "list_gns3_projects" - description: str = """ - Retrieves a list of all GNS3 projects with their details. - Returns a dictionary containing a list of project information including name, - project_name, project_id, nodes count, links count, and status. - Example output: - { - "projects": [ - ("mylab", "ff8e059c-c33d-47f4-bc11-c7dda8a1d500", 0, 0, "closed"), - ("network_ai", "f2f7ed27-7aa3-4b11-a64c-da947a2c7210", 6, 8, "opened") - ] - } - """ - - def _run(self, tool_input: Any = None, run_manager: Any = None) -> dict: - # Log received input - logger.info("Received input: %s", tool_input) - - try: - # Initialize Gns3Connector using factory function - logger.info("Connecting to GNS3 server...") - server = get_gns3_connector() - - if server is None: - logger.error("Failed to create GNS3 connector") - return { - "error": "Failed to connect to GNS3 server. Please check your configuration." - } - - # Return the projects data in a structured format - projects = server.projects_summary(is_print=False) - - # Prepare result - result = {"projects": projects} - - # Log result - logger.info("Projects list result: %s", result) - - return result - - except Exception as e: - logger.error("Error retrieving GNS3 project list: %s", str(e)) - return {"error": f"Failed to retrieve GNS3 project list: {str(e)}"} - - -if __name__ == "__main__": - from pprint import pprint - - # Test the tool - tool = GNS3ProjectList() - - print("Testing GNS3ProjectList - retrieving all projects...") - result = tool._run() - pprint(result) diff --git a/gns3server/agent/gns3_copilot/prompts/prompt_loader.py b/gns3server/agent/gns3_copilot/prompts/prompt_loader.py new file mode 100644 index 000000000..a9f1f7ef7 --- /dev/null +++ b/gns3server/agent/gns3_copilot/prompts/prompt_loader.py @@ -0,0 +1,28 @@ +""" +Prompt loader for GNS3 Copilot. + +This module provides utilities for loading system prompts. +""" + +import logging +import os + +from .base_prompt import SYSTEM_PROMPT + +logger = logging.getLogger(__name__) + + +def load_system_prompt() -> str: + """ + Load the system prompt for GNS3 Copilot. + + In the future, this can be extended to support multiple prompt variants + based on environment variables (e.g., ENGLISH_LEVEL). + + Returns: + str: The system prompt string. + """ + # For now, just return the base system prompt + # Future enhancement: Load different prompts based on ENGLISH_LEVEL env var + # english_level = os.getenv("ENGLISH_LEVEL", "native") + return SYSTEM_PROMPT diff --git a/gns3server/agent/gns3_copilot/tools_v2/__init__.py b/gns3server/agent/gns3_copilot/tools_v2/__init__.py index 2111252c8..bf8a655fd 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/__init__.py +++ b/gns3server/agent/gns3_copilot/tools_v2/__init__.py @@ -1,5 +1,5 @@ """ -FlowNet-Lab Tools Package +GNS3-Copilot Tools Package This package provides various tools for interacting with GNS3 network simulator, including: - Device configuration command execution @@ -7,8 +7,6 @@ This package provides various tools for interacting with GNS3 network simulator, - 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 @@ -18,14 +16,7 @@ Main modules: - 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) +- gns3_update_node_name: GNS3 node name update tool Note: GNS3TopologyTool is now available from gns3_client package @@ -35,15 +26,12 @@ 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: @@ -68,11 +56,7 @@ __all__ = [ "GNS3StartNodeQuickTool", "GNS3UpdateNodeNameTool", "GNS3TemplateTool", - "GNS3CreateAreaDrawingTool", - "LinuxTelnetBatchTool", - "WindowControllerTool", - "TextInputTool", ] # Package initialization message -# print(f"FlowNet-Lab Tools package loaded (version {__version__})") +# print(f"GNS3-Copilot Tools package loaded (version {__version__})") \ No newline at end of file diff --git a/gns3server/agent/gns3_copilot/tools_v2/gns3_create_area_drawing.py b/gns3server/agent/gns3_copilot/tools_v2/gns3_create_area_drawing.py deleted file mode 100644 index dd7779068..000000000 --- a/gns3server/agent/gns3_copilot/tools_v2/gns3_create_area_drawing.py +++ /dev/null @@ -1,403 +0,0 @@ -""" -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 -} -""" diff --git a/gns3server/agent/gns3_copilot/utils/__init__.py b/gns3server/agent/gns3_copilot/utils/__init__.py index 6c4da8b89..4b780ffbb 100644 --- a/gns3server/agent/gns3_copilot/utils/__init__.py +++ b/gns3server/agent/gns3_copilot/utils/__init__.py @@ -1,5 +1,5 @@ """ -FlowNet-Lab Public Model Package +GNS3-Copilot Public Model Package This package provides reusable public models and utilities for GNS3 network automation tasks. It contains common functionality that can be shared across different tools and modules.