From f14d30cb7ecca18654237fb710b8fbe857795b5c Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Sun, 14 Jun 2026 21:36:24 +0800 Subject: [PATCH] feat: Add batch link_ids to link_capture_download --- gns3server/api/routes/mcp/__init__.py | 14 +++++++++----- gns3server/api/routes/mcp/links.py | 26 +++++++++++++++++++++----- 2 files changed, 30 insertions(+), 10 deletions(-) diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index b99b4783c..0c6b037b0 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -952,12 +952,16 @@ async def link_capture_stop( @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. The URL includes a short-lived JWT (10 min). 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 ───────────────────────────────────────────────────── diff --git a/gns3server/api/routes/mcp/links.py b/gns3server/api/routes/mcp/links.py index 2b5713427..0e9be5e29 100644 --- a/gns3server/api/routes/mcp/links.py +++ b/gns3server/api/routes/mcp/links.py @@ -225,13 +225,29 @@ 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") - 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" - # Short-lived JWT (10 min) — username stored during auth, never exposes raw key + 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 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" result = { "link_id": link_id, "download_url": download_url,