feat: open-api-specs-describe-oauth

api/api/openapi.py

@@ -11,6 +11,8 @@
 from rest_framework.request import Request
 from typing_extensions import is_typeddict
 
+from oauth2_metadata.dataclasses import OAuthConfig
+
 
 def append_meta(schema: dict[str, Any], meta: dict[str, Any]) -> dict[str, Any]:
     """
@@ -73,21 +75,21 @@ class MCPSchemaGenerator(SchemaGenerator):
     """
 
     MCP_TAG = "mcp"
-    MCP_SERVER_URL = "https://api.flagsmith.com"
 
     def get_schema(
         self, request: Request | None = None, public: bool = False
     ) -> dict[str, Any]:
+        oauth = OAuthConfig.from_settings()
         schema = super().get_schema(request, public)
         schema["paths"] = self._filter_paths(schema.get("paths", {}))
-        schema = self._update_security_for_mcp(schema)
+        schema = self._update_security_for_mcp(schema, oauth)
         schema.pop("$schema", None)
         info = schema.pop("info").copy()
         info["title"] = "mcp_openapi"
         return {
             "openapi": schema.pop("openapi"),
             "info": info,
-            "servers": [{"url": self.MCP_SERVER_URL}],
+            "servers": [{"url": oauth.api_url}],
             **schema,
         }
 
@@ -121,19 +123,34 @@ def _transform_for_mcp(self, operation: dict[str, Any]) -> dict[str, Any]:
         operation.pop("security", None)
         return operation
 
-    def _update_security_for_mcp(self, schema: dict[str, Any]) -> dict[str, Any]:
-        """Update security schemes for MCP (Organisation API Key)."""
+    def _update_security_for_mcp(
+        self, schema: dict[str, Any], oauth: OAuthConfig
+    ) -> dict[str, Any]:
+        """Update security schemes for MCP (OAuth + API Key fallback)."""
         schema = schema.copy()
         schema["components"] = schema.get("components", {}).copy()
         schema["components"]["securitySchemes"] = {
+            "oauth2": {
+                "type": "oauth2",
+                "flows": {
+                    "authorizationCode": {
+                        "authorizationUrl": f"{oauth.frontend_url}/oauth/authorize/",
+                        "tokenUrl": f"{oauth.api_url}/o/token/",
+                        "scopes": oauth.scopes,
+                    },
+                },
+            },
             "TOKEN_AUTH": {
                 "type": "apiKey",
                 "in": "header",
                 "name": "Authorization",
                 "description": "Organisation API Key. Format: Api-Key <key>",
             },
         }
-        schema["security"] = [{"TOKEN_AUTH": []}]
+        schema["security"] = [
+            {"oauth2": list(oauth.scopes.keys())},
+            {"TOKEN_AUTH": []},
+        ]
         return schema
 
 

api/oauth2_metadata/dataclasses.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from django.conf import settings
+
+
+@dataclass(frozen=True)
+class OAuthConfig:
+    """Base OAuth configuration derived from Django settings."""
+
+    api_url: str
+    frontend_url: str
+    scopes: dict[str, str]
+
+    @classmethod
+    def from_settings(cls) -> OAuthConfig:
+        oauth2_provider: dict[str, Any] = settings.OAUTH2_PROVIDER
+        return cls(
+            api_url=settings.FLAGSMITH_API_URL.rstrip("/"),
+            frontend_url=settings.FLAGSMITH_FRONTEND_URL.rstrip("/"),
+            scopes=oauth2_provider.get("SCOPES", {}),
+        )

api/oauth2_metadata/views.py

@@ -1,7 +1,6 @@
 from typing import Any
 from urllib.parse import urlencode, urlparse, urlunparse
 
-from django.conf import settings
 from django.http import HttpRequest, JsonResponse, QueryDict
 from django.views.decorators.csrf import csrf_exempt
 from django.views.decorators.http import require_GET
@@ -17,6 +16,7 @@
 from rest_framework.throttling import ScopedRateThrottle
 from rest_framework.views import APIView
 
+from oauth2_metadata.dataclasses import OAuthConfig
 from oauth2_metadata.serializers import DCRRequestSerializer, OAuthConsentSerializer
 from oauth2_metadata.services import create_oauth2_application
 
@@ -25,19 +25,16 @@
 @require_GET
 def authorization_server_metadata(request: HttpRequest) -> JsonResponse:
     """RFC 8414 OAuth 2.0 Authorization Server Metadata."""
