NgTech-LTD/gns3-server
2
0
Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2026-08-27 20:40:13 +03:00
Code Issues Packages Projects Releases Wiki Activity

feat: rename FlowNet-Lab to GNS3-Copilot across codebase

Browse Source 
Update all references from FlowNet-Lab to GNS3-Copilot in package names, documentation, and logging. This includes:
- Module and package __init__.py files
- License headers and file descriptions
- Log messages and internal comments
- Remove deprecated tools: GNS3CreateAreaDrawingTool and LinuxTelnetBatchTool

The renaming aligns with the project's new branding while maintaining all existing functionality.
This commit is contained in:
YueGuobin 2026-03-04 00:55:29 +08:00
parent 95b309c7af
commit d525e5cad0
24 changed files with 47 additions and 3487 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
gns3server/agent/gns3_copilot/__init__.py
View File

@ -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.
"""

4
gns3server/agent/gns3_copilot/agent/__init__.py
View File

@ -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

18
gns3server/agent/gns3_copilot/agent/gns3_copilot.py
View File

@ -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 <https://www.gnu.org/licenses/>.
# with GNS3-Copilot. If not, see <https://www.gnu.org/licenses/>.
# 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.

2
gns3server/agent/gns3_copilot/agent/model_factory.py
View File

@ -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.

44
gns3server/agent/gns3_copilot/gns3_client/__init__.py
View File

@ -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",
]

292
gns3server/agent/gns3_copilot/gns3_client/gns3_create_drawing.py
View File

@ -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": "<svg>...</svg>",
"x": 100,
"y": -200,
"z": 0,
"locked": false,
"rotation": 0
},
{
"svg": "<svg>...</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": "<svg>...</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": '<svg xmlns="http://www.w3.org/2000/svg" width="100" height="50"><text x="10" y="30" font-size="14">Label 1</text></svg>',
"x": 100,
"y": -200,
"z": 0,
"locked": False,
"rotation": 0,
},
{
"svg": '<svg xmlns="http://www.w3.org/2000/svg" width="100" height="50"><text x="10" y="30" font-size="14">Label 2</text></svg>',
"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}
"""

160
gns3server/agent/gns3_copilot/gns3_client/gns3_delete_drawing.py
View File

@ -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'}
"""

257
gns3server/agent/gns3_copilot/gns3_client/gns3_file_index.py
View File

@ -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()

236
gns3server/agent/gns3_copilot/gns3_client/gns3_get_drawings.py
View File

@ -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": "<svg>...</svg>",
"x": 100,
"y": 200,
"z": 0,
"locked": false,
"rotation": 0
},
{
"drawing_id": "uuid-of-drawing2",
"svg": "<svg>...</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': '<svg height="131" width="391"><rect fill="#ffffff" '
'fill-opacity="1" height="131" width="391" '
'stroke="#000000" stroke-width="2" '
'stroke-dasharray="undefined" rx="0" ry="0" /></svg>',
'x': -376,
'y': -381,
'z': 1},
{'drawing_id': '7d2f2411-efe8-4f3a-9882-cc0ad7798ee0',
'locked': False,
'rotation': 0,
'svg': '<svg height="100" width="100"><text fill="#000000" '
'fill-opacity="1.0" font-family="Noto Sans" '
'font-size="11" font-weight="bold"></text></svg>',
'x': -394,
'y': 4,
'z': 1},
{'drawing_id': '264be3ab-002c-4b1d-886d-b4c323de845a',
'locked': False,
'rotation': 0,
'svg': '<svg height="100" width="100"><text fill="#000000" '
'fill-opacity="1.0" font-family="Noto Sans" '
'font-size="11" font-weight="bold">哈哈哈\n'
'哈和</text></svg>',
'x': -596,
'y': -43,
'z': 1},
{'drawing_id': '44210a41-0100-46a0-8962-a73c72171c47',
'locked': False,
'rotation': 0,
'svg': '<svg height="100" width="100"><text fill="#000000" '
'fill-opacity="1.0" font-family="Noto Sans" '
'font-size="11" font-weight="bold">Area 0</text></svg>',
'x': -573,
'y': -272,
'z': 1},
{'drawing_id': 'efd1add9-a798-489d-af26-58b2d73c89dc',
'locked': False,
'rotation': 0,
'svg': '<svg height="0" width="100"><line stroke="#000000" '
'stroke-width="2" x1="0" x2="200" y1="0" y2="0" '
'stroke-dasharray="none" /></svg>',
'x': -636,
'y': -152,
'z': 1},
{'drawing_id': '89913ef3-1041-4b11-94e1-c5d65c3a52d4',
'locked': False,
'rotation': 0,
'svg': '<svg height="119" width="488"><ellipse fill="#ffffff" '
'fill-opacity="1" cx="244" cy="59.5" rx="244" ry="59.5" '
'stroke="#000000" stroke-width="2" '
'stroke-dasharray="undefined" /></svg>',
'x': -891,
'y': 66,
'z': 1}],
'project_id': '2245149a-71c8-4387-9d1f-441a683ef7e7',
'total_drawings': 8}
"""

182
gns3server/agent/gns3_copilot/gns3_client/gns3_get_nodes.py
View File

@ -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)

167
gns3server/agent/gns3_copilot/gns3_client/gns3_project_create.py
View File

@ -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)}",
}

152
gns3server/agent/gns3_copilot/gns3_client/gns3_project_delete.py
View File

@ -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)}",
}

239
gns3server/agent/gns3_copilot/gns3_client/gns3_project_list_files.py
View File

@ -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()

305
gns3server/agent/gns3_copilot/gns3_client/gns3_project_lock.py
View File

@ -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}")

158
gns3server/agent/gns3_copilot/gns3_client/gns3_project_open.py
View File

@ -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)}",
}

160
gns3server/agent/gns3_copilot/gns3_client/gns3_project_path.py
View File

@ -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)}",
}

191
gns3server/agent/gns3_copilot/gns3_client/gns3_project_read_file.py
View File

@ -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)

224
gns3server/agent/gns3_copilot/gns3_client/gns3_project_update.py
View File

@ -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)}",
}

207
gns3server/agent/gns3_copilot/gns3_client/gns3_project_write_file.py
View File

@ -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)

77
gns3server/agent/gns3_copilot/gns3_client/gns3_projects_list.py
View File

@ -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)

28
gns3server/agent/gns3_copilot/prompts/prompt_loader.py Normal file
View File

@ -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

22
gns3server/agent/gns3_copilot/tools_v2/__init__.py
View File

@ -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__})")

403
gns3server/agent/gns3_copilot/tools_v2/gns3_create_area_drawing.py
View File

@ -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
}
"""

2
gns3server/agent/gns3_copilot/utils/__init__.py
View File

@ -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.