feat: add Quizzical Beats MCP automation interface

2026-05-21 18:12:17 +02:00
3 changed files with 0 additions and 171 deletions
@@ -13,16 +13,6 @@ 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,
@@ -1,92 +0,0 @@
 """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
 _BASE_ALLOWED_HOSTS = tuple(mcp.settings.transport_security.allowed_hosts)
 _BASE_ALLOWED_ORIGINS = tuple(mcp.settings.transport_security.allowed_origins)
 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."},
                headers={"WWW-Authenticate": "Bearer"},
                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 = list(dict.fromkeys([*_BASE_ALLOWED_HOSTS, *allowed_hosts]))
    security.allowed_origins = list(
        dict.fromkeys([*_BASE_ALLOWED_ORIGINS, *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)
@@ -1,69 +0,0 @@
 """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, mcp  # 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 missing.headers["WWW-Authenticate"] == "Bearer"
    assert wrong.status_code == 401
    assert accepted.status_code != 401
 def test_mcp_falls_back_to_automation_token(monkeypatch):
    monkeypatch.delenv("MCP_BEARER_TOKEN", raising=False)
    monkeypatch.setenv("AUTOMATION_TOKEN", "automation-token")
    client = TestClient(BearerAuthMiddleware(_dummy_app))
    accepted = client.get("/mcp", headers={"Authorization": "Bearer automation-token"})
    assert accepted.status_code == 204
 def test_mcp_reports_missing_bearer_configuration(monkeypatch):
    monkeypatch.delenv("MCP_BEARER_TOKEN", raising=False)
    monkeypatch.delenv("AUTOMATION_TOKEN", raising=False)
    client = TestClient(BearerAuthMiddleware(_dummy_app))
    response = client.get("/mcp", headers={"Authorization": "Bearer anything"})
    assert response.status_code == 500
 def test_allowed_hosts_are_replaced_between_builds(monkeypatch):
    monkeypatch.setenv("MCP_ALLOWED_HOSTS", "first.example")
    build_app()
    monkeypatch.setenv("MCP_ALLOWED_HOSTS", "second.example")
    build_app()
    assert "first.example" not in mcp.settings.transport_security.allowed_hosts
    assert "second.example" in mcp.settings.transport_security.allowed_hosts