feat: authenticate MCP with managed API keys

backend/app/mcp_auth.py

@@ -3,9 +3,11 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
 from functools import partial
 from typing import Any
 
+import asyncpg
 from fastapi import HTTPException
 from mcp.server.auth.provider import AccessToken, TokenVerifier
 
@@ -27,9 +29,65 @@ class KeycloakMcpTokenVerifier(TokenVerifier):
 
     def __init__(self, settings: Settings) -> None:
         self._settings = settings
+        self._api_key_pool: Any | None = None
+
+    def bind_api_key_pool(self, pool: Any) -> None:
+        """Bind the process-lifetime pool used by LineageWeave API keys."""
+        self._api_key_pool = pool
+
+    def unbind_api_key_pool(self, pool: Any) -> None:
+        """Release the pool only when it is the currently bound instance."""
+        if self._api_key_pool is pool:
+            self._api_key_pool = None
+
+    async def _verify_api_key(self, token: str) -> AccessToken | None:
+        """Resolve one hashed application key to its provisioned account subject."""
+        if self._api_key_pool is None:
+            return None
+        key_hash = hashlib.sha256(token.encode("utf-8")).hexdigest()
+        try:
+            row = await self._api_key_pool.fetchrow(
+                """
+                select api_key.mcp_api_key_id,
+                       api_key.user_account_id,
+                       account.external_subject_id,
+                       extract(epoch from api_key.expires_at) as expires_at
+                  from mcp_api_key api_key
+                  join user_account account
+                    on account.user_account_id = api_key.user_account_id
+                 where api_key.key_hash = $1
+                   and api_key.revoked_at is null
+                   and (api_key.expires_at is null or api_key.expires_at > now())
+                """,
+                key_hash,
+            )
+        except asyncpg.UndefinedTableError:
+            # The key-management stack may be deployed after this MCP stack.
+            return None
+        if row is None:
+            return None
+        subject = row["external_subject_id"]
+        if not isinstance(subject, str) or not subject:
+            return None
+        expires_at = row["expires_at"]
+        return AccessToken(
+            token=token,
+            client_id="lineageweave-mcp-api-key",
+            scopes=list(self._settings.mcp_required_scopes),
+            expires_at=int(expires_at) if expires_at is not None else None,
+            resource=self._settings.mcp_audience,
+            subject=subject,
+            claims={
+                "auth_method": "mcp_api_key",
+                "mcp_api_key_id": str(row["mcp_api_key_id"]),
+                "user_account_id": str(row["user_account_id"]),
+            },
+        )
 
     async def verify_token(self, token: str) -> AccessToken | None:
         """Return MCP access metadata for a valid token; otherwise fail closed."""
+        if token.startswith("lw_mcp_"):
+            return await self._verify_api_key(token)
         try:
             claims = await asyncio.to_thread(
                 partial(
@@ -54,4 +112,4 @@ async def verify_token(self, token: str) -> AccessToken | None:
             resource=self._settings.mcp_audience,
             subject=subject,
             claims={"iss": claims.get("iss"), "aud": claims.get("aud")},
-        )
+        )

backend/app/mcp_server.py

