feat(MCP): OAuth 2.0 support for HTTP transport

mcp/pyproject.toml

@@ -6,22 +6,22 @@ authors = [{ name = "Flagsmith", email = "support@flagsmith.com" }]
 readme = "README.md"
 requires-python = ">=3.10"
 dependencies = [
-    "fastmcp>=3.3.1,<4.0.0",  # Base MCP functionality
-    "pydantic-settings>=2.0.0,<3.0.0",  # Environment-driven configuration
+    "fastmcp>=3.3.1,<4.0.0",           # Base MCP functionality
+    "pydantic-settings>=2.0.0,<3.0.0", # Environment-driven configuration
 ]
 
 [project.scripts]
 flagsmith-mcp = "flagsmith_mcp.server:run"
 
 [dependency-groups]
 dev = [
-    "mypy>=2.1.0,<3.0.0",  # Static type checking
-    "openapi-pydantic>=0.5.0,<1.0.0",  # Build OpenAPI specs as fixtures
-    "pytest>=9.0.3,<10.0.0",  # Run tests
-    "pytest-asyncio>=1.3.0,<2.0.0",  # Run asynchronous tests
-    "pytest-cov>=7.0.0,<8.0.0",  # Measure test coverage
-    "pytest-httpx>=0.35.0,<1.0.0",  # Mock HTTP interactions
-    "ruff>=0.15.12,<0.16.0",  # Lint and format
+    "mypy>=2.1.0,<3.0.0",             # Static type checking
+    "openapi-pydantic>=0.5.0,<1.0.0", # Build OpenAPI specs as fixtures
+    "pytest>=9.0.3,<10.0.0",          # Run tests
+    "pytest-asyncio>=1.3.0,<2.0.0",   # Run asynchronous tests
+    "pytest-cov>=7.0.0,<8.0.0",       # Measure test coverage
+    "respx>=0.22,<1.0",               # Mock HTTP interactions
+    "ruff>=0.15.12,<0.16.0",          # Lint and format
 ]
 
 [build-system]

mcp/src/flagsmith_mcp/config.py

@@ -21,6 +21,11 @@ class Settings(BaseSettings):
         default="http",
     )
     """MCP transport to use."""
+    mcp_server_url: str = Field(
+        default="http://127.0.0.1:8000",
+    )
+    """Public base URL of this MCP server, advertised in OAuth protected-resource
+    metadata. Override for HTTP deployments behind a proxy/public hostname."""
 
     @model_validator(mode="after")
     def validate_stdio_token(self) -> "Settings":

mcp/src/flagsmith_mcp/constants.py

@@ -1,2 +1,3 @@
 # TODO: consume a version-controlled schema — https://github.com/Flagsmith/flagsmith/issues/7669
 OPENAPI_SPEC_URL = "https://api.flagsmith.com/api/v1/swagger.json"
+OAUTH_SCOPES = ["mcp"]

mcp/src/flagsmith_mcp/oauth.py

@@ -0,0 +1,65 @@
+from fastmcp.server.auth.auth import AccessToken, RemoteAuthProvider, TokenVerifier
+from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
+from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
+from pydantic import AnyHttpUrl
+from starlette.authentication import AuthCredentials, AuthenticationBackend
+from starlette.middleware import Middleware
+from starlette.middleware.authentication import AuthenticationMiddleware
+from starlette.requests import HTTPConnection
+
+from flagsmith_mcp.constants import OAUTH_SCOPES
+
+
+class _AnySchemeBackend(AuthenticationBackend):
+    """Authenticates a request on the mere presence of an `Authorization`
+    header, regardless of scheme (`Bearer` OAuth token or `Api-Key`). The
+    credential is forwarded upstream verbatim and validated by the API."""
+
+    async def authenticate(
+        self, conn: HTTPConnection
+    ) -> tuple[AuthCredentials, AuthenticatedUser] | None:
+        if not (
+            header := next(
+                (
+                    conn.headers.get(k)
+                    for k in conn.headers
+                    if k.lower() == "authorization"
+                ),
+                None,
+            )
+        ):
+            return None
+
+        token = header.split(" ", 1)[-1]
+        access = AccessToken(
+            token=token, client_id="flagsmith-mcp", scopes=OAUTH_SCOPES
+        )
+        return AuthCredentials(OAUTH_SCOPES), AuthenticatedUser(access)
+
+
+class FlagsmithResourceAuth(RemoteAuthProvider):
+    """OAuth 2.0 protected resource for HTTP transport.
+
+    Serves Protected Resource Metadata (RFC 9728) pointing at the Flagsmith
+    authorization server and returns 401 + `WWW-Authenticate` when a request
+    carries no credential, so MCP clients can discover and complete the OAuth
+    flow. Any `Authorization` header is accepted and passed through — the API
+    validates it (no introspection here).
+    """
+
+    def __init__(self, *, resource_url: str, authorization_server: str) -> None:
+        token_verifier = TokenVerifier(
+            required_scopes=[]
+        )  # never consulted — introspection done by Core API.
+        super().__init__(
+            token_verifier=token_verifier,
+            authorization_servers=[AnyHttpUrl(authorization_server)],
+            base_url=resource_url,
+            scopes_supported=OAUTH_SCOPES,
+        )
+
+    def get_middleware(self) -> list[Middleware]:
+        return [
+            Middleware(AuthenticationMiddleware, backend=_AnySchemeBackend()),
+            Middleware(AuthContextMiddleware),
+        ]

