From acce79d2431c88c17fdbd1db2608def227530e3b Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Wed, 10 Jun 2026 13:47:10 +0800 Subject: [PATCH] refactor: unify MCP handlers to use http_call directly, relocate node file ops to Node class - 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 --- .../gns3_copilot/gns3_client/custom_gns3fy.py | 223 ++++++++++++------ gns3server/api/routes/mcp/__init__.py | 54 +++++ gns3server/api/routes/mcp/computes.py | 6 +- gns3server/api/routes/mcp/links.py | 119 +++++++++- gns3server/api/routes/mcp/nodes.py | 29 ++- gns3server/api/routes/mcp/projects.py | 18 +- gns3server/api/routes/mcp/templates.py | 57 +++-- 7 files changed, 387 insertions(+), 119 deletions(-) diff --git a/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py b/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py index 13c17436c..3f714fd36 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py +++ b/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py @@ -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: """ diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index 7af5029b7..51bdc2f30 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -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): diff --git a/gns3server/api/routes/mcp/computes.py b/gns3server/api/routes/mcp/computes.py index fa6c70674..e95fbd45c 100644 --- a/gns3server/api/routes/mcp/computes.py +++ b/gns3server/api/routes/mcp/computes.py @@ -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)} diff --git a/gns3server/api/routes/mcp/links.py b/gns3server/api/routes/mcp/links.py index 1d5a87da6..3ae7924bf 100644 --- a/gns3server/api/routes/mcp/links.py +++ b/gns3server/api/routes/mcp/links.py @@ -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, + }, ] diff --git a/gns3server/api/routes/mcp/nodes.py b/gns3server/api/routes/mcp/nodes.py index f74213e7f..6c680ec3c 100644 --- a/gns3server/api/routes/mcp/nodes.py +++ b/gns3server/api/routes/mcp/nodes.py @@ -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} diff --git a/gns3server/api/routes/mcp/projects.py b/gns3server/api/routes/mcp/projects.py index bcf30d947..5d6023f64 100644 --- a/gns3server/api/routes/mcp/projects.py +++ b/gns3server/api/routes/mcp/projects.py @@ -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} diff --git a/gns3server/api/routes/mcp/templates.py b/gns3server/api/routes/mcp/templates.py index e2cc71e76..6ed9eaf30 100644 --- a/gns3server/api/routes/mcp/templates.py +++ b/gns3server/api/routes/mcp/templates.py @@ -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"}