mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-08-27 12:30:13 +03:00
Merge pull request #2785 from yueguobin/mcp/batch-operations
MCP batch operations, field filtering, Jinja2 templates, download security, and WebSocket fixes
This commit is contained in:
Jeremy Grossmann
GitHub
commit
b605d71a8a
@ -95,15 +95,15 @@ Both JWT tokens and API keys work for MCP and REST API endpoints interchangeably
|
||||
|
||||
| Tool | Description |
|
||||
|------|-------------|
|
||||
| `node_list` | List all nodes in a project |
|
||||
| `node_get` | Get node details |
|
||||
| `node_create` | Create a node from template |
|
||||
| `node_list` | List all nodes (`fields` to filter columns, e.g. `["name","status"]`) |
|
||||
| `node_get` | Get node details (`fields` to filter columns) |
|
||||
| `node_create` | Create node(s) — single via `template_id` or batch via `nodes` array |
|
||||
| `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_start` | Start node(s) — `node_id` or `node_ids` array |
|
||||
| `node_stop` | Stop node(s) — `node_id` or `node_ids` array |
|
||||
| `node_reload` | Reload node(s) — `node_id` or `node_ids` array |
|
||||
| `node_suspend` | Suspend node(s) — `node_id` or `node_ids` array |
|
||||
| `node_console` | Get WebSocket console URL |
|
||||
| `node_file_list` | List files in node directory |
|
||||
| `node_file_get` | Read a file (with offset/limit) |
|
||||
@ -122,15 +122,15 @@ Both JWT tokens and API keys work for MCP and REST API endpoints interchangeably
|
||||
|
||||
| Tool | Description |
|
||||
|------|-------------|
|
||||
| `link_list` | List all links in a project |
|
||||
| `link_list` | List all links (`fields` to filter columns) |
|
||||
| `link_get` | Get link details |
|
||||
| `link_create` | Create a link between nodes |
|
||||
| `link_delete` | Delete a link |
|
||||
| `link_create` | Create link(s) — single via `nodes` or batch via `links` array |
|
||||
| `link_delete` | Delete link(s) — `link_id` or `link_ids` array |
|
||||
| `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 |
|
||||
| `link_reset` | Reset link(s) — `link_id` or `link_ids` array |
|
||||
| `link_capture_start` | Start capture(s) — `link_id` or `link_ids` array |
|
||||
| `link_capture_stop` | Stop capture(s) — `link_id` or `link_ids` array |
|
||||
| `link_capture_download` | Get PCAP download URL(s) — `link_id` or `link_ids` array |
|
||||
|
||||
### Template (5)
|
||||
|
||||
@ -184,7 +184,7 @@ Both JWT tokens and API keys work for MCP and REST API endpoints interchangeably
|
||||
|
||||
| Tool | Description |
|
||||
|------|-------------|
|
||||
| `appliance_list` | List appliances from template library |
|
||||
| `appliance_list` | List appliances (`fields` to filter, e.g. `["name","category"]`) |
|
||||
| `appliance_get` | Get appliance details |
|
||||
| `appliance_install` | Create template from appliance (images must exist locally) |
|
||||
|
||||
@ -209,11 +209,101 @@ Both JWT tokens and API keys work for MCP and REST API endpoints interchangeably
|
||||
|
||||
| Tool | Description |
|
||||
|------|-------------|
|
||||
| `device_config_send` | Push config commands to devices via console (Nornir + Netmiko) |
|
||||
| `device_command_run` | Run read-only show commands on devices |
|
||||
| `device_config_send` | Push config commands to devices via console (Nornir + Netmiko). Supports Jinja2 `template` + `vars` |
|
||||
| `device_command_run` | Run read-only show commands on devices. Supports Jinja2 `template` + `vars` |
|
||||
| `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.
|
||||
The tool connects to each device's console via telnet/SSH. Nodes must be in the `started` state (use `node_start` or `node_start_all`). Device type is auto-detected from the node's `device_type:<type>` tag in GNS3.
|
||||
|
||||
#### Jinja2 Template Mode
|
||||
|
||||
Both `device_config_send` and `device_command_run` support an optional `template` parameter. When provided, each device's `vars` dict is rendered against the template to produce commands. Entries with the same `device_name` are merged into a single device session.
|
||||
|
||||
```python
|
||||
# Direct commands (single/batch)
|
||||
device_config_send(project_id, device_configs=[
|
||||
{"device_name": "R1", "config_commands": ["int lo0", "ip add 1.1.1.1 255.255.255.255"]},
|
||||
])
|
||||
|
||||
# Jinja2 template (reduces token usage for batch)
|
||||
device_config_send(project_id,
|
||||
template="interface lo{{ n }}\nip address {{ ip }} 255.255.255.255",
|
||||
device_configs=[
|
||||
{"device_name": "R1", "vars": {"n": 0, "ip": "1.1.1.1"}},
|
||||
{"device_name": "R2", "vars": {"n": 0, "ip": "2.2.2.2"}},
|
||||
])
|
||||
|
||||
# Show commands with template
|
||||
device_command_run(project_id,
|
||||
template="show ip route {{ protocol }}",
|
||||
device_configs=[
|
||||
{"device_name": "R1", "vars": {"protocol": "ospf"}},
|
||||
{"device_name": "R2", "vars": {"protocol": "bgp"}},
|
||||
])
|
||||
```
|
||||
|
||||
### Best Practices
|
||||
|
||||
**Prefer template over direct commands for batch.** When ≥2 nodes share the same config structure with different values, use `template`+`vars` instead of writing `config_commands` per node. This reduces token usage and transcription errors.
|
||||
|
||||
**Batch merging.** Multiple entries with the same `device_name` are merged into a single Nornir session. The output contains all commands' results in one block. Match results by `device_name`, not list index.
|
||||
|
||||
**Don't rely on `status: success` alone.** It only means commands entered config mode. IOS errors (`% Invalid input`, `% overlaps`, `% Incomplete command`) appear inside `output` text — always scan for `%` lines.
|
||||
|
||||
**Pilot before full rollout.** Test template + vars on 1–2 devices first to verify rendering and syntax, then expand to all nodes.
|
||||
|
||||
**Config backup via file operations.** IOU and Dynamips nodes save startup config as a plain text file (`startup-config.cfg`) in the node directory after `write memory`. These can be backed up and restored via `node_file_get`/`node_file_write`.
|
||||
|
||||
```python
|
||||
# Save config on device
|
||||
device_command_run(project_id, device_configs=[
|
||||
{"device_name": "R1", "commands": ["write memory"]},
|
||||
])
|
||||
# Backup
|
||||
config = node_file_get(project_id, node_id, "startup-config.cfg")
|
||||
# Restore if config breaks
|
||||
node_file_write(project_id, node_id, "startup-config.cfg", config)
|
||||
node_reload(project_id, node_id)
|
||||
```
|
||||
|
||||
### Device Config Workflow
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant AI as AI Agent
|
||||
participant MCP as MCP Handler
|
||||
participant TM as Template Renderer
|
||||
participant DP as Device Discovery
|
||||
participant NR as Nornir
|
||||
participant NM as Netmiko
|
||||
participant D as Device Console
|
||||
|
||||
Note over AI: Decide: template or direct commands?
|
||||
|
||||
alt Direct commands
|
||||
AI->>MCP: device_config_send(config_commands=[...])
|
||||
else Jinja2 template
|
||||
AI->>MCP: device_config_send(template + vars)
|
||||
MCP->>TM: Render template per device
|
||||
TM->>TM: Jinja2.render(**vars)
|
||||
TM-->>MCP: device_configs with rendered commands
|
||||
end
|
||||
|
||||
MCP->>DP: get_device_ports_from_topology()
|
||||
DP-->>MCP: hosts_data (console port, device_type)
|
||||
|
||||
Note over MCP: Prepare Nornir inventory
|
||||
|
||||
MCP->>NR: InitNornir(hosts, threaded runner)
|
||||
par Device 1 to N (parallel, max 10)
|
||||
NR->>NM: netmiko_send_config(commands)
|
||||
NM->>D: telnet/SSH console session
|
||||
D-->>NM: command output
|
||||
NM-->>NR: execution result
|
||||
end
|
||||
NR-->>MCP: aggregated results
|
||||
MCP-->>AI: per-device results with output
|
||||
```
|
||||
|
||||
## Configuration
|
||||
|
||||
@ -236,20 +326,6 @@ claude mcp add --transport sse My_GNS3_Server \
|
||||
-H "Authorization: Bearer $TOKEN"
|
||||
```
|
||||
|
||||
### Claude Desktop
|
||||
|
||||
Add to `claude_desktop_config.json`:
|
||||
|
||||
```json
|
||||
{
|
||||
"mcpServers": {
|
||||
"My_GNS3_Server": {
|
||||
"url": "http://localhost:3080/v3/mcp/transport/sse?token=your_jwt_or_api_key"
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Transport Security
|
||||
|
||||
MCP server uses FastMCP's DNS rebinding protection to prevent attackers from
|
||||
@ -310,12 +386,12 @@ For public-facing MCP servers, set `allowed_hosts` to your server's domain name.
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant Client as Claude Code / Claude Desktop
|
||||
participant Client as Claude Code
|
||||
participant MCP as MCP Service
|
||||
participant Auth as JWT Auth
|
||||
participant Auth as Auth
|
||||
participant GNS3 as GNS3 REST API
|
||||
|
||||
Note over Client: 1. Connect with JWT
|
||||
Note over Client: 1. Connect with credential (JWT or API Key)
|
||||
Client->>MCP: GET /sse (token in header or query)
|
||||
MCP->>Auth: Validate Token
|
||||
Auth-->>MCP: Token Valid
|
||||
@ -329,7 +405,7 @@ sequenceDiagram
|
||||
Client->>MCP: POST /messages/ (tools/list)
|
||||
MCP-->>Client: event: message (tools list)
|
||||
|
||||
Client->>MCP: POST /messages/ (tools/call list_projects)
|
||||
Client->>MCP: POST /messages/ (tools/call project_list)
|
||||
MCP->>GNS3: Gns3Connector HTTP request
|
||||
GNS3-->>MCP: Projects data
|
||||
MCP-->>Client: event: message (tool result)
|
||||
@ -345,7 +421,7 @@ sequenceDiagram
|
||||
|
||||
### Console WebSocket
|
||||
|
||||
The `get_node_console_info` tool returns a WebSocket URL for connecting to a node's console. This endpoint is protocol-agnostic — it works for **telnet**, **ssh**, and **vnc** console types alike. The WebSocket simply proxies raw byte streams between the client and the compute node; protocol negotiation (e.g. SSH key exchange) happens on the compute side.
|
||||
The `node_console` tool returns a WebSocket URL for connecting to a node's console. The URL includes a short-lived JWT (10 min) — reconnect if it expires. This endpoint is protocol-agnostic — it works for **telnet**, **ssh**, and **vnc** console types alike. The WebSocket simply proxies raw byte streams between the client and the compute node; protocol negotiation (e.g. SSH key exchange) happens on the compute side.
|
||||
|
||||
The WebSocket URL is constructed using the server's `_server_url()`, which resolves the host as follows:
|
||||
|
||||
|
||||
@ -53,6 +53,7 @@ import gns3server.db.models as models
|
||||
from gns3server.services import auth_service
|
||||
from gns3server.utils.request_utils import extract_client_info
|
||||
from gns3server.db.repositories.api_keys import ApiKeysRepository
|
||||
from gns3server.db.repositories.users import UsersRepository
|
||||
from .projects import (
|
||||
list_projects_handler, get_project_handler, create_project_handler,
|
||||
delete_project_handler, open_project_handler, close_project_handler,
|
||||
@ -118,6 +119,9 @@ from .drawings import (
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# Suppress noisy telnet connection logs from device config tools.
|
||||
logging.getLogger("telnetlib3").setLevel(logging.WARNING)
|
||||
|
||||
# FastAPI app reference — used to lazily access app.state._db_engine for API key validation.
|
||||
# The db engine is initialized during the lifespan startup, which runs AFTER
|
||||
# register_starlette_routes() is called, so we cannot capture it at registration time.
|
||||
@ -184,6 +188,11 @@ async def wait_for_mcp_ready() -> bool:
|
||||
_jwt_token_var: contextvars.ContextVar[str | None] = contextvars.ContextVar(
|
||||
"mcp_jwt_token", default=None
|
||||
)
|
||||
# Username extracted during token validation — used by handlers to generate
|
||||
# short-lived JWTs for download/console URLs without exposing the raw key.
|
||||
_jwt_username_var: contextvars.ContextVar[str | None] = contextvars.ContextVar(
|
||||
"mcp_jwt_username", default=None
|
||||
)
|
||||
|
||||
|
||||
# ── Token validation ──────────────────────────────────────────────────
|
||||
@ -198,12 +207,13 @@ async def _resolve_token(token: str) -> str | None:
|
||||
"""
|
||||
# Try JWT first
|
||||
try:
|
||||
auth_service.get_username_from_token(token)
|
||||
username = auth_service.get_username_from_token(token)
|
||||
_jwt_username_var.set(username)
|
||||
return token
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Try API key — pass through directly; REST API auth already supports gns3_ keys
|
||||
# Try API key
|
||||
if token.startswith("gns3_") and _app is not None:
|
||||
db_engine = getattr(_app.state, "_db_engine", None)
|
||||
if db_engine is not None:
|
||||
@ -215,8 +225,10 @@ async def _resolve_token(token: str) -> str | None:
|
||||
for db_key in result.scalars().all():
|
||||
if bcrypt.checkpw(token.encode(), db_key.key_hash.encode()):
|
||||
await repo.update_last_used(db_key.api_key_id)
|
||||
# Return the raw API key — it will be passed as Bearer token
|
||||
# and validated by the REST API auth layer
|
||||
user_repo = UsersRepository(db_session)
|
||||
user = await user_repo.get_user(db_key.user_id)
|
||||
if user:
|
||||
_jwt_username_var.set(user.username)
|
||||
return token
|
||||
except Exception:
|
||||
pass
|
||||
@ -274,6 +286,7 @@ def _run_handler_sync(handler, params: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
ctx = {
|
||||
"server_url": _server_url(),
|
||||
"jwt_token": _jwt_token_var.get(),
|
||||
"jwt_username": _jwt_username_var.get(),
|
||||
}
|
||||
result = handler(params, ctx)
|
||||
return [{"type": "text", "text": json.dumps(result, ensure_ascii=False, default=str)}]
|
||||
@ -297,8 +310,8 @@ async def project_get(
|
||||
async def project_create(
|
||||
name: Annotated[str, Field(description="Project name")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Create a new GNS3 project."""
|
||||
params = {"name": name}
|
||||
"""Create a new GNS3 project. auto_close is set to False so the project stays open when clients disconnect."""
|
||||
params = {"name": name, "auto_close": False}
|
||||
return await asyncio.to_thread(_run_handler_sync, create_project_handler, params)
|
||||
|
||||
|
||||
@ -396,61 +409,100 @@ async def project_readme_update(
|
||||
# ── Node tools ────────────────────────────────────────────────────────
|
||||
|
||||
@mcp.tool()
|
||||
async def node_list(project_id: str) -> list[dict[str, Any]]:
|
||||
"""List all nodes in a project."""
|
||||
return await asyncio.to_thread(_run_handler_sync, get_nodes_handler, {"project_id": project_id})
|
||||
async def node_list(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
fields: Annotated[list[str] | None, Field(description="Optional: return only these fields per node. e.g. [\"name\",\"status\"]. Available: name, status, node_type, console, console_type, console_host, node_id, project_id, compute_id, symbol, x, y, z, locked, ports, properties, command_line, node_directory, label, tags, template_id, width, height, aux, aux_type")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""List all nodes in a project. Use fields=[] to return only what you need."""
|
||||
return await asyncio.to_thread(_run_handler_sync, get_nodes_handler, {"project_id": project_id, "fields": fields})
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def node_get(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str, Field(description="UUID of the node")],
|
||||
fields: Annotated[list[str] | None, Field(description="Optional: return only these fields. e.g. [\"name\",\"status\"]. Available: name, status, node_type, console, console_type, console_host, node_id, project_id, compute_id, symbol, x, y, z, locked, ports, properties, command_line, node_directory, label, tags, template_id, width, height, aux, aux_type")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Get detailed information about a specific node."""
|
||||
return await asyncio.to_thread(_run_handler_sync, get_node_handler, {"project_id": project_id, "node_id": node_id})
|
||||
"""Get detailed information about a specific node. Use fields=[] to return only what you need."""
|
||||
return await asyncio.to_thread(_run_handler_sync, get_node_handler, {
|
||||
"project_id": project_id, "node_id": node_id, "fields": fields,
|
||||
})
|
||||
|
||||
@mcp.tool()
|
||||
async def node_start(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str, Field(description="UUID of the node to start")],
|
||||
node_id: Annotated[str | None, Field(description="Node UUID (single mode)")] = None,
|
||||
node_ids: Annotated[list[str] | None, Field(description="Batch mode: [\"uuid1\",\"uuid2\"] — start multiple nodes in parallel")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Start a node in a project."""
|
||||
return await asyncio.to_thread(_run_handler_sync, start_node_handler, {"project_id": project_id, "node_id": node_id})
|
||||
"""Start one or more nodes. Provide node_id for single, or node_ids for batch."""
|
||||
params = {"project_id": project_id}
|
||||
if node_ids:
|
||||
params["node_ids"] = node_ids
|
||||
else:
|
||||
params["node_id"] = node_id
|
||||
return await asyncio.to_thread(_run_handler_sync, start_node_handler, params)
|
||||
|
||||
@mcp.tool()
|
||||
async def node_stop(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str, Field(description="UUID of the node to stop")],
|
||||
node_id: Annotated[str | None, Field(description="Node UUID (single mode)")] = None,
|
||||
node_ids: Annotated[list[str] | None, Field(description="Batch mode: [\"uuid1\",\"uuid2\"] — stop multiple nodes in parallel")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Stop a node in a project."""
|
||||
return await asyncio.to_thread(_run_handler_sync, stop_node_handler, {"project_id": project_id, "node_id": node_id})
|
||||
"""Stop one or more nodes. Provide node_id for single, or node_ids for batch."""
|
||||
params = {"project_id": project_id}
|
||||
if node_ids:
|
||||
params["node_ids"] = node_ids
|
||||
else:
|
||||
params["node_id"] = node_id
|
||||
return await asyncio.to_thread(_run_handler_sync, stop_node_handler, params)
|
||||
|
||||
@mcp.tool()
|
||||
async def node_reload(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str, Field(description="UUID of the node to reload")],
|
||||
node_id: Annotated[str | None, Field(description="Node UUID (single mode)")] = None,
|
||||
node_ids: Annotated[list[str] | None, Field(description="Batch mode: [\"uuid1\",\"uuid2\"] — reload multiple nodes in parallel")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Reload (restart) a node in a project."""
|
||||
return await asyncio.to_thread(_run_handler_sync, reload_node_handler, {"project_id": project_id, "node_id": node_id})
|
||||
"""Reload (restart) one or more nodes. Provide node_id for single, or node_ids for batch."""
|
||||
params = {"project_id": project_id}
|
||||
if node_ids:
|
||||
params["node_ids"] = node_ids
|
||||
else:
|
||||
params["node_id"] = node_id
|
||||
return await asyncio.to_thread(_run_handler_sync, reload_node_handler, params)
|
||||
|
||||
@mcp.tool()
|
||||
async def node_suspend(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str, Field(description="UUID of the node to suspend")],
|
||||
node_id: Annotated[str | None, Field(description="Node UUID (single mode)")] = None,
|
||||
node_ids: Annotated[list[str] | None, Field(description="Batch mode: [\"uuid1\",\"uuid2\"] — suspend multiple nodes in parallel")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Suspend a node in a project."""
|
||||
return await asyncio.to_thread(_run_handler_sync, suspend_node_handler, {"project_id": project_id, "node_id": node_id})
|
||||
"""Suspend one or more nodes. Provide node_id for single, or node_ids for batch."""
|
||||
params = {"project_id": project_id}
|
||||
if node_ids:
|
||||
params["node_ids"] = node_ids
|
||||
else:
|
||||
params["node_id"] = node_id
|
||||
return await asyncio.to_thread(_run_handler_sync, suspend_node_handler, params)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def node_create(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
template_id: Annotated[str, Field(description="UUID of the template to create the node from")],
|
||||
x: Annotated[int, Field(description="X coordinate on the project canvas")] = 0,
|
||||
y: Annotated[int, Field(description="Y coordinate on the project canvas")] = 0,
|
||||
template_id: Annotated[str | None, Field(description="Template UUID (required for single mode)")] = None,
|
||||
x: Annotated[int, Field(description="X coordinate")] = 0,
|
||||
y: Annotated[int, Field(description="Y coordinate")] = 0,
|
||||
compute_id: Annotated[str, Field(description="Compute ID (default: local)")] = "local",
|
||||
nodes: Annotated[list | None, Field(description="Batch mode: [{template_id, x?, y?, name?, compute_id?}] — creates multiple nodes in parallel")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Create a new node from a template in a project."""
|
||||
"""Create one or more nodes from templates.
|
||||
|
||||
Single mode: provide template_id, x, y (optional compute_id)
|
||||
Batch mode: provide nodes=[{template_id, x, y, name?, compute_id?}] — creates up to 10 in parallel
|
||||
"""
|
||||
if nodes is not None:
|
||||
return await asyncio.to_thread(_run_handler_sync, create_node_handler, {
|
||||
"project_id": project_id, "nodes": nodes,
|
||||
})
|
||||
return await asyncio.to_thread(_run_handler_sync, create_node_handler, {
|
||||
"project_id": project_id, "template_id": template_id,
|
||||
"x": x, "y": y, "compute_id": compute_id,
|
||||
@ -460,10 +512,16 @@ async def node_create(
|
||||
@mcp.tool()
|
||||
async def node_delete(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
node_id: Annotated[str, Field(description="UUID of the node to delete")],
|
||||
node_id: Annotated[str | None, Field(description="Node UUID (single mode)")] = None,
|
||||
node_ids: Annotated[list[str] | None, Field(description="Batch mode: [\"uuid1\",\"uuid2\"] — delete multiple nodes in parallel")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Delete a node from a project."""
|
||||
return await asyncio.to_thread(_run_handler_sync, delete_node_handler, {"project_id": project_id, "node_id": node_id})
|
||||
"""Delete one or more nodes from a project. Provide node_id for single, or node_ids for batch."""
|
||||
params = {"project_id": project_id}
|
||||
if node_ids:
|
||||
params["node_ids"] = node_ids
|
||||
else:
|
||||
params["node_id"] = node_id
|
||||
return await asyncio.to_thread(_run_handler_sync, delete_node_handler, params)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
@ -486,6 +544,7 @@ async def node_console(
|
||||
|
||||
Returns the WebSocket URL, console type (telnet/ssh/vnc), and other
|
||||
connection details needed to interact with a node's console via WebSocket.
|
||||
The URL includes a short-lived JWT (10 min) — reconnect if it expires.
|
||||
|
||||
Complete workflow:
|
||||
1. Call this tool with project_id and node_id to get the WebSocket URL
|
||||
@ -510,9 +569,12 @@ async def node_console(
|
||||
# ── Link tools ────────────────────────────────────────────────────────
|
||||
|
||||
@mcp.tool()
|
||||
async def link_list(project_id: str) -> list[dict[str, Any]]:
|
||||
"""List all links in a project."""
|
||||
return await asyncio.to_thread(_run_handler_sync, get_links_handler, {"project_id": project_id})
|
||||
async def link_list(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
fields: Annotated[list[str] | None, Field(description="Optional: return only these fields. e.g. [\"link_id\",\"nodes\"]. Available: link_id, project_id, link_type, nodes, suspend, filters, capturing, capture_file_name, link_style")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""List all links in a project. Use fields=[] to return only what you need."""
|
||||
return await asyncio.to_thread(_run_handler_sync, get_links_handler, {"project_id": project_id, "fields": fields})
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
@ -527,21 +589,20 @@ async def link_get(
|
||||
@mcp.tool()
|
||||
async def link_create(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
nodes: Annotated[list, Field(description="List of node connections, e.g. [{\"node_id\": \"...\", \"adapter_number\": 0, \"port_number\": 0}]")],
|
||||
nodes: Annotated[list | None, Field(description="Single mode: [{node_id, adapter_number, port_number}]")] = None,
|
||||
link_type: Annotated[str, Field(description="Link type - ethernet or serial")] = "ethernet",
|
||||
filters: Annotated[dict, Field(description="Optional packet filters (must use array format): frequency_drop: [N], packet_loss: [rate], delay: [ms, jitter], corrupt: [rate], bpf: [expression]")] = None,
|
||||
filters: Annotated[dict | None, Field(description="Optional packet filters")] = None,
|
||||
links: Annotated[list | None, Field(description="Batch mode: [{nodes, link_type?, filters?}] — creates multiple links in parallel")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Create a link between two nodes in a project.
|
||||
"""Create one or more links between nodes.
|
||||
|
||||
Filters must use array format:
|
||||
- frequency_drop: [N] - Drop every Nth packet (N: -1 to 32767)
|
||||
- packet_loss: [rate] - Packet loss percentage (rate: 0 to 100)
|
||||
- delay: [ms, jitter] - Latency and jitter in milliseconds
|
||||
- corrupt: [rate] - Packet corruption percentage (rate: 0 to 100)
|
||||
- bpf: [expression] - Berkeley Packet Filter expression
|
||||
|
||||
Example: {"filters": {"delay": [100, 10], "packet_loss": [5]}}
|
||||
Single mode: provide nodes, link_type (optional filters)
|
||||
Batch mode: provide links=[{nodes, link_type?, filters?}] — up to 10 in parallel
|
||||
"""
|
||||
if links:
|
||||
return await asyncio.to_thread(_run_handler_sync, create_link_handler, {
|
||||
"project_id": project_id, "links": links,
|
||||
})
|
||||
params = {"project_id": project_id, "nodes": nodes, "link_type": link_type}
|
||||
if filters:
|
||||
params["filters"] = filters
|
||||
@ -551,10 +612,16 @@ async def link_create(
|
||||
@mcp.tool()
|
||||
async def link_delete(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
link_id: Annotated[str, Field(description="UUID of the link to delete")],
|
||||
link_id: Annotated[str | None, Field(description="Link UUID (single mode)")] = None,
|
||||
link_ids: Annotated[list[str] | None, Field(description="Batch mode: [\"uuid1\",\"uuid2\"] — delete multiple links in parallel")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Delete a link from a project."""
|
||||
return await asyncio.to_thread(_run_handler_sync, delete_link_handler, {"project_id": project_id, "link_id": link_id})
|
||||
"""Delete one or more links from a project."""
|
||||
params = {"project_id": project_id}
|
||||
if link_ids:
|
||||
params["link_ids"] = link_ids
|
||||
else:
|
||||
params["link_id"] = link_id
|
||||
return await asyncio.to_thread(_run_handler_sync, delete_link_handler, params)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
@ -848,9 +915,10 @@ async def node_links(
|
||||
@mcp.tool()
|
||||
async def link_reset(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
link_id: Annotated[str, Field(description="UUID of the link")],
|
||||
link_id: Annotated[str | None, Field(description="Link UUID (single mode)")] = None,
|
||||
link_ids: Annotated[list[str] | None, Field(description="Batch mode: [\"uuid1\",\"uuid2\"] — reset multiple links in parallel")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Reset a link by tearing down the underlying UDP connection and recreating it.
|
||||
"""Reset one or more links by tearing down and recreating the UDP connection.
|
||||
|
||||
Use cases:
|
||||
- Clear accumulated packet errors/drops from the link's UDP connection
|
||||
@ -859,47 +927,60 @@ async def link_reset(
|
||||
|
||||
Filters are preserved but their internal application state resets.
|
||||
"""
|
||||
return await asyncio.to_thread(_run_handler_sync, reset_link_handler, {
|
||||
"project_id": project_id, "link_id": link_id,
|
||||
})
|
||||
params = {"project_id": project_id}
|
||||
if link_ids:
|
||||
params["link_ids"] = link_ids
|
||||
else:
|
||||
params["link_id"] = link_id
|
||||
return await asyncio.to_thread(_run_handler_sync, reset_link_handler, params)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def link_capture_start(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
link_id: Annotated[str, Field(description="UUID of the link")],
|
||||
link_id: Annotated[str | None, Field(description="Link UUID (single mode)")] = None,
|
||||
data_link_type: Annotated[str, Field(description="Data link type (default: DLT_EN10MB)")] = "DLT_EN10MB",
|
||||
capture_file_name: Annotated[str | None, Field(description="Capture file name (optional)")] = None,
|
||||
wireshark: Annotated[bool, Field(description="Open Wireshark automatically (default: false)")] = False,
|
||||
link_ids: Annotated[list[str] | None, Field(description="Batch mode: [\"uuid1\",\"uuid2\"] — start capture on multiple links in parallel")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Start packet capture on a link. The capture file can later be downloaded with download_capture_file."""
|
||||
return await asyncio.to_thread(_run_handler_sync, start_capture_handler, {
|
||||
"project_id": project_id, "link_id": link_id,
|
||||
"data_link_type": data_link_type, "capture_file_name": capture_file_name,
|
||||
"wireshark": wireshark,
|
||||
})
|
||||
"""Start packet capture on one or more links."""
|
||||
params = {"project_id": project_id, "data_link_type": data_link_type, "capture_file_name": capture_file_name, "wireshark": wireshark}
|
||||
if link_ids:
|
||||
params["link_ids"] = link_ids
|
||||
else:
|
||||
params["link_id"] = link_id
|
||||
return await asyncio.to_thread(_run_handler_sync, start_capture_handler, params)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def link_capture_stop(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
link_id: Annotated[str, Field(description="UUID of the link")],
|
||||
link_id: Annotated[str | None, Field(description="Link UUID (single mode)")] = None,
|
||||
link_ids: Annotated[list[str] | None, Field(description="Batch mode: [\"uuid1\",\"uuid2\"] — stop capture on multiple links in parallel")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Stop packet capture on a link. After stopping, the capture file can be downloaded."""
|
||||
return await asyncio.to_thread(_run_handler_sync, stop_capture_handler, {
|
||||
"project_id": project_id, "link_id": link_id,
|
||||
})
|
||||
"""Stop packet capture on one or more links."""
|
||||
params = {"project_id": project_id}
|
||||
if link_ids:
|
||||
params["link_ids"] = link_ids
|
||||
else:
|
||||
params["link_id"] = link_id
|
||||
return await asyncio.to_thread(_run_handler_sync, stop_capture_handler, params)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def link_capture_download(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
link_id: Annotated[str, Field(description="UUID of the link")],
|
||||
link_id: Annotated[str | None, Field(description="Link UUID (single mode)")] = None,
|
||||
link_ids: Annotated[list[str] | None, Field(description="Batch mode: [\"uuid1\",\"uuid2\"] — get download URLs for multiple captures")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Get the download URL and instructions for a PCAP capture file. Use curl to download."""
|
||||
return await asyncio.to_thread(_run_handler_sync, download_capture_file_handler, {
|
||||
"project_id": project_id, "link_id": link_id,
|
||||
})
|
||||
"""Get download URL(s) for PCAP capture file(s). The URL includes a short-lived JWT (10 min). Use curl to download."""
|
||||
params = {"project_id": project_id}
|
||||
if link_ids:
|
||||
params["link_ids"] = link_ids
|
||||
else:
|
||||
params["link_id"] = link_id
|
||||
return await asyncio.to_thread(_run_handler_sync, download_capture_file_handler, params)
|
||||
|
||||
|
||||
# ── Snapshot tools ─────────────────────────────────────────────────────
|
||||
@ -920,7 +1001,12 @@ async def snapshot_create(
|
||||
project_id: Annotated[str, Field(description="UUID of the project")],
|
||||
name: Annotated[str, Field(description="Name for the new snapshot")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Create a new snapshot of a project."""
|
||||
"""Create a new snapshot of a project.
|
||||
|
||||
Prerequisite: All stoppable nodes (qemu, docker, dynamips, vpcs, iou, etc.)
|
||||
must be stopped first. Use node_stop_all before creating a snapshot.
|
||||
Cloud, NAT, and switch nodes are always-running and can be ignored.
|
||||
"""
|
||||
return await asyncio.to_thread(_run_handler_sync, create_snapshot_handler, {
|
||||
"project_id": project_id, "name": name,
|
||||
})
|
||||
@ -1105,7 +1191,7 @@ async def symbol_list() -> list[dict[str, Any]]:
|
||||
async def symbol_get(
|
||||
symbol_id: Annotated[str, Field(description="Symbol ID (e.g. ':/symbols/router.svg')")],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Get a download URL for a symbol file (SVG). Use curl to download."""
|
||||
"""Get a download URL for a symbol file (SVG). The URL includes a short-lived JWT (10 min). Use curl to download."""
|
||||
return await asyncio.to_thread(_run_handler_sync, get_symbol_handler, {
|
||||
"symbol_id": symbol_id,
|
||||
})
|
||||
@ -1157,9 +1243,11 @@ async def symbol_delete(
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def appliance_list() -> list[dict[str, Any]]:
|
||||
"""List all available appliances (template library)."""
|
||||
return await asyncio.to_thread(_run_handler_sync, get_appliances_handler, {})
|
||||
async def appliance_list(
|
||||
fields: Annotated[list[str] | None, Field(description="Optional: return only these fields. e.g. [\"name\",\"category\"]. Available: name, category, description, vendor_name, product_name, status, availability, images, versions, tags, symbol, usage, builtin")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""List all available appliances (template library). Use fields=[] to return only what you need."""
|
||||
return await asyncio.to_thread(_run_handler_sync, get_appliances_handler, {"fields": fields} if fields else {})
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
@ -1256,16 +1344,23 @@ async def device_config_send(
|
||||
device_configs: Annotated[list, Field(
|
||||
description="List of device configs. Each entry: {\"device_name\": \"R1\", \"config_commands\": [\"int lo0\", \"ip add 1.1.1.1 255.255.255.255\"]}"
|
||||
)],
|
||||
template: Annotated[str | None, Field(description="Optional Jinja2 template. Use with vars in each device to reduce token usage for batch config. Example: \"interface lo{{ n }}\\nip address {{ ip }} 255.255.255.255\"")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Send configuration commands to network devices via console (telnet/SSH).
|
||||
|
||||
Two modes:
|
||||
1. Direct commands: each device has config_commands=[...]
|
||||
2. Jinja2 template: provide template + vars per device — template is rendered for each
|
||||
Example: device_configs=[{\"device_name\": \"R1\", \"vars\": {\"n\": 0, \"ip\": \"1.1.1.1\"}}]
|
||||
|
||||
Devices must be started first (use node_start or node_start_all).
|
||||
Device type is auto-detected from the 'device_type:<type>' tag on each node.
|
||||
Common device types: cisco_ios_telnet, cisco_xr_telnet, huawei_telnet, gns3_huawei_telnet_ce
|
||||
"""
|
||||
return await asyncio.to_thread(_run_handler_sync, device_config_send_handler, {
|
||||
"project_id": project_id, "device_configs": device_configs,
|
||||
})
|
||||
params = {"project_id": project_id, "device_configs": device_configs}
|
||||
if template is not None:
|
||||
params["template"] = template
|
||||
return await asyncio.to_thread(_run_handler_sync, device_config_send_handler, params)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
@ -1274,15 +1369,21 @@ async def device_command_run(
|
||||
device_configs: Annotated[list, Field(
|
||||
description="List of device commands. Each entry: {\"device_name\": \"R1\", \"show_commands\": [\"show ip int brief\", \"show running-config\"]}"
|
||||
)],
|
||||
template: Annotated[str | None, Field(description="Optional Jinja2 template. Use with vars per device. Example: \"show ip route {{ protocol }}\"")] = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Run read-only diagnostic (show) commands on network devices via console.
|
||||
|
||||
Two modes:
|
||||
1. Direct commands: each device has show_commands=[...]
|
||||
2. Jinja2 template: provide template + vars per device
|
||||
|
||||
Use this to inspect device status, view configurations, or verify changes.
|
||||
Devices must be started first.
|
||||
"""
|
||||
return await asyncio.to_thread(_run_handler_sync, device_command_run_handler, {
|
||||
"project_id": project_id, "device_configs": device_configs,
|
||||
})
|
||||
params = {"project_id": project_id, "device_configs": device_configs}
|
||||
if template is not None:
|
||||
params["template"] = template
|
||||
return await asyncio.to_thread(_run_handler_sync, device_command_run_handler, params)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
|
||||
@ -40,9 +40,30 @@ def _get_connector(gns3_ctx: dict[str, Any]):
|
||||
|
||||
# ── Tool handlers ──────────────────────────────────────────────────────────
|
||||
|
||||
VALID_APPLIANCE_FIELDS = {
|
||||
"appliance_id", "name", "category", "description", "vendor_name",
|
||||
"vendor_url", "product_name", "product_url", "documentation_url",
|
||||
"status", "availability", "maintainer", "usage", "symbol",
|
||||
"images", "versions", "tags", "builtin",
|
||||
"first_port_name", "port_name_format", "port_segment_size",
|
||||
"linked_clone", "docker", "iou", "dynamips", "qemu",
|
||||
}
|
||||
|
||||
|
||||
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()
|
||||
fields = params.get("fields")
|
||||
if fields:
|
||||
if not isinstance(fields, list):
|
||||
return {"error": "fields must be a list, e.g. [\"name\", \"category\"]"}
|
||||
invalid = [f for f in fields if f not in VALID_APPLIANCE_FIELDS]
|
||||
if invalid:
|
||||
return {
|
||||
"error": f"Unknown fields: {invalid}",
|
||||
"available_fields": sorted(VALID_APPLIANCE_FIELDS),
|
||||
}
|
||||
appliances = [{k: a[k] for k in fields if k in a} for a in appliances]
|
||||
return {"appliances": appliances, "count": len(appliances)}
|
||||
|
||||
|
||||
|
||||
@ -33,18 +33,61 @@ import json
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from jinja2 import Template as JinjaTemplate, TemplateError as JinjaError
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _render_template(template: str, device_configs: list[dict], commands_field: str = "config_commands") -> list[dict]:
|
||||
"""Render a Jinja2 template for each device's vars into the specified commands field.
|
||||
|
||||
Entries with the same device_name are merged into a single entry
|
||||
so they share one Nornir session and avoid output fragmentation.
|
||||
|
||||
Each device in device_configs can have:
|
||||
- "vars": dict of template variables (rendered into commands_field)
|
||||
- commands_field: existing commands merged after rendering if present
|
||||
|
||||
Args:
|
||||
commands_field: field name for the rendered commands, e.g. "config_commands", "commands"
|
||||
"""
|
||||
jinja = JinjaTemplate(template)
|
||||
merged: dict[str, dict] = {}
|
||||
for dev in device_configs:
|
||||
name = dev.get("device_name")
|
||||
if not name:
|
||||
continue
|
||||
vars_data = dev.get("vars", {})
|
||||
if name not in merged:
|
||||
merged[name] = {"device_name": name, commands_field: list(dev.get(commands_field, []))}
|
||||
entry = merged[name]
|
||||
if vars_data:
|
||||
try:
|
||||
output = jinja.render(**vars_data)
|
||||
lines = [l for l in output.splitlines() if l.strip()]
|
||||
entry[commands_field].extend(lines)
|
||||
except JinjaError as e:
|
||||
error_msg = f"Template rendering failed for '{name}': {e}"
|
||||
log.error(error_msg)
|
||||
return [{"error": error_msg}]
|
||||
return list(merged.values())
|
||||
|
||||
|
||||
# ── 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")
|
||||
template = params.get("template")
|
||||
if not project_id or not device_configs:
|
||||
return [{"error": "project_id and device_configs are required"}]
|
||||
|
||||
if template:
|
||||
device_configs = _render_template(template, device_configs, commands_field="config_commands")
|
||||
if len(device_configs) == 1 and "error" in device_configs[0]:
|
||||
return device_configs
|
||||
|
||||
from gns3server.agent.gns3_copilot.tools_v2.config_tools_nornir import ExecuteMultipleDeviceConfigCommands
|
||||
|
||||
tool = ExecuteMultipleDeviceConfigCommands()
|
||||
@ -63,9 +106,15 @@ def device_command_run_handler(params: dict[str, Any], gns3_ctx: dict[str, Any])
|
||||
"""Run read-only diagnostic (show) commands on network devices."""
|
||||
project_id = params.get("project_id")
|
||||
device_configs = params.get("device_configs")
|
||||
template = params.get("template")
|
||||
if not project_id or not device_configs:
|
||||
return [{"error": "project_id and device_configs (list of {device_name, show_commands}) are required"}]
|
||||
|
||||
if template:
|
||||
device_configs = _render_template(template, device_configs, commands_field="commands")
|
||||
if len(device_configs) == 1 and "error" in device_configs[0]:
|
||||
return device_configs
|
||||
|
||||
from gns3server.agent.gns3_copilot.tools_v2.display_tools_nornir import ExecuteMultipleDeviceCommands
|
||||
|
||||
tool = ExecuteMultipleDeviceCommands()
|
||||
|
||||
@ -23,11 +23,16 @@ via Gns3Connector (from custom_gns3fy).
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
import logging
|
||||
|
||||
from gns3server.services import auth_service
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
BATCH_MAX_WORKERS = 10
|
||||
|
||||
|
||||
# ── Helper ─────────────────────────────────────────────────────────────────
|
||||
|
||||
@ -43,12 +48,31 @@ def _get_connector(gns3_ctx: dict[str, Any]):
|
||||
|
||||
# ── Tool handlers ──────────────────────────────────────────────────────────
|
||||
|
||||
VALID_LINK_FIELDS = {
|
||||
"link_id", "project_id", "link_type", "nodes", "suspend",
|
||||
"link_style", "filters", "show_filters_icon",
|
||||
"capturing", "capture_file_name", "capture_file_path",
|
||||
"capture_compute_id", "wireshark",
|
||||
}
|
||||
|
||||
|
||||
def get_links_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)
|
||||
links = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/links").json()
|
||||
fields = params.get("fields")
|
||||
if fields:
|
||||
if not isinstance(fields, list):
|
||||
return {"error": "fields must be a list, e.g. [\"link_id\", \"nodes\"]"}
|
||||
invalid = [f for f in fields if f not in VALID_LINK_FIELDS]
|
||||
if invalid:
|
||||
return {
|
||||
"error": f"Unknown fields: {invalid}",
|
||||
"available_fields": sorted(VALID_LINK_FIELDS),
|
||||
}
|
||||
links = [{k: l[k] for k in fields if k in l} for l in links]
|
||||
return {"links": links, "count": len(links)}
|
||||
|
||||
|
||||
@ -63,9 +87,42 @@ def get_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[s
|
||||
|
||||
def create_link_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"}
|
||||
|
||||
links = params.get("links")
|
||||
# Batch mode: links=[{nodes, link_type?, filters?, suspend?}]
|
||||
if links is not None:
|
||||
if not isinstance(links, list) or not links:
|
||||
return {"error": "links must be a non-empty array"}
|
||||
results = []
|
||||
conn = _get_connector(gns3_ctx)
|
||||
def _create_one(link_data):
|
||||
if not link_data.get("nodes"):
|
||||
return {"status": "error", "error": "nodes is required for each link"}
|
||||
try:
|
||||
body = {"nodes": link_data["nodes"]}
|
||||
if link_data.get("link_type"):
|
||||
body["link_type"] = link_data["link_type"]
|
||||
if link_data.get("filters"):
|
||||
body["filters"] = link_data["filters"]
|
||||
if link_data.get("suspend"):
|
||||
body["suspend"] = link_data["suspend"]
|
||||
url = f"{conn.base_url}/projects/{project_id}/links"
|
||||
resp = conn.http_call("post", url, json_data=body).json()
|
||||
return {"status": "success", "link": resp}
|
||||
except Exception as e:
|
||||
return {"status": "error", "error": str(e)}
|
||||
with ThreadPoolExecutor(max_workers=min(len(links), BATCH_MAX_WORKERS)) as pool:
|
||||
futures = {pool.submit(_create_one, l): l for l in links}
|
||||
for future in as_completed(futures):
|
||||
results.append(future.result())
|
||||
return results
|
||||
|
||||
# Single mode
|
||||
nodes = params.get("nodes")
|
||||
if not project_id or not nodes:
|
||||
return {"error": "project_id and nodes are required"}
|
||||
if not nodes:
|
||||
return {"error": "nodes is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
data = {"nodes": nodes}
|
||||
if "link_type" in params:
|
||||
@ -80,9 +137,24 @@ def create_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dic
|
||||
|
||||
def delete_link_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"}
|
||||
link_ids = params.get("link_ids")
|
||||
if link_ids:
|
||||
if not isinstance(link_ids, list):
|
||||
return {"error": "link_ids must be a list"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
def _del(lid):
|
||||
try:
|
||||
conn.http_call("delete", f"{conn.base_url}/projects/{project_id}/links/{lid}")
|
||||
return {"link_id": lid, "status": "deleted"}
|
||||
except Exception as e:
|
||||
return {"link_id": lid, "status": "error", "error": str(e)}
|
||||
with ThreadPoolExecutor(max_workers=min(len(link_ids), BATCH_MAX_WORKERS)) as pool:
|
||||
return list(pool.map(_del, link_ids))
|
||||
link_id = params.get("link_id")
|
||||
if not project_id or not link_id:
|
||||
return {"error": "project_id and link_id are required"}
|
||||
if not link_id:
|
||||
return {"error": "link_id or link_ids is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("delete", f"{conn.base_url}/projects/{project_id}/links/{link_id}")
|
||||
return {"message": f"Link {link_id} deleted", "link_id": link_id}
|
||||
@ -110,20 +182,66 @@ def update_link_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dic
|
||||
|
||||
def reset_link_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"}
|
||||
link_ids = params.get("link_ids")
|
||||
if link_ids:
|
||||
if not isinstance(link_ids, list):
|
||||
return {"error": "link_ids must be a list"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
def _rst(lid):
|
||||
try:
|
||||
url = f"{conn.base_url}/projects/{project_id}/links/{lid}/reset"
|
||||
r = conn.http_call("post", url).json()
|
||||
return {"link_id": lid, "status": "reset", "link": r}
|
||||
except Exception as e:
|
||||
return {"link_id": lid, "status": "error", "error": str(e)}
|
||||
with ThreadPoolExecutor(max_workers=min(len(link_ids), BATCH_MAX_WORKERS)) as pool:
|
||||
return list(pool.map(_rst, link_ids))
|
||||
link_id = params.get("link_id")
|
||||
if not project_id or not link_id:
|
||||
return {"error": "project_id and link_id are required"}
|
||||
if not link_id:
|
||||
return {"error": "link_id or link_ids is 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 _batch_capture(project_id, link_ids, action, data_builder, conn):
|
||||
"""Helper for batch capture start/stop."""
|
||||
def _act(lid):
|
||||
try:
|
||||
url = f"{conn.base_url}/projects/{project_id}/links/{lid}/capture/{action}"
|
||||
kwargs = data_builder(lid) if data_builder else {}
|
||||
conn.http_call("post", url, **kwargs)
|
||||
return {"link_id": lid, "status": "success"}
|
||||
except Exception as e:
|
||||
return {"link_id": lid, "status": "error", "error": str(e)}
|
||||
with ThreadPoolExecutor(max_workers=min(len(link_ids), BATCH_MAX_WORKERS)) as pool:
|
||||
return list(pool.map(_act, link_ids))
|
||||
|
||||
|
||||
def start_capture_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"}
|
||||
link_ids = params.get("link_ids")
|
||||
if link_ids:
|
||||
if not isinstance(link_ids, list):
|
||||
return {"error": "link_ids must be a list"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
dlt = params.get("data_link_type", "DLT_EN10MB")
|
||||
ws = params.get("wireshark", False)
|
||||
fname = params.get("capture_file_name")
|
||||
def _build(lid):
|
||||
data = {"data_link_type": dlt, "wireshark": ws}
|
||||
if fname:
|
||||
data["capture_file_name"] = fname
|
||||
return {"json_data": data}
|
||||
return _batch_capture(project_id, link_ids, "start", _build, conn)
|
||||
link_id = params.get("link_id")
|
||||
if not project_id or not link_id:
|
||||
return {"error": "project_id and link_id are required"}
|
||||
if not link_id:
|
||||
return {"error": "link_id or link_ids is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
data = {
|
||||
"data_link_type": params.get("data_link_type", "DLT_EN10MB"),
|
||||
@ -138,9 +256,17 @@ def start_capture_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> d
|
||||
|
||||
def stop_capture_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"}
|
||||
link_ids = params.get("link_ids")
|
||||
if link_ids:
|
||||
if not isinstance(link_ids, list):
|
||||
return {"error": "link_ids must be a list"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return _batch_capture(project_id, link_ids, "stop", None, conn)
|
||||
link_id = params.get("link_id")
|
||||
if not project_id or not link_id:
|
||||
return {"error": "project_id and link_id are required"}
|
||||
if not link_id:
|
||||
return {"error": "link_id or link_ids is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
url = f"{conn.base_url}/projects/{project_id}/links/{link_id}/capture/stop"
|
||||
conn.http_call("post", url)
|
||||
@ -149,18 +275,38 @@ def stop_capture_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> di
|
||||
|
||||
def download_capture_file_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"}
|
||||
username = gns3_ctx.get("jwt_username")
|
||||
download_token = auth_service.create_access_token(username, expires_in=10) if username else None
|
||||
|
||||
link_ids = params.get("link_ids")
|
||||
if link_ids:
|
||||
if not isinstance(link_ids, list):
|
||||
return {"error": "link_ids must be a list"}
|
||||
results = []
|
||||
for lid in link_ids:
|
||||
url = f"{gns3_ctx['server_url']}/v3/projects/{project_id}/links/{lid}/capture/file"
|
||||
entry = {"link_id": lid, "download_url": url}
|
||||
if download_token:
|
||||
cmd = f"curl -L -o capture_{lid}.pcap -H 'Authorization: Bearer {download_token}' '{url}'"
|
||||
entry["curl_command"] = cmd
|
||||
results.append(entry)
|
||||
return {"downloads": results, "count": len(results), "note": "Files are in pcap format. Links include a 10-minute token."}
|
||||
|
||||
link_id = params.get("link_id")
|
||||
if not project_id or not link_id:
|
||||
return {"error": "project_id and link_id are required"}
|
||||
if not link_id:
|
||||
return {"error": "link_id or link_ids is required"}
|
||||
download_url = f"{gns3_ctx['server_url']}/v3/projects/{project_id}/links/{link_id}/capture/file"
|
||||
auth_token = gns3_ctx['jwt_token']
|
||||
return {
|
||||
result = {
|
||||
"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.",
|
||||
"note": "The file is in pcap format and can be analyzed with Wireshark or tcpdump.",
|
||||
}
|
||||
if download_token:
|
||||
result["curl_command"] = f"curl -L -o capture.pcap -H 'Authorization: Bearer {download_token}' '{download_url}'"
|
||||
result["note"] += " The download link includes a 10-minute token."
|
||||
return result
|
||||
|
||||
|
||||
# ── Tool definitions ───────────────────────────────────────────────────────
|
||||
|
||||
@ -23,11 +23,16 @@ via Gns3Connector (from custom_gns3fy).
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
import logging
|
||||
|
||||
from gns3server.services import auth_service
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
BATCH_MAX_WORKERS = 10
|
||||
|
||||
# ── Constants ──────────────────────────────────────────────────────────────
|
||||
|
||||
# Maximum bytes to return from get_node_file (safety net).
|
||||
@ -55,23 +60,82 @@ def get_nodes_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[
|
||||
return {"error": "project_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
nodes = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/nodes").json()
|
||||
fields = params.get("fields")
|
||||
if fields:
|
||||
if not isinstance(fields, list):
|
||||
return {"error": "fields must be a list of field names, e.g. [\"name\", \"status\"]"}
|
||||
invalid = [f for f in fields if f not in VALID_NODE_FIELDS]
|
||||
if invalid:
|
||||
return {
|
||||
"error": f"Unknown fields: {invalid}",
|
||||
"available_fields": sorted(VALID_NODE_FIELDS),
|
||||
}
|
||||
nodes = [{k: n[k] for k in fields if k in n} for n in nodes]
|
||||
return {"nodes": nodes, "count": len(nodes)}
|
||||
|
||||
|
||||
VALID_NODE_FIELDS = {
|
||||
# NodeBase
|
||||
"compute_id", "name", "node_type", "node_id",
|
||||
"console", "console_type", "console_auto_start",
|
||||
"aux", "aux_type", "properties", "label", "symbol",
|
||||
"x", "y", "z", "locked",
|
||||
"port_name_format", "port_segment_size", "first_port_name",
|
||||
"custom_adapters", "tags",
|
||||
# Node
|
||||
"template_id", "project_id", "node_directory", "status",
|
||||
"command_line", "width", "height", "ports", "console_host",
|
||||
}
|
||||
|
||||
|
||||
def get_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)
|
||||
return conn.http_call("get", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}").json()
|
||||
node = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}").json()
|
||||
|
||||
fields = params.get("fields")
|
||||
if fields:
|
||||
if not isinstance(fields, list):
|
||||
return {"error": "fields must be a list of field names, e.g. [\"name\", \"status\"]"}
|
||||
invalid = [f for f in fields if f not in VALID_NODE_FIELDS]
|
||||
if invalid:
|
||||
return {
|
||||
"error": f"Unknown fields: {invalid}",
|
||||
"available_fields": sorted(VALID_NODE_FIELDS),
|
||||
}
|
||||
return {k: node[k] for k in fields if k in node}
|
||||
|
||||
return node
|
||||
|
||||
|
||||
def _batch_lifecycle(project_id, node_ids, action, conn, action_label):
|
||||
"""Helper to run a lifecycle action on multiple nodes in parallel."""
|
||||
def _act(nid):
|
||||
try:
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/{nid}/{action}")
|
||||
return {"node_id": nid, "status": "success", "message": f"Node {nid} {action_label}"}
|
||||
except Exception as e:
|
||||
return {"node_id": nid, "status": "error", "error": str(e)}
|
||||
with ThreadPoolExecutor(max_workers=min(len(node_ids), BATCH_MAX_WORKERS)) as pool:
|
||||
return list(pool.map(_act, node_ids))
|
||||
|
||||
|
||||
def start_node_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"}
|
||||
node_ids = params.get("node_ids")
|
||||
if node_ids:
|
||||
if not isinstance(node_ids, list):
|
||||
return {"error": "node_ids must be a list"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return _batch_lifecycle(project_id, node_ids, "start", conn, "started")
|
||||
node_id = params.get("node_id")
|
||||
if not project_id or not node_id:
|
||||
return {"error": "project_id and node_id are required"}
|
||||
if not node_id:
|
||||
return {"error": "node_id or node_ids is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/start", json_data={})
|
||||
return {"message": f"Node {node_id} started", "node_id": node_id}
|
||||
@ -79,9 +143,17 @@ def start_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict
|
||||
|
||||
def stop_node_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"}
|
||||
node_ids = params.get("node_ids")
|
||||
if node_ids:
|
||||
if not isinstance(node_ids, list):
|
||||
return {"error": "node_ids must be a list"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return _batch_lifecycle(project_id, node_ids, "stop", conn, "stopped")
|
||||
node_id = params.get("node_id")
|
||||
if not project_id or not node_id:
|
||||
return {"error": "project_id and node_id are required"}
|
||||
if not node_id:
|
||||
return {"error": "node_id or node_ids is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/stop", json_data={})
|
||||
return {"message": f"Node {node_id} stopped", "node_id": node_id}
|
||||
@ -89,9 +161,17 @@ def stop_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[
|
||||
|
||||
def reload_node_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"}
|
||||
node_ids = params.get("node_ids")
|
||||
if node_ids:
|
||||
if not isinstance(node_ids, list):
|
||||
return {"error": "node_ids must be a list"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return _batch_lifecycle(project_id, node_ids, "reload", conn, "reloaded")
|
||||
node_id = params.get("node_id")
|
||||
if not project_id or not node_id:
|
||||
return {"error": "project_id and node_id are required"}
|
||||
if not node_id:
|
||||
return {"error": "node_id or node_ids is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/reload")
|
||||
return {"message": f"Node {node_id} reloaded", "node_id": node_id}
|
||||
@ -99,9 +179,17 @@ def reload_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dic
|
||||
|
||||
def suspend_node_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"}
|
||||
node_ids = params.get("node_ids")
|
||||
if node_ids:
|
||||
if not isinstance(node_ids, list):
|
||||
return {"error": "node_ids must be a list"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
return _batch_lifecycle(project_id, node_ids, "suspend", conn, "suspended")
|
||||
node_id = params.get("node_id")
|
||||
if not project_id or not node_id:
|
||||
return {"error": "project_id and node_id are required"}
|
||||
if not node_id:
|
||||
return {"error": "node_id or node_ids is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("post", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}/suspend")
|
||||
return {"message": f"Node {node_id} suspended", "node_id": node_id}
|
||||
@ -109,9 +197,41 @@ def suspend_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> di
|
||||
|
||||
def create_node_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"}
|
||||
|
||||
nodes = params.get("nodes")
|
||||
# Batch mode: nodes=[{template_id, x, y, name?, compute_id?}]
|
||||
if nodes is not None:
|
||||
if not isinstance(nodes, list) or not nodes:
|
||||
return {"error": "nodes must be a non-empty array"}
|
||||
results = []
|
||||
conn = _get_connector(gns3_ctx)
|
||||
def _create_one(node_data):
|
||||
tid = node_data.get("template_id")
|
||||
if not tid:
|
||||
return {"template_id": tid, "status": "error", "error": "template_id is required"}
|
||||
try:
|
||||
url = f"{conn.base_url}/projects/{project_id}/templates/{tid}"
|
||||
body = {
|
||||
"x": node_data.get("x", 0),
|
||||
"y": node_data.get("y", 0),
|
||||
"compute_id": node_data.get("compute_id", "local"),
|
||||
}
|
||||
resp = conn.http_call("post", url, json_data=body).json()
|
||||
return {"template_id": tid, "status": "success", "node": resp}
|
||||
except Exception as e:
|
||||
return {"template_id": tid, "status": "error", "error": str(e)}
|
||||
with ThreadPoolExecutor(max_workers=min(len(nodes), BATCH_MAX_WORKERS)) as pool:
|
||||
futures = {pool.submit(_create_one, n): n for n in nodes}
|
||||
for future in as_completed(futures):
|
||||
results.append(future.result())
|
||||
return results
|
||||
|
||||
# Single mode
|
||||
template_id = params.get("template_id")
|
||||
if not project_id or not template_id:
|
||||
return {"error": "project_id and template_id are required"}
|
||||
if not template_id:
|
||||
return {"error": "template_id is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
data = {
|
||||
"x": params.get("x", 0),
|
||||
@ -124,9 +244,24 @@ def create_node_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dic
|
||||
|
||||
def delete_node_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"}
|
||||
node_ids = params.get("node_ids")
|
||||
if node_ids:
|
||||
if not isinstance(node_ids, list):
|
||||
return {"error": "node_ids must be a list"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
def _del(nid):
|
||||
try:
|
||||
conn.http_call("delete", f"{conn.base_url}/projects/{project_id}/nodes/{nid}")
|
||||
return {"node_id": nid, "status": "deleted"}
|
||||
except Exception as e:
|
||||
return {"node_id": nid, "status": "error", "error": str(e)}
|
||||
with ThreadPoolExecutor(max_workers=min(len(node_ids), BATCH_MAX_WORKERS)) as pool:
|
||||
return list(pool.map(_del, node_ids))
|
||||
node_id = params.get("node_id")
|
||||
if not project_id or not node_id:
|
||||
return {"error": "project_id and node_id are required"}
|
||||
if not node_id:
|
||||
return {"error": "node_id or node_ids is required"}
|
||||
conn = _get_connector(gns3_ctx)
|
||||
conn.http_call("delete", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}")
|
||||
return {"message": f"Node {node_id} deleted", "node_id": node_id}
|
||||
@ -158,7 +293,14 @@ def get_node_console_info_handler(params: dict[str, Any], gns3_ctx: dict[str, An
|
||||
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']}"
|
||||
# Short-lived JWT for the WebSocket URL (10 min)
|
||||
username = gns3_ctx.get("jwt_username")
|
||||
ws_token = auth_service.create_access_token(username, expires_in=10) if username else None
|
||||
raw_url = f"{gns3_ctx['server_url']}/v3/projects/{project_id}/nodes/{node_id}/console/ws"
|
||||
if ws_token:
|
||||
raw_url += f"?token={ws_token}"
|
||||
# Convert http scheme to ws for direct websocat usage
|
||||
ws_url = raw_url.replace("https://", "wss://").replace("http://", "ws://")
|
||||
|
||||
result = {
|
||||
"node_id": node_id,
|
||||
|
||||
@ -23,6 +23,8 @@ from typing import Any
|
||||
|
||||
import logging
|
||||
|
||||
from gns3server.services import auth_service
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@ -51,13 +53,18 @@ def get_symbol_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict
|
||||
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 {
|
||||
username = gns3_ctx.get("jwt_username")
|
||||
download_token = auth_service.create_access_token(username, expires_in=10) if username else None
|
||||
result = {
|
||||
"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.",
|
||||
"note": "Symbol files are SVG images.",
|
||||
}
|
||||
if download_token:
|
||||
safe_name = symbol_id.replace(':', '').replace('/', '_')
|
||||
result["curl_command"] = f"curl -L -o '{safe_name}.svg' -H 'Authorization: Bearer {download_token}' '{download_url}'"
|
||||
result["note"] += " Download link includes a 10-minute token."
|
||||
return result
|
||||
|
||||
|
||||
def get_symbol_dimensions_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user