mcp/src/flagsmith_mcp/server.py

@@ -9,6 +9,7 @@
 
 from flagsmith_mcp import config, constants
 from flagsmith_mcp.auth import FlagsmithAuth
+from flagsmith_mcp.oauth import FlagsmithResourceAuth
 
 ROUTE_MAPS = [
     RouteMap(tags={"mcp"}, mcp_type=MCPType.TOOL),
@@ -40,6 +41,16 @@ def _fetch_spec() -> dict[str, Any]:
 
 
 def create_server(settings: config.Settings) -> FastMCP[None]:
+    # OAuth discovery is the credential fallback for HTTP transport: only when
+    # the server holds no static token does it advertise the AS and gate on a
+    # missing Authorization header. Otherwise (stdio, static token, or a
+    # forwarded --header) it's pure pass-through.
+    auth = None
+    if settings.transport == "http" and settings.flagsmith_api_token is None:
+        auth = FlagsmithResourceAuth(
+            resource_url=settings.mcp_server_url,
+            authorization_server=settings.flagsmith_api_url,
+        )
     return FastMCP.from_openapi(
         openapi_spec=_fetch_spec(),
         client=httpx.AsyncClient(
@@ -50,6 +61,7 @@ def create_server(settings: config.Settings) -> FastMCP[None]:
         route_maps=ROUTE_MAPS,
         mcp_component_fn=_customise,
         validate_output=False,  # TODO https://github.com/Flagsmith/flagsmith/issues/7679
+        auth=auth,
     )
 
 

mcp/tests/integration/conftest.py

@@ -1,19 +1,19 @@
 from collections.abc import AsyncIterator
-from typing import Any
 
 import openapi_pydantic as openapi
 import pytest
 from fastmcp import Client, FastMCP
 from fastmcp.client.transports import FastMCPTransport
+from respx import MockRouter
 
-from flagsmith_mcp import config
+from flagsmith_mcp import config, constants
 from flagsmith_mcp import server as server_module
 
 
 @pytest.fixture
-def openapi_spec() -> dict[str, Any]:
+def openapi_spec() -> openapi.OpenAPI:
     ok = openapi.Response(description="OK")
-    spec = openapi.OpenAPI(
+    return openapi.OpenAPI(
         info=openapi.Info(title="Flagsmith API", version="1.0.0"),
         paths={
             "/environments/": openapi.PathItem(
@@ -35,16 +35,19 @@ def openapi_spec() -> dict[str, Any]:
             ),
         },
     )
-    return spec.model_dump(by_alias=True, exclude_none=True, mode="json")
+
+
+@pytest.fixture(autouse=True)
+def openapi_spec_mock(respx_mock: MockRouter, openapi_spec: openapi.OpenAPI) -> None:
+    # create_server fetches the OpenAPI spec over HTTP; mock that call (respx
+    # leaves the in-memory ASGI transport used by the tests untouched).
+    respx_mock.get(constants.OPENAPI_SPEC_URL).respond(
+        json=openapi_spec.model_dump(by_alias=True, exclude_none=True, mode="json")
+    )
 
 
 @pytest.fixture
-def server(
-    monkeypatch: pytest.MonkeyPatch,
-    openapi_spec: dict[str, Any],
-) -> FastMCP:
-    monkeypatch.setenv("FLAGSMITH_API_URL", "https://flagsmith.example.com")
-    monkeypatch.setattr(server_module, "_fetch_spec", lambda: openapi_spec)
+def server() -> FastMCP:
     return server_module.create_server(config.Settings())
 
 

mcp/tests/integration/test_oauth.py

@@ -0,0 +1,98 @@
+from collections.abc import AsyncIterator
+from typing import Callable
+
+import httpx
+import pytest
+from fastmcp import FastMCP
+
+from flagsmith_mcp import config
+from flagsmith_mcp import server as server_module
+
+HTTPClientFactoryFixture = Callable[[FastMCP], AsyncIterator[httpx.AsyncClient]]
+
+
+PRM_PATH = "/.well-known/oauth-protected-resource/mcp"
+
+
+@pytest.fixture
+def http_client_factory() -> HTTPClientFactoryFixture:
+    async def factory(server: FastMCP) -> AsyncIterator[httpx.AsyncClient]:
+        transport = httpx.ASGITransport(app=server.http_app())
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://testserver"
+        ) as connected:
+            yield connected
+
+    return factory
+
+
+@pytest.fixture
+async def http_client(
+    server: FastMCP,
+    http_client_factory: HTTPClientFactoryFixture,
+) -> AsyncIterator[httpx.AsyncClient]:
+    async for client in http_client_factory(server):
+        yield client
+
+
+@pytest.fixture
+def server_with_flagsmith_api_token() -> FastMCP:
+    return server_module.create_server(
+        config.Settings(flagsmith_api_token="secret.token")
+    )
+
+
+@pytest.fixture
+async def http_client_with_flagsmith_api_token(
+    server_with_flagsmith_api_token: FastMCP,
+    http_client_factory: HTTPClientFactoryFixture,
+) -> AsyncIterator[httpx.AsyncClient]:
+    async for client in http_client_factory(server_with_flagsmith_api_token):
+        yield client
+
+
+async def test_http_no_token__serves_protected_resource_metadata(
+    http_client: httpx.AsyncClient,
+) -> None:
+    # Given OAuth discovery is active (server fixture: http, no static token)
+    response = await http_client.get(PRM_PATH)
+
+    # Then it advertises the Flagsmith AS and the mcp scope (RFC 9728)
+    assert response.status_code == 200
+    body = response.json()
+    assert body["authorization_servers"] == ["https://api.flagsmith.com/"]
+    assert body["scopes_supported"] == ["mcp"]
+
+
+async def test_http_no_token__missing_authorization__401_points_at_prm(
+    http_client: httpx.AsyncClient,
+) -> None:
+    # When a request reaches the MCP endpoint with no credential
+    response = await http_client.get("/mcp")
+
+    # Then it 401s (no API round-trip) and points the client at the PRM
+    assert response.status_code == 401
+    assert PRM_PATH in response.headers["www-authenticate"]
+
+
+async def test_http_no_token__non_bearer_credential__accepted_by_gate(
+    http_client: httpx.AsyncClient,
+) -> None:
+    # When a request carries a non-Bearer (Api-Key) credential
+    response = await http_client.get(
+        PRM_PATH, headers={"Authorization": "Api-Key ser.secret"}
+    )
+
+    # Then the gate authenticates it (scheme-agnostic); end-to-end pass-through
+    # to the API is exercised against live SaaS.
+    assert response.status_code == 200
+
+
+async def test_http_static_token__no_oauth_resource(
+    http_client_with_flagsmith_api_token: httpx.AsyncClient,
+) -> None:
+    # Given a static token (pure pass-through, OAuth disabled)
+    response = await http_client_with_flagsmith_api_token.get(PRM_PATH)
+
+    # Then no protected-resource metadata is served
+    assert response.status_code == 404

mcp/tests/unit/test_server.py

@@ -4,7 +4,7 @@
 import pytest
 from fastmcp import Client
 from mcp.types import ToolAnnotations
-from pytest_httpx import HTTPXMock
+from respx import MockRouter
 
 from flagsmith_mcp import config, constants, server
 
@@ -63,7 +63,7 @@
     ],
 )
 async def test_create_server__mcp_route__annotates_tool_per_method(
-    httpx_mock: HTTPXMock,
+    respx_mock: MockRouter,
     method: str,
     expected: ToolAnnotations,
 ) -> None:
