feat: add authenticated remote MCP HTTP entrypoint

docs/developer-guide/mcp.md

@@ -13,6 +13,16 @@ pip install -r requirements.txt
 python -m musicround.mcp_server
 ```
 
+For a production streamable HTTP endpoint, run the authenticated ASGI entrypoint:
+
+```bash
+MCP_BEARER_TOKEN=... uvicorn musicround.mcp_http:app --host 0.0.0.0 --port 8000
+```
+
+If `MCP_BEARER_TOKEN` is not set, the HTTP entrypoint falls back to
+`AUTOMATION_TOKEN`. Set `MCP_ALLOWED_HOSTS` and `MCP_ALLOWED_ORIGINS` when the
+server is exposed behind a reverse proxy or ingress.
+
 The server uses the normal Quizzical Beats Flask configuration. Set the same
 environment variables you use for the web app, including `SECRET_KEY`,
 `AUTOMATION_TOKEN`, database configuration, mail settings, and any Spotify,

musicround/mcp_http.py

@@ -0,0 +1,88 @@
+"""Authenticated streamable HTTP entrypoint for the Quizzical Beats MCP server."""
+
+from __future__ import annotations
+
+import os
+from secrets import compare_digest
+
+from starlette.responses import JSONResponse
+from starlette.types import ASGIApp, Receive, Scope, Send
+
+from musicround.mcp_server import mcp
+
+
+class BearerAuthMiddleware:
+    """Require a bearer token for MCP HTTP traffic."""
+
+    def __init__(self, app: ASGIApp) -> None:
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        if scope.get("path") == "/healthz":
+            await JSONResponse({"ok": True})(scope, receive, send)
+            return
+
+        expected = os.getenv("MCP_BEARER_TOKEN") or os.getenv("AUTOMATION_TOKEN")
+        if not expected:
+            await JSONResponse(
+                {"error": "MCP bearer token is not configured."},
+                status_code=500,
+            )(scope, receive, send)
+            return
+
+        headers = dict(scope.get("headers", []))
+        authorization = headers.get(b"authorization", b"").decode("latin1")
+        scheme, _, token = authorization.partition(" ")
+        if scheme.lower() != "bearer" or not compare_digest(token.strip(), expected):
+            await JSONResponse({"error": "Unauthorized."}, status_code=401)(
+                scope, receive, send
+            )
+            return
+
+        await self.app(scope, receive, send)
+
+
+def _configure_server() -> None:
+    host = os.getenv("MCP_HOST", "0.0.0.0")
+    port = int(os.getenv("MCP_PORT", "8000"))
+    mcp.settings.host = host
+    mcp.settings.port = port
+
+    allowed_hosts = [
+        value.strip()
+        for value in os.getenv("MCP_ALLOWED_HOSTS", "qb.kaufdeinquiz.com").split(",")
+        if value.strip()
+    ]
+    allowed_origins = [
+        value.strip()
+        for value in os.getenv(
+            "MCP_ALLOWED_ORIGINS", "https://qb.kaufdeinquiz.com"
+        ).split(",")
+        if value.strip()
+    ]
+    security = mcp.settings.transport_security
+    security.allowed_hosts.extend(
+        host for host in allowed_hosts if host not in security.allowed_hosts
+    )
+    security.allowed_origins.extend(
+        origin for origin in allowed_origins if origin not in security.allowed_origins
+    )
+
+
+def build_app() -> ASGIApp:
+    """Build the authenticated streamable HTTP MCP ASGI app."""
+    _configure_server()
+    return BearerAuthMiddleware(mcp.streamable_http_app())
+
+
+app = build_app()
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run(app, host=mcp.settings.host, port=mcp.settings.port)

tests/test_mcp_http.py

@@ -0,0 +1,38 @@
+"""Tests for the authenticated MCP HTTP entrypoint."""
+
+import os
+
+from starlette.responses import Response
+from starlette.testclient import TestClient
+
+os.environ.setdefault("SECRET_KEY", "test-secret-key-for-testing-only")
+os.environ.setdefault("AUTOMATION_TOKEN", "test-automation-token-for-testing")
+
+from musicround.mcp_http import BearerAuthMiddleware, build_app  # noqa: E402
+
+
+async def _dummy_app(scope, receive, send):
+    await Response(status_code=204)(scope, receive, send)
+
+
+def test_healthz_does_not_require_auth(monkeypatch):
+    monkeypatch.setenv("MCP_BEARER_TOKEN", "test-mcp-token")
+    client = TestClient(build_app())
+
+    response = client.get("/healthz")
+
+    assert response.status_code == 200
+    assert response.json() == {"ok": True}
+
+
+def test_mcp_requires_bearer_token(monkeypatch):
+    monkeypatch.setenv("MCP_BEARER_TOKEN", "test-mcp-token")
+    client = TestClient(BearerAuthMiddleware(_dummy_app))
+
+    missing = client.get("/mcp")
+    wrong = client.get("/mcp", headers={"Authorization": "Bearer wrong"})
+    accepted = client.get("/mcp", headers={"Authorization": "Bearer test-mcp-token"})
+
+    assert missing.status_code == 401
+    assert wrong.status_code == 401
+    assert accepted.status_code != 401