feat: Add node file operations as MCP tools (list, get, write, delete)

- Add list_node_files, get_node_file, write_node_file, delete_node_file methods to Gns3Connector
- Add MCP handlers with offset/limit line-based reading for get_node_file
- Auto-truncate files >50KB with truncated flag in response
- Rich metadata returned (total_lines, total_bytes, has_more, etc.)
- Tool docstrings guide AI to check file sizes before reading chunked
This commit is contained in:
YueGuobin 2026-06-10 12:16:35 +08:00
parent 066b7076c0
commit 28b06f37c4
No known key found for this signature in database
3 changed files with 301 additions and 0 deletions

View File

@ -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.

View File

@ -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):

View File

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