-    api_url: str = settings.FLAGSMITH_API_URL.rstrip("/")
-    frontend_url: str = settings.FLAGSMITH_FRONTEND_URL.rstrip("/")
-    oauth2_settings: dict[str, Any] = settings.OAUTH2_PROVIDER
-    scopes: dict[str, str] = oauth2_settings.get("SCOPES", {})
+    oauth = OAuthConfig.from_settings()
 
     metadata = {
-        "issuer": api_url,
-        "authorization_endpoint": f"{frontend_url}/oauth/authorize/",
-        "token_endpoint": f"{api_url}/o/token/",
-        "registration_endpoint": f"{api_url}/o/register/",
-        "revocation_endpoint": f"{api_url}/o/revoke_token/",
-        "introspection_endpoint": f"{api_url}/o/introspect/",
-        "scopes_supported": list(scopes.keys()),
+        "issuer": oauth.api_url,
+        "authorization_endpoint": f"{oauth.frontend_url}/oauth/authorize/",
+        "token_endpoint": f"{oauth.api_url}/o/token/",
+        "registration_endpoint": f"{oauth.api_url}/o/register/",
+        "revocation_endpoint": f"{oauth.api_url}/o/revoke_token/",
+        "introspection_endpoint": f"{oauth.api_url}/o/introspect/",
+        "scopes_supported": list(oauth.scopes.keys()),
         "response_types_supported": ["code"],
         "grant_types_supported": ["authorization_code", "refresh_token"],
         "code_challenge_methods_supported": ["S256"],

api/tests/unit/api/test_mcp_openapi.py

@@ -2,9 +2,11 @@
 from unittest.mock import MagicMock
 
 import pytest
+from pytest_django.fixtures import SettingsWrapper
 
 from api.openapi import MCPSchemaGenerator, SchemaGenerator
 from api.openapi_views import CustomSpectacularJSONAPIView, CustomSpectacularYAMLAPIView
+from oauth2_metadata.dataclasses import OAuthConfig
 
 
 def test_mcp_filter_paths__mcp_tagged_operation__includes_path() -> None:
@@ -143,10 +145,15 @@ def test_mcp_transform_for_mcp__security_present__removes_operation_level_securi
     assert "security" not in transformed
 
 
-def test_mcp_update_security_for_mcp__existing_scheme__sets_token_auth_security() -> (
+def test_mcp_update_security_for_mcp__existing_scheme__sets_oauth_and_token_auth() -> (
     None
 ):
     # Given
+    oauth = OAuthConfig(
+        api_url="https://api.flagsmith.example.com",
+        frontend_url="https://app.flagsmith.example.com",
+        scopes={"mcp": "MCP access"},
+    )
     schema: dict[str, Any] = {
         "components": {
             "securitySchemes": {
@@ -158,12 +165,21 @@ def test_mcp_update_security_for_mcp__existing_scheme__sets_token_auth_security(
     generator = MCPSchemaGenerator()
 
     # When
-    updated = generator._update_security_for_mcp(schema)
+    updated = generator._update_security_for_mcp(schema, oauth)
 
     # Then
-    assert "TOKEN_AUTH" in updated["components"]["securitySchemes"]
-    assert updated["security"] == [{"TOKEN_AUTH": []}]
     assert "Private" not in updated["components"]["securitySchemes"]
+    assert "TOKEN_AUTH" in updated["components"]["securitySchemes"]
+    oauth2_scheme = updated["components"]["securitySchemes"]["oauth2"]
+    assert oauth2_scheme["type"] == "oauth2"
+    auth_code_flow = oauth2_scheme["flows"]["authorizationCode"]
+    assert (
+        auth_code_flow["authorizationUrl"]
+        == "https://app.flagsmith.example.com/oauth/authorize/"
+    )
+    assert auth_code_flow["tokenUrl"] == "https://api.flagsmith.example.com/o/token/"
+    assert auth_code_flow["scopes"] == {"mcp": "MCP access"}
+    assert updated["security"] == [{"oauth2": ["mcp"]}, {"TOKEN_AUTH": []}]
 
 
 @pytest.mark.parametrize(
@@ -267,24 +283,32 @@ def test_mcp_schema__full_schema_generated__includes_expected_endpoints_only() -
     assert "/api/v1/users/" not in paths
 
 
-def test_mcp_schema__full_schema_generated__includes_https_server() -> None:
+def test_mcp_schema__full_schema_generated__includes_server_from_settings(
+    settings: SettingsWrapper,
+) -> None:
     # Given
+    settings.FLAGSMITH_API_URL = "https://flagsmith.example.com/"
     generator = MCPSchemaGenerator()
 
     # When
     schema = generator.get_schema(request=None, public=True)
 
     # Then
-    assert schema["servers"] == [{"url": "https://api.flagsmith.com"}]
+    assert schema["servers"] == [{"url": "https://flagsmith.example.com"}]
 
 
-def test_mcp_schema__full_schema_generated__includes_token_auth_security() -> None:
+def test_mcp_schema__full_schema_generated__includes_oauth_and_token_auth(
+    settings: SettingsWrapper,
+) -> None:
     # Given
+    settings.FLAGSMITH_API_URL = "https://api.flagsmith.example.com/"
+    settings.FLAGSMITH_FRONTEND_URL = "https://app.flagsmith.example.com/"
     generator = MCPSchemaGenerator()
 
     # When
     schema = generator.get_schema(request=None, public=True)
 
     # Then
     assert "TOKEN_AUTH" in schema["components"]["securitySchemes"]
-    assert schema["security"] == [{"TOKEN_AUTH": []}]
+    assert "oauth2" in schema["components"]["securitySchemes"]
+    assert schema["security"] == [{"oauth2": ["mcp"]}, {"TOKEN_AUTH": []}]