fix(auth): update API token last_used for MCP Streamable HTTP requests

The MCP-specific auth handler (_StreamableHttpAuthHandler._auth_jwt)
performs its own JWT verification via verify_credentials() but did not
call _update_api_token_last_used_sync(), causing API token last_used
timestamps to remain stale when accessing virtual servers via
/servers/{id}/mcp.

Additionally, auth_method was hardcoded to "jwt" regardless of token
type, preventing TokenUsageMiddleware from recognising and logging
API token requests on MCP transport paths.

Changes:
- Detect API tokens (auth_provider == "api_token") and legacy tokens
  (DB JTI lookup) in _auth_jwt()
- Call _update_api_token_last_used_sync() for API tokens (rate-limited)
- Set auth_method dynamically ("api_token" vs "jwt")
- Propagate auth_method, jti, and user_email to ASGI scope state for
  TokenUsageMiddleware

Signed-off-by: kimsehwan96 <sktpghks138@gmail.com>

Author: kimsehwan96

mcpgateway/transports/streamablehttp_transport.py

@@ -3642,12 +3642,51 @@ async def _auth_jwt(self, *, token: str) -> bool:
                 else:
                     _record_mcp_auth_cache_event("team_membership_cache_hit")
 
+            # Detect API token auth and update last_used timestamp.
+            # API tokens carry user.auth_provider == "api_token" in their JWT payload.
+            # Legacy tokens (created before auth_provider was added) are detected via DB lookup.
+            _nested_user = user_payload.get("user", {})
+            _auth_provider = _nested_user.get("auth_provider") if isinstance(_nested_user, dict) else None
+            _is_api_token = _auth_provider == "api_token"
+
+            if not _is_api_token and jti and _auth_provider is None:
+                # Legacy API token fallback: only when auth_provider is absent (pre-auth_provider tokens).
+                # Tokens with an explicit auth_provider (email, oauth, saml, etc.) are never legacy API tokens.
+                try:
+                    # First-Party
+                    from mcpgateway.auth import _is_api_token_jti_sync  # pylint: disable=import-outside-toplevel
+
+                    _is_api_token = await asyncio.to_thread(_is_api_token_jti_sync, jti)
+                except Exception:
+                    pass  # Best-effort detection; default to "jwt" auth_method
+
+            resolved_auth_method = "api_token" if _is_api_token else "jwt"
+
+            # Update last_used timestamp for API tokens (rate-limited internally)
+            if _is_api_token and jti:
+                try:
+                    # First-Party
+                    from mcpgateway.auth import _update_api_token_last_used_sync  # pylint: disable=import-outside-toplevel
+
+                    await asyncio.to_thread(_update_api_token_last_used_sync, jti)
+                except Exception:
+                    logger.debug("Failed to update API token last_used in MCP auth for jti=...%s", jti[-8:] if jti else "")
+
+            # Propagate auth_method and jti into ASGI scope state so that
+            # TokenUsageMiddleware can recognise API token requests and log usage.
+            state = self.scope.setdefault("state", {})
+            state["auth_method"] = resolved_auth_method
+            if _is_api_token and jti:
+                state["jti"] = jti
+            if user_email:
+                state["user_email"] = user_email
+
             auth_user_ctx: dict[str, Any] = {
                 "email": user_email,
                 "teams": final_teams,
                 "is_authenticated": True,
                 "is_admin": is_admin,
-                "auth_method": "jwt",
+                "auth_method": resolved_auth_method,
                 "permission_is_admin": db_user_is_admin or is_admin,
                 "token_use": token_use,  # propagated for downstream RBAC (check_any_team)
             }
@@ -3667,7 +3706,7 @@ async def _auth_jwt(self, *, token: str) -> bool:
                 final_teams,
                 user_email=user_email,
                 is_admin=bool(db_user_is_admin or is_admin),
-                auth_method="jwt",
+                auth_method=resolved_auth_method,
                 team_name=trace_team_name,
             )
         except HTTPException:

tests/playwright/security/test_mcp_transport_auth_matrix.py

