diff --git a/gns3server/api/openapi.py b/gns3server/api/openapi.py
new file mode 100644
index 000000000..cf5dce209
--- /dev/null
+++ b/gns3server/api/openapi.py
@@ -0,0 +1,47 @@
+#
+# Copyright (C) 2026 GNS3 Technologies Inc.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see .
+
+from typing import Any, Dict
+
+BINARY_SCHEMA: Dict[str, Any] = {"type": "string", "format": "binary"}
+
+NOTIFICATION_LINE_SCHEMA: Dict[str, Any] = {
+ "type": "object",
+ "description": "One notification event per line",
+ "properties": {
+ "action": {"type": "string"},
+ "event": {"description": "Event payload, its shape depends on the action"},
+ },
+ "required": ["action", "event"],
+ "additionalProperties": True,
+}
+
+PCAP_MEDIA_TYPE = "application/vnd.tcpdump.pcap"
+NDJSON_MEDIA_TYPE = "application/x-ndjson"
+OCTET_STREAM_MEDIA_TYPE = "application/octet-stream"
+ZIP_MEDIA_TYPE = "application/zip"
+
+
+def binary_response(media_type: str = OCTET_STREAM_MEDIA_TYPE, description: str = "Binary content") -> Dict[str, Any]:
+ return {"description": description, "content": {media_type: {"schema": BINARY_SCHEMA}}}
+
+
+def ndjson_response(description: str = "Stream of newline-delimited JSON notifications") -> Dict[str, Any]:
+ return {"description": description, "content": {NDJSON_MEDIA_TYPE: {"schema": NOTIFICATION_LINE_SCHEMA}}}
+
+
+def binary_request_body(media_type: str = OCTET_STREAM_MEDIA_TYPE) -> Dict[str, Any]:
+ return {"requestBody": {"required": True, "content": {media_type: {"schema": BINARY_SCHEMA}}}}
diff --git a/gns3server/api/routes/controller/controller.py b/gns3server/api/routes/controller/controller.py
index e1f03b7e3..c662333be 100644
--- a/gns3server/api/routes/controller/controller.py
+++ b/gns3server/api/routes/controller/controller.py
@@ -30,6 +30,7 @@ from websockets.exceptions import ConnectionClosed, WebSocketException
from gns3server import schemas
from gns3server.agent.web_wireshark.stats import collect_webwireshark_stats
+from gns3server.api.openapi import ndjson_response
from gns3server.config import Config
from gns3server.controller import Controller
from gns3server.controller.controller_error import ControllerError, ControllerForbiddenError
@@ -254,7 +255,12 @@ async def statistics() -> dict:
}
-@router.get("/notifications", dependencies=[Depends(get_current_active_user)])
+@router.get(
+ "/notifications",
+ response_class=StreamingResponse,
+ responses={200: ndjson_response()},
+ dependencies=[Depends(get_current_active_user)],
+)
async def controller_http_notifications(request: Request) -> StreamingResponse:
"""
Receive controller notifications about the controller from HTTP stream.
diff --git a/gns3server/api/routes/controller/images.py b/gns3server/api/routes/controller/images.py
index c7f6bf3f5..0f534e377 100644
--- a/gns3server/api/routes/controller/images.py
+++ b/gns3server/api/routes/controller/images.py
@@ -32,6 +32,7 @@ from starlette.requests import ClientDisconnect
import gns3server.db.models as models
from gns3server import schemas
+from gns3server.api.openapi import binary_request_body
from gns3server.compute.qemu import Qemu
from gns3server.config import Config
from gns3server.controller import Controller
@@ -210,6 +211,7 @@ async def get_images(
response_model=schemas.ImageUpload,
response_model_exclude_unset=True,
status_code=status.HTTP_201_CREATED,
+ openapi_extra=binary_request_body(),
dependencies=[Depends(has_privilege("Image.Allocate"))],
)
async def upload_image(
diff --git a/gns3server/api/routes/controller/links.py b/gns3server/api/routes/controller/links.py
index b7fa2b40d..053366046 100644
--- a/gns3server/api/routes/controller/links.py
+++ b/gns3server/api/routes/controller/links.py
@@ -31,6 +31,7 @@ from fastapi.responses import FileResponse, StreamingResponse
from gns3server import schemas
from gns3server.agent.web_wireshark.manager import WebWiresharkManager
+from gns3server.api.openapi import PCAP_MEDIA_TYPE, binary_response
from gns3server.controller import Controller
from gns3server.controller.controller_error import ControllerError
from gns3server.controller.link import _UNSET, Link
@@ -275,7 +276,12 @@ async def restart_wireshark(http_request: Request, link: Link = Depends(dep_link
return {"status": "restarted"}
-@router.get("/{link_id}/capture/stream", dependencies=[Depends(has_privilege("Link.Capture"))])
+@router.get(
+ "/{link_id}/capture/stream",
+ response_class=StreamingResponse,
+ responses={200: binary_response(PCAP_MEDIA_TYPE, "Packet capture stream")},
+ dependencies=[Depends(has_privilege("Link.Capture"))],
+)
async def stream_pcap(request: Request, link: Link = Depends(dep_link)) -> StreamingResponse:
"""
Stream the PCAP capture file from compute.
diff --git a/gns3server/api/routes/controller/nodes.py b/gns3server/api/routes/controller/nodes.py
index 8b22a9ce4..851175f73 100644
--- a/gns3server/api/routes/controller/nodes.py
+++ b/gns3server/api/routes/controller/nodes.py
@@ -32,6 +32,7 @@ from fastapi.responses import StreamingResponse
from fastapi.routing import APIRoute
from gns3server import schemas
+from gns3server.api.openapi import binary_request_body, binary_response
from gns3server.config import Config
from gns3server.controller import Controller
from gns3server.controller.controller_error import ControllerBadRequestError, ControllerForbiddenError
@@ -546,7 +547,12 @@ async def list_node_files(
return res.json
-@router.get("/{node_id}/files/{file_path:path}", dependencies=[Depends(has_privilege("Node.Audit"))])
+@router.get(
+ "/{node_id}/files/{file_path:path}",
+ response_class=StreamingResponse,
+ responses={200: binary_response(description="File content")},
+ dependencies=[Depends(has_privilege("Node.Audit"))],
+)
async def get_file(file_path: str, node: Node = Depends(dep_node)) -> Response:
"""
Return a file from the node directory.
@@ -587,6 +593,13 @@ async def get_file(file_path: str, node: Node = Depends(dep_node)) -> Response:
@router.post(
"/{node_id}/files/{file_path:path}",
status_code=status.HTTP_201_CREATED,
+ responses={
+ 201: {
+ "description": "File written",
+ "content": {"application/json": {"schema": {"type": "null"}}},
+ }
+ },
+ openapi_extra=binary_request_body(),
dependencies=[Depends(has_privilege("Node.Modify"))],
)
async def post_file(file_path: str, request: Request, node: Node = Depends(dep_node)):
diff --git a/gns3server/api/routes/controller/projects.py b/gns3server/api/routes/controller/projects.py
index fab2691d2..f7fb3eb72 100644
--- a/gns3server/api/routes/controller/projects.py
+++ b/gns3server/api/routes/controller/projects.py
@@ -40,6 +40,7 @@ from fastapi.responses import FileResponse, StreamingResponse
from websockets.exceptions import ConnectionClosed, WebSocketException
from gns3server import schemas
+from gns3server.api.openapi import ZIP_MEDIA_TYPE, binary_request_body, binary_response, ndjson_response
from gns3server.controller import Controller, marker_replay
from gns3server.controller.controller_error import ControllerBadRequestError, ControllerError
from gns3server.controller.export_project import export_project as export_controller_project
@@ -523,7 +524,12 @@ async def load_project(path: str = Body(..., embed=True)) -> schemas.Project:
return project.asdict()
-@router.get("/{project_id}/notifications", dependencies=[Depends(has_privilege("Project.Audit"))])
+@router.get(
+ "/{project_id}/notifications",
+ response_class=StreamingResponse,
+ responses={200: ndjson_response()},
+ dependencies=[Depends(has_privilege("Project.Audit"))],
+)
async def project_http_notifications(project_id: UUID) -> StreamingResponse:
"""
Receive project notifications about the controller from HTTP stream.
@@ -632,7 +638,12 @@ async def project_marker_ws_notifications(
log.warning(f"Error while sending marker event to WebSocket client: {e}")
-@router.get("/{project_id}/export", dependencies=[Depends(has_privilege("Project.Audit"))])
+@router.get(
+ "/{project_id}/export",
+ response_class=StreamingResponse,
+ responses={200: binary_response(ZIP_MEDIA_TYPE, "Portable project archive")},
+ dependencies=[Depends(has_privilege("Project.Audit"))],
+)
async def export_project(
project: Project = Depends(dep_project),
include_snapshots: bool = False,
@@ -718,6 +729,7 @@ async def export_project(
"/{project_id}/import",
status_code=status.HTTP_201_CREATED,
response_model=schemas.Project,
+ openapi_extra=binary_request_body(ZIP_MEDIA_TYPE),
dependencies=[Depends(has_privilege("Project.Allocate"))],
)
async def import_project(project_id: UUID, request: Request, name: Optional[str] = None) -> schemas.Project:
@@ -842,7 +854,12 @@ async def unlock_project(project: Project = Depends(dep_project)) -> None:
project.unlock()
-@router.get("/{project_id}/files/{file_path:path}", dependencies=[Depends(has_privilege("Project.Audit"))])
+@router.get(
+ "/{project_id}/files/{file_path:path}",
+ response_class=FileResponse,
+ responses={200: binary_response(description="File content")},
+ dependencies=[Depends(has_privilege("Project.Audit"))],
+)
async def get_file(file_path: str, project: Project = Depends(dep_project)) -> FileResponse:
"""
Return a file from a project.
@@ -864,7 +881,17 @@ async def get_file(file_path: str, project: Project = Depends(dep_project)) -> F
return FileResponse(path, media_type="application/octet-stream")
-@router.get("/{project_id}/gns3file", dependencies=[Depends(has_privilege("Project.Audit"))])
+@router.get(
+ "/{project_id}/gns3file",
+ response_class=FileResponse,
+ responses={
+ 200: {
+ "description": "Project topology file",
+ "content": {"application/json": {"schema": {"type": "object"}}},
+ }
+ },
+ dependencies=[Depends(has_privilege("Project.Audit"))],
+)
async def get_project_gns3_file(project: Project = Depends(dep_project)) -> FileResponse:
"""
Return the .gns3 topology file of a project.
@@ -882,6 +909,7 @@ async def get_project_gns3_file(project: Project = Depends(dep_project)) -> File
@router.post(
"/{project_id}/files/{file_path:path}",
status_code=status.HTTP_204_NO_CONTENT,
+ openapi_extra=binary_request_body(),
dependencies=[Depends(has_privilege("Project.Modify"))],
)
async def write_file(file_path: str, request: Request, project: Project = Depends(dep_project)) -> None:
diff --git a/gns3server/api/routes/controller/symbols.py b/gns3server/api/routes/controller/symbols.py
index 98d6a0c91..a9c80741a 100644
--- a/gns3server/api/routes/controller/symbols.py
+++ b/gns3server/api/routes/controller/symbols.py
@@ -27,6 +27,7 @@ from fastapi import APIRouter, Depends, Request, Response, status
from fastapi.responses import FileResponse
from gns3server import schemas
+from gns3server.api.openapi import binary_request_body, binary_response
from gns3server.controller import Controller
from gns3server.controller.controller_error import ControllerError, ControllerForbiddenError, ControllerNotFoundError
from gns3server.utils.get_resource import get_resource
@@ -53,7 +54,11 @@ def get_symbols() -> List[dict]:
@router.get(
"/{symbol_id:path}/raw",
- responses={404: {"model": schemas.ErrorMessage, "description": "Could not find symbol"}},
+ response_class=FileResponse,
+ responses={
+ 200: binary_response(description="Symbol file"),
+ 404: {"model": schemas.ErrorMessage, "description": "Could not find symbol"},
+ },
dependencies=[Depends(has_privilege("Symbol.Audit"))],
)
async def get_symbol(symbol_id: str, request: Request) -> Response:
@@ -112,6 +117,7 @@ def get_default_symbols() -> dict:
@router.post(
"/{symbol_id:path}/raw",
status_code=status.HTTP_204_NO_CONTENT,
+ openapi_extra=binary_request_body(),
dependencies=[Depends(has_privilege("Symbol.Allocate"))],
)
async def upload_symbol(symbol_id: str, request: Request) -> None:
diff --git a/tests/api/routes/test_openapi.py b/tests/api/routes/test_openapi.py
new file mode 100644
index 000000000..9ff6a2f6f
--- /dev/null
+++ b/tests/api/routes/test_openapi.py
@@ -0,0 +1,68 @@
+#
+# Copyright (C) 2026 GNS3 Technologies Inc.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see .
+
+import pytest
+from fastapi import FastAPI
+
+pytestmark = pytest.mark.asyncio
+
+STREAM_AND_BINARY_RESPONSES = [
+ ("/v3/notifications", "get", "200", "application/x-ndjson"),
+ ("/v3/projects/{project_id}/notifications", "get", "200", "application/x-ndjson"),
+ ("/v3/projects/{project_id}/export", "get", "200", "application/zip"),
+ ("/v3/projects/{project_id}/gns3file", "get", "200", "application/json"),
+ ("/v3/projects/{project_id}/files/{file_path}", "get", "200", "application/octet-stream"),
+ ("/v3/projects/{project_id}/nodes/{node_id}/files/{file_path}", "get", "200", "application/octet-stream"),
+ ("/v3/projects/{project_id}/nodes/{node_id}/files/{file_path}", "post", "201", "application/json"),
+ ("/v3/projects/{project_id}/links/{link_id}/capture/stream", "get", "200", "application/vnd.tcpdump.pcap"),
+ ("/v3/symbols/{symbol_id}/raw", "get", "200", "application/octet-stream"),
+]
+
+RAW_BODY_REQUESTS = [
+ ("/v3/images/upload/{image_path}", "application/octet-stream"),
+ ("/v3/symbols/{symbol_id}/raw", "application/octet-stream"),
+ ("/v3/projects/{project_id}/files/{file_path}", "application/octet-stream"),
+ ("/v3/projects/{project_id}/nodes/{node_id}/files/{file_path}", "application/octet-stream"),
+ ("/v3/projects/{project_id}/import", "application/zip"),
+]
+
+
+class TestOpenAPIBinaryRoutes:
+ @pytest.mark.parametrize("path, method, status_code, media_type", STREAM_AND_BINARY_RESPONSES)
+ async def test_response_content(
+ self, app: FastAPI, path: str, method: str, status_code: str, media_type: str
+ ) -> None:
+
+ content = app.openapi()["paths"][path][method]["responses"][status_code]["content"]
+ assert media_type in content
+ assert content[media_type]["schema"] != {}
+ if media_type != "application/json":
+ assert "application/json" not in content
+
+ @pytest.mark.parametrize("path, media_type", RAW_BODY_REQUESTS)
+ async def test_request_body(self, app: FastAPI, path: str, media_type: str) -> None:
+
+ request_body = app.openapi()["paths"][path]["post"]["requestBody"]
+ assert request_body["required"] is True
+ assert request_body["content"][media_type]["schema"] == {"type": "string", "format": "binary"}
+
+ async def test_binary_responses_use_binary_schema(self, app: FastAPI) -> None:
+
+ paths = app.openapi()["paths"]
+ for path, method, status_code, media_type in STREAM_AND_BINARY_RESPONSES:
+ if media_type in ("application/octet-stream", "application/zip", "application/vnd.tcpdump.pcap"):
+ schema = paths[path][method]["responses"][status_code]["content"][media_type]["schema"]
+ assert schema == {"type": "string", "format": "binary"}