@@ -77,9 +77,8 @@ async def test_create_server__mcp_route__annotates_tool_per_method(
         info=openapi.Info(title="Flagsmith API", version="1.0.0"),
         paths={"/things/": openapi.PathItem.model_validate({method: operation})},
     )
-    httpx_mock.add_response(
-        url=constants.OPENAPI_SPEC_URL,
-        json=spec.model_dump(by_alias=True, exclude_none=True, mode="json"),
+    respx_mock.get(constants.OPENAPI_SPEC_URL).respond(
+        json=spec.model_dump(by_alias=True, exclude_none=True, mode="json")
     )
 
     # When
@@ -91,7 +90,7 @@ async def test_create_server__mcp_route__annotates_tool_per_method(
 
 
 async def test_create_server__untagged_route__excluded_from_tools(
-    httpx_mock: HTTPXMock,
+    respx_mock: MockRouter,
 ) -> None:
     # Given a spec with one mcp-tagged route and one untagged route
     spec = openapi.OpenAPI(
@@ -112,9 +111,8 @@ async def test_create_server__untagged_route__excluded_from_tools(
             ),
         },
     )
-    httpx_mock.add_response(
-        url=constants.OPENAPI_SPEC_URL,
-        json=spec.model_dump(by_alias=True, exclude_none=True, mode="json"),
+    respx_mock.get(constants.OPENAPI_SPEC_URL).respond(
+        json=spec.model_dump(by_alias=True, exclude_none=True, mode="json")
     )
 
     # When