Fix vulnerable musicround dependencies

Add generic MCP CRUD access for mapped datastore objects and include song usage frequency in song query results.

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,
@@ -27,6 +37,12 @@ The MCP server exposes these tools:
 | --- | --- |
 | `find_songs` | Search the existing Quizzical Beats catalog before adding duplicates. |
 | `add_song` | Add or update a catalog song, including platform IDs and tags. |
+| `datastore_schema` | Describe all mapped datastore object types, columns, and primary keys. |
+| `list_datastore_objects` | List persisted objects with optional exact-match filters, ordering, limit, and offset. |
+| `get_datastore_object` | Fetch one persisted object by primary key. |
+| `create_datastore_object` | Create one persisted object from scalar column fields. |
+| `update_datastore_object` | Update scalar column fields on one persisted object. |
+| `delete_datastore_object` | Delete one persisted object by primary key. |
 | `import_catalog_item` | Import a Spotify or Deezer track, album, or playlist. |
 | `compile_round` | Create a named round from explicit song IDs or selection criteria. |
 | `rename_round` | Set or clear a round name. |
@@ -37,6 +53,15 @@ The MCP server exposes these tools:
 | `send_round_email` | Generate assets and email the finished round bundle. |
 | `generate_tts_snippet` | Generate and assign custom intro, replay, or outro TTS MP3s. |
 
+`find_songs` includes `used_count`, `usage_frequency`, and `last_used` for each
+result so agents can see how often songs have already appeared in rounds.
+
+The generic datastore CRUD tools operate on mapped SQLAlchemy models, including
+`song`, `round`, `tag`, `song_tag`, `user`, `role`, `user_preferences`,
+`round_export`, `system_setting`, and `import_job_record`. Read results redact
+fields whose names contain `password`, `token`, or `secret` unless
+`include_sensitive` is explicitly set.
+
 ## Intended Workflow
 
 1. Search with `find_songs` to avoid duplicates.

musicround/mcp_http.py

