mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-08-27 20:40:13 +03:00
Merge pull request #2782 from yueguobin/mcp/device-config
MCP (Model Context Protocol) integration — 82 tools with device config, API key auth, and bug fixes
This commit is contained in:
Jeremy Grossmann
GitHub
commit
16d3f28cf5
@ -16,19 +16,19 @@ The MCP service exposes GNS3 project management operations as MCP tools that can
|
||||
|
||||
## Authentication
|
||||
|
||||
The SSE endpoint requires a valid GNS3 JWT token. It supports two ways to pass the token:
|
||||
The SSE endpoint supports two types of credentials, passed the same way.
|
||||
|
||||
1. **Authorization header** (recommended for Claude Code):
|
||||
1. **Authorization header** (recommended):
|
||||
```
|
||||
Authorization: Bearer <jwt>
|
||||
Authorization: Bearer <jwt_or_api_key>
|
||||
```
|
||||
|
||||
2. **Query parameter** (required for Claude Desktop, since EventSource does not support custom headers):
|
||||
2. **Query parameter** (for clients that don't support custom headers):
|
||||
```
|
||||
GET /v3/mcp/transport/sse?token=<jwt>
|
||||
GET /v3/mcp/transport/sse?token=<jwt_or_api_key>
|
||||
```
|
||||
|
||||
### Getting a Token
|
||||
### Option 1: JWT Token (24h expiry)
|
||||
|
||||
```bash
|
||||
curl -X POST http://localhost:3080/v3/access/users/authenticate \
|
||||
@ -36,85 +36,201 @@ curl -X POST http://localhost:3080/v3/access/users/authenticate \
|
||||
-d '{"username": "admin", "password": "admin"}'
|
||||
```
|
||||
|
||||
### Token Expiry
|
||||
|
||||
Default JWT token lifetime is **1440 minutes (24 hours)**. This can be configured in `gns3_server.conf`:
|
||||
|
||||
Default lifetime is **1440 minutes (24 hours)**. Configurable in `gns3_server.conf`:
|
||||
```ini
|
||||
jwt_access_token_expire_minutes = 1440 ; 24 hours
|
||||
```
|
||||
|
||||
### Option 2: API Key (permanent, revocable) — Recommended for MCP
|
||||
|
||||
API keys never expire and can be revoked individually. Create one via the REST API:
|
||||
|
||||
```bash
|
||||
# Create an API key (requires a JWT to authenticate)
|
||||
curl -X POST http://localhost:3080/v3/access/api-keys \
|
||||
-H "Authorization: Bearer <your_jwt>" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"name": "MCP Production"}'
|
||||
# Response: {"api_key": "gns3_a1b2c3d4...", "api_key_id": "...", ...}
|
||||
# ⚠️ The key is only shown once — save it immediately.
|
||||
```
|
||||
|
||||
API key management endpoints:
|
||||
|
||||
| Endpoint | Description |
|
||||
|----------|-------------|
|
||||
| `POST /v3/access/api-keys` | Create a new key (returns plaintext once) |
|
||||
| `GET /v3/access/api-keys` | List all your keys |
|
||||
| `POST /v3/access/api-keys/{id}/revoke` | Revoke a key (can be restored) |
|
||||
| `POST /v3/access/api-keys/{id}/restore` | Restore a revoked key |
|
||||
| `DELETE /v3/access/api-keys/{id}` | Permanently delete a key |
|
||||
|
||||
Both JWT tokens and API keys work for MCP and REST API endpoints interchangeably.
|
||||
|
||||
## Available Tools
|
||||
|
||||
**30 tools** across 5 categories:
|
||||
**82 tools** across 12 categories:
|
||||
|
||||
### Project (7)
|
||||
### Project (15)
|
||||
|
||||
| Tool | Description | Required Parameters |
|
||||
|------|-------------|-------------------|
|
||||
| `list_projects` | List all projects | none |
|
||||
| `get_project` | Get project details | `project_id` |
|
||||
| `create_project` | Create a project | `name` |
|
||||
| `delete_project` | Delete a project | `project_id` |
|
||||
| `open_project` | Open a project | `project_id` |
|
||||
| `close_project` | Close a project | `project_id` |
|
||||
| `get_project_stats` | Get project statistics | `project_id` |
|
||||
| Tool | Description |
|
||||
|------|-------------|
|
||||
| `project_list` | List all projects |
|
||||
| `project_get` | Get project details |
|
||||
| `project_create` | Create a project |
|
||||
| `project_delete` | Delete a project |
|
||||
| `project_open` | Open a closed project |
|
||||
| `project_close` | Close an open project |
|
||||
| `project_stats` | Get project statistics |
|
||||
| `project_update` | Update project properties |
|
||||
| `project_duplicate` | Duplicate a project |
|
||||
| `project_readme_get` | Get project README content |
|
||||
| `project_readme_update` | Update project README |
|
||||
| `project_lock` | Lock project (prevent edits) |
|
||||
| `project_unlock` | Unlock project |
|
||||
| `project_load` | Load project from path |
|
||||
| `project_locked` | Check if project is locked |
|
||||
|
||||
### Node (10)
|
||||
### Node (22)
|
||||
|
||||
| Tool | Description | Required Parameters |
|
||||
|------|-------------|-------------------|
|
||||
| `get_nodes` | List all nodes in a project | `project_id` |
|
||||
| `get_node` | Get node details | `project_id`, `node_id` |
|
||||
| `start_node` | Start a node | `project_id`, `node_id` |
|
||||
| `stop_node` | Stop a node | `project_id`, `node_id` |
|
||||
| `reload_node` | Reload a node | `project_id`, `node_id` |
|
||||
| `suspend_node` | Suspend a node | `project_id`, `node_id` |
|
||||
| `create_node` | Create a node from template | `project_id`, `template_id` |
|
||||
| `delete_node` | Delete a node | `project_id`, `node_id` |
|
||||
| `update_node` | Update node properties | `project_id`, `node_id` |
|
||||
| `get_node_console_info` | Get WebSocket console URL | `project_id`, `node_id` |
|
||||
| Tool | Description |
|
||||
|------|-------------|
|
||||
| `node_list` | List all nodes in a project |
|
||||
| `node_get` | Get node details |
|
||||
| `node_create` | Create a node from template |
|
||||
| `node_delete` | Delete a node |
|
||||
| `node_update` | Update node properties |
|
||||
| `node_start` | Start a node |
|
||||
| `node_stop` | Stop a node |
|
||||
| `node_reload` | Reload a node |
|
||||
| `node_suspend` | Suspend a node |
|
||||
| `node_console` | Get WebSocket console URL |
|
||||
| `node_file_list` | List files in node directory |
|
||||
| `node_file_get` | Read a file (with offset/limit) |
|
||||
| `node_file_write` | Write a file |
|
||||
| `node_file_delete` | Delete a file |
|
||||
| `node_start_all` | Start all nodes |
|
||||
| `node_stop_all` | Stop all nodes |
|
||||
| `node_suspend_all` | Suspend all nodes |
|
||||
| `node_reload_all` | Reload all nodes |
|
||||
| `node_duplicate` | Duplicate a node |
|
||||
| `node_isolate` | Isolate a node (suspend links) |
|
||||
| `node_unisolate` | Un-isolate a node (resume links) |
|
||||
| `node_links` | List links connected to a node |
|
||||
|
||||
### Link (5)
|
||||
### Link (9)
|
||||
|
||||
| Tool | Description | Required Parameters |
|
||||
|------|-------------|-------------------|
|
||||
| `get_links` | List all links in a project | `project_id` |
|
||||
| `get_link` | Get link details | `project_id`, `link_id` |
|
||||
| `create_link` | Create a link between nodes | `project_id`, `nodes` |
|
||||
| `delete_link` | Delete a link | `project_id`, `link_id` |
|
||||
| `update_link` | Update link properties | `project_id`, `link_id` |
|
||||
| Tool | Description |
|
||||
|------|-------------|
|
||||
| `link_list` | List all links in a project |
|
||||
| `link_get` | Get link details |
|
||||
| `link_create` | Create a link between nodes |
|
||||
| `link_delete` | Delete a link |
|
||||
| `link_update` | Update link (suspend, filters) |
|
||||
| `link_reset` | Reset link (delete + recreate) |
|
||||
| `link_capture_start` | Start packet capture |
|
||||
| `link_capture_stop` | Stop packet capture |
|
||||
| `link_capture_download` | Get PCAP download URL |
|
||||
|
||||
### Template (5)
|
||||
|
||||
| Tool | Description | Required Parameters |
|
||||
|------|-------------|-------------------|
|
||||
| `list_templates` | List all templates | none |
|
||||
| `get_template` | Get template details | `template_id` or `name` |
|
||||
| `create_template` | Create a template | `name`, `template_type` |
|
||||
| `update_template` | Update a template | `template_id` or `name` |
|
||||
| `delete_template` | Delete a template | `template_id` or `name` |
|
||||
| Tool | Description |
|
||||
|------|-------------|
|
||||
| `template_list` | List all templates |
|
||||
| `template_get` | Get template details |
|
||||
| `template_create` | Create a template (Docker needs `image`) |
|
||||
| `template_update` | Update a template |
|
||||
| `template_delete` | Delete a template |
|
||||
|
||||
### Compute (3)
|
||||
|
||||
| Tool | Description | Required Parameters |
|
||||
|------|-------------|-------------------|
|
||||
| `list_computes` | List all compute nodes | none |
|
||||
| `get_compute` | Get compute details | `compute_id` |
|
||||
| `get_compute_images` | List available images | `emulator` |
|
||||
| Tool | Description |
|
||||
|------|-------------|
|
||||
| `compute_list` | List registered remote computes |
|
||||
| `compute_get` | Get compute details (requires UUID) |
|
||||
| `compute_images` | List emulator images on a compute |
|
||||
|
||||
### Snapshot (4)
|
||||
|
||||
| Tool | Description |
|
||||
|------|-------------|
|
||||
| `snapshot_list` | List snapshots |
|
||||
| `snapshot_create` | Create a snapshot |
|
||||
| `snapshot_delete` | Delete a snapshot |
|
||||
| `snapshot_restore` | Restore a snapshot |
|
||||
|
||||
### Drawing (5)
|
||||
|
||||
| Tool | Description |
|
||||
|------|-------------|
|
||||
| `drawing_list` | List drawings on canvas |
|
||||
| `drawing_get` | Get drawing details |
|
||||
| `drawing_create` | Create drawing (SVG label/shape/image) |
|
||||
| `drawing_update` | Update drawing (position, rotation, SVG) |
|
||||
| `drawing_delete` | Delete a drawing |
|
||||
|
||||
### Symbol (6)
|
||||
|
||||
| Tool | Description |
|
||||
|------|-------------|
|
||||
| `symbol_list` | List all symbols |
|
||||
| `symbol_get` | Get symbol download URL |
|
||||
| `symbol_dimensions` | Get symbol dimensions |
|
||||
| `symbol_defaults` | Get default symbol mapping |
|
||||
| `symbol_upload` | Upload a custom symbol (SVG content) |
|
||||
| `symbol_delete` | Delete a custom symbol (built-in: 403) |
|
||||
|
||||
### Appliance (3)
|
||||
|
||||
| Tool | Description |
|
||||
|------|-------------|
|
||||
| `appliance_list` | List appliances from template library |
|
||||
| `appliance_get` | Get appliance details |
|
||||
| `appliance_install` | Create template from appliance (images must exist locally) |
|
||||
|
||||
### Image (5)
|
||||
|
||||
| Tool | Description |
|
||||
|------|-------------|
|
||||
| `image_list` | List all images |
|
||||
| `image_get` | Get image details |
|
||||
| `image_delete` | Delete an image |
|
||||
| `image_prune` | Remove images not referenced by any template |
|
||||
| `image_install` | Auto-create templates from uploaded images by checksum |
|
||||
|
||||
### Server (2)
|
||||
|
||||
| Tool | Description |
|
||||
|------|-------------|
|
||||
| `server_version` | Get GNS3 server version |
|
||||
| `server_statistics` | Get server statistics (computes, projects, nodes) |
|
||||
|
||||
### Device Config (3)
|
||||
|
||||
| Tool | Description |
|
||||
|------|-------------|
|
||||
| `device_config_send` | Push config commands to devices via console (Nornir + Netmiko) |
|
||||
| `device_command_run` | Run read-only show commands on devices |
|
||||
| `vpcs_config_set` | Configure VPCS devices (IP, gateway, etc.) |
|
||||
|
||||
Requires nodes to be started first. Device type is auto-detected from the node's `device_type:<type>` tag.
|
||||
|
||||
## Configuration
|
||||
|
||||
### Claude Code (CLI)
|
||||
|
||||
```bash
|
||||
# Get a JWT token
|
||||
# Option A: Using API key (recommended — never expires)
|
||||
claude mcp add --transport sse My_GNS3_Server \
|
||||
http://localhost:3080/v3/mcp/transport/sse \
|
||||
-H "Authorization: Bearer gns3_a1b2c3d4..."
|
||||
|
||||
# Option B: Using JWT token (expires after 24h)
|
||||
TOKEN=$(curl -s -X POST http://localhost:3080/v3/access/users/authenticate \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"username": "admin", "password": "admin"}' | python3 -c \
|
||||
"import sys,json; print(json.load(sys.stdin)['access_token'])")
|
||||
|
||||
# Add MCP server
|
||||
claude mcp add --transport sse My_GNS3_Server \
|
||||
http://localhost:3080/v3/mcp/transport/sse \
|
||||
-H "Authorization: Bearer $TOKEN"
|
||||
@ -128,7 +244,7 @@ Add to `claude_desktop_config.json`:
|
||||
{
|
||||
"mcpServers": {
|
||||
"My_GNS3_Server": {
|
||||
"url": "http://localhost:3080/v3/mcp/transport/sse?token=your_jwt_token"
|
||||
"url": "http://localhost:3080/v3/mcp/transport/sse?token=your_jwt_or_api_key"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -29,6 +29,10 @@ GNS3 Connector Factory Module
|
||||
This module provides factory functions for creating Gns3Connector instances
|
||||
with JWT token authentication and context-aware configuration management.
|
||||
|
||||
⚠️ WARNING: This module is shared with the MCP (Model Context Protocol) service.
|
||||
The get_gns3_connector() function is used by both gns3-copilot and MCP.
|
||||
Modifications must be tested with BOTH.
|
||||
|
||||
Features:
|
||||
- Context variable based request-scoped data management (JWT tokens, LLM
|
||||
config)
|
||||
|
||||
@ -29,6 +29,10 @@ Adapted gns3fy module for GNS3-Copilot
|
||||
This module is based on the upstream gns3fy project
|
||||
(https://github.com/davidban77/gns3fy).
|
||||
|
||||
⚠️ WARNING: This module is shared with the MCP (Model Context Protocol) service.
|
||||
The Gns3Connector class is used by MCP handlers to make HTTP calls.
|
||||
Modifications to this file must be tested with BOTH gns3-copilot AND MCP.
|
||||
|
||||
Modifications made for GNS3-Copilot:
|
||||
- Adjusted pydantic usages and dataclass configuration to reduce dependency
|
||||
conflicts with langchain (pydantic version/api differences)
|
||||
@ -782,81 +786,6 @@ class Gns3Connector:
|
||||
_url = f"{self.base_url}/projects/{project_id}/files/{encoded_path}"
|
||||
self.http_call("post", _url, data=content, headers={"Content-Type": "text/plain"})
|
||||
|
||||
def list_node_files(
|
||||
self, project_id: str, node_id: str,
|
||||
path: str = "", recursive: bool = False
|
||||
) -> list[dict[str, Any]]:
|
||||
"""
|
||||
List files in a node directory with metadata.
|
||||
|
||||
**Required Attributes:**
|
||||
|
||||
- `project_id`
|
||||
- `node_id`
|
||||
- `path` Subdirectory path within node directory (optional)
|
||||
- `recursive` Whether to recursively list all files (optional)
|
||||
|
||||
**Returns:**
|
||||
|
||||
List of file objects with name, path, size, modified time, type, etc.
|
||||
"""
|
||||
_url = f"{self.base_url}/projects/{project_id}/nodes/{node_id}/files"
|
||||
_params = {}
|
||||
if path:
|
||||
_params["path"] = path
|
||||
if recursive:
|
||||
_params["recursive"] = "true"
|
||||
_response_data = self.http_call("get", _url, params=_params if _params else None)
|
||||
return cast(list[dict[str, Any]], _response_data.json())
|
||||
|
||||
def get_node_file(self, project_id: str, node_id: str, file_path: str) -> str:
|
||||
"""
|
||||
Get the content of a file in a node directory.
|
||||
|
||||
**Required Attributes:**
|
||||
|
||||
- `project_id`
|
||||
- `node_id`
|
||||
- `file_path`
|
||||
|
||||
**Returns**
|
||||
|
||||
File content as text string
|
||||
"""
|
||||
encoded_path = quote(file_path, safe="/")
|
||||
_url = f"{self.base_url}/projects/{project_id}/nodes/{node_id}/files/{encoded_path}"
|
||||
_response = self.http_call("get", _url)
|
||||
return _response.text
|
||||
|
||||
def write_node_file(self, project_id: str, node_id: str, file_path: str, content: str) -> None:
|
||||
"""
|
||||
Write content to a file in a node directory. Creates the file if it doesn't exist.
|
||||
|
||||
**Required Attributes:**
|
||||
|
||||
- `project_id`
|
||||
- `node_id`
|
||||
- `file_path`
|
||||
- `content`
|
||||
"""
|
||||
encoded_path = quote(file_path, safe="/")
|
||||
_url = f"{self.base_url}/projects/{project_id}/nodes/{node_id}/files/{encoded_path}"
|
||||
self.http_call("post", _url, data=content, headers={"Content-Type": "text/plain"})
|
||||
|
||||
def delete_node_file(self, project_id: str, node_id: str, file_path: str) -> None:
|
||||
"""
|
||||
Delete a file from the node directory.
|
||||
|
||||
**Required Attributes:**
|
||||
|
||||
- `project_id`
|
||||
- `node_id`
|
||||
- `file_path`
|
||||
"""
|
||||
encoded_path = quote(file_path, safe="/")
|
||||
_url = f"{self.base_url}/projects/{project_id}/nodes/{node_id}/files/{encoded_path}"
|
||||
self.http_call("delete", _url)
|
||||
|
||||
def get_computes(self) -> list[dict[str, Any]]:
|
||||
"""
|
||||
Returns a list of computes.
|
||||
@ -1251,6 +1180,96 @@ class Link:
|
||||
|
||||
return cast(list[dict[str, Any]], _response.json())
|
||||
|
||||
def reset(self) -> None:
|
||||
"""
|
||||
Reset the link, clearing its state (counters, filters, etc.).
|
||||
|
||||
**Required Attributes:**
|
||||
|
||||
- `project_id`
|
||||
- `connector`
|
||||
- `link_id`
|
||||
"""
|
||||
_conn = self.connector
|
||||
_project_id = self.project_id
|
||||
|
||||
if _conn is None:
|
||||
raise ValueError("Gns3Connector not assigned under 'connector'")
|
||||
if _project_id is None:
|
||||
raise ValueError("Need to submit project_id")
|
||||
|
||||
_url = f"{_conn.base_url}/projects/{_project_id}/links/{self.link_id}/reset"
|
||||
_response = _conn.http_call("post", _url)
|
||||
self._update(_response.json())
|
||||
|
||||
def start_capture(
|
||||
self,
|
||||
data_link_type: str = "DLT_EN10MB",
|
||||
capture_file_name: str | None = None,
|
||||
wireshark: bool = False,
|
||||
) -> None:
|
||||
"""
|
||||
Start packet capture on the link.
|
||||
|
||||
**Required Attributes:**
|
||||
|
||||
- `project_id`
|
||||
- `connector`
|
||||
- `link_id`
|
||||
|
||||
**Optional Attributes:**
|
||||
|
||||
- `data_link_type` Data link type (default: DLT_EN10MB)
|
||||
- `capture_file_name` Name of the capture file (optional)
|
||||
- `wireshark` Open Wireshark automatically (default: False)
|
||||
"""
|
||||
_conn = self.connector
|
||||
_project_id = self.project_id
|
||||
|
||||
if _conn is None:
|
||||
raise ValueError("Gns3Connector not assigned under 'connector'")
|
||||
if _project_id is None:
|
||||
raise ValueError("Need to submit project_id")
|
||||
if not self.link_id:
|
||||
raise ValueError("Need to submit link_id")
|
||||
|
||||
_url = (
|
||||
f"{_conn.base_url}/projects/{_project_id}/links/"
|
||||
f"{self.link_id}/capture/start"
|
||||
)
|
||||
_data: dict[str, Any] = {
|
||||
"data_link_type": data_link_type,
|
||||
"wireshark": wireshark,
|
||||
}
|
||||
if capture_file_name:
|
||||
_data["capture_file_name"] = capture_file_name
|
||||
_response = _conn.http_call("post", _url, json_data=_data)
|
||||
self._update(_response.json())
|
||||
|
||||
def stop_capture(self) -> None:
|
||||
"""
|
||||
Stop packet capture on the link.
|
||||
|
||||
**Required Attributes:**
|
||||
|
||||
- `project_id`
|
||||
- `connector`
|
||||
- `link_id`
|
||||
"""
|
||||
_conn = self.connector
|
||||
_project_id = self.project_id
|
||||
|
||||
if _conn is None:
|
||||
raise ValueError("Gns3Connector not assigned under 'connector'")
|
||||
if _project_id is None:
|
||||
raise ValueError("Need to submit project_id")
|
||||
|
||||
_url = (
|
||||
f"{_conn.base_url}/projects/{_project_id}/links/"
|
||||
f"{self.link_id}/capture/stop"
|
||||
)
|
||||
_conn.http_call("post", _url)
|
||||
|
||||
|
||||
@dataclass(config=config)
|
||||
class Node:
|
||||
@ -1763,6 +1782,64 @@ class Node:
|
||||
|
||||
return cast(str, _conn.http_call("get", _url).text)
|
||||
|
||||
@verify_connector_and_id
|
||||
def list_files(self, path: str = "", recursive: bool = False) -> list[dict[str, Any]]:
|
||||
"""
|
||||
List files in the node directory with metadata (name, size, type, modified time).
|
||||
|
||||
**Required Attributes:**
|
||||
|
||||
- `project_id`
|
||||
- `connector`
|
||||
- `node_id`
|
||||
|
||||
**Optional Attributes:**
|
||||
|
||||
- `path`: Subdirectory path within node directory (default: "")
|
||||
- `recursive`: Recursively list all files (default: False)
|
||||
|
||||
**Returns:**
|
||||
|
||||
List of file objects with metadata.
|
||||
"""
|
||||
_conn = self.connector
|
||||
assert _conn is not None
|
||||
_project_id = self.project_id
|
||||
assert _project_id is not None
|
||||
_node_id = self.node_id
|
||||
assert _node_id is not None
|
||||
|
||||
_url = f"{_conn.base_url}/projects/{_project_id}/nodes/{_node_id}/files"
|
||||
_params: dict[str, str] = {}
|
||||
if path:
|
||||
_params["path"] = path
|
||||
if recursive:
|
||||
_params["recursive"] = "true"
|
||||
_response = _conn.http_call("get", _url, params=_params if _params else None)
|
||||
return cast(list[dict[str, Any]], _response.json())
|
||||
|
||||
@verify_connector_and_id
|
||||
def delete_file(self, path: str) -> None:
|
||||
"""
|
||||
Delete a file from the node directory.
|
||||
|
||||
**Required Attributes:**
|
||||
|
||||
- `project_id`
|
||||
- `connector`
|
||||
- `node_id`
|
||||
- `path`: Node's relative path of the file to delete
|
||||
"""
|
||||
_conn = self.connector
|
||||
assert _conn is not None
|
||||
_project_id = self.project_id
|
||||
assert _project_id is not None
|
||||
_node_id = self.node_id
|
||||
assert _node_id is not None
|
||||
|
||||
_url = f"{_conn.base_url}/projects/{_project_id}/nodes/{_node_id}/files/{path}"
|
||||
_conn.http_call("delete", _url)
|
||||
|
||||
@verify_connector_and_id
|
||||
def write_file(self, path: str, data: Any) -> None:
|
||||
"""
|
||||
|
||||
@ -30,6 +30,10 @@ This module provides a LangChain BaseTool to retrieve the topology of a
|
||||
specific GNS3 project by project ID. Returns nodes, links, and project
|
||||
metadata.
|
||||
|
||||
⚠️ WARNING: This module is shared with the MCP (Model Context Protocol) service.
|
||||
GNS3TopologyTool._run() is called by MCP device config handlers.
|
||||
The jwt_token/url parameters were added for MCP compatibility.
|
||||
Modifications must be tested with BOTH gns3-copilot AND MCP.
|
||||
"""
|
||||
|
||||
import copy
|
||||
@ -70,6 +74,8 @@ class GNS3TopologyTool(BaseTool):
|
||||
tool_input: Any = None,
|
||||
run_manager: Any = None,
|
||||
project_id: str | None = None,
|
||||
jwt_token: str | None = None,
|
||||
url: str | None = None,
|
||||
) -> dict:
|
||||
"""
|
||||
Synchronous method to retrieve the topology of a specific GNS3 project.
|
||||
@ -80,6 +86,8 @@ class GNS3TopologyTool(BaseTool):
|
||||
run_manager: Callback manager for tool run.
|
||||
project_id: The UUID of the specific GNS3 project to retrieve
|
||||
topology from.
|
||||
jwt_token: JWT token for authentication (used by MCP handlers).
|
||||
url: GNS3 server URL (used by MCP handlers).
|
||||
|
||||
Returns:
|
||||
dict: A dictionary containing the project ID, name, status, nodes,
|
||||
@ -102,8 +110,10 @@ class GNS3TopologyTool(BaseTool):
|
||||
}
|
||||
|
||||
# Initialize Gns3Connector using factory function
|
||||
# jwt_token/url can be passed explicitly (e.g. from MCP handlers)
|
||||
# or auto-detected (e.g. from gns3-copilot agent)
|
||||
logger.debug("Connecting to GNS3 server...")
|
||||
server = get_gns3_connector()
|
||||
server = get_gns3_connector(jwt_token=jwt_token, url=url)
|
||||
|
||||
if server is None:
|
||||
logger.error("Failed to create GNS3 connector")
|
||||
|
||||
@ -26,6 +26,11 @@
|
||||
|
||||
This module provides a tool to execute configuration commands on multiple
|
||||
devices in a GNS3 topology using Nornir.
|
||||
|
||||
⚠️ WARNING: This module is shared with the MCP (Model Context Protocol) service.
|
||||
ExecuteMultipleDeviceConfigCommands._run() is called by the MCP device_config_send handler.
|
||||
The jwt_token/url parameters were added for MCP compatibility.
|
||||
Modifications must be tested with BOTH gns3-copilot AND MCP.
|
||||
"""
|
||||
|
||||
import json
|
||||
@ -169,6 +174,8 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
|
||||
self,
|
||||
tool_input: str, # or Union[str, List[Any], Dict[str, Any]]
|
||||
run_manager: CallbackManagerForToolRun | None = None,
|
||||
jwt_token: str | None = None,
|
||||
url: str | None = None,
|
||||
**kwargs: Any,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""
|
||||
@ -177,6 +184,8 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
|
||||
Args:
|
||||
tool_input (str): A JSON string containing project_id and device
|
||||
configuration commands to execute.
|
||||
jwt_token: JWT token for GNS3 API auth (MCP handlers).
|
||||
url: GNS3 server URL (MCP handlers).
|
||||
|
||||
Returns:
|
||||
List[Dict[str, Any]]: A list of dicts containing device names and
|
||||
@ -214,7 +223,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
|
||||
# Prepare device hosts data
|
||||
try:
|
||||
hosts_data = self._prepare_device_hosts_data(
|
||||
device_configs_list, project_id
|
||||
device_configs_list, project_id, jwt_token=jwt_token, url=url
|
||||
)
|
||||
except ValueError as e:
|
||||
logger.error("Failed to prepare device hosts data: %s", e)
|
||||
@ -547,6 +556,8 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
|
||||
self,
|
||||
device_config_list: list[dict[str, Any]],
|
||||
project_id: str | None = None,
|
||||
jwt_token: str | None = None,
|
||||
url: str | None = None,
|
||||
) -> dict[str, dict[str, Any]]:
|
||||
"""Prepare device hosts data from topology information."""
|
||||
# Extract device names list
|
||||
@ -556,7 +567,9 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
|
||||
]
|
||||
|
||||
# Get device port information with project_id
|
||||
hosts_data = get_device_ports_from_topology(device_names, project_id)
|
||||
hosts_data = get_device_ports_from_topology(
|
||||
device_names, project_id, jwt_token=jwt_token, url=url
|
||||
)
|
||||
|
||||
if not hosts_data:
|
||||
error_msg = (
|
||||
|
||||
@ -26,6 +26,11 @@
|
||||
|
||||
This module provides a tool to execute display commands on multiple devices
|
||||
in a GNS3 topology using Nornir.
|
||||
|
||||
⚠️ WARNING: This module is shared with the MCP (Model Context Protocol) service.
|
||||
ExecuteMultipleDeviceCommands._run() is called by the MCP device_command_run handler.
|
||||
The jwt_token/url parameters were added for MCP compatibility.
|
||||
Modifications must be tested with BOTH gns3-copilot AND MCP.
|
||||
"""
|
||||
|
||||
import json
|
||||
@ -171,6 +176,8 @@ class ExecuteMultipleDeviceCommands(BaseTool):
|
||||
self,
|
||||
tool_input: str | bytes | list[Any] | dict[str, Any],
|
||||
run_manager: CallbackManagerForToolRun | None = None,
|
||||
jwt_token: str | None = None,
|
||||
url: str | None = None,
|
||||
**kwargs: Any,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""
|
||||
@ -181,6 +188,8 @@ class ExecuteMultipleDeviceCommands(BaseTool):
|
||||
|
||||
Args:
|
||||
tool_input: JSON string with project_id and diagnostic commands.
|
||||
jwt_token: JWT token for GNS3 API auth (MCP handlers).
|
||||
url: GNS3 server URL (MCP handlers).
|
||||
|
||||
Returns:
|
||||
List[Dict]: A list of dicts with device names and outputs.
|
||||
@ -210,7 +219,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
|
||||
# Prepare device hosts data
|
||||
try:
|
||||
hosts_data = self._prepare_device_hosts_data(
|
||||
device_configs_list, project_id
|
||||
device_configs_list, project_id, jwt_token=jwt_token, url=url
|
||||
)
|
||||
except ValueError as e:
|
||||
logger.error("Failed to prepare device hosts data: %s", e)
|
||||
@ -490,6 +499,8 @@ class ExecuteMultipleDeviceCommands(BaseTool):
|
||||
self,
|
||||
device_config_list: list[dict[str, Any]],
|
||||
project_id: str | None = None,
|
||||
jwt_token: str | None = None,
|
||||
url: str | None = None,
|
||||
) -> dict[str, dict[str, Any]]:
|
||||
"""Prepare device hosts data from topology information."""
|
||||
# Extract device names list
|
||||
@ -499,7 +510,9 @@ class ExecuteMultipleDeviceCommands(BaseTool):
|
||||
]
|
||||
|
||||
# Get device port information with project_id
|
||||
hosts_data = get_device_ports_from_topology(device_names, project_id)
|
||||
hosts_data = get_device_ports_from_topology(
|
||||
device_names, project_id, jwt_token=jwt_token, url=url
|
||||
)
|
||||
|
||||
if not hosts_data:
|
||||
error_msg = (
|
||||
|
||||
@ -26,6 +26,11 @@
|
||||
"""
|
||||
This module provides a tool to execute commands on VPCS devices
|
||||
in a GNS3 topology using Nornir with Netmiko.
|
||||
|
||||
⚠️ WARNING: This module is shared with the MCP (Model Context Protocol) service.
|
||||
VPCSCommands._run() is called by the MCP vpcs_config_set handler.
|
||||
The jwt_token/url parameters were added for MCP compatibility.
|
||||
Modifications must be tested with BOTH gns3-copilot AND MCP.
|
||||
"""
|
||||
|
||||
import json
|
||||
@ -154,6 +159,8 @@ class VPCSCommands(BaseTool):
|
||||
self,
|
||||
tool_input: str | bytes | list[Any] | dict[str, Any],
|
||||
run_manager: CallbackManagerForToolRun | None = None,
|
||||
jwt_token: str | None = None,
|
||||
url: str | None = None,
|
||||
**kwargs: Any,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""
|
||||
@ -161,6 +168,8 @@ class VPCSCommands(BaseTool):
|
||||
|
||||
Args:
|
||||
tool_input: JSON string with project_id and VPCS commands.
|
||||
jwt_token: JWT token for GNS3 API auth (MCP handlers).
|
||||
url: GNS3 server URL (MCP handlers).
|
||||
|
||||
Returns:
|
||||
List of dicts with device names and command outputs.
|
||||
@ -183,7 +192,7 @@ class VPCSCommands(BaseTool):
|
||||
# Prepare device hosts data
|
||||
try:
|
||||
hosts_data = self._prepare_device_hosts_data(
|
||||
device_configs_list, project_id
|
||||
device_configs_list, project_id, jwt_token=jwt_token, url=url
|
||||
)
|
||||
except ValueError as e:
|
||||
logger.error("Failed to prepare device hosts data: %s", e)
|
||||
@ -408,7 +417,11 @@ class VPCSCommands(BaseTool):
|
||||
}
|
||||
|
||||
def _prepare_device_hosts_data(
|
||||
self, device_configs_list: list[dict[str, Any]], project_id: str
|
||||
self,
|
||||
device_configs_list: list[dict[str, Any]],
|
||||
project_id: str,
|
||||
jwt_token: str | None = None,
|
||||
url: str | None = None,
|
||||
) -> dict[str, dict[str, Any]]:
|
||||
"""
|
||||
Prepare Nornir inventory hosts data for VPCS devices.
|
||||
@ -416,6 +429,8 @@ class VPCSCommands(BaseTool):
|
||||
Args:
|
||||
device_configs_list: List of device configurations
|
||||
project_id: GNS3 project ID
|
||||
jwt_token: JWT token for GNS3 API auth (MCP handlers).
|
||||
url: GNS3 server URL (MCP handlers).
|
||||
|
||||
Returns:
|
||||
Dictionary mapping device names to their host data
|
||||
@ -431,7 +446,7 @@ class VPCSCommands(BaseTool):
|
||||
|
||||
# Get device port mappings from topology
|
||||
device_ports = get_device_ports_from_topology(
|
||||
device_names, project_id=project_id
|
||||
device_names, project_id=project_id, jwt_token=jwt_token, url=url
|
||||
)
|
||||
|
||||
# Build Nornir inventory hosts data
|
||||
|
||||
@ -25,6 +25,11 @@
|
||||
"""
|
||||
|
||||
Public module for getting device port information from GNS3 topology
|
||||
|
||||
⚠️ WARNING: This module is shared with the MCP (Model Context Protocol) service.
|
||||
get_device_ports_from_topology() is called by MCP device config handlers.
|
||||
The jwt_token/url parameters were added for MCP compatibility.
|
||||
Modifications must be tested with BOTH gns3-copilot AND MCP.
|
||||
"""
|
||||
|
||||
import logging
|
||||
@ -36,6 +41,8 @@ logger = logging.getLogger(__name__)
|
||||
def get_device_ports_from_topology(
|
||||
device_names: list[str],
|
||||
project_id: str | None = None,
|
||||
jwt_token: str | None = None,
|
||||
url: str | None = None,
|
||||
) -> dict[str, dict[str, Any]]:
|
||||
"""
|
||||
Get device connection information from GNS3 topology
|
||||
@ -43,6 +50,8 @@ def get_device_ports_from_topology(
|
||||
Args:
|
||||
device_names: List of device names to look up
|
||||
project_id: UUID of the specific GNS3 project to retrieve topology from
|
||||
jwt_token: JWT token for authentication (used by MCP handlers).
|
||||
url: GNS3 server URL (used by MCP handlers).
|
||||
|
||||
Returns:
|
||||
Dictionary mapping device names to their connection data:
|
||||
@ -71,7 +80,7 @@ def get_device_ports_from_topology(
|
||||
|
||||
# Get topology information
|
||||
topo = GNS3TopologyTool()
|
||||
topology = topo._run(project_id=project_id)
|
||||
topology = topo._run(project_id=project_id, jwt_token=jwt_token, url=url)
|
||||
|
||||
# Dynamically build hosts_data from topology
|
||||
hosts_data: dict[str, dict[str, Any]] = {}
|
||||
|
||||
@ -60,6 +60,7 @@ from . import roles
|
||||
from . import acl
|
||||
from . import pools
|
||||
from . import privileges
|
||||
from . import api_keys
|
||||
|
||||
from .dependencies.authentication import get_current_active_user
|
||||
|
||||
@ -192,3 +193,9 @@ router.include_router(
|
||||
dependencies=[Depends(get_current_active_user)],
|
||||
tags=["GNS3 Copilot"]
|
||||
)
|
||||
|
||||
router.include_router(
|
||||
api_keys.router,
|
||||
dependencies=[Depends(get_current_active_user)],
|
||||
tags=["API Keys"]
|
||||
)
|
||||
|
||||
146
gns3server/api/routes/controller/api_keys.py
Normal file
146
gns3server/api/routes/controller/api_keys.py
Normal file
@ -0,0 +1,146 @@
|
||||
#
|
||||
# Copyright (C) 2026 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 API key management.
|
||||
"""
|
||||
|
||||
import secrets
|
||||
import bcrypt
|
||||
from uuid import uuid4, UUID
|
||||
|
||||
from fastapi import APIRouter, Depends, status, HTTPException
|
||||
|
||||
from gns3server import schemas
|
||||
from gns3server.db.repositories.api_keys import ApiKeysRepository
|
||||
from .dependencies.database import get_repository
|
||||
from .dependencies.authentication import get_current_active_user
|
||||
|
||||
import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(prefix="/access/api-keys", tags=["API Keys"])
|
||||
|
||||
API_KEY_PREFIX = "gns3_"
|
||||
API_KEY_BYTES = 32
|
||||
|
||||
|
||||
def _generate_api_key() -> tuple[str, str, str]:
|
||||
random_bytes = secrets.token_hex(API_KEY_BYTES)
|
||||
raw_key = API_KEY_PREFIX + random_bytes
|
||||
key_hash = bcrypt.hashpw(raw_key.encode(), bcrypt.gensalt()).decode()
|
||||
key_prefix = raw_key[: len(API_KEY_PREFIX) + 8]
|
||||
return raw_key, key_hash, key_prefix
|
||||
|
||||
|
||||
@router.post("", status_code=status.HTTP_201_CREATED)
|
||||
async def create_api_key(
|
||||
api_key_data: schemas.ApiKeyCreate,
|
||||
current_user: schemas.User = Depends(get_current_active_user),
|
||||
api_keys_repo: ApiKeysRepository = Depends(get_repository(ApiKeysRepository)),
|
||||
) -> dict:
|
||||
"""Create a new API key. The full key is returned only once."""
|
||||
|
||||
raw_key, key_hash, key_prefix = _generate_api_key()
|
||||
db_key = await api_keys_repo.create_api_key(
|
||||
api_key_id=uuid4(),
|
||||
user_id=current_user.user_id,
|
||||
name=api_key_data.name,
|
||||
key_hash=key_hash,
|
||||
key_prefix=key_prefix,
|
||||
)
|
||||
return {
|
||||
"api_key_id": str(db_key.api_key_id),
|
||||
"api_key": raw_key,
|
||||
"name": db_key.name,
|
||||
"key_prefix": db_key.key_prefix,
|
||||
"created_at": db_key.created_at.isoformat() if db_key.created_at else None,
|
||||
}
|
||||
|
||||
|
||||
@router.get("")
|
||||
async def list_api_keys(
|
||||
current_user: schemas.User = Depends(get_current_active_user),
|
||||
api_keys_repo: ApiKeysRepository = Depends(get_repository(ApiKeysRepository)),
|
||||
) -> list[dict]:
|
||||
"""List all API keys for the current user."""
|
||||
|
||||
keys = await api_keys_repo.get_api_keys_by_user(current_user.user_id)
|
||||
return [
|
||||
{
|
||||
"api_key_id": str(k.api_key_id),
|
||||
"name": k.name,
|
||||
"key_prefix": k.key_prefix,
|
||||
"created_at": k.created_at.isoformat() if k.created_at else None,
|
||||
"last_used_at": k.last_used_at.isoformat() if k.last_used_at else None,
|
||||
"revoked": k.revoked,
|
||||
}
|
||||
for k in keys
|
||||
]
|
||||
|
||||
|
||||
@router.post("/{api_key_id}/revoke", status_code=status.HTTP_200_OK)
|
||||
async def revoke_api_key(
|
||||
api_key_id: UUID,
|
||||
current_user: schemas.User = Depends(get_current_active_user),
|
||||
api_keys_repo: ApiKeysRepository = Depends(get_repository(ApiKeysRepository)),
|
||||
) -> dict:
|
||||
"""Revoke an API key. It will immediately stop working, but can be restored."""
|
||||
|
||||
key = await api_keys_repo.get_api_key(api_key_id)
|
||||
if not key:
|
||||
raise HTTPException(status_code=404, detail="API key not found")
|
||||
if key.user_id != current_user.user_id:
|
||||
raise HTTPException(status_code=403, detail="Cannot modify another user's API key")
|
||||
|
||||
await api_keys_repo.revoke_api_key(api_key_id)
|
||||
return {"message": f"API key '{key.name}' revoked"}
|
||||
|
||||
|
||||
@router.post("/{api_key_id}/restore", status_code=status.HTTP_200_OK)
|
||||
async def restore_api_key(
|
||||
api_key_id: UUID,
|
||||
current_user: schemas.User = Depends(get_current_active_user),
|
||||
api_keys_repo: ApiKeysRepository = Depends(get_repository(ApiKeysRepository)),
|
||||
) -> dict:
|
||||
"""Restore a previously revoked API key."""
|
||||
|
||||
key = await api_keys_repo.get_api_key(api_key_id)
|
||||
if not key:
|
||||
raise HTTPException(status_code=404, detail="API key not found")
|
||||
if key.user_id != current_user.user_id:
|
||||
raise HTTPException(status_code=403, detail="Cannot modify another user's API key")
|
||||
|
||||
await api_keys_repo.restore_api_key(api_key_id)
|
||||
return {"message": f"API key '{key.name}' restored"}
|
||||
|
||||
|
||||
@router.delete("/{api_key_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
async def delete_api_key(
|
||||
api_key_id: UUID,
|
||||
current_user: schemas.User = Depends(get_current_active_user),
|
||||
api_keys_repo: ApiKeysRepository = Depends(get_repository(ApiKeysRepository)),
|
||||
) -> None:
|
||||
"""Permanently delete an API key. Cannot be undone."""
|
||||
|
||||
key = await api_keys_repo.get_api_key(api_key_id)
|
||||
if not key:
|
||||
raise HTTPException(status_code=404, detail="API key not found")
|
||||
if key.user_id != current_user.user_id:
|
||||
raise HTTPException(status_code=403, detail="Cannot delete another user's API key")
|
||||
|
||||
await api_keys_repo.delete_api_key(api_key_id)
|
||||
@ -15,12 +15,16 @@
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import logging
|
||||
import bcrypt
|
||||
|
||||
from fastapi import Request, Query, Depends, HTTPException, WebSocket, status
|
||||
from fastapi.security import OAuth2PasswordBearer
|
||||
from typing import Optional
|
||||
from sqlalchemy import select
|
||||
|
||||
from gns3server import schemas
|
||||
import gns3server.db.models as models
|
||||
from gns3server.db.repositories.api_keys import ApiKeysRepository
|
||||
from gns3server.db.repositories.users import UsersRepository
|
||||
from gns3server.db.repositories.rbac import RbacRepository
|
||||
from gns3server.services import auth_service
|
||||
@ -33,6 +37,7 @@ oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/v3/access/users/login", auto_err
|
||||
async def get_user_from_token(
|
||||
bearer_token: str = Depends(oauth2_scheme),
|
||||
user_repo: UsersRepository = Depends(get_repository(UsersRepository)),
|
||||
api_keys_repo: ApiKeysRepository = Depends(get_repository(ApiKeysRepository)),
|
||||
token: Optional[str] = Query(None, include_in_schema=False)
|
||||
) -> schemas.User:
|
||||
|
||||
@ -47,6 +52,29 @@ async def get_user_from_token(
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
|
||||
# API Key authentication
|
||||
if token.startswith("gns3_"):
|
||||
query = select(models.ApiKey).where(models.ApiKey.revoked == False)
|
||||
result = await api_keys_repo._db_session.execute(query)
|
||||
for db_key in result.scalars().all():
|
||||
if bcrypt.checkpw(token.encode(), db_key.key_hash.encode()):
|
||||
await api_keys_repo.update_last_used(db_key.api_key_id)
|
||||
user = await user_repo.get_user(db_key.user_id)
|
||||
if user:
|
||||
if not user.is_active:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Not an active user",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
return user
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Invalid API key",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
|
||||
# JWT authentication
|
||||
token_data = auth_service.get_token_data(token)
|
||||
user = await user_repo.get_user_by_username(token_data.username)
|
||||
if user is None:
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
63
gns3server/api/routes/mcp/appliances.py
Normal file
63
gns3server/api/routes/mcp/appliances.py
Normal file
@ -0,0 +1,63 @@
|
||||
#
|
||||
# Copyright (C) 2026 GNS3 Technologies Inc.
|
||||
# Author: Yue Guobin
|
||||
#
|
||||
# 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/>.
|
||||
|
||||
"""
|
||||
MCP tool handlers for GNS3 appliance management.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ── Helper ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def _get_connector(gns3_ctx: dict[str, Any]):
|
||||
from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector
|
||||
return Gns3Connector(
|
||||
url=gns3_ctx["server_url"],
|
||||
jwt_token=gns3_ctx["jwt_token"],
|
||||
api_version=3,
|
||||
verify=False,
|
||||
)
|
||||
|
||||
|
||||
# ── Tool handlers ──────────────────────────────────────────────────────────
|
||||
|
||||
def get_appliances_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
appliances = conn.http_call("get", f"{conn.base_url}/appliances").json()
|
||||
return {"appliances": appliances, "count": len(appliances)}
|
||||
|
||||
|
||||
def get_appliance_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
appliance_id = params.get("appliance_id")
|
||||
if not appliance_id:
|
||||
return {"error": "appliance_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return conn.http_call("get", f"{conn.base_url}/appliances/{appliance_id}").json()
|
||||
|
||||
|
||||
def install_appliance_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
appliance_id = params.get("appliance_id")
|
||||
if not appliance_id:
|
||||
return {"error": "appliance_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
result = conn.http_call("post", f"{conn.base_url}/appliances/{appliance_id}/install").json()
|
||||
return {"message": f"Appliance {appliance_id} installation requested", "result": result}
|
||||
@ -37,23 +37,27 @@ def _get_connector(gns3_ctx: dict[str, Any]):
|
||||
|
||||
def list_computes_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
computes = conn.get_computes()
|
||||
computes = conn.http_call("get", f"{conn.base_url}/computes").json()
|
||||
return {"computes": computes, "count": len(computes)}
|
||||
|
||||
|
||||
def get_compute_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
compute_id = params.get("compute_id", "local")
|
||||
compute_id = params.get("compute_id")
|
||||
if not compute_id:
|
||||
return {"error": "compute_id is required (use compute_list to get the UUID)"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return conn.get_compute(compute_id=compute_id)
|
||||
return conn.http_call("get", f"{conn.base_url}/computes/{compute_id}").json()
|
||||
|
||||
|
||||
def get_compute_images_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
emulator = params.get("emulator")
|
||||
compute_id = params.get("compute_id", "local")
|
||||
compute_id = params.get("compute_id")
|
||||
if not emulator:
|
||||
return {"error": "emulator is required (e.g. qemu, iou, docker)"}
|
||||
if not compute_id:
|
||||
return {"error": "compute_id is required (use compute_list to get the UUID)"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
images = conn.get_compute_images(emulator=emulator, compute_id=compute_id)
|
||||
images = conn.http_call("get", f"{conn.base_url}/computes/{compute_id}/{emulator}/images").json()
|
||||
return {"images": images, "count": len(images)}
|
||||
|
||||
|
||||
|
||||
101
gns3server/api/routes/mcp/device_config.py
Normal file
101
gns3server/api/routes/mcp/device_config.py
Normal file
@ -0,0 +1,101 @@
|
||||
#
|
||||
# Copyright (C) 2026 GNS3 Technologies Inc.
|
||||
# Author: Yue Guobin
|
||||
#
|
||||
# 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/>.
|
||||
|
||||
"""
|
||||
MCP tool handlers for device configuration via Nornir + Netmiko.
|
||||
|
||||
These tools connect to network device consoles via telnet/SSH and execute
|
||||
configuration or diagnostic commands. Device connection info is automatically
|
||||
discovered from the project topology using the device's tags for device_type.
|
||||
|
||||
Prerequisites:
|
||||
- Device must be started (use node_start / node_start_all)
|
||||
- Device must have a 'device_type:<type>' tag set in GNS3
|
||||
(right-click → Configure → Tags → add 'device_type:cisco_ios_telnet')
|
||||
- Device must have a console port assigned
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ── Tool handlers ──────────────────────────────────────────────────────────
|
||||
|
||||
def device_config_send_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
"""Send configuration commands to network devices via console."""
|
||||
project_id = params.get("project_id")
|
||||
device_configs = params.get("device_configs")
|
||||
if not project_id or not device_configs:
|
||||
return [{"error": "project_id and device_configs are required"}]
|
||||
|
||||
from gns3server.agent.gns3_copilot.tools_v2.config_tools_nornir import ExecuteMultipleDeviceConfigCommands
|
||||
|
||||
tool = ExecuteMultipleDeviceConfigCommands()
|
||||
input_data = json.dumps({
|
||||
"project_id": project_id,
|
||||
"device_configs": device_configs,
|
||||
})
|
||||
return tool._run(
|
||||
input_data,
|
||||
jwt_token=gns3_ctx["jwt_token"],
|
||||
url=gns3_ctx["server_url"],
|
||||
)
|
||||
|
||||
|
||||
def device_command_run_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
"""Run read-only diagnostic (show) commands on network devices."""
|
||||
project_id = params.get("project_id")
|
||||
device_configs = params.get("device_configs")
|
||||
if not project_id or not device_configs:
|
||||
return [{"error": "project_id and device_configs (list of {device_name, show_commands}) are required"}]
|
||||
|
||||
from gns3server.agent.gns3_copilot.tools_v2.display_tools_nornir import ExecuteMultipleDeviceCommands
|
||||
|
||||
tool = ExecuteMultipleDeviceCommands()
|
||||
input_data = json.dumps({
|
||||
"project_id": project_id,
|
||||
"device_configs": device_configs,
|
||||
})
|
||||
return tool._run(
|
||||
input_data,
|
||||
jwt_token=gns3_ctx["jwt_token"],
|
||||
url=gns3_ctx["server_url"],
|
||||
)
|
||||
|
||||
|
||||
def vpcs_config_set_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
"""Configure VPCS devices (set IP, gateway, etc.)."""
|
||||
project_id = params.get("project_id")
|
||||
device_configs = params.get("device_configs")
|
||||
if not project_id or not device_configs:
|
||||
return [{"error": "project_id and device_configs are required"}]
|
||||
|
||||
from gns3server.agent.gns3_copilot.tools_v2.vpcs_tools_netmiko import VPCSCommands
|
||||
|
||||
tool = VPCSCommands()
|
||||
input_data = json.dumps({
|
||||
"project_id": project_id,
|
||||
"device_configs": device_configs,
|
||||
})
|
||||
return tool._run(
|
||||
input_data,
|
||||
jwt_token=gns3_ctx["jwt_token"],
|
||||
url=gns3_ctx["server_url"],
|
||||
)
|
||||
96
gns3server/api/routes/mcp/drawings.py
Normal file
96
gns3server/api/routes/mcp/drawings.py
Normal file
@ -0,0 +1,96 @@
|
||||
#
|
||||
# Copyright (C) 2026 GNS3 Technologies Inc.
|
||||
# Author: Yue Guobin
|
||||
#
|
||||
# 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/>.
|
||||
|
||||
"""
|
||||
MCP tool handlers for GNS3 drawing management.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ── Helper ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def _get_connector(gns3_ctx: dict[str, Any]):
|
||||
from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector
|
||||
return Gns3Connector(
|
||||
url=gns3_ctx["server_url"],
|
||||
jwt_token=gns3_ctx["jwt_token"],
|
||||
api_version=3,
|
||||
verify=False,
|
||||
)
|
||||
|
||||
|
||||
# ── Tool handlers ──────────────────────────────────────────────────────────
|
||||
|
||||
def get_drawings_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
if not project_id:
|
||||
return {"error": "project_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
drawings = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/drawings").json()
|
||||
return {"drawings": drawings, "count": len(drawings)}
|
||||
|
||||
|
||||
def create_drawing_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
svg = params.get("svg")
|
||||
if not project_id or not svg:
|
||||
return {"error": "project_id and svg are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
data = {
|
||||
"svg": svg,
|
||||
"x": params.get("x", 0),
|
||||
"y": params.get("y", 0),
|
||||
"z": params.get("z", 0),
|
||||
"locked": params.get("locked", False),
|
||||
"rotation": params.get("rotation", 0),
|
||||
}
|
||||
result = conn.http_call("post", f"{conn.base_url}/projects/{project_id}/drawings", json_data=data).json()
|
||||
return {"message": "Drawing created", "drawing": result}
|
||||
|
||||
|
||||
def get_drawing_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
drawing_id = params.get("drawing_id")
|
||||
if not project_id or not drawing_id:
|
||||
return {"error": "project_id and drawing_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return conn.http_call("get", f"{conn.base_url}/projects/{project_id}/drawings/{drawing_id}").json()
|
||||
|
||||
|
||||
def update_drawing_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
drawing_id = params.get("drawing_id")
|
||||
if not project_id or not drawing_id:
|
||||
return {"error": "project_id and drawing_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
data = {k: v for k, v in params.items() if k not in ("project_id", "drawing_id") and v is not None}
|
||||
return conn.http_call("put", f"{conn.base_url}/projects/{project_id}/drawings/{drawing_id}", json_data=data).json()
|
||||
|
||||
|
||||
def delete_drawing_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
drawing_id = params.get("drawing_id")
|
||||
if not project_id or not drawing_id:
|
||||
return {"error": "project_id and drawing_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("delete", f"{conn.base_url}/projects/{project_id}/drawings/{drawing_id}")
|
||||
return {"message": f"Drawing {drawing_id} deleted", "drawing_id": drawing_id}
|
||||
77
gns3server/api/routes/mcp/images.py
Normal file
77
gns3server/api/routes/mcp/images.py
Normal file
@ -0,0 +1,77 @@
|
||||
#
|
||||
# Copyright (C) 2026 GNS3 Technologies Inc.
|
||||
# Author: Yue Guobin
|
||||
#
|
||||
# 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/>.
|
||||
|
||||
"""
|
||||
MCP tool handlers for GNS3 image management.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ── Helper ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def _get_connector(gns3_ctx: dict[str, Any]):
|
||||
from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector
|
||||
return Gns3Connector(
|
||||
url=gns3_ctx["server_url"],
|
||||
jwt_token=gns3_ctx["jwt_token"],
|
||||
api_version=3,
|
||||
verify=False,
|
||||
)
|
||||
|
||||
|
||||
# ── Tool handlers ──────────────────────────────────────────────────────────
|
||||
|
||||
def get_images_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
images = conn.http_call("get", f"{conn.base_url}/images").json()
|
||||
return {"images": images, "count": len(images)}
|
||||
|
||||
|
||||
def get_image_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
image_id = params.get("image_id")
|
||||
if not image_id:
|
||||
return {"error": "image_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return conn.http_call("get", f"{conn.base_url}/images/{image_id}").json()
|
||||
|
||||
|
||||
def delete_image_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
image_id = params.get("image_id")
|
||||
if not image_id:
|
||||
return {"error": "image_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("delete", f"{conn.base_url}/images/{image_id}")
|
||||
return {"message": f"Image {image_id} deleted", "image_id": image_id}
|
||||
|
||||
|
||||
def prune_images_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
# Returns 204 No Content on success (empty body, no .json())
|
||||
conn.http_call("delete", f"{conn.base_url}/images/prune")
|
||||
return {"message": "Unused images pruned"}
|
||||
|
||||
|
||||
def install_images_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
# Returns 204 No Content on success (empty body, no .json())
|
||||
conn.http_call("post", f"{conn.base_url}/images/install")
|
||||
return {"message": "Image installation completed"}
|
||||
@ -48,7 +48,7 @@ def get_links_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[
|
||||
if not project_id:
|
||||
return {"error": "project_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
links = conn.get_links(project_id=project_id)
|
||||
links = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/links").json()
|
||||
return {"links": links, "count": len(links)}
|
||||
|
||||
|
||||
@ -58,7 +58,7 @@ def get_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[s
|
||||
if not project_id or not link_id:
|
||||
return {"error": "project_id and link_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return conn.get_link(project_id=project_id, link_id=link_id)
|
||||
return conn.http_call("get", f"{conn.base_url}/projects/{project_id}/links/{link_id}").json()
|
||||
|
||||
|
||||
def create_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
@ -105,6 +105,64 @@ def update_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dic
|
||||
return conn.http_call("put", url, json_data=update_data).json()
|
||||
|
||||
|
||||
# ── Link capture / reset handlers ──────────────────────────────────────
|
||||
|
||||
|
||||
def reset_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
link_id = params.get("link_id")
|
||||
if not project_id or not link_id:
|
||||
return {"error": "project_id and link_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
url = f"{conn.base_url}/projects/{project_id}/links/{link_id}/reset"
|
||||
result = conn.http_call("post", url).json()
|
||||
return {"message": f"Link {link_id} reset", "link": result}
|
||||
|
||||
|
||||
def start_capture_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
link_id = params.get("link_id")
|
||||
if not project_id or not link_id:
|
||||
return {"error": "project_id and link_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
data = {
|
||||
"data_link_type": params.get("data_link_type", "DLT_EN10MB"),
|
||||
"wireshark": params.get("wireshark", False),
|
||||
}
|
||||
if params.get("capture_file_name"):
|
||||
data["capture_file_name"] = params["capture_file_name"]
|
||||
url = f"{conn.base_url}/projects/{project_id}/links/{link_id}/capture/start"
|
||||
result = conn.http_call("post", url, json_data=data).json()
|
||||
return {"message": f"Capture started on link {link_id}", "link": result}
|
||||
|
||||
|
||||
def stop_capture_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
link_id = params.get("link_id")
|
||||
if not project_id or not link_id:
|
||||
return {"error": "project_id and link_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
url = f"{conn.base_url}/projects/{project_id}/links/{link_id}/capture/stop"
|
||||
conn.http_call("post", url)
|
||||
return {"message": f"Capture stopped on link {link_id}", "link_id": link_id}
|
||||
|
||||
|
||||
def download_capture_file_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
link_id = params.get("link_id")
|
||||
if not project_id or not link_id:
|
||||
return {"error": "project_id and link_id are required"}
|
||||
download_url = f"{gns3_ctx['server_url']}/v3/projects/{project_id}/links/{link_id}/capture/file"
|
||||
auth_token = gns3_ctx['jwt_token']
|
||||
return {
|
||||
"link_id": link_id,
|
||||
"download_url": download_url,
|
||||
"curl_command": f"curl -L -o capture.pcap -H 'Authorization: Bearer {auth_token}' '{download_url}'",
|
||||
"note": "Use the curl command to download the PCAP capture file. "
|
||||
"The file is in pcap format and can be analyzed with Wireshark or tcpdump.",
|
||||
}
|
||||
|
||||
|
||||
# ── Tool definitions ───────────────────────────────────────────────────────
|
||||
|
||||
LINK_TOOLS = [
|
||||
@ -193,4 +251,61 @@ LINK_TOOLS = [
|
||||
},
|
||||
"handler": update_link_handler,
|
||||
},
|
||||
{
|
||||
"name": "reset_link",
|
||||
"description": "Reset a link, clearing its state (counters, filters, etc.)",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {"type": "string", "description": "Project UUID"},
|
||||
"link_id": {"type": "string", "description": "Link UUID"},
|
||||
},
|
||||
"required": ["project_id", "link_id"],
|
||||
},
|
||||
"handler": reset_link_handler,
|
||||
},
|
||||
{
|
||||
"name": "start_capture",
|
||||
"description": "Start packet capture on a link. The capture file can later be downloaded with download_capture_file.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {"type": "string", "description": "Project UUID"},
|
||||
"link_id": {"type": "string", "description": "Link UUID"},
|
||||
"data_link_type": {"type": "string", "description": "Data link type (optional, default: DLT_EN10MB)"},
|
||||
"capture_file_name": {"type": "string", "description": "Capture file name (optional)"},
|
||||
"wireshark": {"type": "boolean", "description": "Open Wireshark automatically (optional, default: false)"},
|
||||
},
|
||||
"required": ["project_id", "link_id"],
|
||||
},
|
||||
"handler": start_capture_handler,
|
||||
},
|
||||
{
|
||||
"name": "stop_capture",
|
||||
"description": "Stop packet capture on a link. After stopping, the capture file can be downloaded.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {"type": "string", "description": "Project UUID"},
|
||||
"link_id": {"type": "string", "description": "Link UUID"},
|
||||
},
|
||||
"required": ["project_id", "link_id"],
|
||||
},
|
||||
"handler": stop_capture_handler,
|
||||
},
|
||||
{
|
||||
"name": "download_capture_file",
|
||||
"description": "Get the download URL and instructions for a PCAP capture file from a link. "
|
||||
"Use the returned curl command to download the file. "
|
||||
"The PCAP file can be analyzed with Wireshark or tcpdump.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {"type": "string", "description": "Project UUID"},
|
||||
"link_id": {"type": "string", "description": "Link UUID"},
|
||||
},
|
||||
"required": ["project_id", "link_id"],
|
||||
},
|
||||
"handler": download_capture_file_handler,
|
||||
},
|
||||
]
|
||||
|
||||
@ -54,7 +54,7 @@ def get_nodes_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[
|
||||
if not project_id:
|
||||
return {"error": "project_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
nodes = conn.get_nodes(project_id=project_id)
|
||||
nodes = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/nodes").json()
|
||||
return {"nodes": nodes, "count": len(nodes)}
|
||||
|
||||
|
||||
@ -64,7 +64,7 @@ def get_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[s
|
||||
if not project_id or not node_id:
|
||||
return {"error": "project_id and node_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return conn.get_node(project_id=project_id, node_id=node_id)
|
||||
return conn.http_call("get", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}").json()
|
||||
|
||||
|
||||
def start_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
@ -155,7 +155,7 @@ def get_node_console_info_handler(params: dict[str, Any], gns3_ctx: dict[str, An
|
||||
if not project_id or not node_id:
|
||||
return {"error": "project_id and node_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
node = conn.get_node(project_id=project_id, node_id=node_id)
|
||||
node = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}").json()
|
||||
|
||||
console_type = node.get("console_type", "unknown")
|
||||
ws_url = f"{gns3_ctx['server_url']}/v3/projects/{project_id}/nodes/{node_id}/console/ws?token={gns3_ctx['jwt_token']}"
|
||||
@ -181,12 +181,13 @@ def list_node_files_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) ->
|
||||
if not project_id or not node_id:
|
||||
return {"error": "project_id and node_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
files = conn.list_node_files(
|
||||
project_id=project_id,
|
||||
node_id=node_id,
|
||||
path=params.get("path", ""),
|
||||
recursive=params.get("recursive", False),
|
||||
)
|
||||
url = f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/files"
|
||||
query = {}
|
||||
if params.get("path"):
|
||||
query["path"] = params["path"]
|
||||
if params.get("recursive"):
|
||||
query["recursive"] = "true"
|
||||
files = conn.http_call("get", url, params=query if query else None).json()
|
||||
return {"files": files, "count": len(files)}
|
||||
|
||||
|
||||
@ -201,7 +202,8 @@ def get_node_file_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> d
|
||||
limit = params.get("limit", 200)
|
||||
|
||||
conn = _get_connector(gns3_ctx)
|
||||
raw = conn.get_node_file(project_id=project_id, node_id=node_id, file_path=file_path)
|
||||
url = f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/files/{file_path}"
|
||||
raw = conn.http_call("get", url).text
|
||||
|
||||
total_bytes = len(raw.encode("utf-8"))
|
||||
truncated = False
|
||||
@ -240,7 +242,8 @@ def write_node_file_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) ->
|
||||
if not project_id or not node_id or not file_path or content is None:
|
||||
return {"error": "project_id, node_id, file_path and content are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.write_node_file(project_id=project_id, node_id=node_id, file_path=file_path, content=content)
|
||||
url = f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/files/{file_path}"
|
||||
conn.http_call("post", url, data=content, headers={"Content-Type": "text/plain"})
|
||||
return {"message": f"File {file_path} written to node {node_id}", "file_path": file_path, "node_id": node_id}
|
||||
|
||||
|
||||
@ -251,10 +254,91 @@ def delete_node_file_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -
|
||||
if not project_id or not node_id or not file_path:
|
||||
return {"error": "project_id, node_id and file_path are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.delete_node_file(project_id=project_id, node_id=node_id, file_path=file_path)
|
||||
url = f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/files/{file_path}"
|
||||
conn.http_call("delete", url)
|
||||
return {"message": f"File {file_path} deleted from node {node_id}", "file_path": file_path, "node_id": node_id}
|
||||
|
||||
|
||||
# ── Node bulk / advanced handlers ────────────────────────────────────
|
||||
|
||||
|
||||
def start_all_nodes_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
if not project_id:
|
||||
return {"error": "project_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/start")
|
||||
return {"message": "All nodes started", "project_id": project_id}
|
||||
|
||||
|
||||
def stop_all_nodes_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
if not project_id:
|
||||
return {"error": "project_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/stop")
|
||||
return {"message": "All nodes stopped", "project_id": project_id}
|
||||
|
||||
|
||||
def suspend_all_nodes_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
if not project_id:
|
||||
return {"error": "project_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/suspend")
|
||||
return {"message": "All nodes suspended", "project_id": project_id}
|
||||
|
||||
|
||||
def reload_all_nodes_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
if not project_id:
|
||||
return {"error": "project_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/reload")
|
||||
return {"message": "All nodes reloaded", "project_id": project_id}
|
||||
|
||||
|
||||
def duplicate_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
node_id = params.get("node_id")
|
||||
if not project_id or not node_id:
|
||||
return {"error": "project_id and node_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
data = {k: v for k, v in params.items() if k not in ("project_id", "node_id") and v is not None}
|
||||
result = conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/duplicate", json_data=data).json()
|
||||
return {"message": f"Node {node_id} duplicated", "node": result}
|
||||
|
||||
|
||||
def isolate_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
node_id = params.get("node_id")
|
||||
if not project_id or not node_id:
|
||||
return {"error": "project_id and node_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/isolate")
|
||||
return {"message": f"Node {node_id} isolated (all links suspended)", "node_id": node_id}
|
||||
|
||||
|
||||
def unisolate_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
node_id = params.get("node_id")
|
||||
if not project_id or not node_id:
|
||||
return {"error": "project_id and node_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/unisolate")
|
||||
return {"message": f"Node {node_id} unisolated (links resumed)", "node_id": node_id}
|
||||
|
||||
|
||||
def get_node_links_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
node_id = params.get("node_id")
|
||||
if not project_id or not node_id:
|
||||
return {"error": "project_id and node_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
links = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/links").json()
|
||||
return {"links": links, "count": len(links)}
|
||||
|
||||
|
||||
# ── Tool definitions ───────────────────────────────────────────────────────
|
||||
|
||||
NODE_TOOLS = [
|
||||
|
||||
@ -46,7 +46,7 @@ def _get_connector(gns3_ctx: dict[str, Any]):
|
||||
|
||||
def list_projects_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
projects = conn.get_projects()
|
||||
projects = conn.http_call("get", f"{conn.base_url}/projects").json()
|
||||
return {"projects": projects, "count": len(projects)}
|
||||
|
||||
|
||||
@ -55,7 +55,7 @@ def get_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dic
|
||||
if not project_id:
|
||||
return {"error": "project_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
project = conn.get_project(project_id=project_id)
|
||||
project = conn.http_call("get", f"{conn.base_url}/projects/{project_id}").json()
|
||||
if project is None:
|
||||
return {"error": f"Project '{project_id}' not found"}
|
||||
return project
|
||||
@ -66,10 +66,7 @@ def create_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) ->
|
||||
if not name:
|
||||
return {"error": "name is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
project_data = {"name": name}
|
||||
if "description" in params:
|
||||
project_data["description"] = params["description"]
|
||||
return conn.create_project(**project_data)
|
||||
return conn.http_call("post", f"{conn.base_url}/projects", json_data={"name": name}).json()
|
||||
|
||||
|
||||
def delete_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
@ -77,7 +74,7 @@ def delete_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) ->
|
||||
if not project_id:
|
||||
return {"error": "project_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.delete_project(project_id=project_id)
|
||||
conn.http_call("delete", f"{conn.base_url}/projects/{project_id}")
|
||||
return {"message": f"Project '{project_id}' deleted", "project_id": project_id}
|
||||
|
||||
|
||||
@ -115,7 +112,7 @@ def update_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) ->
|
||||
return {"error": "project_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
kwargs = {k: v for k, v in params.items() if k != "project_id" and v is not None}
|
||||
return conn.update_project(project_id=project_id, **kwargs)
|
||||
return conn.http_call("put", f"{conn.base_url}/projects/{project_id}", json_data=kwargs).json()
|
||||
|
||||
|
||||
def duplicate_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
@ -127,7 +124,7 @@ def duplicate_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any])
|
||||
return {"error": "name is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
kwargs = {k: v for k, v in params.items() if k not in ("project_id",) and v is not None}
|
||||
return conn.duplicate_project(project_id=project_id, **kwargs)
|
||||
return conn.http_call("post", f"{conn.base_url}/projects/{project_id}/duplicate", json_data=kwargs).json()
|
||||
|
||||
|
||||
def get_project_readme_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
@ -136,7 +133,8 @@ def get_project_readme_handler(params: dict[str, Any], gns3_ctx: dict[str, Any])
|
||||
return {"error": "project_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
try:
|
||||
content = conn.get_project_file(project_id=project_id, file_path="README.txt")
|
||||
url = f"{conn.base_url}/projects/{project_id}/files/README.txt"
|
||||
content = conn.http_call("get", url).text
|
||||
return {"project_id": project_id, "file": "README.txt", "content": content}
|
||||
except Exception as e:
|
||||
if "404" in str(e):
|
||||
@ -152,10 +150,47 @@ def update_project_readme_handler(params: dict[str, Any], gns3_ctx: dict[str, An
|
||||
if content is None:
|
||||
return {"error": "content is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.write_project_file(project_id=project_id, file_path="README.txt", content=content)
|
||||
url = f"{conn.base_url}/projects/{project_id}/files/README.txt"
|
||||
conn.http_call("post", url, data=content, headers={"Content-Type": "text/plain"})
|
||||
return {"message": "README.txt updated", "project_id": project_id}
|
||||
|
||||
|
||||
def lock_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
if not project_id:
|
||||
return {"error": "project_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/lock")
|
||||
return {"message": f"Project {project_id} locked", "project_id": project_id}
|
||||
|
||||
|
||||
def unlock_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
if not project_id:
|
||||
return {"error": "project_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/unlock")
|
||||
return {"message": f"Project {project_id} unlocked", "project_id": project_id}
|
||||
|
||||
|
||||
def load_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
path = params.get("path")
|
||||
if not path:
|
||||
return {"error": "path is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
result = conn.http_call("post", f"{conn.base_url}/projects/load", json_data={"path": path}).json()
|
||||
return {"message": f"Project loaded from {path}", "project": result}
|
||||
|
||||
|
||||
def get_locked_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
if not project_id:
|
||||
return {"error": "project_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
locked = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/locked").json()
|
||||
return {"project_id": project_id, "locked": locked}
|
||||
|
||||
|
||||
# ── Tool definitions (consumed by mcp/__init__.py) ─────────────────────────
|
||||
|
||||
PROJECT_TOOLS = [
|
||||
|
||||
50
gns3server/api/routes/mcp/server.py
Normal file
50
gns3server/api/routes/mcp/server.py
Normal file
@ -0,0 +1,50 @@
|
||||
#
|
||||
# Copyright (C) 2026 GNS3 Technologies Inc.
|
||||
# Author: Yue Guobin
|
||||
#
|
||||
# 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/>.
|
||||
|
||||
"""
|
||||
MCP tool handlers for GNS3 server information.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ── Helper ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def _get_connector(gns3_ctx: dict[str, Any]):
|
||||
from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector
|
||||
return Gns3Connector(
|
||||
url=gns3_ctx["server_url"],
|
||||
jwt_token=gns3_ctx["jwt_token"],
|
||||
api_version=3,
|
||||
verify=False,
|
||||
)
|
||||
|
||||
|
||||
# ── Tool handlers ──────────────────────────────────────────────────────────
|
||||
|
||||
def get_version_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return conn.http_call("get", f"{conn.base_url}/version").json()
|
||||
|
||||
|
||||
def get_statistics_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return conn.http_call("get", f"{conn.base_url}/statistics").json()
|
||||
79
gns3server/api/routes/mcp/snapshots.py
Normal file
79
gns3server/api/routes/mcp/snapshots.py
Normal file
@ -0,0 +1,79 @@
|
||||
#
|
||||
# Copyright (C) 2026 GNS3 Technologies Inc.
|
||||
# Author: Yue Guobin
|
||||
#
|
||||
# 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/>.
|
||||
|
||||
"""
|
||||
MCP tool handlers for GNS3 snapshot management.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ── Helper ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def _get_connector(gns3_ctx: dict[str, Any]):
|
||||
from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector
|
||||
return Gns3Connector(
|
||||
url=gns3_ctx["server_url"],
|
||||
jwt_token=gns3_ctx["jwt_token"],
|
||||
api_version=3,
|
||||
verify=False,
|
||||
)
|
||||
|
||||
|
||||
# ── Tool handlers ──────────────────────────────────────────────────────────
|
||||
|
||||
def get_snapshots_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
if not project_id:
|
||||
return {"error": "project_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
snapshots = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/snapshots").json()
|
||||
return {"snapshots": snapshots, "count": len(snapshots)}
|
||||
|
||||
|
||||
def create_snapshot_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
name = params.get("name")
|
||||
if not project_id or not name:
|
||||
return {"error": "project_id and name are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
result = conn.http_call("post", f"{conn.base_url}/projects/{project_id}/snapshots", json_data={"name": name}).json()
|
||||
return {"message": f"Snapshot '{name}' created", "snapshot": result}
|
||||
|
||||
|
||||
def delete_snapshot_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
snapshot_id = params.get("snapshot_id")
|
||||
if not project_id or not snapshot_id:
|
||||
return {"error": "project_id and snapshot_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("delete", f"{conn.base_url}/projects/{project_id}/snapshots/{snapshot_id}")
|
||||
return {"message": f"Snapshot {snapshot_id} deleted", "snapshot_id": snapshot_id}
|
||||
|
||||
|
||||
def restore_snapshot_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
project_id = params.get("project_id")
|
||||
snapshot_id = params.get("snapshot_id")
|
||||
if not project_id or not snapshot_id:
|
||||
return {"error": "project_id and snapshot_id are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
result = conn.http_call("post", f"{conn.base_url}/projects/{project_id}/snapshots/{snapshot_id}/restore").json()
|
||||
return {"message": f"Snapshot {snapshot_id} restored", "project": result}
|
||||
94
gns3server/api/routes/mcp/symbols.py
Normal file
94
gns3server/api/routes/mcp/symbols.py
Normal file
@ -0,0 +1,94 @@
|
||||
#
|
||||
# Copyright (C) 2026 GNS3 Technologies Inc.
|
||||
# Author: Yue Guobin
|
||||
#
|
||||
# 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/>.
|
||||
|
||||
"""
|
||||
MCP tool handlers for GNS3 symbol management.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ── Helper ─────────────────────────────────────────────────────────────────
|
||||
|
||||
def _get_connector(gns3_ctx: dict[str, Any]):
|
||||
from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector
|
||||
return Gns3Connector(
|
||||
url=gns3_ctx["server_url"],
|
||||
jwt_token=gns3_ctx["jwt_token"],
|
||||
api_version=3,
|
||||
verify=False,
|
||||
)
|
||||
|
||||
|
||||
# ── Tool handlers ──────────────────────────────────────────────────────────
|
||||
|
||||
def get_symbols_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
symbols = conn.http_call("get", f"{conn.base_url}/symbols").json()
|
||||
return {"symbols": symbols, "count": len(symbols)}
|
||||
|
||||
|
||||
def get_symbol_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
symbol_id = params.get("symbol_id")
|
||||
if not symbol_id:
|
||||
return {"error": "symbol_id is required"}
|
||||
download_url = f"{gns3_ctx['server_url']}/v3/symbols/{symbol_id}/raw"
|
||||
auth_token = gns3_ctx['jwt_token']
|
||||
return {
|
||||
"symbol_id": symbol_id,
|
||||
"download_url": download_url,
|
||||
"curl_command": f"curl -L -o '{symbol_id.replace(':', '').replace('/', '_')}.svg' -H 'Authorization: Bearer {auth_token}' '{download_url}'",
|
||||
"note": "Symbol files are SVG images. Use curl to download.",
|
||||
}
|
||||
|
||||
|
||||
def get_symbol_dimensions_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
symbol_id = params.get("symbol_id")
|
||||
if not symbol_id:
|
||||
return {"error": "symbol_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return conn.http_call("get", f"{conn.base_url}/symbols/{symbol_id}/dimensions").json()
|
||||
|
||||
|
||||
def get_default_symbols_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
symbols = conn.http_call("get", f"{conn.base_url}/symbols/default_symbols").json()
|
||||
return {"default_symbols": symbols}
|
||||
|
||||
|
||||
def upload_symbol_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
symbol_id = params.get("symbol_id")
|
||||
content = params.get("content")
|
||||
if not symbol_id or content is None:
|
||||
return {"error": "symbol_id and content (SVG data) are required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
url = f"{conn.base_url}/symbols/{symbol_id}/raw"
|
||||
conn.http_call("post", url, data=content, headers={"Content-Type": "image/svg+xml"})
|
||||
return {"message": f"Symbol {symbol_id} uploaded", "symbol_id": symbol_id}
|
||||
|
||||
|
||||
def delete_symbol_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
symbol_id = params.get("symbol_id")
|
||||
if not symbol_id:
|
||||
return {"error": "symbol_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("delete", f"{conn.base_url}/symbols/{symbol_id}")
|
||||
return {"message": f"Symbol {symbol_id} deleted", "symbol_id": symbol_id}
|
||||
@ -45,7 +45,7 @@ def _get_connector(gns3_ctx: dict[str, Any]):
|
||||
|
||||
def list_templates_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
conn = _get_connector(gns3_ctx)
|
||||
templates = conn.get_templates()
|
||||
templates = conn.http_call("get", f"{conn.base_url}/templates").json()
|
||||
return {"templates": templates, "count": len(templates)}
|
||||
|
||||
|
||||
@ -57,10 +57,16 @@ def get_template_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> di
|
||||
return {"error": "template_id or name is required"}
|
||||
|
||||
conn = _get_connector(gns3_ctx)
|
||||
template = conn.get_template(name=name, template_id=template_id)
|
||||
|
||||
if template is None:
|
||||
return {"error": "Template not found"}
|
||||
if template_id:
|
||||
template = conn.http_call("get", f"{conn.base_url}/templates/{template_id}").json()
|
||||
else:
|
||||
# Find template by name
|
||||
all_templates = conn.http_call("get", f"{conn.base_url}/templates").json()
|
||||
matches = [t for t in all_templates if t.get("name") == name]
|
||||
if not matches:
|
||||
return {"error": f"Template '{name}' not found"}
|
||||
template = matches[0]
|
||||
|
||||
return template
|
||||
|
||||
@ -72,14 +78,16 @@ def create_template_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) ->
|
||||
return {"error": "name and template_type are required"}
|
||||
|
||||
conn = _get_connector(gns3_ctx)
|
||||
|
||||
# Handle nested kwargs structure from MCP clients
|
||||
if "kwargs" in params and isinstance(params["kwargs"], dict):
|
||||
create_params = params["kwargs"]
|
||||
else:
|
||||
create_params = params
|
||||
|
||||
return conn.create_template(**create_params)
|
||||
data = {
|
||||
"name": name,
|
||||
"template_type": template_type,
|
||||
"compute_id": params.get("compute_id", "local"),
|
||||
}
|
||||
# Pass through optional template-type-specific fields (image, qemu_path, etc.)
|
||||
for key in ("image",):
|
||||
if key in params:
|
||||
data[key] = params[key]
|
||||
return conn.http_call("post", f"{conn.base_url}/templates", json_data=data).json()
|
||||
|
||||
|
||||
def update_template_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
@ -91,13 +99,20 @@ def update_template_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) ->
|
||||
|
||||
conn = _get_connector(gns3_ctx)
|
||||
|
||||
# Extract update parameters - handle nested kwargs structure from MCP clients
|
||||
if "kwargs" in params and isinstance(params["kwargs"], dict):
|
||||
update_params = params["kwargs"]
|
||||
else:
|
||||
update_params = {k: v for k, v in params.items() if k not in ("template_id", "name", "kwargs")}
|
||||
# Resolve name to ID if needed
|
||||
if not template_id and name:
|
||||
all_templates = conn.http_call("get", f"{conn.base_url}/templates").json()
|
||||
matches = [t for t in all_templates if t.get("name") == name]
|
||||
if not matches:
|
||||
return {"error": f"Template '{name}' not found"}
|
||||
template_id = matches[0]["template_id"]
|
||||
|
||||
return conn.update_template(name=name, template_id=template_id, **update_params)
|
||||
update_data = {k: v for k, v in params.items() if k not in ("template_id", "name", "kwargs")}
|
||||
# Support nested kwargs from MCP clients
|
||||
if "kwargs" in params and isinstance(params["kwargs"], dict):
|
||||
update_data = params["kwargs"]
|
||||
|
||||
return conn.http_call("put", f"{conn.base_url}/templates/{template_id}", json_data=update_data).json()
|
||||
|
||||
|
||||
def delete_template_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
@ -108,7 +123,15 @@ def delete_template_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) ->
|
||||
return {"error": "template_id or name is required"}
|
||||
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.delete_template(name=name, template_id=template_id)
|
||||
|
||||
if not template_id and name:
|
||||
all_templates = conn.http_call("get", f"{conn.base_url}/templates").json()
|
||||
matches = [t for t in all_templates if t.get("name") == name]
|
||||
if not matches:
|
||||
return {"error": f"Template '{name}' not found"}
|
||||
template_id = matches[0]["template_id"]
|
||||
|
||||
conn.http_call("delete", f"{conn.base_url}/templates/{template_id}")
|
||||
return {"message": f"Template deleted"}
|
||||
|
||||
|
||||
|
||||
@ -1532,21 +1532,23 @@ class Project:
|
||||
@open_required
|
||||
async def start_all(self):
|
||||
"""
|
||||
Start all nodes
|
||||
Start all nodes (except always-running types like Ethernet switch, Cloud, NAT, etc.)
|
||||
"""
|
||||
pool = Pool(concurrency=3)
|
||||
for node in self.nodes.values():
|
||||
pool.append(node.start)
|
||||
if not node.is_always_running():
|
||||
pool.append(node.start)
|
||||
await pool.join()
|
||||
|
||||
@open_required
|
||||
async def stop_all(self):
|
||||
"""
|
||||
Stop all nodes
|
||||
Stop all nodes (except always-running types like Ethernet switch, Cloud, NAT, etc.)
|
||||
"""
|
||||
pool = Pool(concurrency=3)
|
||||
for node in self.nodes.values():
|
||||
pool.append(node.stop)
|
||||
if not node.is_always_running():
|
||||
pool.append(node.stop)
|
||||
await pool.join()
|
||||
|
||||
@open_required
|
||||
|
||||
@ -24,6 +24,7 @@ from .computes import Compute
|
||||
from .images import Image
|
||||
from .pools import Resource, ResourcePool
|
||||
from .llm_model_configs import LLMModelConfig
|
||||
from .api_keys import ApiKey
|
||||
from .templates import (
|
||||
Template,
|
||||
CloudTemplate,
|
||||
|
||||
33
gns3server/db/models/api_keys.py
Normal file
33
gns3server/db/models/api_keys.py
Normal file
@ -0,0 +1,33 @@
|
||||
#
|
||||
# Copyright (C) 2026 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/>.
|
||||
|
||||
from sqlalchemy import Column, String, Boolean, DateTime, ForeignKey, func
|
||||
|
||||
from .base import BaseTable, GUID
|
||||
|
||||
|
||||
class ApiKey(BaseTable):
|
||||
|
||||
__tablename__ = "api_keys"
|
||||
|
||||
api_key_id = Column(GUID, primary_key=True)
|
||||
user_id = Column(GUID, ForeignKey("users.user_id", ondelete="CASCADE"), nullable=False, index=True)
|
||||
name = Column(String(128), nullable=False)
|
||||
key_hash = Column(String(128), nullable=False)
|
||||
key_prefix = Column(String(8), nullable=False)
|
||||
last_used_at = Column(DateTime, nullable=True)
|
||||
created_at = Column(DateTime, server_default=func.current_timestamp(), nullable=False)
|
||||
revoked = Column(Boolean, default=False, nullable=False)
|
||||
104
gns3server/db/repositories/api_keys.py
Normal file
104
gns3server/db/repositories/api_keys.py
Normal file
@ -0,0 +1,104 @@
|
||||
#
|
||||
# Copyright (C) 2026 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/>.
|
||||
|
||||
from uuid import UUID
|
||||
from typing import Optional, List
|
||||
from datetime import datetime, timezone
|
||||
from sqlalchemy import select, update, delete, func
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from .base import BaseRepository
|
||||
import gns3server.db.models as models
|
||||
|
||||
import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ApiKeysRepository(BaseRepository):
|
||||
|
||||
def __init__(self, db_session: AsyncSession) -> None:
|
||||
super().__init__(db_session)
|
||||
|
||||
async def create_api_key(
|
||||
self, api_key_id: UUID, user_id: UUID, name: str, key_hash: str, key_prefix: str
|
||||
) -> models.ApiKey:
|
||||
db_api_key = models.ApiKey(
|
||||
api_key_id=api_key_id,
|
||||
user_id=user_id,
|
||||
name=name,
|
||||
key_hash=key_hash,
|
||||
key_prefix=key_prefix,
|
||||
)
|
||||
self._db_session.add(db_api_key)
|
||||
await self._db_session.commit()
|
||||
await self._db_session.refresh(db_api_key)
|
||||
return db_api_key
|
||||
|
||||
async def get_api_key(self, api_key_id: UUID) -> Optional[models.ApiKey]:
|
||||
query = select(models.ApiKey).where(models.ApiKey.api_key_id == api_key_id)
|
||||
result = await self._db_session.execute(query)
|
||||
return result.scalars().first()
|
||||
|
||||
async def get_api_keys_by_user(self, user_id: UUID) -> List[models.ApiKey]:
|
||||
query = (
|
||||
select(models.ApiKey)
|
||||
.where(models.ApiKey.user_id == user_id)
|
||||
.order_by(models.ApiKey.created_at.desc())
|
||||
)
|
||||
result = await self._db_session.execute(query)
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def get_api_key_by_hash(self, key_hash: str) -> Optional[models.ApiKey]:
|
||||
query = select(models.ApiKey).where(models.ApiKey.key_hash == key_hash)
|
||||
result = await self._db_session.execute(query)
|
||||
return result.scalars().first()
|
||||
|
||||
async def revoke_api_key(self, api_key_id: UUID) -> bool:
|
||||
query = (
|
||||
update(models.ApiKey)
|
||||
.where(models.ApiKey.api_key_id == api_key_id)
|
||||
.values(revoked=True)
|
||||
)
|
||||
result = await self._db_session.execute(query)
|
||||
await self._db_session.commit()
|
||||
return result.rowcount > 0
|
||||
|
||||
async def restore_api_key(self, api_key_id: UUID) -> bool:
|
||||
query = (
|
||||
update(models.ApiKey)
|
||||
.where(models.ApiKey.api_key_id == api_key_id)
|
||||
.values(revoked=False)
|
||||
)
|
||||
result = await self._db_session.execute(query)
|
||||
await self._db_session.commit()
|
||||
return result.rowcount > 0
|
||||
|
||||
async def update_last_used(self, api_key_id: UUID) -> None:
|
||||
query = (
|
||||
update(models.ApiKey)
|
||||
.where(models.ApiKey.api_key_id == api_key_id)
|
||||
.values(last_used_at=func.now())
|
||||
)
|
||||
await self._db_session.execute(query)
|
||||
await self._db_session.commit()
|
||||
|
||||
async def delete_api_key(self, api_key_id: UUID) -> bool:
|
||||
query = delete(models.ApiKey).where(models.ApiKey.api_key_id == api_key_id)
|
||||
result = await self._db_session.execute(query)
|
||||
await self._db_session.commit()
|
||||
return result.rowcount > 0
|
||||
@ -0,0 +1,43 @@
|
||||
"""add api_keys table
|
||||
|
||||
Revision ID: f0b0de2a9
|
||||
Revises: a8829e6c069b
|
||||
Create Date: 2026-06-11 10:00:00.000000
|
||||
|
||||
"""
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
import gns3server.db.models.base as models
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision = 'f0b0de2a9'
|
||||
down_revision = 'a8829e6c069b'
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
|
||||
op.create_table(
|
||||
'api_keys',
|
||||
sa.Column('api_key_id', models.GUID(), nullable=False),
|
||||
sa.Column('user_id', models.GUID(), nullable=False),
|
||||
sa.Column('name', sa.String(128), nullable=False),
|
||||
sa.Column('key_hash', sa.String(128), nullable=False),
|
||||
sa.Column('key_prefix', sa.String(8), nullable=False),
|
||||
sa.Column('last_used_at', sa.DateTime(), nullable=True),
|
||||
sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
|
||||
sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
|
||||
sa.Column('revoked', sa.Boolean(), default=False, nullable=False),
|
||||
sa.ForeignKeyConstraint(['user_id'], ['users.user_id'], ondelete='CASCADE'),
|
||||
sa.PrimaryKeyConstraint('api_key_id'),
|
||||
)
|
||||
op.create_index('ix_api_keys_user_id', 'api_keys', ['user_id'])
|
||||
op.create_index('ix_api_keys_key_hash', 'api_keys', ['key_hash'])
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
|
||||
op.drop_index('ix_api_keys_key_hash', table_name='api_keys')
|
||||
op.drop_index('ix_api_keys_user_id', table_name='api_keys')
|
||||
op.drop_table('api_keys')
|
||||
@ -57,7 +57,7 @@ except ImportError:
|
||||
|
||||
from .controller.rbac import RoleCreate, RoleUpdate, Role, Privilege, ACECreate, ACEUpdate, ACE
|
||||
from .controller.pools import Resource, ResourceCreate, ResourcePoolCreate, ResourcePoolUpdate, ResourcePool
|
||||
from .controller.tokens import Token
|
||||
from .controller.tokens import Token, ApiKeyCreate
|
||||
from .controller.snapshots import SnapshotCreate, Snapshot
|
||||
from .controller.iou_license import IOULicense
|
||||
from .controller.capabilities import Capabilities
|
||||
|
||||
@ -28,3 +28,9 @@ class TokenData(BaseModel):
|
||||
|
||||
username: Optional[str] = None
|
||||
token_version: int = 0
|
||||
|
||||
|
||||
class ApiKeyCreate(BaseModel):
|
||||
"""Schema for creating a new API key."""
|
||||
|
||||
name: str
|
||||
|
||||
@ -17,6 +17,7 @@
|
||||
from joserfc import jwt
|
||||
from joserfc.jwk import OctKey
|
||||
from joserfc.errors import JoseError
|
||||
import time
|
||||
from datetime import datetime, timedelta, timezone
|
||||
import bcrypt
|
||||
|
||||
@ -80,6 +81,10 @@ class AuthService:
|
||||
username: str = payload.claims.get("sub")
|
||||
if username is None:
|
||||
raise credentials_exception
|
||||
# Validate the exp claim — joserfc does not validate time-based claims by default
|
||||
token_exp: int = payload.claims.get("exp", 0)
|
||||
if token_exp and time.time() > token_exp:
|
||||
raise credentials_exception
|
||||
token_version: int = payload.claims.get("ver", 0)
|
||||
token_data = TokenData(username=username, token_version=token_version)
|
||||
except (JoseError, ValidationError, ValueError):
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user