feat(MCP): Authentication, configuration and HTTP transport (#7680)

Author: khvn26

mcp/pyproject.toml

@@ -4,7 +4,7 @@ version = "0.1.0"
 description = "An MCP server connecting to the Flagsmith API."
 authors = [{ name = "Flagsmith", email = "support@flagsmith.com" }]
 readme = "README.md"
-requires-python = ">=3.14,<3.15"
+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
@@ -37,7 +37,7 @@ branch = true
 
 [tool.ruff]
 line-length = 88
-target-version = "py314"
+target-version = "py310"
 
 [tool.ruff.lint]
 # Establish parity with flake8 + isort

mcp/src/flagsmith_mcp/auth.py

@@ -0,0 +1,28 @@
+from collections.abc import Generator
+
+import httpx
+from fastmcp.server.dependencies import get_http_headers
+
+
+class FlagsmithAuth(httpx.Auth):
+    def __init__(self, global_master_api_key: str | None = None) -> None:
+        self._global_master_api_key = global_master_api_key
+
+    def auth_flow(
+        self, request: httpx.Request
+    ) -> Generator[httpx.Request, httpx.Response, None]:
+        if "authorization" not in request.headers:
+            # Prefer the caller's forwarded MCP `--header`; fall back to the
+            # server's own static token (the only credential under stdio).
+            forwarded = get_http_headers(include={"authorization"})
+            if (
+                authorization := forwarded.get("authorization")
+                or self._global_authorization_value()
+            ):
+                request.headers["authorization"] = authorization
+        yield request
+
+    def _global_authorization_value(self) -> str | None:
+        if self._global_master_api_key:
+            return f"Api-Key {self._global_master_api_key}"
+        return None

mcp/src/flagsmith_mcp/config.py

@@ -1,6 +1,6 @@
 from typing import Literal
 
-from pydantic import Field
+from pydantic import Field, model_validator
 from pydantic_settings import BaseSettings
 
 Transport = Literal["http", "stdio"]
@@ -16,8 +16,18 @@ class Settings(BaseSettings):
     flagsmith_api_token: str | None = Field(
         default=None,
     )
-    """Flagsmith Master API Key. Required for stdio transport, optional for http transport (caller-supplied credential will be used instead)."""
+    """Flagsmith Master API Key. Required for stdio transport."""
     transport: Transport = Field(
         default="http",
     )
     """MCP transport to use."""
+
+    @model_validator(mode="after")
+    def validate_stdio_token(self) -> "Settings":
+        # stdio has no inbound request to forward a credential from, so the
+        # server must hold its own master API key.
+        if self.transport == "stdio" and self.flagsmith_api_token is None:
+            raise ValueError(
+                "FLAGSMITH_API_TOKEN is required when TRANSPORT is 'stdio'"
+            )
+        return self

mcp/src/flagsmith_mcp/server.py

@@ -8,6 +8,7 @@
 from mcp.types import ToolAnnotations
 
 from flagsmith_mcp import config, constants
+from flagsmith_mcp.auth import FlagsmithAuth
 
 ROUTE_MAPS = [
     RouteMap(tags={"mcp"}, mcp_type=MCPType.TOOL),
@@ -41,10 +42,14 @@ def _fetch_spec() -> dict[str, Any]:
 def create_server(settings: config.Settings) -> FastMCP[None]:
     return FastMCP.from_openapi(
         openapi_spec=_fetch_spec(),
-        client=httpx.AsyncClient(base_url=settings.flagsmith_api_url),
+        client=httpx.AsyncClient(
+            base_url=settings.flagsmith_api_url,
+            auth=FlagsmithAuth(settings.flagsmith_api_token),
+        ),
         name="Flagsmith",
         route_maps=ROUTE_MAPS,
         mcp_component_fn=_customise,
+        validate_output=False,  # TODO https://github.com/Flagsmith/flagsmith/issues/7679
     )
 
 

mcp/tests/unit/test_auth.py

@@ -0,0 +1,78 @@
+import httpx
+from fastmcp.server.http import set_http_request
+from starlette.requests import Request
+
+from flagsmith_mcp import auth
+
+
+def test_flagsmith_auth__header_in_http_request__forwards_to_upstream() -> None:
+    # Given an inbound HTTP request carrying an Authorization header, exactly as
+    # FastMCP's RequestContextMiddleware sets it under HTTP transport (--header)
+    inbound = Request(
+        {"type": "http", "headers": [(b"authorization", b"Api-Key caller")]}
+    )
+    upstream = httpx.Request("GET", "https://api.flagsmith.com/api/v1/organisations/")
+
+    # When the upstream auth flow runs within that request context
+    with set_http_request(inbound):
+        next(auth.FlagsmithAuth().auth_flow(upstream))
+
+    # Then the real get_http_headers picks it up and forwards it
+    assert upstream.headers["authorization"] == "Api-Key caller"
+
+
+def test_flagsmith_auth__no_http_request__leaves_upstream_unchanged() -> None:
+    # Given no active HTTP request (e.g. stdio transport)
+    upstream = httpx.Request("GET", "https://api.flagsmith.com/api/v1/organisations/")
+
+    # When
+    next(auth.FlagsmithAuth().auth_flow(upstream))
+
+    # Then nothing is forwarded
+    assert "authorization" not in upstream.headers
+
+
+def test_flagsmith_auth__static_token_no_http_request__uses_api_key() -> None:
+    # Given a static token and no active HTTP request (e.g. stdio transport)
+    upstream = httpx.Request("GET", "https://api.flagsmith.com/api/v1/organisations/")
+
+    # When
+    next(auth.FlagsmithAuth("ser.secret").auth_flow(upstream))
+
+    # Then the server's own token is sent as an Api-Key credential
+    assert upstream.headers["authorization"] == "Api-Key ser.secret"
+
+
+def test_flagsmith_auth__forwarded_header__wins_over_static_token() -> None:
+    # Given both a forwarded caller header and a static token
+    inbound = Request(
+        {"type": "http", "headers": [(b"authorization", b"Api-Key caller")]}
+    )
+    upstream = httpx.Request("GET", "https://api.flagsmith.com/api/v1/organisations/")
+
+    # When
+    with set_http_request(inbound):
+        next(auth.FlagsmithAuth("ser.secret").auth_flow(upstream))
+
+    # Then the caller's forwarded credential takes precedence
+    assert upstream.headers["authorization"] == "Api-Key caller"
+
+
+def test_flagsmith_auth__upstream_already_authorized__does_not_override() -> None:
+    # Given an upstream request that already carries a credential, and an inbound
+    # header that differs
+    inbound = Request(
+        {"type": "http", "headers": [(b"authorization", b"Api-Key context")]}
+    )
+    upstream = httpx.Request(
+        "GET",
+        "https://api.flagsmith.com/api/v1/organisations/",
+        headers={"authorization": "Api-Key static"},
+    )
+
+    # When
+    with set_http_request(inbound):
+        next(auth.FlagsmithAuth().auth_flow(upstream))
+
+    # Then the existing credential wins
+    assert upstream.headers["authorization"] == "Api-Key static"

mcp/tests/unit/test_config.py

@@ -42,3 +42,15 @@ def test_settings__unsupported_transport__raises(
     # When / Then
     with pytest.raises(ValidationError):
         config.Settings()
+
+
+def test_settings__stdio_without_token__raises(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # Given stdio transport but no API token
+    monkeypatch.setenv("TRANSPORT", "stdio")
+    monkeypatch.delenv("FLAGSMITH_API_TOKEN", raising=False)
+
+    # When / Then
+    with pytest.raises(ValidationError, match="FLAGSMITH_API_TOKEN is required"):
+        config.Settings()

mcp/tests/unit/test_server.py

@@ -137,6 +137,7 @@ def run(self, transport: str) -> None:
 
     monkeypatch.setattr(server, "create_server", lambda settings: FakeServer())
     monkeypatch.setenv("TRANSPORT", "stdio")
+    monkeypatch.setenv("FLAGSMITH_API_TOKEN", "ser.secret")
 
     # When
     server.run()