@@ -170,3 +170,80 @@ def test_websocket_auth_handshake_behavior(self):
 
         assert isinstance(response, str)
         assert "Parse error" in response or "jsonrpc" in response
+
+
+class TestApiTokenLastUsedViaMCP:
+    """Verify API token last_used is updated when accessing virtual servers via MCP Streamable HTTP."""
+
+    @pytest.fixture(autouse=True)
+    def _api_token(self, admin_api: APIRequestContext, playwright: Playwright):
+        """Create an API token via session JWT and expose its access_token and id."""
+        # admin_api uses a session JWT, which CAN create tokens
+        resp = admin_api.post("/tokens", data={"name": f"last-used-test-{uuid.uuid4().hex[:8]}", "expires_in_days": 1})
+        if resp.status == 404:
+            pytest.skip("/tokens endpoint unavailable")
+        assert resp.status in (200, 201), f"Failed to create token: {resp.status} {resp.text()}"
+        payload = resp.json()
+        self._access_token = payload["access_token"]
+        token_obj = payload.get("token", payload)
+        self._token_id = token_obj.get("id") or token_obj.get("token_id")
+        yield
+        # cleanup: revoke the token
+        with suppress(Exception):
+            admin_api.delete(f"/tokens/{self._token_id}")
+
+    def test_mcp_streamable_http_updates_last_used(self, admin_api: APIRequestContext, playwright: Playwright, public_server_id: str):
+        """Accessing /servers/{id}/mcp with an API token should update last_used."""
+        # 1. Check initial last_used (should be None for new token)
+        detail = admin_api.get(f"/tokens/{self._token_id}")
+        if detail.status == 404:
+            pytest.skip("Token detail endpoint unavailable")
+        initial_last_used = detail.json().get("last_used")
+
+        # 2. Make MCP Streamable HTTP request with the API token
+        token_ctx = _api_context(playwright, self._access_token)
+        try:
+            mcp_resp = token_ctx.post(
+                f"/servers/{public_server_id}/mcp",
+                data={"jsonrpc": "2.0", "id": "1", "method": "initialize", "params": {"protocolVersion": "2025-03-26", "capabilities": {}, "clientInfo": {"name": "e2e-test", "version": "1.0.0"}}},
+                headers={"Content-Type": "application/json", "Accept": "application/json, text/event-stream"},
+            )
+        finally:
+            token_ctx.dispose()
+
+        if mcp_resp.status == 404:
+            pytest.skip("Streamable HTTP endpoint unavailable")
+        assert mcp_resp.status != 401, f"API token auth rejected: {mcp_resp.text()}"
+
+        # 3. Verify last_used was updated
+        # Standard
+        import time
+
+        time.sleep(2)  # Allow async update to complete
+        detail2 = admin_api.get(f"/tokens/{self._token_id}")
+        updated_last_used = detail2.json().get("last_used")
+
+        assert updated_last_used is not None, f"last_used not updated after MCP access. Initial: {initial_last_used}, After: {updated_last_used}"
+
+    def test_mcp_request_records_token_usage_log(self, admin_api: APIRequestContext, playwright: Playwright, public_server_id: str):
+        """MCP requests with API tokens should appear in the token usage log."""
+        token_ctx = _api_context(playwright, self._access_token)
+        try:
+            token_ctx.post(
+                f"/servers/{public_server_id}/mcp",
+                data={"jsonrpc": "2.0", "id": "2", "method": "ping", "params": {}},
+                headers={"Content-Type": "application/json", "Accept": "application/json, text/event-stream"},
+            )
+        finally:
+            token_ctx.dispose()
+
+        # Standard
+        import time
+
+        time.sleep(2)
+        usage_resp = admin_api.get(f"/tokens/{self._token_id}/usage")
+        if usage_resp.status == 404:
+            pytest.skip("Token usage endpoint unavailable")
+        assert usage_resp.status == 200, f"Usage stats failed: {usage_resp.status}"
+        total = usage_resp.json().get("total_requests", 0)
+        assert total > 0, f"Token usage log should have entries after MCP access, got {total}"