From 19e7533cd77a54147f2f5867c328ef872b4b72f8 Mon Sep 17 00:00:00 2001
From: YueGuobin <yueguobin@outlook.com>
Date: Thu, 4 Jun 2026 22:38:49 +0800
Subject: [PATCH] feat: support Authorization header and query param for MCP
 token

- SSE endpoint supports both Authorization: Bearer header and ?token= query param
- Claude Code can use headers (no URL exposure)
- Claude Desktop (EventSource) can use ?token= URL param
---
 gns3server/api/routes/mcp/__init__.py | 36 +++++++++++++++++++--------
 1 file changed, 25 insertions(+), 11 deletions(-)

diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py
index e50931872..6dcfbe56e 100644
--- a/gns3server/api/routes/mcp/__init__.py
+++ b/gns3server/api/routes/mcp/__init__.py
@@ -171,23 +171,37 @@ async def get_project_stats(project_id: str) -> list[dict[str, Any]]:
 # ── Auth‑wrapped SSE app ──────────────────────────────────────────────
 
 def _make_auth_wrapper(sse_app):
-    """Wrap the SSE app with JWT validation from ?token= query parameter.
+    """Wrap the SSE app with JWT validation.
 
-    The wrapper intercepts GET requests (SSE connections), validates the
-    JWT token, and stores it in a context variable so tool handlers can
-    use it to call the GNS3 REST API.  POST messages are passed through
-    unchanged (they are authenticated by their session association).
+    Supports two ways to pass the token (checked in order):
+      1. Authorization: Bearer <jwt> header
+      2. ?token=<jwt> query parameter
+
+    POST messages are passed through (authenticated by their session).
     """
 
     async def auth_wrapper(scope, receive, send):
         if scope["type"] == "http" and scope["method"] == "GET":
-            params = parse_qs(scope.get("query_string", b"").decode())
-            tokens = params.get("token", [])
-            if not tokens or not await _validate_token(tokens[0]):
+            token = None
+
+            # 1. Try Authorization header first
+            headers = dict(scope.get("headers", []))
+            auth_header = headers.get(b"authorization", b"").decode()
+            if auth_header.startswith("Bearer "):
+                token = auth_header[7:]
+
+            # 2. Fall back to ?token= query param
+            if not token:
+                params = parse_qs(scope.get("query_string", b"").decode())
+                tokens = params.get("token", [])
+                if tokens:
+                    token = tokens[0]
+
+            if not token or not await _validate_token(token):
                 response = Response("Missing or invalid token", status_code=401)
                 await response(scope, receive, send)
                 return
-            _jwt_token_var.set(tokens[0])
+            _jwt_token_var.set(token)
         await sse_app(scope, receive, send)
 
     return auth_wrapper
@@ -206,9 +220,9 @@ async def mcp_root():
         "version": "1.0.0",
         "protocol": "Model Context Protocol",
         "transport": "SSE",
-        "authentication": "?token=<jwt>",
+        "authentication": ["Authorization: Bearer <jwt>", "?token=<jwt>"],
         "endpoints": {
-            "sse": "/v3/mcp/transport/sse?token=<jwt>",
+            "sse": "/v3/mcp/transport/sse",
             "messages": "/v3/mcp/transport/messages/",
         },
     }