@@ -156,11 +156,14 @@ def build_mcp_server(
     """Build a testable OAuth resource server with one read-only Global Ask tool."""
     resolved_settings = settings or load_settings()
     resolved_external_verifier = external_verifier or _external_verifier(resolved_settings)
+    resolved_token_verifier = token_verifier or KeycloakMcpTokenVerifier(resolved_settings)
 
     @asynccontextmanager
     async def lifespan(_: MCPServer) -> AsyncIterator[McpAppContext]:
         """Open and close the MCP process-wide database and client context."""
         pool = await pool_factory(resolved_settings.database_url)
+        if isinstance(resolved_token_verifier, KeycloakMcpTokenVerifier):
+            resolved_token_verifier.bind_api_key_pool(pool)
         try:
             yield McpAppContext(
                 pool=pool,
@@ -172,6 +175,8 @@ async def lifespan(_: MCPServer) -> AsyncIterator[McpAppContext]:
                 external_verifier=resolved_external_verifier,
             )
         finally:
+            if isinstance(resolved_token_verifier, KeycloakMcpTokenVerifier):
+                resolved_token_verifier.unbind_api_key_pool(pool)
             await pool.close()
 
     mcp = MCPServer(
@@ -191,7 +196,7 @@ async def lifespan(_: MCPServer) -> AsyncIterator[McpAppContext]:
         ),
         version="1.0.1",
         lifespan=lifespan,
-        token_verifier=token_verifier or KeycloakMcpTokenVerifier(resolved_settings),
+        token_verifier=resolved_token_verifier,
         auth=AuthSettings(
             issuer_url=AnyHttpUrl(resolved_settings.oidc_issuer),
             resource_server_url=AnyHttpUrl(resolved_settings.mcp_resource_url),

docs/adr/0110-mcp-api-key-authentication.md

@@ -0,0 +1,43 @@
+# ADR 0110: Authenticate MCP with LineageWeave-managed API keys
+
+- Status: Accepted
+- Date: 2026-08-21
+- Depends on: ADR 0103 and PR #333 (`mcp_api_key` lifecycle)
+
+## Context
+
+LineageWeave exposes an authenticated MCP server. ADR 0103 establishes that
+Keyverse is the identity boundary while LineageWeave owns the normalized,
+owner-scoped application key resource. A key that can be created in the
+LineageWeave settings screen but cannot authenticate the MCP server is not a
+usable buyer capability.
+
+## Decision
+
+The MCP resource server accepts two bearer forms:
+
+1. Keyverse/Keycloak OIDC JWTs continue through the existing signature,
+   issuer, audience, expiry, and subject validation.
+2. A LineageWeave application key with the `lw_mcp_` prefix is hashed with
+   SHA-256 and looked up in `mcp_api_key`. Only a non-revoked, non-expired key
+   joined to a provisioned `user_account` yields an MCP `AccessToken`.
+
+The API key never becomes a new identity. Its `user_account.external_subject_id`
+is the MCP subject, so the existing account resolver and ABAC/RBAC checks remain
+the authorization authority. The raw key is never stored or logged.
+
+The MCP lifespan binds its existing PostgreSQL pool to the verifier and clears
+that binding during shutdown. If the key-management migration is not deployed,
+API-key authentication is unavailable while OIDC authentication remains intact;
+the verifier does not turn a schema deployment race into a server-wide outage.
+
+## Consequences
+
+- Settings-created keys are directly usable by MCP clients after PR #333 and
+  this authentication change are deployed together.
+- Revocation and expiry take effect at the database lookup boundary.
+- Key issuance remains LineageWeave UI/API behavior; Keyverse remains the
+  identity and account-provisioning authority.
+- A future Keyverse-native application-key resource can replace the lookup
+  without changing the MCP tool contract, but no second credential source is
+  introduced now.

tests/test_mcp_api_key_auth.py

@@ -0,0 +1,80 @@
+"""Regression tests for the LineageWeave-managed MCP API-key boundary."""
+
+from __future__ import annotations
+
+import hashlib
+from types import SimpleNamespace
+
+import asyncpg
+import pytest
+
+from backend.app.mcp_auth import KeycloakMcpTokenVerifier
+
+
+class FakePool:
+    """Small asyncpg-pool substitute that records the hashed lookup input."""
+
+    def __init__(self, row: dict[str, object] | None) -> None:
+        self.row = row
+        self.received_hash: str | None = None
+
+    async def fetchrow(self, _query: str, key_hash: str) -> dict[str, object] | None:
+        self.received_hash = key_hash
+        return self.row
+
+
+class MissingKeyTablePool:
+    """Pool substitute for a deployment where PR #333 is not applied yet."""
+
+    async def fetchrow(self, _query: str, _key_hash: str) -> None:
+        raise asyncpg.UndefinedTableError("mcp_api_key is not installed")
+
+
+def _settings() -> SimpleNamespace:
+    return SimpleNamespace(
+        mcp_required_scopes=("post_read",),
+        mcp_audience="https://lineageweave.example/mcp",
+    )
+
+
+@pytest.mark.asyncio
+async def test_api_key_resolves_to_the_provisioned_account_subject() -> None:
+    token = "lw_mcp_test-secret"
+    pool = FakePool(
+        {
+            "mcp_api_key_id": "key-id",
+            "user_account_id": "account-id",
+            "external_subject_id": "keyverse-subject",
+            "expires_at": 1_900_000_000,
+        }
+    )
+    verifier = KeycloakMcpTokenVerifier(_settings())  # type: ignore[arg-type]
+    verifier.bind_api_key_pool(pool)
+
+    access_token = await verifier.verify_token(token)
+
+    assert pool.received_hash == hashlib.sha256(token.encode()).hexdigest()
+    assert access_token is not None
+    assert access_token.subject == "keyverse-subject"
+    assert access_token.scopes == ["post_read"]
+    assert access_token.expires_at == 1_900_000_000
+    assert access_token.claims == {
+        "auth_method": "mcp_api_key",
+        "mcp_api_key_id": "key-id",
+        "user_account_id": "account-id",
+    }
+
+
+@pytest.mark.asyncio
+async def test_api_key_is_unavailable_before_the_mcp_lifespan_binds_a_pool() -> None:
+    verifier = KeycloakMcpTokenVerifier(_settings())  # type: ignore[arg-type]
+
+    assert await verifier.verify_token("lw_mcp_not-bound") is None
+
+
+@pytest.mark.asyncio
+async def test_missing_key_table_fails_closed_without_affecting_oidc_verification() -> None:
+    verifier = KeycloakMcpTokenVerifier(_settings())  # type: ignore[arg-type]
+    verifier.bind_api_key_pool(MissingKeyTablePool())
+
+    assert await verifier.verify_token("lw_mcp_schema-not-ready") is None