feat: add fault injection system and external skills repository

## Summary

Add a complete fault injection system for GNS3 Copilot, migrate all
skills from local Python files to an external Git repository with
hot reload support, and restructure Copilot API under /copilot/.

## Key Changes

### Fault Injection
- New troubleshooting_injection mode with InjectionSkillsTool
- 368 fault scenarios across 39 protocol categories
- Context-based filtering (LLM must pass topology protocols)

### External Skills Repository
- SkillsManager: Git clone/pull, version tracking, smart updates
- SkillsLoader: YAML skills + Markdown prompts from external repo
- Hot reload via POST /copilot/reload/skills
- Configurable via gns3_server.conf

### Architecture
- API unified under /copilot/ prefix
- SkillsManager moved from Controller to agent module
- Lazy initialization with startup background preload
- Per-command Git timeout, smart update checks
- Forbidden commands hot-reloadable from external repo
- 32 INFO logs downgraded to DEBUG
This commit is contained in:
YueGuobin 2026-05-11 01:24:55 +08:00
parent 534bb5eded
commit e68cc17ed5
No known key found for this signature in database
42 changed files with 2182 additions and 1260 deletions

View File

@ -414,20 +414,30 @@ function handleToolCallEvent(chunk) {
## API Endpoints
All endpoints are under `/v3/projects/{project_id}/chat/` path.
All Copilot endpoints are under `/v3/copilot/` path.
### Chat Endpoints
| Method | Endpoint | Description |
|--------|----------|-------------|
| POST | `/stream` | Streaming Chat (main interface) |
| GET | `/sessions` | List sessions (sorted by pin and update time) |
| GET | `/sessions/{session_id}/history` | Get session history |
| PATCH | `/sessions/{session_id}` | Rename session |
| DELETE | `/sessions/{session_id}` | Delete session |
| POST | `/sessions/{session_id}/abort` | Abort ongoing streaming session |
| PUT | `/sessions/{session_id}/pin` | Pin session |
| DELETE | `/sessions/{session_id}/pin` | Unpin session |
| POST | `/copilot/projects/{project_id}/chat/stream` | Streaming Chat (main interface) |
| GET | `/copilot/projects/{project_id}/chat/sessions` | List sessions (sorted by pin and update time) |
| GET | `/copilot/projects/{project_id}/chat/sessions/{session_id}/history` | Get session history |
| PATCH | `/copilot/projects/{project_id}/chat/sessions/{session_id}` | Rename session |
| DELETE | `/copilot/projects/{project_id}/chat/sessions/{session_id}` | Delete session |
| POST | `/copilot/projects/{project_id}/chat/sessions/{session_id}/abort` | Abort ongoing streaming session |
| PUT | `/copilot/projects/{project_id}/chat/sessions/{session_id}/pin` | Pin session |
| DELETE | `/copilot/projects/{project_id}/chat/sessions/{session_id}/pin` | Unpin session |
### POST /v3/projects/{project_id}/chat/stream
### Skills Endpoints
| Method | Endpoint | Description |
|--------|----------|-------------|
| POST | `/copilot/reload/skills` | Hot reload skills, prompts, and forbidden commands |
See [Skills Repository](skills-repository.md) for details on the reload process and configuration.
### POST /v3/copilot/projects/{project_id}/chat/stream
**Function**: Streaming conversation interface
@ -478,7 +488,7 @@ data: {"type": "content", "content": "! I can help", "session_id": "d7e76375-696
data: {"type": "done", "session_id": "d7e76375-6960-419a-9367-211ef64af877"}
```
### GET /v3/projects/{project_id}/chat/sessions
### GET /v3/copilot/projects/{project_id}/chat/sessions
**Function**: List all sessions in a project
@ -514,7 +524,7 @@ data: {"type": "done", "session_id": "d7e76375-6960-419a-9367-211ef64af877"}
]
```
### GET /v3/projects/{project_id}/chat/sessions/{session_id}/history
### GET /v3/copilot/projects/{project_id}/chat/sessions/{session_id}/history
**Function**: Get complete history of a session
@ -563,7 +573,7 @@ data: {"type": "done", "session_id": "d7e76375-6960-419a-9367-211ef64af877"}
}
```
### PATCH /v3/projects/{project_id}/chat/sessions/{session_id}
### PATCH /v3/copilot/projects/{project_id}/chat/sessions/{session_id}
**Function**: Rename session
@ -581,13 +591,13 @@ data: {"type": "done", "session_id": "d7e76375-6960-419a-9367-211ef64af877"}
**Response**: Updated session information
### DELETE /v3/projects/{project_id}/chat/sessions/{session_id}
### DELETE /v3/copilot/projects/{project_id}/chat/sessions/{session_id}
**Function**: Delete session and all its checkpoint data
**Response**: 204 No Content
### POST /v3/projects/{project_id}/chat/sessions/{session_id}/abort
### POST /v3/copilot/projects/{project_id}/chat/sessions/{session_id}/abort
**Function**: Abort an ongoing streaming session
@ -615,7 +625,7 @@ Graph conditional edge checks flag
Stream ends → yields aborted tool_end events → done
```
### PUT /v3/projects/{project_id}/chat/sessions/{session_id}/pin
### PUT /v3/copilot/projects/{project_id}/chat/sessions/{session_id}/pin
**Function**: Pin session to top of list
@ -630,7 +640,7 @@ Stream ends → yields aborted tool_end events → done
}
```
### DELETE /v3/projects/{project_id}/chat/sessions/{session_id}/pin
### DELETE /v3/copilot/projects/{project_id}/chat/sessions/{session_id}/pin
**Function**: Unpin session

View File

@ -86,14 +86,14 @@ flowchart TD
### Forbidden Commands Configuration
**File:** `gns3server/agent/gns3_copilot/config/forbidden_commands.txt`
The forbidden commands list is loaded from the external [GNS3-Skills](https://github.com/yueguobin/GNS3-Skills) repository at `config/forbidden_commands.txt`.
**Format:**
- One command pattern per line
- **Prefix matching** (case-insensitive) — matches the beginning of each command
- Empty lines and lines starting with `#` are ignored
**Default patterns (used when config file is missing):**
**Default patterns (used when skills repository is unavailable):**
| Pattern | Reason |
|---------|--------|
@ -140,7 +140,7 @@ flowchart TD
| `filter_forbidden_commands(commands)` | Returns `(allowed_commands, blocked_commands_info)` |
| `is_command_forbidden(command)` | Check if a single command matches a forbidden pattern |
| `get_forbidden_commands()` | Get current forbidden patterns list |
| `reload_forbidden_commands()` | Clear cache, reload from file on next access |
| `reload_forbidden_commands()` | Directly load and cache commands from skills repository |
**Integration points:**
- `display_tools_nornir.py``_filter_forbidden_commands_from_device_configs()`
@ -168,7 +168,7 @@ Applies to any command with embedded newlines: `banner`, multi-line ACLs, route-
### Customizing Forbidden Commands
Edit `gns3server/agent/gns3_copilot/config/forbidden_commands.txt` and either restart the server or call `reload_forbidden_commands()` to apply changes without restart.
Edit `config/forbidden_commands.txt` in the [GNS3-Skills repository](https://github.com/yueguobin/GNS3-Skills) and push the changes, then call `POST /copilot/reload/skills` to apply them without restarting the server.
## Implementation Verification
@ -214,11 +214,11 @@ Edit `gns3server/agent/gns3_copilot/config/forbidden_commands.txt` and either re
|---------|----------|
| Command blocked unexpectedly | Check `blocked_commands` in result, identify matching pattern, edit `forbidden_commands.txt` |
| "File not found, using defaults" | Verify `config/forbidden_commands.txt` exists and is readable |
| Changes not taking effect | Restart server or call `reload_forbidden_commands()` |
| Changes not taking effect | Call `POST /copilot/reload/skills` or restart server |
## Related Documentation
- [Tool Implementation](../gns3-copilot/tools_v2/README.md)
- [GNS3-Copilot Documentation](../README.md)
- [Contributing Guide](../../CONTRIBUTING.md)
- [Skills Repository](skills-repository.md)
- [Fault Injection](fault-injection.md)
- [Chat API](chat-api.md)

View File

@ -0,0 +1,159 @@
<!--
SPDX-License-Identifier: CC-BY-SA-4.0
See LICENSE file for licensing information.
-->
> This documentation is organized by AI with reference to actual code. AI can make mistakes — please verify against the source code when in doubt.
# Fault Injection Feature
## Overview
The fault injection feature enables GNS3 Copilot to automatically inject realistic network faults into GNS3 labs for troubleshooting training. The agent analyzes the topology, selects an appropriate fault based on the configured protocols, injects it, and documents the results.
## Architecture
```mermaid
graph TD
subgraph "Agent Workflow"
TG[Topology Gathering]
CFG[Config Collection]
FA[Fault Analysis]
FI[Fault Injection]
FD[Fault Documentation]
end
subgraph "Skills Repository"
IS[INJECTION_SKILLS_REGISTRY<br/>39 categories, 368 issues]
SK[injection_skills Tool]
end
subgraph "Tools"
DC[execute_multiple_device_commands]
CC[execute_multiple_device_config_commands]
GK[GNS3TopologyTool]
end
TG -->|Get topology| GK
CFG -->|Get configs| DC
FA -->|Query filtered skills| IS
FA -->|Select fault| SK
FI -->|Inject config changes| CC
FD -->|Output report| FD
```
## Injection Skills Tool
The `InjectionSkillsTool` (LangChain `BaseTool`) is the primary interface for querying available faults.
### Listing Faults (with context filter)
The LLM MUST pass topology context when listing faults:
```json
{"action": "list", "context": ["ospf", "bgp", "vlan"]}
```
This returns only faults matching the protocols found in the topology. The tool rejects calls without `context`:
```json
{
"error": "context parameter is required when action='list'",
"hint": "Analyze the topology and device configurations first...",
"available_categories": ["bgp", "interface", "mpls", "ospf", ...]
}
```
### Getting Fault Details
Token-efficient usage pattern:
```json
// Step 1: List issue names only (~300 tokens)
{"device_type": "injection_ospf", "detail": "index"}
// Step 2: Get single issue detail (~500 tokens)
{"device_type": "injection_ospf", "issue": "ospf_hello_dead_mismatch"}
```
### Parameters
| Parameter | Required | Description |
|-----------|----------|-------------|
| `action` | No (`"get"`) | `"list"` to browse, `"get"` for details |
| `context` | **Yes** for `action="list"` | Protocols found in topology: `["ospf", "bgp"]` |
| `device_type` | Yes for `action="get"` | e.g. `"injection_ospf"` |
| `detail` | No (`"full"`) | `"index"` (names), `"summary"` (+desc), `"full"` (all) |
| `issue` | No | Single issue key for targeted detail |
## Fault Injection Workflow
```mermaid
sequenceDiagram
participant LLM
participant InjectionSkillsTool
participant TopologyTool
participant DeviceCommands
LLM->>TopologyTool: Get topology info
TopologyTool-->>LLM: Topology + device types
LLM->>DeviceCommands: Get device configs
DeviceCommands-->>LLM: Running configurations
Note over LLM: Analyze which protocols are in use
LLM->>InjectionSkillsTool: {"action": "list", "context": ["ospf", "vlan"]}
InjectionSkillsTool-->>LLM: Matching fault categories
LLM->>InjectionSkillsTool: {"device_type": "injection_ospf", "detail": "index"}
InjectionSkillsTool-->>LLM: Issue names (token-efficient)
LLM->>InjectionSkillsTool: {"device_type": "injection_ospf", "issue": "ospf_hello_dead_mismatch"}
InjectionSkillsTool-->>LLM: Full fault detail + config commands
LLM->>DeviceCommands: Inject fault configuration
DeviceCommands-->>LLM: Execution result
Note over LLM: Document fault in response
```
## Injection Skills Repository
Skills are organized by protocol/category in the external [GNS3-Skills](https://github.com/yueguobin/GNS3-Skills) repository:
| Category | File | Example Issues |
|----------|------|----------------|
| OSPF | `injection/ospf_issues.yaml` | Hello/Dead mismatch, MTU mismatch, area mismatch |
| BGP | `injection/bgp_issues.yaml` | AS-path prepend, next-hop unreachable, route filtering |
| VLAN | `injection/vlan_issues.yaml` | Trunk allowed mismatch, native VLAN mismatch |
| STP | `injection/stp_issues.yaml` | Root guard, loop guard, port priority |
| MPLS | `injection/mpls_issues.yaml` | LDP session down, label binding failure |
| ... | 34 more files | 368 issues total |
## Recovery
Each injected fault includes restore commands in the documentation. The LLM always provides commands to fully revert all changes.
## API Endpoint
### POST /copilot/projects/{project_id}/chat/inject
Dedicated endpoint for fault injection. Internally sets `copilot_mode` to `troubleshooting_injection` and runs the agent.
**Request:**
```json
{
"message": "Inject a network fault for troubleshooting practice",
"session_id": "optional-session-uuid"
}
```
**Response:** Server-Sent Events (SSE) stream identical to the chat stream endpoint.
## Related Documentation
- [Skills Repository](skills-repository.md)
- [Command Security](command-security.md)
- [Chat API](chat-api.md)

View File

@ -0,0 +1,171 @@
<!--
SPDX-License-Identifier: CC-BY-SA-4.0
See LICENSE file for licensing information.
-->
> This documentation is organized by AI with reference to actual code. AI can make mistakes — please verify against the source code when in doubt.
# External Skills Repository
## Overview
GNS3 Copilot loads all skills, prompts, and security configurations from an external Git repository at [github.com/yueguobin/GNS3-Skills](https://github.com/yueguobin/GNS3-Skills). This enables dynamic updates without server redeployment.
The repository provides:
- **Injection skills** (39 categories): Network fault scenarios for troubleshooting practice
- **Device skills**: Device-specific command knowledge (VPCS, etc.)
- **Feature skills**: Topology planning, network design
- **System prompts**: Agent personality and behavior definitions
- **Forbidden commands**: Security rules for command filtering
## Architecture
```mermaid
graph TD
subgraph "GNS3-Skills Repository"
YAML[injection/*.yaml<br/>device/*.yaml<br/>feature/*.yaml]
MD[prompts/*.md]
CFG[config/forbidden_commands.txt]
end
subgraph "GNS3 Server"
SM[SkillsManager]
SL[SkillsLoader]
REG[Registry<br/>SKILLS_REGISTRY<br/>INJECTION_SKILLS_REGISTRY]
PROMPT[PROMPTS_CACHE]
FC[command_filter]
end
YAML --> SL
MD --> SL
CFG --> FC
SL --> REG
SL --> PROMPT
SM --> SL
SM -->|git pull| YAML
```
## Repository Structure
```
GNS3-Skills/
├── injection/ # 39 YAML files, one per protocol/category
│ ├── ospf_issues.yaml
│ ├── bgp_issues.yaml
│ ├── vlan_issues.yaml
│ └── ...
├── device/ # Device-specific skills
│ └── vpcs.yaml
├── feature/ # Feature skills
│ └── topology_planner.yaml
├── prompts/ # System prompts (Markdown)
│ ├── teaching_assistant.md
│ ├── lab_automation_assistant.md
│ ├── troubleshooting_injection.md
│ └── title.md
└── config/ # Security configuration
└── forbidden_commands.txt
```
## Configuration
Skills repository settings are configured in `gns3_server.conf` under the `[Server]` section:
```ini
[Server]
skills_repo_url = https://github.com/yueguobin/GNS3-Skills.git
skills_repo_branch = main
skills_auto_update = true
```
| Setting | Default | Description |
|---------|---------|-------------|
| `skills_repo_url` | `https://github.com/yueguobin/GNS3-Skills.git` | Git repository URL |
| `skills_repo_branch` | `main` | Git branch to track |
| `skills_auto_update` | `true` | Automatically pull on reload |
## Initialization Flow
```mermaid
sequenceDiagram
participant Server
participant agent/__init__.py
participant SkillsManager
participant Git
Server->>agent/__init__.py: Import module
agent/__init__.py->>SkillsManager: Start background init thread
Note over SkillsManager: Thread.join(5s timeout)
SkillsManager->>Git: Check local repo exists?
alt No local repo
Git->>SkillsManager: git clone (with timeout env)
else Local repo exists
SkillsManager->>Git: Check uncommitted changes?
alt Has uncommitted changes
Git-->>SkillsManager: Warn, skip pull
else No changes
SkillsManager->>Git: git fetch (timeout 10s)
SkillsManager->>Git: Behind remote?
alt Behind
Git->>SkillsManager: git pull
else Up to date
Git-->>SkillsManager: Nothing to do
end
end
end
SkillsManager->>SkillsManager: reload_skills() - load YAML files
SkillsManager->>SkillsManager: reload_prompts() - load Markdown files
```
### Git Timeout Configuration
Git operations use per-command environment variables to prevent hanging:
```python
_GIT_TIMEOUT_ENV = {
'GIT_HTTP_TIMEOUT': '10', # Connection timeout (default: 120s)
'GIT_HTTP_LOW_SPEED_TIME': '5', # Slow speed threshold window
'GIT_HTTP_LOW_SPEED_LIMIT': '1000', # < 1 KB/s = slow abort
}
```
These apply only to the specific `clone`/`fetch`/`pull` subprocess, not to the global environment.
## API Endpoint
### POST /copilot/reload/skills
Triggers a full reload of the skills repository. Performs one git update check, then reloads all skills, prompts, and forbidden commands from local files.
**Response:**
```json
{
"success": true,
"skills": true,
"skill_count": 39,
"prompts": true,
"prompt_count": 4,
"forbidden_commands": 6,
"version": "abc123def456..."
}
```
| Field | Description |
|-------|-------------|
| `success` | Overall success (true if skills or prompts loaded) |
| `skills` | Skills reload result |
| `skill_count` | Number of injection skills loaded |
| `prompts` | Prompts reload result |
| `prompt_count` | Number of prompts loaded |
| `forbidden_commands` | Number of forbidden command patterns |
| `version` | Git commit hash of the repository |
## Related Documentation
- [Fault Injection](fault-injection.md)
- [Command Security](command-security.md)
- [Chat API](chat-api.md)

View File

@ -26,6 +26,7 @@ Installation:
"""
import logging
import threading
# Feature flag: AI Copilot is available
AI_COPILOT_AVAILABLE = False
@ -35,6 +36,28 @@ try:
from .gns3_copilot.project_agent_manager import get_project_agent_manager
from .gns3_copilot.project_agent_manager import ProjectAgentManager
AI_COPILOT_AVAILABLE = True
# Start skills repository initialization in background with 5s timeout.
# This clones/pulls GNS3-Skills during server startup.
# If GitHub is unreachable (common in some regions), the timeout
# ensures the server starts without delay, using fallback defaults.
def _init_skills_background():
"""Initialize skills repo - called from background thread."""
from gns3server.agent.gns3_copilot.skills.registry import _ensure_skills_manager
t = threading.Thread(target=_ensure_skills_manager, daemon=True)
t.start()
t.join(5)
if t.is_alive():
log = logging.getLogger(__name__)
log.warning(
"Skills repository initialization timed out (5s). "
"Will use fallback defaults. "
"The background thread will complete when network is available."
)
threading.Thread(target=_init_skills_background, daemon=True).start()
except ImportError as e:
# AI dependencies not installed, disable AI Copilot feature
logging.warning(

View File

@ -67,13 +67,13 @@ logger = logging.getLogger(__name__)
# Initialize tiktoken encoding (required dependency)
import time
logger.info("Initializing tiktoken encoding (cl100k_base)...")
logger.info(f"Cache directory: {_cache_dir}")
logger.info("This may take a moment on first run (downloading ~1.6MB encoding file from openaipublic.blob.core.windows.net)")
logger.debug("Initializing tiktoken encoding (cl100k_base)...")
logger.debug(f"Cache directory: {_cache_dir}")
logger.debug("This may take a moment on first run (downloading ~1.6MB encoding file from openaipublic.blob.core.windows.net)")
start_time = time.time()
_tiktoken_encoding = tiktoken.get_encoding("cl100k_base")
elapsed = time.time() - start_time
logger.info(f"✓ tiktoken encoding loaded successfully (took {elapsed:.2f}s)")
logger.debug(f"✓ tiktoken encoding loaded successfully (took {elapsed:.2f}s)")
# ============================================================================
# Constants

View File

@ -78,8 +78,8 @@ from gns3server.agent.gns3_copilot.gns3_client import GNS3TopologyTool
from gns3server.agent.gns3_copilot.gns3_client.context_helpers import (
get_current_llm_config,
)
from gns3server.agent.gns3_copilot.prompts import TITLE_PROMPT
from gns3server.agent.gns3_copilot.prompts import load_system_prompt
from gns3server.agent.gns3_copilot.skills.registry import get_prompt
from gns3server.agent.gns3_copilot.tools_v2 import (
ExecuteMultipleDeviceCommands,
)
@ -96,6 +96,7 @@ from gns3server.agent.gns3_copilot.tools_v2 import GNS3UpdateNodeNameTool
from gns3server.agent.gns3_copilot.tools_v2.vpcs_tools_netmiko import VPCSCommands
from gns3server.agent.gns3_copilot.tools_v2 import PacketCaptureTool
from gns3server.agent.gns3_copilot.skills import DeviceSkillsTool
from gns3server.agent.gns3_copilot.skills import InjectionSkillsTool
# Set up logger for GNS3-Copilot
logger = logging.getLogger(__name__)
@ -135,15 +136,30 @@ LAB_AUTOMATION_ASSISTANT_MODE_TOOLS = [
DeviceSkillsTool(), # Get device-specific skills and command knowledge
]
# Troubleshooting injection mode: Tools for injecting network faults
# Focused on configuration commands and injection skills for fault injection
TROUBLESHOOTING_INJECTION_MODE_TOOLS = [
ExecuteMultipleDeviceCommands(), # Get device configurations (READ-ONLY)
ExecuteMultipleDeviceConfigCommands(), # Inject configuration changes
InjectionSkillsTool(), # Query injection skills and fault types
GNS3TopologyTool(), # Get topology information
]
# Default tools (legacy support - will be overridden by mode-specific tools)
tools = LAB_AUTOMATION_ASSISTANT_MODE_TOOLS
# Create combined tool lookup for tool_node (supports both modes)
# Create combined tool lookup for tool_node (supports all modes)
# tool_node will receive tool calls based on mode-specific tools bound to the
# model
# Combine all tools from all modes and deduplicate by tool name
ALL_TOOLS = LAB_AUTOMATION_ASSISTANT_MODE_TOOLS
tools_by_name = {tool.name: tool for tool in ALL_TOOLS}
# Add troubleshooting injection tools to the global tool registry
for tool in TROUBLESHOOTING_INJECTION_MODE_TOOLS:
if tool.name not in tools_by_name:
tools_by_name[tool.name] = tool
# Log application startup
logger.info("GNS3-Copilot application starting up")
@ -243,7 +259,7 @@ def llm_call(state: dict, config: RunnableConfig | None = None):
history preservation in state["messages"].
"""
logger.info("LLM call node invoked")
logger.debug("LLM call node invoked")
# Get llm_config from request-scoped context variable
llm_config = get_current_llm_config()
@ -318,6 +334,11 @@ def llm_call(state: dict, config: RunnableConfig | None = None):
"Using LAB_AUTOMATION_ASSISTANT mode tools (includes "
"configuration tools)"
)
elif copilot_mode == "troubleshooting_injection":
mode_tools = TROUBLESHOOTING_INJECTION_MODE_TOOLS
logger.info(
"Using TROUBLESHOOTING_INJECTION mode tools (fault injection)"
)
else: # teaching_assistant mode (default)
mode_tools = TEACHING_ASSISTANT_MODE_TOOLS
logger.info(
@ -351,7 +372,7 @@ def llm_call(state: dict, config: RunnableConfig | None = None):
# trimming)
# Note: LangGraph's pre_model_hook only works with prebuilt agents, not
# custom StateGraph
logger.info("Calling pre_hook to prepare %d messages", len(messages))
logger.debug("Calling pre_hook to prepare %d messages", len(messages))
prepared_state = pre_hook(
{"messages": messages, "topology_info": topology_info}
)
@ -411,12 +432,18 @@ def generate_title(
# Only generate a title if it hasn't been set yet
current_title = state.get("conversation_title")
if current_title in [None, "New Conversation"]:
logger.info("Title generation triggered for session")
logger.debug("Title generation triggered for session")
messages = state["messages"]
# Load title prompt from external repository
title_prompt = get_prompt("title")
if not title_prompt:
logger.error("Title prompt not found in external repository")
return {"conversation_title": UNTITLED_SESSION_FALLBACK, "session_id": state.get("session_id")}
# Build the prompt for title generation
title_prompt_messages = [
SystemMessage(content=TITLE_PROMPT),
SystemMessage(content=title_prompt),
messages[0], # User's first message
messages[-1], # Assistant's final response in this turn
]
@ -456,7 +483,7 @@ def generate_title(
new_title.replace("\n", " ").replace('"', "").replace("'", "")
)
logger.info("Generated new title: %s", new_title)
logger.debug("Generated new title: %s", new_title)
return {"conversation_title": new_title, "session_id": state.get("session_id")}
except Exception as e:
@ -500,7 +527,7 @@ def tool_node(state: dict, config: RunnableConfig | None = None):
"""Performs the tool call"""
tool_calls = state["messages"][-1].tool_calls
logger.info("Tool node invoked: tool_calls=%d", len(tool_calls))
logger.debug("Tool node invoked: tool_calls=%d", len(tool_calls))
result = []
for tool_call in tool_calls:

View File

@ -127,11 +127,11 @@ def create_base_model(
# message handling and avoid 400 errors, we disable it here.
if config_vars["model_provider"] == "deepseek":
init_params["extra_body"] = {"thinking": {"type": "disabled"}}
logger.info("DeepSeek thinking mode disabled")
logger.debug("DeepSeek thinking mode disabled")
model = init_chat_model(**init_params)
logger.info("Base model created successfully")
logger.debug("Base model created successfully")
return model
except Exception as e:
@ -195,11 +195,11 @@ def create_title_model(
# message handling and avoid 400 errors, we disable it here.
if config_vars["model_provider"] == "deepseek":
init_params["extra_body"] = {"thinking": {"type": "disabled"}}
logger.info("DeepSeek thinking mode disabled for title model")
logger.debug("DeepSeek thinking mode disabled for title model")
model = init_chat_model(**init_params)
logger.info("Title model created successfully")
logger.debug("Title model created successfully")
return model
except Exception as e:
@ -226,7 +226,7 @@ def create_model_with_tools(
"""
try:
model_with_tools = model.bind_tools(tools)
logger.info("Model bound with %d tools successfully", len(tools))
logger.debug("Model bound with %d tools successfully", len(tools))
return model_with_tools
except Exception as e:
logger.error("Failed to bind tools to model: %s", e)

View File

@ -285,13 +285,19 @@ class AgentService:
if is_new_session:
# Create new session
copilot_mode = llm_config.get("copilot_mode", "teaching_assistant").lower()
session = await repo.create_session(
thread_id=session_id,
user_id=user_id or "",
project_id=project_id or "",
title="New Conversation",
copilot_mode=copilot_mode,
)
log.debug(
"Created new chat session: thread_id=%s, copilot_mode=%s",
session_id,
copilot_mode
)
log.debug("Created new chat session: thread_id=%s", session_id)
# Set request-scoped context variables (memory-only, not persisted)
if jwt_token:
@ -306,6 +312,14 @@ class AgentService:
)
# Build config - only thread-safe identifiers
# Determine recursion_limit based on copilot_mode
copilot_mode = llm_config.get("copilot_mode", "teaching_assistant").lower()
if copilot_mode == "troubleshooting_injection":
recursion_limit = 100 # Need more recursion depth for fault injection workflow
log.debug("Using extended recursion_limit for troubleshooting_injection mode: 100")
else:
recursion_limit = 25 # Default recursion limit
config = {
"configurable": {
"thread_id": session_id,
@ -313,10 +327,20 @@ class AgentService:
},
"metadata": {
"user_id": user_id,
"copilot_mode": copilot_mode, # Store mode in checkpoint metadata
},
"recursion_limit": recursion_limit,
}
# Build inputs
# Determine remaining_steps based on copilot_mode
# Troubleshooting injection mode needs more steps for multi-tool workflow
if copilot_mode == "troubleshooting_injection":
remaining_steps = 50 # Need more steps for fault injection workflow
log.debug("Using extended remaining_steps for troubleshooting_injection mode: 50")
else:
remaining_steps = 20 # Default for other modes
inputs = {
"messages": [
HumanMessage(
@ -326,7 +350,7 @@ class AgentService:
)
],
"llm_calls": 0,
"remaining_steps": 20,
"remaining_steps": remaining_steps,
"session_id": session_id,
"abort": False,
}
@ -650,13 +674,14 @@ class AgentService:
return convert_langchain_to_openai(msg)
async def list_sessions(
self, user_id: Optional[str] = None, limit: int = 100
self, user_id: Optional[str] = None, copilot_mode: Optional[str] = None, limit: int = 100
) -> List[Dict[str, Any]]:
"""
List chat sessions for this project.
Args:
user_id: Filter by user ID (optional)
copilot_mode: Filter by copilot mode (optional)
limit: Maximum number of sessions to return
Returns:
@ -666,7 +691,7 @@ class AgentService:
await self._get_checkpointer()
repo = ChatSessionsRepository(self._checkpointer_conn)
sessions = await repo.list_sessions(user_id=user_id, limit=limit)
sessions = await repo.list_sessions(user_id=user_id, copilot_mode=copilot_mode, limit=limit)
return [s.to_dict() for s in sessions]
async def delete_session(self, session_id: str) -> bool:

View File

@ -124,6 +124,7 @@ class ChatSessionsRepository:
user_id: str,
project_id: str,
title: str = "New Conversation",
copilot_mode: Optional[str] = None,
) -> ChatSession:
"""
Create a new chat session.
@ -133,25 +134,31 @@ class ChatSessionsRepository:
user_id: User ID
project_id: Project ID
title: Session title
copilot_mode: Copilot mode (optional)
Returns:
Created ChatSession
"""
now = datetime.utcnow().isoformat()
# Build metadata JSON
metadata = {"copilot_mode": copilot_mode} if copilot_mode else {}
metadata_json = json.dumps(metadata)
cursor = await self.conn.execute(
"""
INSERT INTO chat_sessions (
thread_id, user_id, project_id, title,
created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?)
metadata, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(thread_id, user_id, project_id, title, now, now),
(thread_id, user_id, project_id, title, metadata_json, now, now),
)
await self.conn.commit()
session_id = cursor.lastrowid
log.info(
"Created chat session: id=%s, thread_id=%s", session_id, thread_id
"Created chat session: id=%s, thread_id=%s, copilot_mode=%s",
session_id, thread_id, copilot_mode
)
return await self.get_session_by_id(session_id)
@ -202,6 +209,7 @@ class ChatSessionsRepository:
self,
user_id: Optional[str] = None,
project_id: Optional[str] = None,
copilot_mode: Optional[str] = None,
limit: int = 100,
) -> List[ChatSession]:
"""
@ -210,6 +218,7 @@ class ChatSessionsRepository:
Args:
user_id: Filter by user ID
project_id: Filter by project ID
copilot_mode: Filter by copilot mode (metadata field)
limit: Maximum number of sessions to return
Returns:
@ -225,6 +234,10 @@ class ChatSessionsRepository:
if project_id:
conditions.append("project_id = ?")
params.append(project_id)
if copilot_mode:
# Filter by JSON metadata field
conditions.append("json_extract(metadata, '$.copilot_mode') = ?")
params.append(copilot_mode)
if conditions:
query += " WHERE " + " AND ".join(conditions)

View File

@ -1,63 +0,0 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3
#
# This file is part of GNS3-Copilot project.
#
# GNS3-Copilot is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation, either version 3 of the License, or (at your
# option) any later version.
#
# 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 General Public License
# for more details.
#
# You should have received a copy of the GNU General Public License
# along with GNS3-Copilot. If not, see <https://www.gnu.org/licenses/>.
#
# Copyright (C) 2025 Yue Guobin (岳国宾)
# Author: Yue Guobin (岳国宾)
#
# Project Home: https://github.com/yueguobin/gns3-copilot
#
# Forbidden Commands List
# =======================
#
# This file contains commands that are blocked from execution in GNS3-Copilot tools.
#
# Why block these commands?
# ------------------------
# Commands listed here may:
# - Run longer than the tool timeout (e.g., traceroute)
# - Leave the device console in an unusable state for subsequent commands
# - Generate excessive traffic that could impact lab performance
#
# Format
# ------
# - One command pattern per line
# - Simple substring matching (case-insensitive)
# - Empty lines and lines starting with # are ignored
# - Match is performed on the beginning of each command
#
# Examples
# --------
# traceroute # Blocks "traceroute 8.8.8.8", "traceroute google.com", etc.
# debug # Blocks "debug ip routing", "debug ospf events", etc.
# test # Blocks "test ..., "test memory", etc.
# Network diagnostic commands that may timeout
traceroute
tracepath
tracert
# Ping flood variants
ping -f
# Debug commands that may destabilize devices
debug
# Test commands that may affect device stability
test

View File

@ -0,0 +1,22 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3
#
# Copyright (C) 2025 Yue Guobin
#
"""
Configuration package for GNS3 Copilot.
"""
from .skills_config import (
SKILLS_CONFIG,
get_skills_config,
update_skills_config,
)
__all__ = [
"SKILLS_CONFIG",
"get_skills_config",
"update_skills_config",
]

View File

@ -0,0 +1,86 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3
#
# This file is part of GNS3-Copilot project.
#
# GNS3-Copilot is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation, either version 3 of the License, or (at your
# option) any later version.
#
# 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 General Public License
# for more details.
#
# You should have received a copy of the GNU General Public License
# along with GNS3-Copilot. If not, see <https://www.gnu.org/licenses/>.
#
# Copyright (C) 2025 Yue Guobin
# Author: Yue Guobin
#
# Project Home: https://github.com/yueguobin/gns3-copilot
#
"""
Skills Configuration
This module contains configuration for the external skills repository.
"""
from gns3server.config import Config
# Default skills repository configuration
SKILLS_CONFIG = {
# Git repository URL for skills
"repo_url": "https://github.com/yueguobin/GNS3-Skills.git",
# Git branch to use
"branch": "main",
# Automatically pull updates on reload
"auto_update": True,
# Enable external skills loading
# If False, use the built-in hardcoded skills
"enabled": True,
}
def get_skills_config() -> dict:
"""
Get the skills configuration.
Priority:
1. GNS3 server config (gns3_server.conf [Server] skills_*)
2. Hardcoded defaults
Returns:
Dictionary containing skills configuration
"""
config = SKILLS_CONFIG.copy()
try:
server = Config.instance().settings.Server
if server.skills_repo_url:
config["repo_url"] = server.skills_repo_url
if server.skills_repo_branch:
config["branch"] = server.skills_repo_branch
config["auto_update"] = server.skills_auto_update
except Exception:
pass
return config
def update_skills_config(**kwargs):
"""
Update skills configuration.
Args:
**kwargs: Configuration key-value pairs to update
"""
for key, value in kwargs.items():
if key in SKILLS_CONFIG:
SKILLS_CONFIG[key] = value
else:
raise ValueError(f"Unknown configuration key: {key}")

View File

@ -102,7 +102,7 @@ class GNS3ProjectInfoTool(BaseTool):
}
# Initialize Gns3Connector using factory function
logger.info("Connecting to GNS3 server...")
logger.debug("Connecting to GNS3 server...")
server = get_gns3_connector()
if server is None:

View File

@ -102,7 +102,7 @@ class GNS3TopologyTool(BaseTool):
}
# Initialize Gns3Connector using factory function
logger.info("Connecting to GNS3 server...")
logger.debug("Connecting to GNS3 server...")
server = get_gns3_connector()
if server is None:

View File

@ -26,35 +26,27 @@
"""
Prompts Module for GNS3-Copilot
This package contains system prompts and prompt loading utilities for
the GNS3-Copilot AI agent.
This package provides system prompts loading utilities for the GNS3-Copilot AI agent.
Available prompts:
- teaching_assistant_prompt: Teaching assistant mode (diagnostics only)
- lab_automation_assistant_prompt: Lab automation mode (diagnostics + config)
All system prompts are now loaded from the external GNS3-Skills repository:
https://github.com/yueguobin/GNS3-Skills
Available prompts (loaded from external repository):
- lab_automation_assistant.md: Lab automation mode (diagnostics + config)
- teaching_assistant.md: Teaching assistant mode (diagnostics only)
- troubleshooting_injection.md: Fault injection specialist
- title.md: Title generation prompt template
Use the SkillsManager to load these prompts:
from gns3server.agent.gns3_copilot.skills.manager import skills_manager
# Load prompts
lab_prompt = skills_manager.load_prompt("lab_automation_assistant")
teaching_prompt = skills_manager.load_prompt("teaching_assistant")
injection_prompt = skills_manager.load_prompt("troubleshooting_injection")
title_prompt = skills_manager.load_prompt("title")
"""
from .teaching_assistant_prompt import TEACHING_ASSISTANT_PROMPT
from .lab_automation_assistant_prompt import LAB_AUTOMATION_ASSISTANT_PROMPT
from .prompt_loader import load_system_prompt
from .title_prompt import TITLE_PROMPT
# Dynamic version management
try:
from importlib.metadata import version
__version__ = version("gns3-copilot")
except Exception:
__version__ = "unknown"
__author__ = "Yue Guobin (岳国宾)"
__description__ = "AI-powered network automation assistant for GNS3"
__url__ = "https://github.com/yueguobin/gns3-copilot"
__all__ = [
"TEACHING_ASSISTANT_PROMPT",
"LAB_AUTOMATION_ASSISTANT_PROMPT",
"TITLE_PROMPT",
"load_system_prompt",
]
__all__ = ["load_system_prompt"]

View File

@ -1,203 +0,0 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3
#
# This file is part of GNS3-Copilot project.
#
# GNS3-Copilot is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation, either version 3 of the License, or (at your
# option) any later version.
#
# 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 General Public License
# for more details.
#
# You should have received a copy of the GNU General Public License
# along with GNS3-Copilot. If not, see <https://www.gnu.org/licenses/>.
#
# Copyright (C) 2025 Yue Guobin (岳国宾)
# Author: Yue Guobin (岳国宾)
#
# Project Home: https://github.com/yueguobin/gns3-copilot
#
"""
System Prompt for GNS3 Lab Assistant
This module contains the system prompt for the GNS3 Lab Assistant,
an AI-powered assistant that can perform both diagnostic AND configuration
operations in GNS3 network labs.
This is the full-featured assistant mode with configuration permissions enabled.
"""
# System prompt for GNS3 Lab Automation Assistant
LAB_AUTOMATION_ASSISTANT_PROMPT = """
# ROLE & MISSION
You are a **GNS3 Network Lab Assistant** - an intelligent automation assistant
designed to help users manage, diagnose, and configure network devices in GNS3
virtual laboratories.
**Your Mission**: Understand user intent and use available tools to efficiently
complete network operations, including both diagnostics and configurations.
---
# AVAILABLE TOOLS
You have access to the following tools to help users:
| Tool | Purpose | Usage |
|------|---------|-------|
| `device_skills` | Query device/protocol/feature skills | Get command syntax, troubleshooting |
| `gns3_template_reader` | Get available node templates | List templates |
| `gns3_create_node` | Create new nodes in topology | Add routers, switches, VPCS |
| `gns3_link_tool` | Create links between nodes | Connect topology |
| `gns3_start_node_tool` | Start/stop nodes | Control power state |
| `gns3_update_node_name_tool` | Update node names | Rename devices |
| `execute_multiple_device_commands` | Execute display commands | Diagnostics |
| `execute_multiple_device_config_commands` | Execute config commands | Config changes |
| `execute_vpcs_commands` | Execute VPCS commands | Configure VPCS devices |
---
# TOOL USAGE RULES
1. **Sequential Execution**: Call ONE tool at a time, wait for results
2. **Topology Awareness**: If topology is in context, DO NOT call reader again
3. **Efficient Operations**: Batch commands for multiple devices when possible
4. **Safety First**: Be cautious with destructive operations (reload, erase, format)
5. **CRITICAL - NEVER use 'exit' command**: This disconnects the Telnet/SSH session and causes all subsequent commands to fail. Never include 'exit' in command lists.
---
# WORKFLOW GUIDELINES
## Step 1: Understand Intent
Analyze what the user wants to accomplish:
- Diagnosis? Use `execute_multiple_device_commands`
- Configuration? Use `execute_multiple_device_config_commands`
- Topology changes? Use node/link management tools
## Step 2: Plan & Execute
1. Check current topology context
2. Use appropriate tool(s) for the task
3. Verify results when necessary
4. Provide clear feedback to user
## Step 3: Report Results
Clearly communicate:
- What was done
- Results (success/failure)
- Any errors encountered
- Next steps or recommendations
---
# TOPOLOGY PLANNING WORKFLOW
When user asks to create a network lab/experiment/topology:
1. **Query topology_planner skill**:
```
device_skills({"action": "get", "device_type": "topology_planner"})
```
2. **Follow the skill's guidance**:
- Use IOU image by default
- Plan IP addressing with 10.0.0.0/8 range, /24 for LANs, /30 for P2P links
- Use naming convention: R1, R2 for routers; S1, S2 for switches; PC1, PC2 for PCs
- Position nodes based on topology type (star/ring/bus/mesh/hierarchical)
- Place hub/spine nodes at center, leaf nodes radiating outward
- Use "name" field in create_gns3_node to set names directly (no separate rename step)
- Follow 6-step workflow: read templates create nodes link start verify config
3. **Output topology plan** using the skill's output_template format
---
# COMMAND EXAMPLES
## Diagnostic Commands (READ-ONLY)
```
# Cisco IOS
show version
show running-config
show ip interface brief
show ip route
show ip ospf neighbor
show ip bgp summary
debug ip routing
# Huawei VRP
display current-configuration
display ip routing-table
display ospf peer
display bgp peer
# Verification
ping 192.168.1.1
traceroute 10.0.0.1
```
## Configuration Commands
```
# Interface Configuration
interface GigabitEthernet0/0
ip address 192.168.1.1 255.255.255.0
no shutdown
# Routing Configuration
router ospf 1
network 192.168.1.0 0.0.0.255 area 0
# VLAN Configuration
vlan 10
name Sales
```
---
# RESPONSE GUIDELINES
2. **Clear Structure**:
```markdown
## Operation Summary
**Task**: [What was done]
**Result**: [Success/Failure]
## Details
[Device outputs, configurations, etc.]
```
3. **Error Handling**:
- Report errors clearly
- Suggest troubleshooting steps
- Offer to retry or investigate further
---
# SAFETY REMINDERS
While you have configuration permissions, exercise caution:
- **FORBIDDEN**: AAA/password config (enable secret, username, aaa new-model, service password-encryption, line vty) - Provide guidance only
- Avoid destructive commands (reload, erase, format) unless explicitly requested
- Warn user before making major changes
- Recommend backup for critical configurations
- Verify connectivity before routing protocol changes
---
# CURRENT TOPOLOGY
{{topology_info}}
**Note**: Topology is already retrieved. DO NOT call topology reader again unless
needed.
---
"""

View File

@ -17,7 +17,7 @@
# You should have received a copy of the GNU General Public License
# along with GNS3-Copilot. If not, see <https://www.gnu.org/licenses/>.
#
# Copyright (C) 2025 Yue Guobin (岳国宾)
# Copyright (C) 2025 Yue Guobin (<EFBFBD> Yue Guobin)
# Author: Yue Guobin (岳国宾)
#
# Project Home: https://github.com/yueguobin/gns3-copilot
@ -26,7 +26,8 @@
"""
Prompt Loader for GNS3-Copilot
This module provides utilities for loading system prompts.
This module provides utilities for loading system prompts from the external
GNS3-Skills repository.
Supports multiple prompt variants based on LLM model configuration.
Available Modes (controlled by config.copilot_mode in llm_model_configs):
@ -34,20 +35,21 @@ Available Modes (controlled by config.copilot_mode in llm_model_configs):
no configuration
- "lab_automation_assistant": Full lab automation assistant mode - diagnostics
and configuration enabled
- "troubleshooting_injection": Troubleshooting issue injection mode
All prompts are loaded from the external GNS3-Skills repository via SkillsManager.
"""
import logging
from .teaching_assistant_prompt import TEACHING_ASSISTANT_PROMPT
from .lab_automation_assistant_prompt import LAB_AUTOMATION_ASSISTANT_PROMPT
from gns3server.agent.gns3_copilot.skills.registry import get_prompt
logger = logging.getLogger(__name__)
def load_system_prompt(llm_config: dict | None = None) -> str:
"""
Load the system prompt for GNS3-Copilot.
Load the system prompt for GNS3-Copilot from the external repository.
The prompt mode is controlled by the `copilot_mode` field in the LLM
model config:
@ -55,31 +57,59 @@ def load_system_prompt(llm_config: dict | None = None) -> str:
only, no configuration
- "lab_automation_assistant": Full lab automation assistant mode -
diagnostics and configuration enabled
- "troubleshooting_injection": Troubleshooting issue injection mode -
inject network faults for practice
Prompts are loaded from the external GNS3-Skills repository via SkillsManager.
Args:
llm_config: LLM model configuration dictionary (flattened structure
from get_user_llm_config_full)
Returns:
str: The system prompt string.
str: The system prompt string, or empty string if not found.
"""
if not llm_config:
logger.info(
logger.debug(
"No LLM config provided, using default TEACHING_ASSISTANT "
"prompt mode"
)
return TEACHING_ASSISTANT_PROMPT
return _load_prompt("teaching_assistant")
# llm_config is a flattened dict with copilot_mode at the top level
# Example: {"provider": "...", "model": "...", "copilot_mode": "...", ...}
mode = llm_config.get("copilot_mode", "teaching_assistant").lower()
if mode == "lab_automation_assistant":
logger.info(
logger.debug(
"Using LAB_AUTOMATION_ASSISTANT prompt mode (diagnostics + "
"configuration)"
)
return LAB_AUTOMATION_ASSISTANT_PROMPT
return _load_prompt("lab_automation_assistant")
elif mode == "troubleshooting_injection":
logger.debug(
"Using TROUBLESHOOTING_INJECTION prompt mode (fault injection)"
)
return _load_prompt("troubleshooting_injection")
else:
logger.info("Using TEACHING_ASSISTANT prompt mode (diagnostics only)")
return TEACHING_ASSISTANT_PROMPT
logger.debug("Using TEACHING_ASSISTANT prompt mode (diagnostics only)")
return _load_prompt("teaching_assistant")
def _load_prompt(prompt_name: str) -> str:
"""
Load a prompt from the external skills repository.
Args:
prompt_name: Name of the prompt (e.g., "teaching_assistant")
Returns:
Prompt content as string, or empty string if not found
"""
prompt = get_prompt(prompt_name)
if prompt:
logger.debug(f"Loaded prompt '{prompt_name}' from external repository")
return prompt
logger.warning(f"Prompt '{prompt_name}' not found in external repository")
return ""

View File

@ -1,128 +0,0 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3
#
# This file is part of GNS3-Copilot project.
#
# GNS3-Copilot is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation, either version 3 of the License, or (at your
# option) any later version.
#
# 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 General Public License
# for more details.
#
# You should have received a copy of the GNU General Public License
# along with GNS3-Copilot. If not, see <https://www.gnu.org/licenses/>.
#
# Copyright (C) 2025 Yue Guobin (岳国宾)
# Author: Yue Guobin (岳国宾)
#
# Project Home: https://github.com/yueguobin/gns3-copilot
#
"""
System Prompt for GNS3 Network Lab Teaching Assistant
This module contains the system prompt used by the LangGraph agent
to guide network diagnostics and teaching activities.
CRITICAL: This assistant has DIAGNOSIS permissions only, NO configuration permissions.
"""
# System prompt for LangChain v1.0 agent
TEACHING_ASSISTANT_PROMPT = """
# ROLE & PERMISSIONS
You are a **GNS3 Lab Teaching Assistant**.
**Core Principle**: Teach students HOW to solve problems, not solve problems FOR them.
**Your Permissions**:
- **ALLOWED**: Read-only diagnostics (show/display/debug commands)
- **FORBIDDEN**: Any configuration changes
---
# STRICT PROHIBITIONS
1. **NEVER** call `execute_multiple_device_config_commands`
2. **NEVER** say "I've configured..." / "Configuration complete"
3. **NEVER** execute configuration commands (interface, router, ip address,
vlan, acl, route-map, etc.)
**Before EVERY response, ask yourself**: "Am I about to execute a configuration "
"operation?"
If YES Stop and provide guidance instead
If NO Proceed with diagnosis
---
# TOOL USAGE RULES
| Tool | Permission |
|------|------------|
| `get_gns3_templates` | List available device templates |
| `create_gns3_node` | Create nodes in topology |
| `create_gns3_link` | Connect nodes with links |
| `update_gns3_node_name` | Rename nodes |
| `start_gns3_node` | Start nodes for diagnostics |
| `execute_multiple_device_commands` | Only for show/display/debug |
| `execute_multiple_device_config_commands` | 🚫 **NEVER use** |
**Tool Calling Rules**:
- Call only ONE tool at a time
- Wait for result before calling next tool
- If topology is already in context, DO NOT call topology reader again
**Topology Management Permissions**:
- You CAN create and manage topology (templates, nodes, links, names)
- You CAN start nodes for diagnostic purposes
- You CANNOT stop or suspend nodes (prevents disruption of active labs)
---
# WORKFLOW
## Step 1: Diagnose
Use read-only commands to understand the problem:
```
# Cisco
show running-config, show ip route, show ip ospf neighbor, debug ip routing
# Huawei
display current-configuration, display ip routing-table, display ospf peer
# Linux
ip route, ip addr, tcpdump, ping, traceroute
```
## Step 2: Output Results
```markdown
## 🔍 Problem Diagnosis
**Root Cause**: [What you found]
---
## 💡 Solution
**Configuration Steps**:
[Cisco commands with explanations]
[Huawei commands with explanations]
**Verification**: `show command` to check success
```
---
# CURRENT TOPOLOGY
{{topology_info}}
**Note**: Topology is already retrieved. DO NOT call topology reader again unless
needed.
"""

View File

@ -1,39 +0,0 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3
#
# This file is part of GNS3-Copilot project.
#
# GNS3-Copilot is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation, either version 3 of the License, or (at your
# option) any later version.
#
# 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 General Public License
# for more details.
#
# You should have received a copy of the GNU General Public License
# along with GNS3-Copilot. If not, see <https://www.gnu.org/licenses/>.
#
# Copyright (C) 2025 Yue Guobin (岳国宾)
# Author: Yue Guobin (岳国宾)
#
# Project Home: https://github.com/yueguobin/gns3-copilot
#
"""
Title Generation Prompt for GNS3-Copilot
Prompt template for generating conversation titles.
Generates concise titles matching the conversation language.
"""
TITLE_PROMPT = """
Based on the following conversation records,
analyze the language composition and generate a concise, summary title.
Match the language of the conversation.
Only return the title, do not include any additional explanations or punctuation:
"""

View File

@ -24,26 +24,47 @@
#
"""
Device Skills Package
Skills Package
This package provides device-specific skills for GNS3 Copilot.
Skills are organized by vendor and product series for easy extensibility.
This package provides skills management for GNS3 Copilot.
All skills are loaded from the external GNS3-Skills repository.
Directory Structure:
- skills/
- registry.py # SKILLS_REGISTRY and get_skill()
- cisco/ # Cisco devices
- huawei/ # Huawei devices
- h3c/ # H3C devices
- ruijie/ # Ruijie devices
- vpcs/ # GNS3 VPCS
- generic/ # Base templates
- manager.py # SkillsManager - Git clone/pull and hot reload
- loader.py # SkillsLoader - YAML/Markdown file loading
"""
from .registry import SKILLS_REGISTRY, get_skill, DeviceSkillsTool
from .registry import (
SKILLS_REGISTRY,
INJECTION_SKILLS_REGISTRY,
get_skill,
get_injection_skill,
DeviceSkillsTool,
InjectionSkillsTool,
set_skills_manager,
get_skills_manager,
reload_injection_skills,
reload_forbidden_commands,
reload_skills_repository,
get_skills_repository_info,
)
from .manager import SkillsManager
from .loader import SkillsLoader
__all__ = [
"SKILLS_REGISTRY",
"INJECTION_SKILLS_REGISTRY",
"get_skill",
"get_injection_skill",
"DeviceSkillsTool",
"InjectionSkillsTool",
"SkillsManager",
"SkillsLoader",
"set_skills_manager",
"get_skills_manager",
"reload_injection_skills",
"reload_skills_repository",
"get_skills_repository_info",
]

View File

@ -1,13 +0,0 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# Cisco Skill Package
#
# Placeholder for future Cisco device skill implementations.
#
# Available device types:
# - cisco_ios_telnet: Cisco IOS Router (via Telnet)
# - cisco_iou_telnet: Cisco IOU L2/L3 Switch
#
# TODO: Implement Cisco IOS skill
# TODO: Implement Cisco IOU skill

View File

@ -1,18 +0,0 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# Generic Skill Package
#
# Base templates and generic skill utilities.
#
# Base skill template for reference:
# SKILL_TEMPLATE = {
# "device_type": "",
# "name": "",
# "description": "",
# "config_commands": {},
# "display_commands": {},
# "notes": [],
# "troubleshooting": {},
# "command_aliases": {},
# }

View File

@ -1,11 +0,0 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# H3C Skill Package
#
# Placeholder for future H3C (Hewlett Packard Enterprise) device skill implementations.
#
# Available device types:
# - h3c_telnet: H3C Comware Series (via Telnet)
#
# TODO: Implement H3C Comware skill

View File

@ -1,12 +0,0 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# Huawei Skill Package
#
# Placeholder for future Huawei device skill implementations.
#
# Available device types:
# - huawei_telnet: Huawei NE/AR/CE Series (via Telnet)
#
# TODO: Implement Huawei CloudEngine (CE) skill
# TODO: Implement Huawei AR skill

View File

@ -0,0 +1,239 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3
#
# This file is part of GNS3-Copilot project.
#
# GNS3-Copilot is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation, either version 3 of the License, or (at your
# option) any later version.
#
# 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 General Public License
# for more details.
#
# You should have received a copy of the GNU General Public License
# along with GNS3-Copilot. If not, see <https://www.gnu.org/licenses/>.
#
# Copyright (C) 2025 Yue Guobin
# Author: Yue Guobin
#
# Project Home: https://github.com/yueguobin/gns3-copilot
#
"""
Skills Loader
This module provides functionality to load skills from YAML files.
Supports loading injection skills and device/feature skills.
"""
import logging
import os
from pathlib import Path
from typing import Any, Dict
try:
import yaml
except ImportError:
yaml = None
logger = logging.getLogger(__name__)
class SkillsLoader:
"""
Load skills from YAML files in the skills directory.
This loader reads YAML files and converts them to the skill dictionary
format expected by the skills registry.
"""
def __init__(self, skills_dir: str):
"""
Initialize the skills loader.
Args:
skills_dir: Path to the skills directory containing YAML files
"""
self.skills_dir = Path(skills_dir)
if not self.skills_dir.exists():
logger.warning(f"Skills directory does not exist: {self.skills_dir}")
def load_injection_skills(self) -> Dict[str, Dict[str, Any]]:
"""
Load all injection skills from YAML files.
Returns:
Dictionary mapping skill keys to skill definitions
"""
if yaml is None:
logger.error("PyYAML is not installed. Cannot load skills from YAML.")
return {}
skills = {}
injection_dir = self.skills_dir / "injection"
if not injection_dir.exists():
logger.warning(f"Injection skills directory not found: {injection_dir}")
return {}
for yaml_file in injection_dir.glob("*.yaml"):
try:
skill_data = self._load_yaml(yaml_file)
if not skill_data:
logger.warning(f"Skipping empty YAML file: {yaml_file}")
continue
# Generate key from filename (e.g., "ospf_issues.yaml" -> "injection_ospf")
skill_key = f"injection_{yaml_file.stem}"
skills[skill_key] = skill_data
logger.debug(f"Loaded injection skill: {skill_key} from {yaml_file}")
except Exception as e:
logger.error(f"Failed to load skill from {yaml_file}: {e}")
logger.debug(f"Loaded {len(skills)} injection skills from {injection_dir}")
return skills
def load_device_skills(self) -> Dict[str, Dict[str, Any]]:
"""
Load all device/feature skills from YAML files.
Returns:
Dictionary mapping skill keys to skill definitions
"""
if yaml is None:
logger.error("PyYAML is not installed. Cannot load skills from YAML.")
return {}
skills = {}
device_dir = self.skills_dir / "device"
if not device_dir.exists():
logger.warning(f"Device skills directory not found: {device_dir}")
return {}
for yaml_file in device_dir.glob("*.yaml"):
try:
skill_data = self._load_yaml(yaml_file)
if not skill_data:
logger.warning(f"Skipping empty YAML file: {yaml_file}")
continue
# Use device_type from YAML content as the key
# Fallback to filename stem if device_type not present
skill_key = skill_data.get("device_type") if isinstance(skill_data, dict) else None
if not skill_key:
skill_key = yaml_file.stem
logger.warning(f"No device_type in {yaml_file}, using filename '{skill_key}' as key")
skills[skill_key] = skill_data
logger.debug(f"Loaded device skill: {skill_key} from {yaml_file}")
except Exception as e:
logger.error(f"Failed to load skill from {yaml_file}: {e}")
logger.info(f"Loaded {len(skills)} device skills from {device_dir}")
return skills
def load_prompt(self, prompt_name: str) -> str:
"""
Load a prompt from a markdown file.
Args:
prompt_name: Name of the prompt file (without .md extension)
Returns:
Prompt content as string, or empty string if not found
"""
prompts_dir = self.skills_dir / "prompts"
prompt_file = prompts_dir / f"{prompt_name}.md"
if not prompt_file.exists():
logger.warning(f"Prompt file not found: {prompt_file}")
return ""
try:
with open(prompt_file, "r", encoding="utf-8") as f:
content = f.read()
logger.debug(f"Loaded prompt: {prompt_name} from {prompt_file}")
return content
except Exception as e:
logger.error(f"Failed to load prompt from {prompt_file}: {e}")
return ""
def load_forbidden_commands(self) -> list:
"""
Load forbidden commands from the config directory.
Returns:
List of forbidden command patterns, or empty list if not found
"""
config_dir = self.skills_dir / "config"
config_file = config_dir / "forbidden_commands.txt"
if not config_file.exists():
logger.warning(f"Forbidden commands file not found: {config_file}")
return []
try:
commands = []
with open(config_file, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#"):
continue
commands.append(line.lower())
logger.debug(f"Loaded {len(commands)} forbidden commands from {config_file}")
return commands
except Exception as e:
logger.error(f"Failed to load forbidden commands from {config_file}: {e}")
return []
def _load_yaml(self, file_path: Path) -> Dict[str, Any]:
"""
Load a YAML file and return its content.
Args:
file_path: Path to the YAML file
Returns:
Parsed YAML content as dictionary, or empty dict if file is empty
"""
with open(file_path, "r", encoding="utf-8") as f:
data = yaml.safe_load(f)
if not isinstance(data, dict):
logger.warning(f"YAML file {file_path} is empty or has invalid format")
return {}
return data
def validate_skill_format(self, skill_data: Dict[str, Any]) -> bool:
"""
Validate that a skill dictionary has the required fields.
Args:
skill_data: Skill dictionary to validate
Returns:
True if valid, False otherwise
"""
required_fields = ["name", "description", "issues"]
for field in required_fields:
if field not in skill_data:
logger.error(f"Skill missing required field: {field}")
return False
if not isinstance(skill_data["issues"], dict):
logger.error("Skill 'issues' field must be a dictionary")
return False
for issue_key, issue_data in skill_data["issues"].items():
if not isinstance(issue_data, dict):
logger.error(f"Issue '{issue_key}' must be a dictionary")
return False
issue_required = ["name", "description"]
for field in issue_required:
if field not in issue_data:
logger.error(f"Issue '{issue_key}' missing required field: {field}")
return False
return True

View File

@ -0,0 +1,431 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3
#
# This file is part of GNS3-Copilot project.
#
# GNS3-Copilot is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation, either version 3 of the License, or (at your
# option) any later version.
#
# 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 General Public License
# for more details.
#
# You should have received a copy of the GNU General Public License
# along with GNS3-Copilot. If not, see <https://www.gnu.org/licenses/>.
#
# Copyright (C) 2025 Yue Guobin
# Author: Yue Guobin
#
# Project Home: https://github.com/yueguobin/gns3-copilot
#
"""
Skills Manager
This module provides functionality to manage the skills repository,
including Git operations and hot reload of skills.
"""
import logging
import os
from pathlib import Path
from typing import Optional, Dict, Any
try:
import git
except ImportError:
git = None
from gns3server.config import Config
from .loader import SkillsLoader
logger = logging.getLogger(__name__)
# Git timeout settings for network operations.
# Applied per-command via the `env` parameter to avoid polluting
# the global process environment.
_GIT_TIMEOUT_ENV = {
'GIT_HTTP_TIMEOUT': '10', # Connection timeout (default: 120s)
'GIT_HTTP_LOW_SPEED_TIME': '5', # Slow speed threshold window
'GIT_HTTP_LOW_SPEED_LIMIT': '1000', # < 1 KB/s = slow → abort
}
class SkillsManager:
"""
Manage skills repository and hot reload functionality.
This class handles:
- Cloning the skills repository
- Pulling latest updates
- Hot reloading skills into memory
- Version tracking
"""
def __init__(
self,
repo_url: str = None,
branch: str = "main",
auto_update: bool = False
):
"""
Initialize the skills manager.
Args:
repo_url: Git repository URL (default: https://github.com/yueguobin/GNS3-Skills.git)
branch: Git branch to use (default: "main")
auto_update: Whether to automatically pull updates on reload
"""
if repo_url is None:
repo_url = "https://github.com/yueguobin/GNS3-Skills.git"
# Get local path from GNS3 config directory
config_dir = Config.instance().config_dir
local_path = os.path.join(config_dir, "skills")
self.repo_url = repo_url
self.local_path = Path(local_path)
self.branch = branch
self.auto_update = auto_update
self.loader = SkillsLoader(str(local_path))
self._repo: Optional["git.Repo"] = None
self._prompt_count = 0
if git is None:
logger.warning("GitPython is not installed. Skills management features will be limited.")
def initialize(self) -> bool:
"""
Initialize the skills repository.
Behavior:
- If no local repo clone from remote
- If local repo exists open and run _update_if_needed()
Returns:
True if successful, False otherwise
"""
try:
if not self.local_path.exists():
logger.info(f"Creating skills directory: {self.local_path}")
self.local_path.mkdir(parents=True, exist_ok=True)
if not (self.local_path / ".git").exists():
return self._clone()
# Repo exists, open it
try:
self._repo = git.Repo(self.local_path)
except Exception as e:
logger.error(f"Failed to open existing repository: {e}")
return False
self._update_if_needed()
return True
except Exception as e:
logger.error(f"Failed to initialize skills repository: {e}")
return False
def _update_if_needed(self) -> None:
"""
Check local repo status and pull updates if safe.
- Uncommitted changes warn and skip
- Behind remote pull
- Up to date nothing
- Network error use existing files, log warning
"""
if git is None or self._repo is None:
return
# Check for uncommitted changes
if self._repo.is_dirty(untracked_files=True):
logger.warning(
"Skills repository has uncommitted changes, skipping pull. "
"Commit or stash changes in %s to enable automatic updates.",
self.local_path
)
return
# Fetch remote
try:
origin = self._repo.remotes.origin
origin.fetch(env=_GIT_TIMEOUT_ENV)
except Exception as e:
logger.warning(f"Failed to fetch remote: {e}. Using local files.")
return
# Check if behind and pull
try:
behind_commits = list(self._repo.iter_commits(
f'{self.branch}..origin/{self.branch}'
))
if behind_commits:
logger.info(
"Skills repository is behind by %d commit(s), pulling...",
len(behind_commits)
)
origin.pull(self.branch, env=_GIT_TIMEOUT_ENV)
logger.info(f"Updated to commit {self.get_current_version()}")
else:
logger.info("Skills repository is up to date")
except Exception as e:
logger.warning(f"Failed to pull updates: {e}. Using local files.")
def _clone(self) -> bool:
"""
Clone the skills repository.
Returns:
True if successful, False otherwise
"""
if git is None:
logger.error("GitPython is not installed. Cannot clone repository.")
return False
try:
self._repo = git.Repo.clone_from(
self.repo_url,
self.local_path,
branch=self.branch,
env=_GIT_TIMEOUT_ENV
)
logger.info(f"Successfully cloned skills repository to {self.local_path}")
return True
except git.GitCommandError as e:
logger.error(f"Git clone failed: {e}")
return False
def reload_skills(self) -> bool:
"""
Hot reload skills from YAML files into the registry.
Loads the latest skill definitions from YAML files and updates
the INJECTION_SKILLS_REGISTRY.
Returns:
True if successful, False otherwise
"""
try:
# Import here to avoid circular dependency
from .registry import INJECTION_SKILLS_REGISTRY, SKILLS_REGISTRY
# Load new injection skills from YAML files
new_injection_skills = self.loader.load_injection_skills()
if not new_injection_skills:
logger.warning("No injection skills loaded, keeping existing skills")
return False
# Validate injection skills
for skill_key, skill_data in new_injection_skills.items():
if not self.loader.validate_skill_format(skill_data):
logger.error(f"Invalid skill format for {skill_key}, skipping")
continue
# Load new device/feature skills from YAML files
new_device_skills = self.loader.load_device_skills()
# Update registries (safe replace - never leaves dict empty)
for k in list(INJECTION_SKILLS_REGISTRY):
if k not in new_injection_skills:
del INJECTION_SKILLS_REGISTRY[k]
INJECTION_SKILLS_REGISTRY.update(new_injection_skills)
for k in list(SKILLS_REGISTRY):
if k not in new_device_skills:
del SKILLS_REGISTRY[k]
SKILLS_REGISTRY.update(new_device_skills)
logger.debug(f"Successfully reloaded {len(new_injection_skills)} injection skills and {len(new_device_skills)} device skills")
return True
except Exception as e:
logger.error(f"Failed to reload skills: {e}")
return False
def reload_prompts(self) -> bool:
"""
Hot reload prompts from Markdown files.
Loads the latest prompt definitions from Markdown files in
the skills repository.
Returns:
True if successful, False otherwise
"""
try:
# Available prompt names
prompt_names = [
"lab_automation_assistant",
"teaching_assistant",
"troubleshooting_injection",
"title"
]
# Load all prompts
loaded_count = 0
for prompt_name in prompt_names:
prompt_content = self.loader.load_prompt(prompt_name)
if prompt_content:
loaded_count += 1
else:
logger.warning(f"Failed to load prompt: {prompt_name}")
if loaded_count == 0:
logger.warning("No prompts loaded, keeping existing prompts")
return False
self._prompt_count = loaded_count
logger.debug(f"Successfully reloaded {loaded_count} prompts")
return True
except Exception as e:
logger.error(f"Failed to reload prompts: {e}")
return False
def load_prompt(self, prompt_name: str) -> str:
"""
Load a specific prompt from the skills repository.
Args:
prompt_name: Name of the prompt (without .md extension)
Returns:
Prompt content as string, or empty string if not found
"""
try:
return self.loader.load_prompt(prompt_name)
except Exception as e:
logger.error(f"Failed to load prompt '{prompt_name}': {e}")
return ""
def load_forbidden_commands(self) -> list:
"""
Load forbidden command patterns from the skills repository.
Returns:
List of forbidden command patterns, or empty list if not found
"""
try:
return self.loader.load_forbidden_commands()
except Exception as e:
logger.error(f"Failed to load forbidden commands: {e}")
return []
def get_current_version(self) -> str:
"""
Get the current git commit hash of the skills repository.
Returns:
Commit hash as string, or empty string if not available
"""
if git is None or self._repo is None:
try:
self._repo = git.Repo(self.local_path)
except Exception:
return ""
try:
return self._repo.head.commit.hexsha
except Exception:
return ""
def get_skill_count(self) -> int:
"""
Get the number of currently loaded injection skills.
Returns:
Number of skills in the registry
"""
try:
from .registry import INJECTION_SKILLS_REGISTRY
return len(INJECTION_SKILLS_REGISTRY)
except Exception:
return 0
def get_prompt_count(self) -> int:
"""
Get the number of currently loaded prompts.
Returns:
Number of prompts loaded
"""
return self._prompt_count
def get_repository_info(self) -> Dict[str, Any]:
"""
Get information about the skills repository.
Returns:
Dictionary containing repository information
"""
return {
"repo_url": self.repo_url,
"local_path": str(self.local_path),
"branch": self.branch,
"current_version": self.get_current_version(),
"skill_count": self.get_skill_count(),
"prompt_count": self.get_prompt_count(),
"auto_update": self.auto_update,
"is_initialized": (self.local_path / ".git").exists()
}
def rollback(self, commit_hash: str) -> bool:
"""
Rollback the skills repository to a specific commit.
Args:
commit_hash: Git commit hash to rollback to
Returns:
True if successful, False otherwise
"""
if git is None:
logger.error("GitPython is not installed. Cannot rollback.")
return False
try:
if self._repo is None:
self._repo = git.Repo(self.local_path)
self._repo.git.reset("--hard", commit_hash)
logger.info(f"Successfully rolled back to commit {commit_hash}")
# Reload skills after rollback
return self.reload_skills()
except git.GitCommandError as e:
logger.error(f"Git rollback failed: {e}")
return False
def get_available_versions(self, limit: int = 10) -> list:
"""
Get a list of recent commit hashes.
Args:
limit: Maximum number of commits to return
Returns:
List of commit information dictionaries
"""
if git is None:
return []
try:
if self._repo is None:
self._repo = git.Repo(self.local_path)
commits = []
for commit in self._repo.iter_commits(max_count=limit):
commits.append({
"hash": commit.hexsha,
"message": commit.message.strip(),
"author": str(commit.author),
"date": commit.committed_datetime.isoformat()
})
return commits
except Exception as e:
logger.error(f"Failed to get commit history: {e}")
return []

View File

@ -24,12 +24,15 @@
#
"""
Skill Registry and DeviceSkillsTool
Skill Registry and Tools
This module provides:
- SKILLS_REGISTRY: A dictionary mapping device_type to skill definitions
- get_skill(): Function to retrieve skill for a device_type
- DeviceSkillsTool: LangChain tool for LLM to query skills
- SKILLS_REGISTRY: Device/feature skills (VPCS, topology, etc.)
- INJECTION_SKILLS_REGISTRY: Fault injection skills only
- get_skill(): Function to retrieve device/feature skills
- get_injection_skill(): Function to retrieve injection skills
- DeviceSkillsTool: LangChain tool for device/feature skills
- InjectionSkillsTool: LangChain tool for injection skills
"""
import json
@ -39,47 +42,337 @@ from typing import Any
from langchain.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun
# command_filter imports are done locally in functions
# to avoid circular import (command_filter also imports from registry)
logger = logging.getLogger(__name__)
# Import all skill modules to register them
from gns3server.agent.gns3_copilot.skills.vpcs import VPCS_SKILL
from gns3server.agent.gns3_copilot.skills.topology import TOPOLOGY_PLANNER_SKILL
# Device/Feature skills registry - loaded from external repository
SKILLS_REGISTRY: dict[str, dict[str, Any]] = {}
# Global skill registry - maps device_type to skill definition
SKILLS_REGISTRY: dict[str, dict[str, Any]] = {
"gns3_vpcs_telnet": VPCS_SKILL,
"topology_planner": TOPOLOGY_PLANNER_SKILL,
# Add more skills here as they are implemented
# "huawei_telnet": HUAWEI_SKILL,
# "ruijie_telnet": RUIJIE_SKILL,
# Protocol/Feature skills:
# "ospf": OSPF_SKILL,
# "bgp": BGP_SKILL,
# "mpls": MPLS_SKILL,
}
# Injection skills registry - fault injection skills only
# This registry can be hot-reloaded via SkillsManager
INJECTION_SKILLS_REGISTRY: dict[str, dict[str, Any]] = {}
# Global skills manager instance for hot reload
_skills_manager = None
_init_in_progress = False
_init_complete = False
def set_skills_manager(manager):
"""
Set the global skills manager instance.
The skills manager handles Git operations and hot reload of skills.
Args:
manager: SkillsManager instance
"""
global _skills_manager
_skills_manager = manager
def _ensure_skills_manager():
"""
Initialize the SkillsManager (runs once, idempotent).
Reads config, creates SkillsManager, clones/pulls repo,
and loads skills/prompts into memory. Safe to call from
background threads - uses _init_in_progress to prevent
concurrent initialization.
"""
global _skills_manager, _init_in_progress, _init_complete
if _skills_manager is not None or _init_complete:
return
if _init_in_progress:
return
_init_in_progress = True
try:
from gns3server.agent.gns3_copilot.configs.skills_config import get_skills_config
from gns3server.agent.gns3_copilot.skills.manager import SkillsManager
config = get_skills_config()
if not config.get("enabled", False):
logger.info("External skills repository is disabled")
return
logger.info("Initializing SkillsManager")
manager = SkillsManager(
repo_url=config.get("repo_url"),
branch=config.get("branch", "main"),
auto_update=config.get("auto_update", False)
)
if not manager.initialize():
logger.error("Failed to initialize skills repository")
return
if manager.reload_skills():
logger.debug(f"Loaded {manager.get_skill_count()} injection skills")
else:
logger.warning("Failed to reload injection skills")
if manager.reload_prompts():
logger.debug(f"Loaded {manager.get_prompt_count()} prompts")
else:
logger.warning("Failed to reload prompts")
_skills_manager = manager
logger.debug("SkillsManager initialized successfully")
except Exception as e:
logger.error(f"Error initializing skills manager: {e}", exc_info=True)
finally:
_init_complete = True
def get_skills_manager():
"""
Get the global skills manager instance, initializing on first access.
Returns:
SkillsManager instance or None
"""
_ensure_skills_manager()
return _skills_manager
def reload_skills_repository() -> dict[str, Any]:
"""
Reload the entire skills repository.
Performs one git update check, then reloads all skills, prompts,
and forbidden commands from local files.
Returns:
Dictionary with combined reload results.
"""
manager = get_skills_manager()
if manager is None:
return {
"success": False,
"message": "Skills manager not initialized",
}
# One git update for the entire repository
manager._update_if_needed()
# Reload everything from local files
skills_ok = manager.reload_skills()
prompts_ok = manager.reload_prompts()
# Reload forbidden commands (local import to avoid circular dependency)
from gns3server.agent.gns3_copilot.utils.command_filter import reload_forbidden_commands as _reload_fc
from gns3server.agent.gns3_copilot.utils.command_filter import get_forbidden_commands
_reload_fc()
forbidden_commands = get_forbidden_commands()
return {
"success": skills_ok or prompts_ok,
"skills": skills_ok,
"skill_count": manager.get_skill_count(),
"prompts": prompts_ok,
"prompt_count": manager.get_prompt_count(),
"forbidden_commands": len(forbidden_commands),
"version": manager.get_current_version(),
}
def reload_injection_skills() -> dict[str, Any]:
"""
Trigger hot reload of injection skills.
This function uses the global skills manager to pull latest changes
from the skills repository and reload the INJECTION_SKILLS_REGISTRY.
Returns:
Dictionary with status information:
{
"success": bool,
"message": str,
"skill_count": int,
"version": str
}
"""
manager = get_skills_manager()
if manager is None:
return {
"success": False,
"message": "Skills manager not initialized",
"skill_count": len(INJECTION_SKILLS_REGISTRY),
"version": ""
}
try:
success = manager.reload_skills()
return {
"success": success,
"message": "Skills reloaded successfully" if success else "Failed to reload skills",
"skill_count": manager.get_skill_count(),
"version": manager.get_current_version()
}
except Exception as e:
logger.error(f"Error during skills reload: {e}")
return {
"success": False,
"message": f"Error: {str(e)}",
"skill_count": len(INJECTION_SKILLS_REGISTRY),
"version": ""
}
def reload_prompts() -> dict[str, Any]:
"""
Trigger hot reload of system prompts.
This function uses the global skills manager to pull latest changes
from the skills repository and reload prompts from disk.
Returns:
Dictionary with status information:
{
"success": bool,
"message": str,
"prompt_count": int,
"version": str
}
"""
manager = get_skills_manager()
if manager is None:
return {
"success": False,
"message": "Skills manager not initialized",
"prompt_count": 0,
"version": ""
}
try:
success = manager.reload_prompts()
return {
"success": success,
"message": "Prompts reloaded successfully" if success else "Failed to reload prompts",
"prompt_count": manager.get_prompt_count(),
"version": manager.get_current_version()
}
except Exception as e:
logger.error(f"Error during prompts reload: {e}")
manager = get_skills_manager()
return {
"success": False,
"message": f"Error: {str(e)}",
"prompt_count": manager.get_prompt_count() if manager else 0,
"version": ""
}
def reload_forbidden_commands() -> dict[str, Any]:
"""
Hot reload forbidden commands from the skills repository.
Directly loads and caches commands from the skills repository.
Returns:
Dictionary with status information:
{
"success": bool,
"message": str,
"command_count": int,
"version": str
}
"""
try:
from gns3server.agent.gns3_copilot.utils.command_filter import reload_forbidden_commands as _reload
from gns3server.agent.gns3_copilot.utils.command_filter import get_forbidden_commands
_reload()
commands = get_forbidden_commands()
manager = get_skills_manager()
return {
"success": True,
"message": "Forbidden commands reloaded",
"command_count": len(commands),
"version": manager.get_current_version() if manager else ""
}
except Exception as e:
logger.error(f"Error during forbidden commands reload: {e}")
return {
"success": False,
"message": f"Error: {str(e)}",
"command_count": 0,
"version": ""
}
def get_prompt(prompt_name: str) -> str:
"""
Get a system prompt by name, always loading from disk.
Args:
prompt_name: Name of the prompt (e.g., "teaching_assistant")
Returns:
Prompt content as string, or empty string if not found
"""
# Always load from skills manager (no cache), triggers lazy init
manager = get_skills_manager()
if manager:
try:
prompt = manager.load_prompt(prompt_name)
if prompt:
return prompt
except Exception as e:
logger.error(f"Error loading prompt '{prompt_name}': {e}")
logger.warning(f"Prompt not found: {prompt_name}")
return ""
def get_skills_repository_info() -> dict[str, Any]:
"""
Get information about the skills repository.
Returns:
Dictionary with repository information
"""
manager = get_skills_manager()
if manager is None:
return {
"initialized": False,
"message": "Skills manager not initialized"
}
return manager.get_repository_info()
def get_skill(
device_type: str,
category: str | None = None,
operation: str = "all"
detail: str = "full",
issue: str | None = None,
) -> dict[str, Any]:
"""
Get skill by device_type, optionally filtered by category.
Get skill by device_type, with configurable detail level.
Args:
device_type: The device type identifier (e.g., "gns3_vpcs_telnet", "huawei_telnet")
category: Optional category filter - "device", "protocol", "feature"
operation: Filter by operation type - "config", "diagnosis", or "all" (default)
device_type: The device type identifier
category: Optional category filter
detail: Detail level - "index" (names only), "summary" (+desc/sev/diff), "full" (all)
issue: Optional specific issue key to retrieve
Returns:
Skill dictionary containing commands, notes, and troubleshooting info,
or error dict if device_type not found
Skill dictionary (detail varies by level), or error dict
"""
skill = SKILLS_REGISTRY.get(device_type, {})
if not skill:
# Try to find by name if not found by device_type
for did, s in SKILLS_REGISTRY.items():
if s.get("name", "").lower() == device_type.lower():
skill = s
@ -91,7 +384,6 @@ def get_skill(
"available_device_types": list(SKILLS_REGISTRY.keys()),
}
# Filter by category if specified
if category:
skill_category = skill.get("category", "")
if category.lower() != skill_category.lower():
@ -104,39 +396,56 @@ def get_skill(
],
}
if operation == "config":
issues = skill.get("issues", {})
# Single issue lookup (most token-efficient)
if issue:
issue_def = issues.get(issue)
if not issue_def:
return {
"error": f"Unknown issue '{issue}' in {device_type}",
"available_issues": list(issues.keys()),
}
return {
"device_type": device_type,
"skill_name": skill.get("name"),
"issue": {issue: issue_def},
}
if detail == "index":
# Minimal: only issue keys and names (90%+ token savings)
return {
"device_type": device_type,
"name": skill.get("name"),
"category": skill.get("category"),
"description": skill.get("description"),
"config_commands": skill.get("config_commands", {}),
"issues": {k: v["name"] for k, v in issues.items()},
}
elif operation == "diagnosis":
if detail == "summary":
# Moderate: names + description + severity + difficulty
return {
"device_type": device_type,
"name": skill.get("name"),
"category": skill.get("category"),
"display_commands": skill.get("display_commands", {}),
"troubleshooting": skill.get("troubleshooting", {}),
"description": skill.get("description"),
"issues": {
k: {
"name": v["name"],
"description": v.get("description", ""),
"severity": v.get("severity", ""),
"difficulty": v.get("difficulty", ""),
}
for k, v in issues.items()
},
}
else:
# Return full skill
result = dict(skill)
result["device_type"] = device_type
return result
# Full detail (original behavior)
result = dict(skill)
result["device_type"] = device_type
return result
def list_available_skills(category: str | None = None) -> list[dict[str, str]]:
"""
List all available skills, optionally filtered by category.
Args:
category: Optional category filter - "device", "protocol", "feature"
Returns:
List of dicts with device_type, name, and category
"""
"""List all available device/feature skills, optionally filtered by category."""
skills = []
for did, skill in SKILLS_REGISTRY.items():
if category:
@ -155,68 +464,130 @@ def list_available_skills(category: str | None = None) -> list[dict[str, str]]:
return skills
def get_injection_skill(
device_type: str,
detail: str = "full",
issue: str | None = None,
) -> dict[str, Any]:
"""
Get injection fault skill by device_type, with configurable detail level.
Args:
device_type: The injection fault type (e.g., "injection_ospf")
detail: Detail level - "index" (names only), "summary" (+desc/sev/diff), "full" (all)
issue: Optional specific issue key to retrieve
Returns:
Skill dictionary (detail varies by level), or error dict
"""
skill = INJECTION_SKILLS_REGISTRY.get(device_type, {})
if not skill:
return {
"error": f"Unknown injection fault type: {device_type}",
"available_fault_types": list(INJECTION_SKILLS_REGISTRY.keys()),
"hint": "Use {'action': 'list'} to see all available fault types"
}
issues = skill.get("issues", {})
# Single issue lookup (most token-efficient)
if issue:
issue_def = issues.get(issue)
if not issue_def:
return {
"error": f"Unknown issue '{issue}' in {device_type}",
"available_issues": list(issues.keys()),
}
return {
"device_type": device_type,
"skill_name": skill.get("name"),
"issue": {issue: issue_def},
}
if detail == "index":
# Minimal: only issue keys and names (90%+ token savings)
return {
"device_type": device_type,
"name": skill.get("name"),
"description": skill.get("description"),
"issues": {k: v["name"] for k, v in issues.items()},
}
if detail == "summary":
# Moderate: names + description + severity + difficulty
return {
"device_type": device_type,
"name": skill.get("name"),
"description": skill.get("description"),
"issues": {
k: {
"name": v["name"],
"description": v.get("description", ""),
"severity": v.get("severity", ""),
"difficulty": v.get("difficulty", ""),
}
for k, v in issues.items()
},
}
# Full detail (original behavior)
result = dict(skill)
result["device_type"] = device_type
return result
def list_available_injection_skills(context: list[str] | None = None) -> list[dict[str, str]]:
"""
List available injection fault skills, optionally filtered by context.
Args:
context: List of protocol/service keywords (e.g., ["ospf", "bgp", "vlan"]).
Only returns skills whose category matches any keyword.
Returns:
List of skill info dicts with device_type, name, and category.
"""
skills = []
for did, skill in INJECTION_SKILLS_REGISTRY.items():
category = skill.get("category", "")
if context:
# Match if skill category contains any context keyword
category_lower = category.lower()
if not any(kw.lower() in category_lower or kw.lower() in did.lower() for kw in context):
continue
skills.append({
"device_type": did,
"name": skill.get("name", did),
"category": category,
})
return skills
class DeviceSkillsTool(BaseTool):
"""
LangChain tool for querying device-specific skills.
LangChain tool for querying device/feature skills.
Use this tool to get device-specific command syntax, examples,
and troubleshooting guidance before executing commands.
Example:
# Get VPCS skill by device_type
tool.run('{"device_type": "gns3_vpcs_telnet"}')
# Get OSPF protocol skill
tool.run('{"device_type": "ospf", "category": "protocol"}')
# Get only config commands
tool.run('{"device_type": "gns3_vpcs_telnet", "operation": "config"}')
# Get only diagnosis commands
tool.run('{"device_type": "gns3_vpcs_telnet", "operation": "diagnosis"}')
# List all available skills
tool.run('{"action": "list"}')
# List skills by category
tool.run('{"action": "list", "category": "device"}')
Use this tool to get device command knowledge, topology planning skills, etc.
For fault injection skills, use InjectionSkillsTool instead.
"""
name: str = "device_skills"
description: str = """
Get or list device/ protocol/ feature specific skills and command knowledge.
Get or list device and feature specific skills.
Use this tool BEFORE executing device commands to understand:
- Command syntax for the specific device type
- Configuration command examples
- Display/diagnostic command syntax
- Troubleshooting guidance
Provides access to device command knowledge (VPCS), topology planning, etc.
For fault injection skills, use the injection_skills tool.
INPUT FORMAT (JSON string):
{
"action": "get", # Optional: "get" (default) or "list"
"device_type": "gns3_vpcs_telnet", # Required for action="get": device type identifier
"category": "device", # Optional: "device", "protocol", "feature"
"operation": "all" # Optional: "config", "diagnosis", or "all" (default)
"action": "get", # "get" (default) or "list"
"device_type": "gns3_vpcs_telnet", # Required for action="get"
"detail": "full" # "full" (default) for complete skill information
}
For action="list":
{
"action": "list",
"category": "device" # Optional: filter by category
}
OUTPUT:
- Skill name and description
- Command syntax with parameters
- Usage examples
- Troubleshooting tips
- Important notes
Available categories:
- "device": Device-specific skills (VPCS, routers, switches)
- "protocol": Network protocol skills (OSPF, BGP, MPLS)
- "feature": Feature skills (ACL, QoS, NAT)
{"action": "list"} # Lists all available device/feature skills
"""
def _run(
@ -225,18 +596,9 @@ class DeviceSkillsTool(BaseTool):
run_manager: CallbackManagerForToolRun | None = None,
**kwargs: Any,
) -> str:
"""
Execute the device skills lookup.
"""Execute the device skills lookup."""
logger.debug("DeviceSkillsTool invoked with input: %s", tool_input)
Args:
tool_input: JSON string or dict with device_type and optional operation/category
Returns:
JSON string with skill information or skill list
"""
logger.info("DeviceSkillsTool invoked with input: %s", tool_input)
# Parse input
if isinstance(tool_input, str):
try:
params = json.loads(tool_input)
@ -251,14 +613,12 @@ class DeviceSkillsTool(BaseTool):
action = params.get("action", "get")
if action == "list":
category = params.get("category")
skills = list_available_skills(category)
skills = list_available_skills()
return json.dumps({
"category": category or "all",
"count": len(skills),
"skills": skills
}, ensure_ascii=False, indent=2)
# Default action: "get"
device_type = params.get("device_type")
if not device_type:
return json.dumps({
@ -268,9 +628,100 @@ class DeviceSkillsTool(BaseTool):
}, ensure_ascii=False, indent=2)
category = params.get("category")
operation = params.get("operation", "all")
detail = params.get("detail", "full")
issue = params.get("issue")
# Get skill
skill = get_skill(device_type, category, operation)
skill = get_skill(device_type, category, detail=detail, issue=issue)
return json.dumps(skill, ensure_ascii=False, indent=2)
class InjectionSkillsTool(BaseTool):
"""
LangChain tool for querying fault injection skills.
Use this tool to list available injection fault types and get fault details.
"""
name: str = "injection_skills"
description: str = """
Get or list network fault injection skills for troubleshooting practice.
REQUIRED: When action="list", you MUST always pass "context" with the
protocols/services found in your topology analysis.
Example: {"action": "list", "context": ["ospf", "bgp", "mpls", "vlan", "stp"]}
TOKEN-EFFICIENT USAGE:
1. List faults for YOUR topology protocols (REQUIRED):
{"action": "list", "context": ["ospf", "bgp"]}
2. Get specific fault details:
{"device_type": "injection_ospf", "issue": "ospf_hello_dead_mismatch"}
{"device_type": "injection_ospf", "detail": "index"}
PARAMETERS:
- action: "list" or "get"
- context: [str] - REQUIRED for action="list". Protocols from your topology.
- device_type: Required for action="get" (e.g., "injection_ospf")
- detail: "index" | "summary" | "full"
- issue: Get single fault detail by key
"""
def _run(
self,
tool_input: str | dict[str, Any],
run_manager: CallbackManagerForToolRun | None = None,
**kwargs: Any,
) -> str:
"""Execute the injection skills lookup."""
logger.debug("InjectionSkillsTool invoked with input: %s", tool_input)
if isinstance(tool_input, str):
try:
params = json.loads(tool_input)
except json.JSONDecodeError as e:
return json.dumps({
"error": f"Invalid JSON input: {e}",
"hint": 'Expected format: {"device_type": "xxx"} or {"action": "list"}'
}, ensure_ascii=False, indent=2)
else:
params = tool_input
action = params.get("action", "get")
if action == "list":
context = params.get("context")
if not context or not isinstance(context, list) or len(context) == 0:
return json.dumps({
"error": "context parameter is required when action='list'",
"hint": "Analyze the topology and device configurations first, "
"then pass the protocols/services you found as context. "
'Example: {"action": "list", "context": ["ospf", "bgp", "vlan"]}',
"available_categories": sorted(set(
skill.get("category", "")
for skill in INJECTION_SKILLS_REGISTRY.values()
))
}, ensure_ascii=False, indent=2)
skills = list_available_injection_skills(context=context)
logger.debug(f"Injection skills filtered by context={context}: {len(skills)} matching")
return json.dumps({
"count": len(skills),
"total_available": len(INJECTION_SKILLS_REGISTRY),
"context": context,
"fault_types": skills
}, ensure_ascii=False, indent=2)
device_type = params.get("device_type")
if not device_type:
return json.dumps({
"error": "Missing required field: device_type",
"available_fault_types": list(INJECTION_SKILLS_REGISTRY.keys()),
"hint": 'Use {"action": "list"} to see all available fault types'
}, ensure_ascii=False, indent=2)
detail = params.get("detail", "full")
issue = params.get("issue")
skill = get_injection_skill(device_type, detail=detail, issue=issue)
return json.dumps(skill, ensure_ascii=False, indent=2)

View File

@ -1,11 +0,0 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# Ruijie Skill Package
#
# Placeholder for future Ruijie Networks device skill implementations.
#
# Available device types:
# - ruijie_telnet: Ruijie RGOS (via Telnet)
#
# TODO: Implement Ruijie RGOS skill

View File

@ -1,34 +0,0 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3
#
# This file is part of GNS3-Copilot project.
#
# GNS3-Copilot is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation, either version 3 of the License, or (at your
# option) any later version.
#
# 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 General Public License
# for more details.
#
# You should have received a copy of the GNU General Public License
# along with GNS3-Copilot. If not, see <https://www.gnu.org/licenses/>.
#
# Copyright (C) 2025 Yue Guobin
# Author: Yue Guobin
#
# Project Home: https://github.com/yueguobin/gns3-copilot
#
"""
Topology Planner Skill Package
This package provides the skill for automatic network lab topology planning.
"""
from .topology_planner_skill import TOPOLOGY_PLANNER_SKILL
__all__ = ["TOPOLOGY_PLANNER_SKILL"]

View File

@ -1,235 +0,0 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3
#
# This file is part of GNS3-Copilot project.
#
# GNS3-Copilot is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation, either version 3 of the License, or (at your
# option) any later version.
#
# 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 General Public License
# for more details.
#
# You should have received a copy of the GNU General Public License
# along with GNS3-Copilot. If not, see <https://www.gnu.org/licenses/>.
#
# Copyright (C) 2025 Yue Guobin
# Author: Yue Guobin
#
# Project Home: https://github.com/yueguobin/gns3-copilot
#
"""
Topology Planner Skill for GNS3 Lab Automation
This skill helps users plan and create network lab topologies automatically.
Key Constraints:
- Default image: IOU (L3 device)
- Default IP range: 10.0.0.0/8 (use 10.x.x.x subnets)
- Max 10 nodes recommended
- Node naming: Router=R, Switch=S, PC=PC
"""
# Default node naming convention
NODE_NAMING = {
"router": "R", # e.g., R-1, R-2, R-3
"switch": "SW", # e.g., SW-1, SW-2
"pc": "PC", # e.g., PC-1, PC-2
}
# Default IOU template name
DEFAULT_IOU_TEMPLATE = "IOU"
# Topology Planner Skill Definition
TOPOLOGY_PLANNER_SKILL = {
"device_type": "topology_planner",
"category": "feature",
"name": "GNS3 Topology Planner",
"description": "Automatically plan and create GNS3 network lab topologies",
# Default settings
"defaults": {
"image": "IOU",
"ip_range": "10.0.0.0/8",
"max_nodes": 10,
"naming": NODE_NAMING,
},
# IP planning rules
"ip_planning": {
"subnet_format": "10.0.{node_pair}.0/30 for P2P links, 10.0.{node_pair}.0/24 for LANs",
"subnet_example": "10.0.12.x = R1-R2 (.1=R1, .2=R2)",
"gateway_convention": ".1/.2 for P2P, .254 for LAN gateway",
},
# Node naming conventions
"naming_rules": {
"router": "R{number}, e.g., R-1, R-2, R-3",
"switch": "SW{number}, e.g., SW-1, SW-2",
"pc": "PC{number}, e.g., PC-1, PC-2",
"firewall": "FW{number}, e.g., FW-1",
"loopback": "Lo{number}, e.g., Lo-0, Lo-1",
},
# Node positioning rules based on topology type
"positioning_rules": {
"min_distance_px": 250,
"topology_types": {
"star": {
"description": "Central node with peripherals around it",
"center_node": {"x": 0, "y": 0},
"peripheral_nodes": "Arrange in circle around center, angle = index * (360 / count)",
"radius": 300,
"example_3nodes": {"R-1": (0, 0), "R-2": (-300, 0), "R-3": (300, 0)},
"example_5nodes": {"R-1": (0, 0), "R-2": (-250, -250), "R-3": (250, -250), "R-4": (-250, 250), "R-5": (250, 250)},
},
"ring": {
"description": "Nodes connected in a closed loop",
"arrangement": "Circle arrangement, equal spacing",
"radius": 250,
"example_3nodes": {"R-1": (0, -250), "R-2": (216, 125), "R-3": (-216, 125)},
"example_4nodes": {"R-1": (0, -250), "R-2": (250, 0), "R-3": (0, 250), "R-4": (-250, 0)},
},
"bus": {
"description": "Linear chain of nodes",
"arrangement": "Horizontal line, equal spacing",
"spacing_x": 300,
"spacing_y": 0,
"example_3nodes": {"R-1": (-300, 0), "R-2": (0, 0), "R-3": (300, 0)},
},
"mesh": {
"description": "Fully or partially interconnected nodes",
"arrangement": "Grid pattern, rows and columns",
"cols": 2,
"spacing_x": 300,
"spacing_y": 250,
"example_4nodes": {"R-1": (-150, -125), "R-2": (150, -125), "R-3": (-150, 125), "R-4": (150, 125)},
},
"hierarchical": {
"description": "Three-tier: Core -> Distribution -> Access",
"layers": {
"core": {"y": -250, "x": 0},
"distribution": {"y": 0, "x_offset": 200},
"access": {"y": 250, "x_offset": 300},
},
"example_5nodes": {"Core": (0, -250), "Dist1": (-200, 0), "Dist2": (200, 0), "Acc1": (-300, 250), "Acc2": (300, 250)},
},
"linear_p2p": {
"description": "Point-to-point links in a line (WAN links)",
"arrangement": "Horizontal or vertical line",
"spacing_x": 250,
"spacing_y": 0,
"example_3routers": {"R-1": (-250, 0), "R-2": (0, 0), "R-3": (250, 0)},
},
},
"general_guidelines": [
"Place hub/spine nodes at center (0,0) or top center",
"Leaf/edge nodes radiate outward from center",
"WAN routers typically on left and right sides",
"PCs/terminals placed at outer edges",
"Maintain minimum 250px between any two nodes",
"Adjust positions to reflect actual network topology logic",
"Combine topology types as needed (e.g., star + linear_p2p for WAN segments)",
],
},
# Tool call workflow
"workflow": {
"step_1_read_templates": {
"tool": "gns3_template_reader",
"purpose": "Find available IOU template name",
"example": "List templates to identify IOU template"
},
"step_2_create_nodes": {
"tool": "gns3_create_node",
"purpose": "Create all router/switch/PC nodes",
"params_required": ["project_id", "nodes: [{template_id, x, y, name?}]"],
"positioning": "Choose topology type (star/ring/bus/mesh/hierarchical) based on network design. Place hub/spine at center, leaves at edges. Maintain 250px min distance.",
"note": "Use 'name' field to set node names directly (e.g., R1, R2). No separate rename step needed."
},
"step_3_create_links": {
"tool": "gns3_link_tool",
"purpose": "Connect nodes according to topology design",
"params_required": ["node1_id", "node1_port", "node2_id", "node2_port"]
},
"step_4_start_nodes": {
"tool": "gns3_start_node_tool",
"purpose": "Power on all nodes",
"params_required": ["node_ids"],
"note": "Start nodes before configuration"
},
"step_5_verify": {
"tool": "execute_multiple_device_commands",
"purpose": "Verify connectivity before config",
"example_commands": ["ping <neighbor_ip>"]
},
"step_6_config": {
"tool": "execute_multiple_device_config_commands",
"purpose": "Apply network configuration",
"note": "Only after verifying physical connectivity"
},
},
# Troubleshooting
"troubleshooting": {
"node_creation_failed": [
"1. Check if template name is correct (use gns3_template_reader)",
"2. Verify GNS3 server is running",
"3. Check compute resource availability"
],
"link_creation_failed": [
"1. Verify both nodes exist and have available ports",
"2. Check if link already exists between nodes",
"3. Confirm nodes are stopped before linking (some setups)"
],
"node_start_failed": [
"1. Check if node is already running",
"2. Verify compute resource has enough memory",
"3. Check console port availability"
],
"connectivity_failed": [
"1. Use show ip interface brief to verify IPs are configured",
"2. Check if interfaces are administratively up (no shutdown)",
"3. Verify cable/port mapping in GNS3 topology"
],
},
# Planning output format
"output_template": """
## Topology Plan
### Devices
| Node | Type | Image | Description |
|------|------|-------|-------------|
| R-1 | Router | IOU | Core router |
| ... | ... | ... | ... |
### Connections
| Node1 | Port | Node2 | Port |
|-------|------|-------|------|
| R-1 | {short_name} | R-2 | {short_name} |
| ... | ... | ... | ... |
### IP Addressing
| Device | Interface | IP Address | Subnet |
|--------|-----------|------------|--------|
| R-1 | {short_name} | 10.0.12.1 | /30 |
| R-2 | {short_name} | 10.0.12.2 | /30 |
| ... | ... | ... | ... |
### Configuration Steps
1. Create nodes: gns3_create_node(...)
2. Create links: gns3_link_tool(...)
3. Start nodes: gns3_start_node_tool(...)
4. Verify connectivity: ping ...
5. Apply config: execute_multiple_device_config_commands(...)
Note: Use "name" field in gns3_create_node to set node names directly. No separate rename step needed.
""",
}

View File

@ -1,34 +0,0 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3
#
# This file is part of GNS3-Copilot project.
#
# GNS3-Copilot is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation, either version 3 of the License, or (at your
# option) any later version.
#
# 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 General Public License
# for more details.
#
# You should have received a copy of the GNU General Public License
# along with GNS3-Copilot. If not, see <https://www.gnu.org/licenses/>.
#
# Copyright (C) 2025 Yue Guobin
# Author: Yue Guobin
#
# Project Home: https://github.com/yueguobin/gns3-copilot
#
"""
VPCS Skill Package
This package provides the skill definition for GNS3 VPCS Virtual PC Simulator.
"""
from .vpcs_skill import VPCS_SKILL
__all__ = ["VPCS_SKILL"]

View File

@ -1,158 +0,0 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3
#
# This file is part of GNS3-Copilot project.
#
# GNS3-Copilot is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation, either version 3 of the License, or (at your
# option) any later version.
#
# 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 General Public License
# for more details.
#
# You should have received a copy of the GNU General Public License
# along with GNS3-Copilot. If not, see <https://www.gnu.org/licenses/>.
#
# Copyright (C) 2025 Yue Guobin
# Author: Yue Guobin
#
# Project Home: https://github.com/yueguobin/gns3-copilot
#
"""
VPCS Skill Definition
GNS3 VPCS (Virtual PC Simulator) is a lightweight virtual PC simulator
used in GNS3 labs for testing network connectivity and basic IP configuration.
Key Characteristics:
- NOT a network device (router/switch), it's a simple PC simulator
- No authentication required (direct console access)
- No configuration mode (commands entered directly)
- Simple command set focused on IP configuration and connectivity testing
Device Type Tag: device_type:gns3_vpcs_telnet
"""
# VPCS Skill Definition
VPCS_SKILL = {
"device_type": "gns3_vpcs_telnet",
"category": "device", # device, protocol, or feature
"name": "VPCS Virtual PC Simulator",
"description": "GNS3 VPCS lightweight virtual PC simulator for testing network connectivity and basic IP configuration",
# Configuration Commands (no config mode needed)
"config_commands": {
"ip_config": {
"syntax": "ip <address>/<mask> <gateway>",
"example": "ip 10.10.0.12/24 10.10.0.254",
"description": "Configure PC IP address and default gateway",
"parameters": {
"address": "IP address, e.g., 10.10.0.12",
"mask": "Subnet mask in CIDR notation, e.g., 24 means 255.255.255.0",
"gateway": "Default gateway, e.g., 10.10.0.254"
}
},
"ip_dhcp": {
"syntax": "ip dhcp",
"description": "Obtain IP configuration from DHCP server"
},
"save": {
"syntax": "save",
"description": "Save current configuration to NVRAM (persists after reboot)"
},
"reset": {
"syntax": "reset",
"description": "Reset VPCS configuration (clears all settings)"
},
},
# Display/Diagnostic Commands
"display_commands": {
"show_ip": {
"syntax": "show ip",
"description": "Show current IP configuration (IP address, subnet mask, gateway)"
},
"ping": {
"syntax": "ping <destination>",
"example": "ping 10.10.0.254",
"description": "Test connectivity to destination (sends 4 ICMP echo requests)"
},
"ping_count": {
"syntax": "ping <destination> <count>",
"example": "ping 10.10.0.254 10",
"description": "Send specified number of ICMP packets"
},
"arp": {
"syntax": "arp",
"description": "Display ARP cache table"
},
"version": {
"syntax": "version",
"description": "Show VPCS version information"
},
"show": {
"syntax": "show",
"description": "Display current running configuration"
},
"pc_info": {
"syntax": "pcinfo",
"description": "Display PC hardware information"
},
"route": {
"syntax": "route",
"description": "Display routing table (static routes)"
},
},
# Important Notes
"notes": [
"WARNING: VPCS is NOT a network device (router/switch), it is a lightweight PC simulator!",
"WARNING: Do NOT use router/switch config commands on VPCS (e.g., configure terminal, interface)",
"VPCS has no config mode, commands are entered directly",
"VPCS does not require username/password authentication, direct console access",
"Prompt format: PC1>, PC2>, VPCS>",
"Default sends 4 ICMP packets, use ping <ip> <count> to specify number",
"Must execute save to persist configuration, otherwise lost after reboot",
],
# Troubleshooting Guide
"troubleshooting": {
"ping failed": [
"1. Use show ip to verify IP configuration is correct",
"2. Verify target gateway is reachable (ping gateway IP)",
"3. Ensure source and target are in same subnet or gateway is correct",
"4. Check if link is UP (verify GNS3 topology connections)"
],
"configuration lost": [
"1. VPCS configuration is lost after reboot",
"2. Must execute save command after any configuration change",
"3. Use show command to verify current configuration"
],
"cannot connect to console": [
"1. Check if node is started in GNS3",
"2. Verify console port mapping is correct",
"3. Confirm telnet connection parameters are correct (IP:port)"
],
},
# Command aliases (for LLM understanding)
"command_aliases": {
"show ip": "show ip",
"display ip": "show ip",
"config ip": "ip <address>/<mask> <gateway>",
"set ip": "ip <address>/<mask> <gateway>",
"test connectivity": "ping <destination>",
"ping test": "ping <destination>",
"save config": "save",
"save": "save",
"reset": "reset",
"show route": "route",
"show arp": "arp",
"show version": "version",
},
}

View File

@ -183,7 +183,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
configuration results.
"""
# Log received input
logger.info("Received input: %s", tool_input)
logger.debug("Received input: %s", tool_input)
# Validate input
device_configs_list, project_id = self._validate_tool_input(tool_input)
@ -353,7 +353,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
# Handle models (like DeepSeek) that return a raw JSON string.
try:
parsed_input = json.loads(tool_input)
logger.info("Successfully parsed tool input from JSON string.")
logger.debug("Successfully parsed tool input from JSON string.")
except json.JSONDecodeError as e:
logger.error(
"Invalid JSON string received as tool input: %s", e

View File

@ -114,7 +114,7 @@ class GNS3LinkTool(BaseTool):
list: A list with created link details or error messages.
"""
# Log received input
logger.info("Received input: %s", tool_input)
logger.debug("Received input: %s", tool_input)
try:
# Parse input JSON

View File

@ -29,23 +29,18 @@ Command filter module for GNS3-Copilot.
This module provides functionality to filter out dangerous or long-running
commands that may cause issues with tool execution timeouts or device console
availability.
Forbidden commands are loaded from the external GNS3-Skills repository
(config/forbidden_commands.txt) via SkillsManager.
"""
import logging
from pathlib import Path
from gns3server.agent.gns3_copilot.skills.registry import get_skills_manager
logger = logging.getLogger(__name__)
def _get_gns3_copilot_root() -> Path:
"""Get the root directory of the GNS3-Copilot project."""
# Get the directory containing this file
current_file = Path(__file__).resolve()
# Go up to the gns3_copilot directory (utils parent)
return current_file.parent.parent
# Default forbidden commands (fallback if file not found)
# Default forbidden commands (fallback if skills repository is not available)
DEFAULT_FORBIDDEN_COMMANDS = [
"traceroute",
"tracepath",
@ -55,22 +50,19 @@ DEFAULT_FORBIDDEN_COMMANDS = [
"test",
]
# Cache for forbidden commands to avoid repeated file reads
# Cache for forbidden commands
_forbidden_commands_cache: list[str] | None = None
def _get_forbidden_commands_file_path() -> Path:
"""Get the path to the forbidden commands configuration file."""
return _get_gns3_copilot_root() / "config" / "forbidden_commands.txt"
def _load_forbidden_commands() -> list[str]:
"""
Load forbidden commands from the configuration file.
Load forbidden commands, preferring the skills repository.
Tries to load from the external GNS3-Skills repository first.
Falls back to hardcoded defaults if the repo is unavailable.
Returns:
List of forbidden command patterns. If the file cannot be read,
returns the default list.
List of forbidden command patterns.
"""
global _forbidden_commands_cache
@ -78,69 +70,48 @@ def _load_forbidden_commands() -> list[str]:
if _forbidden_commands_cache is not None:
return _forbidden_commands_cache
file_path = _get_forbidden_commands_file_path()
# Try to load from skills repository
manager = get_skills_manager()
if manager is not None:
commands = manager.load_forbidden_commands()
if commands:
logger.info(
"Loaded %d forbidden command patterns from skills repository",
len(commands),
)
_forbidden_commands_cache = commands
return _forbidden_commands_cache
try:
if not file_path.exists():
logger.warning(
"Forbidden commands file not found: %s. Using default list.",
file_path,
)
_forbidden_commands_cache = DEFAULT_FORBIDDEN_COMMANDS.copy()
return _forbidden_commands_cache
with open(file_path, "r", encoding="utf-8") as f:
forbidden_commands = []
for line in f:
line = line.strip()
# Skip empty lines and comments
if not line or line.startswith("#"):
continue
forbidden_commands.append(line.lower())
if not forbidden_commands:
logger.warning(
"No forbidden commands found in %s. Using default list.",
file_path,
)
_forbidden_commands_cache = DEFAULT_FORBIDDEN_COMMANDS.copy()
else:
logger.info(
"Loaded %d forbidden command patterns from %s",
len(forbidden_commands),
file_path,
)
_forbidden_commands_cache = forbidden_commands
return _forbidden_commands_cache
except Exception as e:
logger.error(
"Error reading forbidden commands file %s: %s. Using default list.",
file_path,
e,
)
_forbidden_commands_cache = DEFAULT_FORBIDDEN_COMMANDS.copy()
return _forbidden_commands_cache
# Fallback to defaults
logger.warning("Using default forbidden commands list")
_forbidden_commands_cache = DEFAULT_FORBIDDEN_COMMANDS.copy()
return _forbidden_commands_cache
def reload_forbidden_commands() -> None:
"""
Reload the forbidden commands list from the configuration file.
Reload the forbidden commands list from the skills repository.
This clears the cache and forces a reload from the file on the next
call to filter_forbidden_commands() or get_forbidden_commands().
Use this after modifying the forbidden_commands.txt file to apply
changes without restarting the GNS3 server.
Directly loads and caches the commands from the repository.
Falls back to defaults if the repository is unavailable.
"""
global _forbidden_commands_cache
_forbidden_commands_cache = None
logger.info(
"Forbidden commands cache cleared. Will reload on next access."
)
from gns3server.agent.gns3_copilot.skills.registry import get_skills_manager
manager = get_skills_manager()
if manager is not None:
commands = manager.load_forbidden_commands()
if commands:
logger.info(
"Loaded %d forbidden command patterns from skills repository",
len(commands),
)
_forbidden_commands_cache = commands
return
logger.warning("Using default forbidden commands list")
_forbidden_commands_cache = DEFAULT_FORBIDDEN_COMMANDS.copy()
def get_forbidden_commands() -> list[str]:

View File

@ -22,15 +22,19 @@ from gns3server.agent import AI_COPILOT_AVAILABLE
# Conditionally import AI-dependent routes
if AI_COPILOT_AVAILABLE:
from . import chat
from . import copilot
from . import llm_model_configs
_chat_router = chat.router
_copilot_router = copilot.router
_llm_router = llm_model_configs.router
else:
# Create stub routers that return 501 for all AI endpoints
_chat_router = APIRouter()
_copilot_router = APIRouter()
_llm_router = APIRouter()
@_chat_router.api_route("/{path:path}", methods=["GET", "POST", "DELETE", "PATCH", "PUT"])
@_copilot_router.api_route("/{path:path}", methods=["GET", "POST", "DELETE", "PATCH", "PUT"])
@_llm_router.api_route("/{path:path}", methods=["GET", "POST", "DELETE", "PATCH", "PUT"])
async def ai_not_available(path: str = ""):
raise HTTPException(
@ -176,8 +180,15 @@ router.include_router(
)
router.include_router(
_chat_router,
prefix="/projects/{project_id}/chat",
_copilot_router,
prefix="/copilot",
dependencies=[Depends(get_current_active_user)],
tags=["Chat"]
tags=["GNS3 Copilot"]
)
router.include_router(
_chat_router,
prefix="/copilot/projects/{project_id}/chat",
dependencies=[Depends(get_current_active_user)],
tags=["GNS3 Copilot"]
)

View File

@ -23,7 +23,7 @@ Nested under projects: /v3/projects/{project_id}/chat/...
import json
import logging
import uuid
from typing import List
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Request, status
from fastapi.responses import StreamingResponse
@ -34,6 +34,7 @@ from gns3server.controller import Controller
from gns3server.controller.project import Project
from gns3server.controller.controller_error import ControllerNotFoundError
from gns3server.agent.gns3_copilot.project_agent_manager import get_project_agent_manager
from gns3server.db.tasks import get_user_llm_config_full
from .dependencies.authentication import get_current_active_user
@ -118,7 +119,6 @@ async def stream_chat(
app = http_request.app
# Get user's LLM config (with decrypted API key)
from gns3server.db.tasks import get_user_llm_config_full
llm_config = await get_user_llm_config_full(user_id, app)
if not llm_config:
log.warning("LLM config not found for user: %s", user_id)
@ -195,14 +195,18 @@ async def stream_chat(
"/sessions",
response_model=List[schemas.ChatSession],
summary="List chat sessions",
description="List all chat sessions for a project."
description="List all chat sessions for a project, optionally filtered by copilot_mode."
)
async def list_sessions(
project: Project = Depends(dep_project),
copilot_mode: Optional[str] = None,
current_user: schemas.User = Depends(get_current_active_user),
) -> list[schemas.ChatSession]:
"""
List chat sessions for a project.
Query Parameters:
- copilot_mode: Optional filter by copilot mode (e.g., "troubleshooting_injection", "teaching_assistant", "lab_automation_assistant")
"""
# Check if project is opened
@ -216,8 +220,11 @@ async def list_sessions(
agent_manager = await get_project_agent_manager()
agent_service = await agent_manager.get_agent(str(project.id), project.path)
# List sessions
sessions = await agent_service.list_sessions(user_id=str(current_user.user_id))
# List sessions with optional copilot_mode filter
sessions = await agent_service.list_sessions(
user_id=str(current_user.user_id),
copilot_mode=copilot_mode
)
# Convert to schemas
return [schemas.ChatSession(**s) for s in sessions]
@ -440,3 +447,130 @@ async def unpin_session(
)
return schemas.ChatSession(**session)
@router.post(
"/inject",
response_model=None,
summary="Inject a network fault for troubleshooting practice",
description="Inject a realistic network fault into the GNS3 lab for troubleshooting training."
)
async def inject_issue(
request: schemas.ChatRequest,
http_request: Request,
project: Project = Depends(dep_project),
current_user: schemas.User = Depends(get_current_active_user),
) -> StreamingResponse:
"""
Inject a network fault for troubleshooting practice.
This endpoint uses the GNS3 Copilot agent in troubleshooting_injection mode
to analyze the topology, select an appropriate fault, and inject it.
The fault injection process is streamed via Server-Sent Events (SSE),
and the complete conversation is stored in the chat history for later review.
The project must be opened to use fault injection.
"""
# Get user authentication info
user_id = str(current_user.user_id)
# Check if project is opened
if project.status != "opened":
log.warning(
"Fault injection rejected: project not opened. user_id=%s, project_id=%s, status=%s",
user_id,
project.id,
project.status
)
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Project must be opened to use fault injection. Current status: {project.status}"
)
log.info(
"Fault injection started: user_id=%s, project_id=%s, project_name=%s, session_id=%s",
user_id,
project.id,
project.name,
request.session_id or "(new)",
)
# Get JWT token from Authorization header
auth_header = http_request.headers.get("Authorization", "")
jwt_token = auth_header.replace("Bearer ", "") if auth_header else None
# Get FastAPI app reference (for database access)
app = http_request.app
# Get user's LLM config (with decrypted API key)
llm_config = await get_user_llm_config_full(user_id, app)
if not llm_config:
log.warning("LLM config not found for user: %s", user_id)
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="LLM configuration not found. Please configure your LLM settings first."
)
# Override copilot_mode to troubleshooting_injection
llm_config["copilot_mode"] = "troubleshooting_injection"
log.debug(
"LLM config loaded and mode set to troubleshooting_injection: user_id=%s, provider=%s, model=%s",
user_id,
llm_config.get("provider"),
llm_config.get("model"),
)
# Get or create AgentService for this project
agent_manager = await get_project_agent_manager()
agent_service = await agent_manager.get_agent(str(project.id), project.path)
log.debug("AgentService obtained for project: %s", project.id)
# Generate session_id if not provided
session_id = request.session_id or str(uuid.uuid4())
if not request.session_id:
log.debug("New session created for fault injection: %s", session_id)
# Use default message if not provided
message = request.message or "Inject a network fault for troubleshooting practice"
async def generate():
"""Generator for SSE streaming."""
try:
log.debug("Starting fault injection stream: session_id=%s", session_id)
async for chunk in agent_service.stream_chat(
message=message,
session_id=session_id,
project_id=str(project.id),
user_id=user_id,
jwt_token=jwt_token,
mode=request.mode,
llm_config=llm_config
):
try:
# Validate and serialize chunk
validated = schemas.ChatResponse(**chunk)
yield f"data: {json.dumps(validated.model_dump(exclude_none=True), ensure_ascii=False)}\n\n"
except Exception as e:
log.warning("Error serializing chunk: %s", e)
# Skip invalid chunks but continue streaming
continue
# Final done message
log.debug("Fault injection stream completed: session_id=%s", session_id)
yield f"data: {json.dumps({'type': 'done', 'session_id': session_id})}\n\n"
except Exception as e:
log.error("Error in fault injection stream: %s", e, exc_info=True)
yield f"data: {json.dumps({'type': 'error', 'error': str(e)})}\n\n"
return StreamingResponse(
generate(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no",
"Connection": "keep-alive",
}
)

View File

@ -0,0 +1,51 @@
#
# Copyright (C) 2025 GNS3 Technologies Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
API routes for GNS3 Copilot global operations (non-project).
"""
import logging
from fastapi import APIRouter, Depends
from .dependencies.authentication import get_current_active_user
log = logging.getLogger(__name__)
router = APIRouter()
@router.post(
"/reload/skills",
dependencies=[Depends(get_current_active_user)],
)
async def reload_skills() -> dict:
"""
Hot reload skills and prompts from the external GNS3-Skills repository.
Reloads injection skills, system prompts, and forbidden commands
from the skills repository without restarting the server.
"""
try:
from gns3server.agent.gns3_copilot.skills.registry import (
reload_skills_repository,
)
result = reload_skills_repository()
except ImportError:
return {"error": "AI Copilot is not available"}
return result

View File

@ -113,6 +113,14 @@ enable_builtin_templates = True
; Install built-in appliances
install_builtin_appliances = True
; Git repository URL for GNS3 Copilot external skills
; This repository provides injection skills, prompts, and device skills
; skills_repo_url = https://github.com/yueguobin/GNS3-Skills.git
; Git branch for the skills repository
; skills_repo_branch = main
; Automatically pull updates from the skills repository when reloading
; skills_auto_update = false
; check if hardware virtualization is used by other emulators (KVM, VMware or VirtualBox)
hardware_virtualization_check = True

View File

@ -159,6 +159,9 @@ class ServerSettings(BaseModel):
allow_remote_console: bool = False
enable_builtin_templates: bool = True
install_builtin_appliances: bool = True
skills_repo_url: str = "https://github.com/yueguobin/GNS3-Skills.git"
skills_repo_branch: str = "main"
skills_auto_update: bool = True
model_config = ConfigDict(validate_assignment=True, str_strip_whitespace=True)
@field_validator("additional_images_paths", mode="before")

View File

@ -1,3 +1,6 @@
[Server]
compute_username = gns3
compute_password = gns3
skills_repo_url = https://github.com/yueguobin/GNS3-Skills.git
skills_repo_branch = main
skills_auto_update = false