diff --git a/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py b/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py index 2c83ddf7a..13c17436c 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py +++ b/gns3server/agent/gns3_copilot/gns3_client/custom_gns3fy.py @@ -782,6 +782,81 @@ 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. diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index 7aae521d7..7af5029b7 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -56,6 +56,8 @@ from .nodes import ( stop_node_handler, reload_node_handler, suspend_node_handler, create_node_handler, delete_node_handler, update_node_handler, get_node_console_info_handler, + list_node_files_handler, get_node_file_handler, + write_node_file_handler, delete_node_file_handler, ) from .links import ( get_links_handler, get_link_handler, create_link_handler, @@ -588,6 +590,74 @@ async def get_compute_images( }) +# ── Node file tools ──────────────────────────────────────────────────── + + +@mcp.tool() +async def list_node_files( + project_id: Annotated[str, Field(description="UUID of the project")], + node_id: Annotated[str, Field(description="UUID of the node")], + path: Annotated[str, Field(description="Subdirectory path within node directory (optional)")] = "", + recursive: Annotated[bool, Field(description="Recursively list all files (optional, default: false)")] = False, +) -> list[dict[str, Any]]: + """List files in a node directory with metadata (name, size, type, modified time). + + Use this first to check file sizes before reading files with get_node_file. + Large config files should be read in chunks using offset/limit. + """ + return await asyncio.to_thread(_run_handler_sync, list_node_files_handler, { + "project_id": project_id, "node_id": node_id, "path": path, "recursive": recursive, + }) + + +@mcp.tool() +async def get_node_file( + project_id: Annotated[str, Field(description="UUID of the project")], + node_id: Annotated[str, Field(description="UUID of the node")], + file_path: Annotated[str, Field(description="Path to the file within the node directory")], + offset: Annotated[int, Field(description="Line offset to start reading from (optional, default: 0)")] = 0, + limit: Annotated[int, Field(description="Maximum number of lines to return (optional, default: 200)")] = 200, +) -> list[dict[str, Any]]: + """Read a text file from a node directory line-by-line with offset/limit support. + + Best practice: + 1. First call list_node_files to see the file size before deciding to read. + 2. Start with offset=0, limit=200 to preview the file. + 3. If metadata.has_more is true, read more by increasing offset. + Large files (>50KB) are auto-truncated; check the metadata.truncated flag. + For binary files, check the file type via list_node_files first. + """ + return await asyncio.to_thread(_run_handler_sync, get_node_file_handler, { + "project_id": project_id, "node_id": node_id, "file_path": file_path, + "offset": offset, "limit": limit, + }) + + +@mcp.tool() +async def write_node_file( + project_id: Annotated[str, Field(description="UUID of the project")], + node_id: Annotated[str, Field(description="UUID of the node")], + file_path: Annotated[str, Field(description="Path to the file within the node directory")], + content: Annotated[str, Field(description="Content to write to the file")], +) -> list[dict[str, Any]]: + """Write content to a file in a node directory. Creates the file if it doesn't exist. Overwrites existing content.""" + return await asyncio.to_thread(_run_handler_sync, write_node_file_handler, { + "project_id": project_id, "node_id": node_id, "file_path": file_path, "content": content, + }) + + +@mcp.tool() +async def delete_node_file( + project_id: Annotated[str, Field(description="UUID of the project")], + node_id: Annotated[str, Field(description="UUID of the node")], + file_path: Annotated[str, Field(description="Path to the file within the node directory")], +) -> list[dict[str, Any]]: + """Delete a file from a node directory. Cannot be undone.""" + return await asyncio.to_thread(_run_handler_sync, delete_node_file_handler, { + "project_id": project_id, "node_id": node_id, "file_path": file_path, + }) + + # ── Auth‑wrapped SSE app ────────────────────────────────────────────── def _make_auth_wrapper(inner_app): diff --git a/gns3server/api/routes/mcp/nodes.py b/gns3server/api/routes/mcp/nodes.py index 7a16a59cd..f74213e7f 100644 --- a/gns3server/api/routes/mcp/nodes.py +++ b/gns3server/api/routes/mcp/nodes.py @@ -28,6 +28,12 @@ import logging log = logging.getLogger(__name__) +# ── Constants ────────────────────────────────────────────────────────────── + +# Maximum bytes to return from get_node_file (safety net). +# Larger files are truncated with a truncated=True flag. +MAX_NODE_FILE_BYTES = 50 * 1024 # 50 KiB + # ── Helper ───────────────────────────────────────────────────────────────── @@ -166,6 +172,89 @@ def get_node_console_info_handler(params: dict[str, Any], gns3_ctx: dict[str, An return result +# ── Node file handlers ──────────────────────────────────────────────────── + + +def list_node_files_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) + files = conn.list_node_files( + project_id=project_id, + node_id=node_id, + path=params.get("path", ""), + recursive=params.get("recursive", False), + ) + return {"files": files, "count": len(files)} + + +def get_node_file_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") + file_path = params.get("file_path") + if not project_id or not node_id or not file_path: + return {"error": "project_id, node_id and file_path are required"} + + offset = params.get("offset", 0) + 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) + + total_bytes = len(raw.encode("utf-8")) + truncated = False + if total_bytes > MAX_NODE_FILE_BYTES: + raw = raw[:MAX_NODE_FILE_BYTES] + truncated = True + + lines = raw.splitlines(keepends=False) + total_lines = len(lines) + + # Apply offset/limit + selected = lines[offset: offset + limit] if offset < total_lines else [] + has_more = (offset + limit) < total_lines or truncated + + return { + "file_path": file_path, + "content": "\n".join(selected), + "metadata": { + "total_lines": total_lines, + "total_bytes": total_bytes, + "offset": offset, + "limit": limit, + "returned_lines": len(selected), + "returned_bytes": len("\n".join(selected).encode("utf-8")), + "truncated": truncated or has_more, + "has_more": has_more, + }, + } + + +def write_node_file_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") + file_path = params.get("file_path") + content = params.get("content") + 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) + return {"message": f"File {file_path} written to node {node_id}", "file_path": file_path, "node_id": node_id} + + +def delete_node_file_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") + file_path = params.get("file_path") + 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) + return {"message": f"File {file_path} deleted from node {node_id}", "file_path": file_path, "node_id": node_id} + + # ── Tool definitions ─────────────────────────────────────────────────────── NODE_TOOLS = [ @@ -305,4 +394,71 @@ NODE_TOOLS = [ }, "handler": get_node_console_info_handler, }, + { + "name": "list_node_files", + "description": "List files in a node directory with metadata (name, size, type, modified time). " + "Use recursive=true for a full recursive listing. " + "Check file sizes before reading large files with get_node_file.", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "node_id": {"type": "string", "description": "Node UUID"}, + "path": {"type": "string", "description": "Subdirectory path within node directory (optional)"}, + "recursive": {"type": "boolean", "description": "Recursively list all files (optional, default: false)"}, + }, + "required": ["project_id", "node_id"], + }, + "handler": list_node_files_handler, + }, + { + "name": "get_node_file", + "description": "Read a text file from a node directory. Returns file content line-by-line with offset/limit support. " + "Best practice: start with offset=0, limit=200 to preview, then increase offset to read more. " + "Large files (>50KB) are auto-truncated; check the metadata.truncated flag. " + "For binary files, check file type via list_node_files first.", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "node_id": {"type": "string", "description": "Node UUID"}, + "file_path": {"type": "string", "description": "Path to the file within the node directory"}, + "offset": {"type": "integer", "description": "Line offset to start reading from (optional, default: 0)"}, + "limit": {"type": "integer", "description": "Maximum number of lines to return (optional, default: 200)"}, + }, + "required": ["project_id", "node_id", "file_path"], + }, + "handler": get_node_file_handler, + }, + { + "name": "write_node_file", + "description": "Write content to a file in a node directory. Creates the file if it doesn't exist. " + "Overwrites existing content. Useful for updating configuration files on nodes.", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "node_id": {"type": "string", "description": "Node UUID"}, + "file_path": {"type": "string", "description": "Path to the file within the node directory"}, + "content": {"type": "string", "description": "Content to write to the file"}, + }, + "required": ["project_id", "node_id", "file_path", "content"], + }, + "handler": write_node_file_handler, + }, + { + "name": "delete_node_file", + "description": "Delete a file from a node directory. Cannot be undone. " + "Use list_node_files to confirm the file path before deleting.", + "parameters": { + "type": "object", + "properties": { + "project_id": {"type": "string", "description": "Project UUID"}, + "node_id": {"type": "string", "description": "Node UUID"}, + "file_path": {"type": "string", "description": "Path to the file within the node directory"}, + }, + "required": ["project_id", "node_id", "file_path"], + }, + "handler": delete_node_file_handler, + }, ]