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

refactor: unify MCP handlers to use http_call directly, relocate node file ops to Node class

Browse Source 
- Move node file operations (list_files, delete_file) from Gns3Connector to Node class
- Add link capture/reset operations (reset, start_capture, stop_capture) to Link class
- Convert all MCP handlers to use conn.http_call() directly instead of
  Gns3Connector/Node/Link abstraction methods
- Register 4 new MCP tools: reset_link, start_capture, stop_capture,
  download_capture_file
This commit is contained in:
YueGuobin 2026-06-10 13:47:10 +08:00
parent 28b06f37c4
commit acce79d243
No known key found for this signature in database
7 changed files with 387 additions and 119 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

223
gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py
View File

@ -782,81 +782,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 +1176,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 +1778,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:
"""

54
gns3server/api/routes/mcp/__init__.py
View File

@ -62,6 +62,8 @@ from .nodes import (
from .links import (
get_links_handler, get_link_handler, create_link_handler,
delete_link_handler, update_link_handler,
reset_link_handler, start_capture_handler, stop_capture_handler,
download_capture_file_handler,
)
from .templates import (
list_templates_handler, get_template_handler, create_template_handler,
@ -658,6 +660,58 @@ async def delete_node_file(
})
# ── Link capture / reset tools ────────────────────────────────────────
@mcp.tool()
async def reset_link(
project_id: Annotated[str, Field(description="UUID of the project")],
link_id: Annotated[str, Field(description="UUID of the link")],
) -> list[dict[str, Any]]:
"""Reset a link, clearing its state (counters, filters, etc.)."""
return await asyncio.to_thread(_run_handler_sync, reset_link_handler, {
"project_id": project_id, "link_id": link_id,
})
@mcp.tool()
async def start_capture(
project_id: Annotated[str, Field(description="UUID of the project")],
link_id: Annotated[str, Field(description="UUID of the link")],
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,
) -> 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,
})
@mcp.tool()
async def stop_capture(
project_id: Annotated[str, Field(description="UUID of the project")],
link_id: Annotated[str, Field(description="UUID of the link")],
) -> 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,
})
@mcp.tool()
async def download_capture_file(
project_id: Annotated[str, Field(description="UUID of the project")],
link_id: Annotated[str, Field(description="UUID of the link")],
) -> 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,
})
# ── Auth‑wrapped SSE app ──────────────────────────────────────────────
def _make_auth_wrapper(inner_app):

6
gns3server/api/routes/mcp/computes.py
View File

@ -37,14 +37,14 @@ 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")
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]:
@ -53,7 +53,7 @@ def get_compute_images_handler(params: dict[str, Any], gns3_ctx: dict[str, Any])
if not emulator:
return {"error": "emulator is required (e.g. qemu, iou, docker)"}
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)}

119
gns3server/api/routes/mcp/links.py
View File

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

29
gns3server/api/routes/mcp/nodes.py
View File

@ -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,7 +254,9 @@ 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}
return {"message": f"File {file_path} deleted from node {node_id}", "file_path": file_path, "node_id": node_id}

18
gns3server/api/routes/mcp/projects.py
View File

@ -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
@ -69,7 +69,7 @@ def create_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) ->
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=project_data).json()
def delete_project_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
@ -77,7 +77,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 +115,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 +127,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 +136,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,7 +153,8 @@ 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}

57
gns3server/api/routes/mcp/templates.py
View File

@ -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,12 @@ 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"),
}
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 +95,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 +119,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"}