diff --git a/docs/developer-guide/mcp.md b/docs/developer-guide/mcp.md index 5400473..01eec15 100644 --- a/docs/developer-guide/mcp.md +++ b/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, diff --git a/musicround/mcp_http.py b/musicround/mcp_http.py new file mode 100644 index 0000000..eba0c9d --- /dev/null +++ b/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) diff --git a/tests/test_mcp_http.py b/tests/test_mcp_http.py new file mode 100644 index 0000000..d282526 --- /dev/null +++ b/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