@@ -0,0 +1,92 @@
+"""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)

musicround/mcp_server.py

@@ -87,6 +87,90 @@ def add_song(
     )
 
 
+@mcp.tool()
+def datastore_schema() -> dict[str, Any]:
+    """Describe every datastore object type available to generic CRUD tools."""
+    return _with_app_context(automation.datastore_schema)
+
+
+@mcp.tool()
+def list_datastore_objects(
+    object_type: str,
+    filters: dict[str, Any] | None = None,
+    limit: int = 50,
+    offset: int = 0,
+    order_by: str | None = None,
+    include_sensitive: bool = False,
+) -> dict[str, Any]:
+    """List datastore objects such as songs, rounds, users, tags, exports, and settings."""
+    return _with_app_context(
+        automation.list_datastore_objects,
+        object_type=object_type,
+        filters=filters,
+        limit=limit,
+        offset=offset,
+        order_by=order_by,
+        include_sensitive=include_sensitive,
+    )
+
+
+@mcp.tool()
+def get_datastore_object(
+    object_type: str,
+    object_id: Any,
+    include_sensitive: bool = False,
+) -> dict[str, Any]:
+    """Fetch one datastore object by primary key."""
+    return _with_app_context(
+        automation.get_datastore_object,
+        object_type=object_type,
+        object_id=object_id,
+        include_sensitive=include_sensitive,
+    )
+
+
+@mcp.tool()
+def create_datastore_object(
+    object_type: str,
+    fields: dict[str, Any],
+    include_sensitive: bool = False,
+) -> dict[str, Any]:
+    """Create one datastore object from scalar column fields."""
+    return _with_app_context(
+        automation.create_datastore_object,
+        object_type=object_type,
+        fields=fields,
+        include_sensitive=include_sensitive,
+    )
+
+
+@mcp.tool()
+def update_datastore_object(
+    object_type: str,
+    object_id: Any,
+    fields: dict[str, Any],
+    include_sensitive: bool = False,
+) -> dict[str, Any]:
+    """Update scalar column fields on one datastore object."""
+    return _with_app_context(
+        automation.update_datastore_object,
+        object_type=object_type,
+        object_id=object_id,
+        fields=fields,
+        include_sensitive=include_sensitive,
+    )
+
+
+@mcp.tool()
+def delete_datastore_object(object_type: str, object_id: Any) -> dict[str, Any]:
+    """Delete one datastore object by primary key."""
+    return _with_app_context(
+        automation.delete_datastore_object,
+        object_type=object_type,
+        object_id=object_id,
+    )
+
+
 @mcp.tool()
 def import_catalog_item(
     service_name: str,

musicround/services/automation.py

@@ -10,12 +10,14 @@ from typing import Any, Iterable
 from flask import current_app
 from flask_login import login_user, logout_user
 from pydub import AudioSegment
+from sqlalchemy import inspect as sa_inspect
 from sqlalchemy import or_
 
 from musicround import db
 from musicround.helpers.email_helper import send_email
 from musicround.helpers.import_helper import ImportHelper
 from musicround.helpers.utils import generate_tts_mp3
+from musicround import models as datastore_models
 from musicround.models import Round, RoundExport, Song, Tag, User
 
 
@@ -36,6 +38,9 @@ def _song_summary(song: Song) -> dict[str, Any]:
         "spotify_id": data["spotify_id"],
         "deezer_id": data["deezer_id"],
         "isrc": data["isrc"],
+        "used_count": data["used_count"] or 0,
+        "usage_frequency": data["used_count"] or 0,
+        "last_used": data["last_used"],
         "tags": data["tags"],
     }
 
@@ -114,6 +119,297 @@ def _attach_tags(song: Song, tag_names: Iterable[str] | None) -> None:
             song.tags.append(tag)
 
 
+def _snake_case(value: str) -> str:
+    value = re.sub(r"(.)([A-Z][a-z]+)", r"\1_\2", value)
+    value = re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", value)
+    return value.lower()
+
+
+def _model_registry() -> dict[str, type[db.Model]]:
+    registry: dict[str, type[db.Model]] = {}
+    for value in vars(datastore_models).values():
+        if not isinstance(value, type):
+            continue
+        if value is db.Model or not issubclass(value, db.Model):
+            continue
+        mapper = sa_inspect(value, raiseerr=False)
+        if mapper is None or getattr(value, "__table__", None) is None:
+            continue
+        canonical = _snake_case(value.__name__)
+        registry[canonical] = value
+        registry[value.__name__] = value
+        registry[value.__name__.lower()] = value
+        registry[value.__tablename__] = value
+    return registry
+
+
+def _canonical_model_key(model: type[db.Model]) -> str:
+    return _snake_case(model.__name__)
+
+
+def _get_model(object_type: str) -> type[db.Model]:
+    if not object_type:
+        raise AutomationError("object_type is required.")
+    model = _model_registry().get(object_type)
+    if not model:
+        allowed = sorted({_canonical_model_key(model) for model in _model_registry().values()})
+        raise AutomationError(f"Unknown object_type '{object_type}'. Allowed values: {allowed}")
+    return model
+
+
+def _column_map(model: type[db.Model]) -> dict[str, Any]:
+    return {column.key: column for column in sa_inspect(model).columns}
+
+
+def _primary_key_columns(model: type[db.Model]) -> list[Any]:
+    return list(sa_inspect(model).primary_key)
+
+
+def _is_sensitive_field(field_name: str) -> bool:
+    lowered = field_name.lower()
+    return any(marker in lowered for marker in ("password", "token", "secret"))
+
+
+def _json_value(value: Any, *, sensitive: bool = False, include_sensitive: bool = False) -> Any:
+    if sensitive and value is not None and not include_sensitive:
+        return "[redacted]"
+    if isinstance(value, datetime):
+        return value.isoformat()
+    return value
+
+
+def _serialize_model(instance: db.Model, *, include_sensitive: bool = False) -> dict[str, Any]:
+    data = {}
+    for column in sa_inspect(instance.__class__).columns:
+        value = getattr(instance, column.key)
+        data[column.key] = _json_value(
+            value,
+            sensitive=_is_sensitive_field(column.key),
+            include_sensitive=include_sensitive,
+        )
+    return data
+
+
+def _coerce_column_value(column: Any, value: Any) -> Any:
+    if value is None:
+        return None
+    try:
+        python_type = column.type.python_type
+    except NotImplementedError:
+        return value
+
+    if python_type is datetime:
+        if isinstance(value, datetime):
+            return value
+        if isinstance(value, str):
+            normalized = value.replace("Z", "+00:00")
+            return datetime.fromisoformat(normalized)
+        raise AutomationError(f"{column.key} must be an ISO datetime string.")
+    if python_type is bool and isinstance(value, str):
+        lowered = value.lower()
+        if lowered in {"true", "1", "yes", "on"}:
+            return True
+        if lowered in {"false", "0", "no", "off"}:
+            return False
+    if python_type in {int, float, str, bool} and not isinstance(value, python_type):
+        return python_type(value)
+    return value
+
+
+def _identity_for_object(model: type[db.Model], object_id: Any) -> Any:
+    primary_key = _primary_key_columns(model)
+    if not primary_key:
+        raise AutomationError(f"{_canonical_model_key(model)} does not have a primary key.")
+
+    if isinstance(object_id, dict):
+        missing = [column.key for column in primary_key if column.key not in object_id]
+        if missing:
+            raise AutomationError(f"Missing primary key field(s): {missing}")
+        values = [_coerce_column_value(column, object_id[column.key]) for column in primary_key]
+    elif len(primary_key) == 1:
+        values = [_coerce_column_value(primary_key[0], object_id)]
+    elif isinstance(object_id, list):
+        if len(object_id) != len(primary_key):
+            raise AutomationError(
+                f"Composite primary key requires {len(primary_key)} values in order."
+            )
+        values = [
+            _coerce_column_value(column, object_id[index])
+            for index, column in enumerate(primary_key)
+        ]
+    else:
+        names = [column.key for column in primary_key]
+        raise AutomationError(f"Composite primary key requires an object with keys {names}.")
+
+    return values[0] if len(values) == 1 else tuple(values)
+
+
+def _get_datastore_instance(model: type[db.Model], object_id: Any) -> db.Model:
+    instance = db.session.get(model, _identity_for_object(model, object_id))
+    if not instance:
+        raise AutomationError(f"{_canonical_model_key(model)} {object_id} was not found.")
+    return instance
+
+
+def _apply_datastore_filters(query: Any, model: type[db.Model], filters: dict[str, Any] | None) -> Any:
+    columns = _column_map(model)
+    for field_name, raw_value in (filters or {}).items():
+        column = columns.get(field_name)
+        if column is None:
+            raise AutomationError(f"Unknown filter field '{field_name}'.")
+        query = query.filter(getattr(model, field_name) == _coerce_column_value(column, raw_value))
+    return query
+
+
+def _assign_datastore_fields(instance: db.Model, fields: dict[str, Any], *, creating: bool) -> None:
+    if not fields:
+        raise AutomationError("fields must not be empty.")
+
+    model = instance.__class__
+    columns = _column_map(model)
+    primary_keys = {column.key for column in _primary_key_columns(model)}
+    for field_name, raw_value in fields.items():
+        column = columns.get(field_name)
+        if column is None:
+            raise AutomationError(f"Unknown field '{field_name}'.")
+        if not creating and field_name in primary_keys:
+            raise AutomationError("Primary key fields cannot be updated.")
+        setattr(instance, field_name, _coerce_column_value(column, raw_value))
+
+
+def datastore_schema() -> dict[str, Any]:
+    """Describe datastore objects available through generic MCP CRUD tools."""
+    models_by_key = {
+        _canonical_model_key(model): model for model in _model_registry().values()
+    }
+    objects = []
+    for object_type, model in sorted(models_by_key.items()):
+        mapper = sa_inspect(model)
+        objects.append(
+            {
+                "object_type": object_type,
+                "table": model.__tablename__,
+                "primary_key": [column.key for column in mapper.primary_key],
+                "columns": [
+                    {
+                        "name": column.key,
+                        "type": str(column.type),
+                        "nullable": column.nullable,
+                        "primary_key": column.primary_key,
+                        "sensitive": _is_sensitive_field(column.key),
+                    }
+                    for column in mapper.columns
+                ],
+            }
+        )
+    return {"object_types": [item["object_type"] for item in objects], "objects": objects}
+
+
+def list_datastore_objects(
+    object_type: str,
+    filters: dict[str, Any] | None = None,
+    limit: int = 50,
+    offset: int = 0,
+    order_by: str | None = None,
+    include_sensitive: bool = False,
+) -> dict[str, Any]:
+    """List persisted rows for a mapped datastore object."""
+    if limit < 1 or limit > 500:
+        raise AutomationError("limit must be between 1 and 500.")
+    if offset < 0:
+        raise AutomationError("offset must not be negative.")
+
+    model = _get_model(object_type)
+    query = _apply_datastore_filters(model.query, model, filters)
+    total = query.count()
+
+    if order_by:
+        descending = order_by.startswith("-")
+        field_name = order_by[1:] if descending else order_by
+        if field_name not in _column_map(model):
+            raise AutomationError(f"Unknown order_by field '{field_name}'.")
+        column = getattr(model, field_name)
+        query = query.order_by(column.desc() if descending else column.asc())
+    else:
+        primary_key = _primary_key_columns(model)
+        if primary_key:
+            query = query.order_by(*[getattr(model, column.key).asc() for column in primary_key])
+
+    rows = query.offset(offset).limit(limit).all()
+    return {
+        "object_type": _canonical_model_key(model),
+        "count": len(rows),
+        "total": total,
+        "limit": limit,
+        "offset": offset,
+        "objects": [_serialize_model(row, include_sensitive=include_sensitive) for row in rows],
+    }
+
+
+def get_datastore_object(
+    object_type: str,
+    object_id: Any,
+    include_sensitive: bool = False,
+) -> dict[str, Any]:
+    """Fetch a single persisted datastore object by primary key."""
+    model = _get_model(object_type)
+    instance = _get_datastore_instance(model, object_id)
+    return {
+        "object_type": _canonical_model_key(model),
+        "object": _serialize_model(instance, include_sensitive=include_sensitive),
+    }
+
+
+def create_datastore_object(
+    object_type: str,
+    fields: dict[str, Any],
+    include_sensitive: bool = False,
+) -> dict[str, Any]:
+    """Create a persisted datastore object from scalar column fields."""
+    model = _get_model(object_type)
+    instance = model()
+    _assign_datastore_fields(instance, fields, creating=True)
+    db.session.add(instance)
+    db.session.commit()
+    return {
+        "created": True,
+        "object_type": _canonical_model_key(model),
+        "object": _serialize_model(instance, include_sensitive=include_sensitive),
+    }
+
+
+def update_datastore_object(
+    object_type: str,
+    object_id: Any,
+    fields: dict[str, Any],
+    include_sensitive: bool = False,
+) -> dict[str, Any]:
+    """Update scalar column fields for a persisted datastore object."""
+    model = _get_model(object_type)
+    instance = _get_datastore_instance(model, object_id)
+    _assign_datastore_fields(instance, fields, creating=False)
+    db.session.commit()
+    return {
+        "updated": True,
+        "object_type": _canonical_model_key(model),
+        "object": _serialize_model(instance, include_sensitive=include_sensitive),
+    }
+
+
+def delete_datastore_object(object_type: str, object_id: Any) -> dict[str, Any]:
+    """Delete a persisted datastore object by primary key."""
+    model = _get_model(object_type)
+    instance = _get_datastore_instance(model, object_id)
+    serialized = _serialize_model(instance)
+    db.session.delete(instance)
+    db.session.commit()
+    return {
+        "deleted": True,
+        "object_type": _canonical_model_key(model),
+        "object": serialized,
+    }
+
+
 def add_song(
     title: str,
     artist: str,

requirements.txt

@@ -5,7 +5,8 @@ flask_migrate
 requests
 pydub
 reportlab
-PyJWT
+PyJWT>=2.10.1
+idna>=3.11
 python-dotenv
 deezer-python
 Flask-Assets

tests/test_automation_service.py

@@ -7,7 +7,10 @@ from unittest.mock import patch
 
 import pytest
 
-from musicround.models import Song, User, db
+os.environ.setdefault("SECRET_KEY", "test-secret-key-for-testing-only")
+os.environ.setdefault("AUTOMATION_TOKEN", "test-automation-token-for-testing")
+
+from musicround.models import Song, SongTag, Tag, User, db
 from musicround.services import automation
 
 
@@ -31,12 +34,20 @@ class TestSongAutomation:
 
     def test_find_songs_by_query(self, app):
         with app.app_context():
-            _create_song(title="Blue Monday", artist="New Order", genre="Synthpop")
+            _create_song(
+                title="Blue Monday",
+                artist="New Order",
+                genre="Synthpop",
+                used_count=3,
+            )
 
             result = automation.find_songs(query="blue")
 
             assert result["count"] == 1
             assert result["songs"][0]["title"] == "Blue Monday"
+            assert result["songs"][0]["used_count"] == 3
+            assert result["songs"][0]["usage_frequency"] == 3
+            assert "last_used" in result["songs"][0]
 
     def test_add_song_reuses_existing_by_isrc_and_adds_tags(self, app):
         with app.app_context():
@@ -127,3 +138,74 @@ class TestTTSAutomation:
 
             assert result["path"] == "custommp3/agentuser/intro.mp3"
             assert User.query.get(user.id).intro_mp3 == "custommp3/agentuser/intro.mp3"
+
+
+class TestDatastoreCrudAutomation:
+    """Tests for generic datastore CRUD operations exposed through MCP."""
+
+    def test_datastore_schema_lists_mapped_models(self, app):
+        with app.app_context():
+            schema = automation.datastore_schema()
+
+            assert "song" in schema["object_types"]
+            assert "round" in schema["object_types"]
+            assert "user" in schema["object_types"]
+            song_schema = next(
+                item for item in schema["objects"] if item["object_type"] == "song"
+            )
+            assert song_schema["primary_key"] == ["id"]
+            assert any(column["name"] == "title" for column in song_schema["columns"])
+
+    def test_crud_lifecycle_for_single_primary_key_object(self, app):
+        with app.app_context():
+            created = automation.create_datastore_object("tag", {"name": "warmup"})
+            tag_id = created["object"]["id"]
+
+            listed = automation.list_datastore_objects(
+                "tag", filters={"name": "warmup"}, order_by="id"
+            )
+            fetched = automation.get_datastore_object("tag", tag_id)
+            updated = automation.update_datastore_object(
+                "tag", tag_id, {"name": "opener"}
+            )
+            deleted = automation.delete_datastore_object("tag", tag_id)
+
+            assert created["object_type"] == "tag"
+            assert listed["total"] == 1
+            assert fetched["object"]["name"] == "warmup"
+            assert updated["object"]["name"] == "opener"
+            assert deleted["deleted"] is True
+            assert Tag.query.get(tag_id) is None
+
+    def test_crud_supports_composite_primary_keys(self, app):
+        with app.app_context():
+            song = _create_song(title="Composite", artist="Key")
+            tag = Tag(name="linked")
+            db.session.add(tag)
+            db.session.commit()
+
+            created = automation.create_datastore_object(
+                "song_tag", {"song_id": song.id, "tag_id": tag.id}
+            )
+            fetched = automation.get_datastore_object(
+                "song_tag", {"song_id": song.id, "tag_id": tag.id}
+            )
+            deleted = automation.delete_datastore_object(
+                "song_tag", {"song_id": song.id, "tag_id": tag.id}
+            )
+
+            assert created["object"]["song_id"] == song.id
+            assert fetched["object"]["tag_id"] == tag.id
+            assert deleted["deleted"] is True
+            assert SongTag.query.count() == 0
+
+    def test_user_sensitive_fields_are_redacted_by_default(self, app):
+        with app.app_context():
+            user = _create_user()
+            user.spotify_token = "secret-token"
+            db.session.commit()
+
+            result = automation.get_datastore_object("user", user.id)
+
+            assert result["object"]["spotify_token"] == "[redacted]"
+            assert result["object"]["password_hash"] == "[redacted]"

tests/test_mcp_http.py

@@ -0,0 +1,69 @@
+"""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

tests/test_security.py

@@ -124,6 +124,30 @@ class TestDependencySecurity:
         assert 'authlib>=1.6.5' in content, \
             "authlib should be pinned to >= 1.6.5 to fix known vulnerabilities"
 
+    def test_pyjwt_version(self):
+        """Test that PyJWT is at least version 2.10.1."""
+        requirements_path = os.path.join(
+            os.path.dirname(__file__), '..', 'requirements.txt'
+        )
+
+        with open(requirements_path, 'r') as f:
+            content = f.read()
+
+        assert 'PyJWT>=2.10.1' in content, \
+            "PyJWT should be pinned to >= 2.10.1 to fix known vulnerabilities"
+
+    def test_idna_version(self):
+        """Test that idna is at least version 3.11."""
+        requirements_path = os.path.join(
+            os.path.dirname(__file__), '..', 'requirements.txt'
+        )
+
+        with open(requirements_path, 'r') as f:
+            content = f.read()
+
+        assert 'idna>=3.11' in content, \
+            "idna should be pinned to >= 3.11 to fix known vulnerabilities"
+
 
 class TestInputValidation:
     """Test that user input is properly validated."""