feat: expose durable Global Ask through authenticated MCP

.env.example

@@ -1,7 +1,8 @@
 # Copy to .env to override. Every value below already has a working
 # default baked into docker-compose.yml (see ${VAR:-default} references) --
 # `docker compose up` succeeds from a clean checkout with no .env file at
-# all. These defaults are throwaway local-dev-only credentials, not
+# all for the default profile. The optional MCP profile requires measured
+# quota inputs below. Other defaults are throwaway local-dev-only credentials, not
 # production secrets; see docs/adr/0001-demo-identity-and-data-boundary.md.
 
 # Host ports deliberately avoid each service's own default (5432, 6379,
@@ -27,6 +28,13 @@ OIDC_AUDIENCE=lineageweave-api
 
 BACKEND_PORT=18420
 
+# Optional authenticated MCP profile. The quota pair is mandatory when the
+# profile is enabled and must come from that deployment's k6 capacity evidence.
+MCP_PORT=18001
+MCP_ALLOWED_ORIGINS=
+MCP_RATE_LIMIT_REQUESTS=
+MCP_RATE_LIMIT_WINDOW_SECONDS=
+
 # Optional. Empty = every LLM/vision channel is unavailable (Null client,
 # dropped and renormalized -- never a placeholder score). Point these at a
 # running contextual-orchestrator to turn the channels on.

CHANGELOG.d/2.18.1-current-contract-mcp-global-ask.md

@@ -0,0 +1,6 @@
+### Added
+
+- Added an authenticated Streamable HTTP MCP adapter that queues and reads the
+  same durable Global Ask jobs as REST, with exact-resource OAuth, bounded
+  pre-auth request admission, owner/affiliation scope preservation, and a
+  fail-closed distributed quota whose capacity inputs are deployment evidence.

README.md

@@ -134,7 +134,7 @@ carrying `corp_code` / `pu_code` as token claims -- these are throwaway
 local-dev credentials in a locally-run realm, never the org's real Keyverse
 tenant (see ADR 0001 for why).
 
-Host ports (15432, 16379, 18080, 18420) deliberately avoid each service's
+Host ports (15432, 16379, 18080, 18001, 18420) deliberately avoid each service's
 own default -- a dev machine commonly already runs its own
 Postgres/Redis/local server on those. Override via `.env` (copy
 `.env.example`) or inline if even those collide, e.g.
@@ -156,6 +156,18 @@ make seed   # scripts/seed_demo_data.py: inserts synthetic corp/account/post
 curl http://localhost:18420/healthz
 ```
 
+The optional authenticated MCP resource server submits and reads the same
+durable Global Ask jobs as REST. Enable it only with quota values established
+by the deployment's k6 capacity evidence; the service intentionally has no
+guessed request/window defaults:
+
+```bash
+MCP_RATE_LIMIT_REQUESTS=<measured-count> \
+MCP_RATE_LIMIT_WINDOW_SECONDS=<measured-window> \
+docker compose --profile mcp up mcp
+# Streamable HTTP resource: http://localhost:18001/mcp
+```
+
 `GET /api/posts`, `GET /api/posts/{post_id}`,
 `GET /api/posts/{post_id}/keymen`, `GET /api/keymen/{person_id}/related`,
 `GET /api/posts/{post_id}/affiliate-tree`,

backend/app/auth.py

@@ -125,8 +125,10 @@ def has_permission(self, permission_code: str) -> bool:
         return permission_code in self.permission_codes
 
 
-def _decode_access_token(token: str, settings: Settings) -> dict:
-    """Validate signature, issuer, resource audience, time claims, and subject."""
+def decode_access_token(
+    token: str, settings: Settings, *, audience: str | None = None
+) -> dict:
+    """Validate a token for the REST or an explicit resource audience."""
     required_claims = ["exp", "sub"]
     if settings.keyverse_claim_binding_required:
         required_claims.insert(1, "iat")
@@ -136,7 +138,7 @@ def _decode_access_token(token: str, settings: Settings) -> dict:
             key=_signing_key(settings, token),
             algorithms=["RS256"],
             issuer=settings.oidc_issuer,
-            audience=settings.oidc_audience,
+            audience=audience or settings.oidc_audience,
             leeway=settings.oidc_clock_skew_seconds,
             options={"require": required_claims},
         )
@@ -150,6 +152,11 @@ def _decode_access_token(token: str, settings: Settings) -> dict:
     return claims
 
 
+def _decode_access_token(token: str, settings: Settings) -> dict:
+    """Validate a REST bearer token against the configured API audience."""
+    return decode_access_token(token, settings)
+
+
 def _keyverse_account_claims(claims: dict) -> tuple[str, str, list[str]]:
     """Return Keyverse's atomic account scope, rejecting ambiguous wire shapes."""
     organization = claims.get("org")
@@ -177,13 +184,10 @@ def _keyverse_account_claims(claims: dict) -> tuple[str, str, list[str]]:
     return organization, workspace, [role.strip() for role in roles]
 
 
-async def get_current_account(
-    credentials: HTTPAuthorizationCredentials = Depends(_bearer_scheme),
-    pool: asyncpg.Pool = Depends(get_pool),
+async def resolve_current_account(
+    pool: asyncpg.Pool, claims: dict, settings: Settings
 ) -> CurrentAccount:
-    """Resolve the bearer token to a provisioned ``user_account`` row."""
-    settings = load_settings()
-    claims = _decode_access_token(credentials.credentials, settings)
+    """Resolve verified claims to database-owned scope and permissions."""
     subject = claims["sub"]
     keyverse_scope = (
         _keyverse_account_claims(claims)
@@ -269,3 +273,13 @@ async def get_current_account(
         process_unit_ids=frozenset(str(row["process_unit_id"]) for row in process_rows),
         permission_codes=frozenset(row["permission_code"] for row in permission_rows),
     )
+
+
+async def get_current_account(
+    credentials: HTTPAuthorizationCredentials = Depends(_bearer_scheme),
+    pool: asyncpg.Pool = Depends(get_pool),
+) -> CurrentAccount:
+    """Resolve the bearer token to a provisioned ``user_account`` row."""
+    settings = load_settings()
+    claims = _decode_access_token(credentials.credentials, settings)
+    return await resolve_current_account(pool, claims, settings)

backend/app/config.py

@@ -6,7 +6,7 @@
 
 import math
 import os
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 # Hard ceiling on one Global Ask job's answer computation, shared with the
 # worker in global_ask_queue.py so config validation and execution can never
@@ -65,6 +65,16 @@ class Settings:
     naruon_calendar_service_token: str
     rankweave_disabled: bool
     ontology_source_cursor_secret: str
+    mcp_resource_url: str = "http://localhost:18001/mcp"
+    mcp_audience: str = "http://localhost:18001/mcp"
+    mcp_required_scopes: list[str] = field(default_factory=list)
+    mcp_allowed_hosts: list[str] = field(
+        default_factory=lambda: ["localhost:*", "127.0.0.1:*", "mcp:8001"]
+    )
+    mcp_allowed_origins: list[str] = field(default_factory=list)
+    mcp_max_request_bytes: int = 65_536
+    mcp_rate_limit_requests: int | None = None
+    mcp_rate_limit_window_seconds: int | None = None
 
     @property
     def keycloak_jwks_uri(self) -> str:
@@ -83,7 +93,9 @@ def _validated_answer_timeout(raw: str) -> float:
     try:
         value = float(raw)
     except ValueError as exc:
-        raise ValueError("ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS must be a number") from exc
+        raise ValueError(
+            "ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS must be a number"
+        ) from exc
     if not math.isfinite(value) or not 0 < value < GLOBAL_ASK_JOB_DEADLINE_SECONDS:
         raise ValueError(
             "ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS must be a finite number greater"
@@ -92,6 +104,20 @@ def _validated_answer_timeout(raw: str) -> float:
     return value
 
 
+def _optional_positive_int(name: str) -> int | None:
+    """Parse an optional positive deployment integer without inventing a default."""
+    raw = os.environ.get(name, "").strip()
+    if not raw:
+        return None
+    try:
+        value = int(raw, 10)
+    except ValueError as exc:
+        raise ValueError(f"{name} must be a base-10 integer") from exc
+    if value <= 0:
+        raise ValueError(f"{name} must be positive")
+    return value
+
+
 def load_settings() -> Settings:
     """Read Settings from the environment, with local-dev defaults only."""
     keycloak_base_url = os.environ.get("KEYCLOAK_BASE_URL", "http://localhost:18080")
@@ -103,7 +129,9 @@ def load_settings() -> Settings:
     keyverse_issuer = os.environ.get("KEYVERSE_ISSUER", "").strip()
     generic_oidc_issuer = os.environ.get("OIDC_ISSUER", "").strip()
     external_oidc = bool(keyverse_issuer or generic_oidc_issuer)
-    oidc_issuer = (keyverse_issuer or generic_oidc_issuer or keycloak_issuer).rstrip("/")
+    oidc_issuer = (keyverse_issuer or generic_oidc_issuer or keycloak_issuer).rstrip(
+        "/"
+    )
     oidc_client_id = (
         os.environ.get("KEYVERSE_CLIENT_ID", "").strip()
         or os.environ.get("OIDC_CLIENT_ID", "").strip()
@@ -119,9 +147,13 @@ def load_settings() -> Settings:
             "do not infer a resource-server audience from the browser client id"
         )
     oidc_audience = configured_audience or "lineageweave-api"
-    oidc_discovery_uri = os.environ.get("KEYVERSE_DISCOVERY_URI", "").strip() or os.environ.get(
-        "OIDC_DISCOVERY_URI", ""
+    mcp_resource_url = os.environ.get(
+        "MCP_RESOURCE_URL", "http://localhost:18001/mcp"
     ).strip()
+    oidc_discovery_uri = (
+        os.environ.get("KEYVERSE_DISCOVERY_URI", "").strip()
+        or os.environ.get("OIDC_DISCOVERY_URI", "").strip()
+    )
     if not oidc_discovery_uri:
         discovery_base = oidc_issuer if external_oidc else keycloak_base_url
         oidc_discovery_uri = (
@@ -161,7 +193,9 @@ def load_settings() -> Settings:
         keyverse_claim_binding_required=bool(keyverse_issuer),
         frontend_origins=[
             origin.strip()
-            for origin in os.environ.get("FRONTEND_ORIGINS", "http://localhost:5173").split(",")
+            for origin in os.environ.get(
+                "FRONTEND_ORIGINS", "http://localhost:5173"
+            ).split(",")
             if origin.strip()
         ],
         orchestrator_base_url=os.environ.get("ORCHESTRATOR_BASE_URL", ""),
@@ -178,9 +212,33 @@ def load_settings() -> Settings:
         naruon_calendar_service_token=os.environ.get(
             "NARUON_CALENDAR_SERVICE_TOKEN", ""
         ).strip(),
-        rankweave_disabled=os.environ.get("RANKWEAVE_DISABLED", "")
-        .strip()
-        .lower()
+        rankweave_disabled=os.environ.get("RANKWEAVE_DISABLED", "").strip().lower()
         in {"1", "true", "yes", "on"},
-        ontology_source_cursor_secret=os.environ.get("ONTOLOGY_SOURCE_CURSOR_SECRET", "").strip(),
+        ontology_source_cursor_secret=os.environ.get(
+            "ONTOLOGY_SOURCE_CURSOR_SECRET", ""
+        ).strip(),
+        mcp_resource_url=mcp_resource_url,
+        mcp_audience=os.environ.get("MCP_AUDIENCE", mcp_resource_url).strip(),
+        mcp_required_scopes=[
+            item.strip()
+            for item in os.environ.get("MCP_REQUIRED_SCOPES", "").split(",")
+            if item.strip()
+        ],
+        mcp_allowed_hosts=[
+            item.strip()
+            for item in os.environ.get(
+                "MCP_ALLOWED_HOSTS", "localhost:*,127.0.0.1:*,mcp:8001"
+            ).split(",")
+            if item.strip()
+        ],
+        mcp_allowed_origins=[
+            item.strip()
+            for item in os.environ.get("MCP_ALLOWED_ORIGINS", "").split(",")
+            if item.strip()
+        ],
+        mcp_max_request_bytes=_optional_positive_int("MCP_MAX_REQUEST_BYTES") or 65_536,
+        mcp_rate_limit_requests=_optional_positive_int("MCP_RATE_LIMIT_REQUESTS"),
+        mcp_rate_limit_window_seconds=_optional_positive_int(
+            "MCP_RATE_LIMIT_WINDOW_SECONDS"
+        ),
     )

backend/app/global_ask_service.py

@@ -0,0 +1,92 @@
+"""Shared durable Global Ask application service for REST and MCP."""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+from uuid import UUID
+
+import asyncpg
+import redis.asyncio as redis
+from fastapi import HTTPException, status
+
+from backend.app.auth import CurrentAccount
+from backend.app.global_ask_queue import enqueue_global_ask_job
+from backend.app.source_post_revision import parse_as_of_clock
+
+
+async def submit_global_ask(
+    *,
+    pool: asyncpg.Pool,
+    valkey: redis.Redis,
+    account: CurrentAccount,
+    question: str,
+    verify_external: bool,
+    knowledge_cutoff: str | None,
+    service_available: bool,
+) -> dict[str, Any]:
+    """Validate and enqueue one durable owner-scoped Global Ask job."""
+    if not account.has_permission("post_read"):
+        raise HTTPException(status.HTTP_403_FORBIDDEN, "post_read permission required")
+    normalized_question = question.strip()
+    if not normalized_question:
+        raise HTTPException(
+            status.HTTP_422_UNPROCESSABLE_CONTENT, "question is required"
+        )
+    cutoff = None
+    if knowledge_cutoff is not None:
+        try:
+            cutoff = parse_as_of_clock(knowledge_cutoff)
+        except ValueError as exc:
+            raise HTTPException(
+                status.HTTP_422_UNPROCESSABLE_CONTENT,
+                "knowledge_cutoff must be an ISO-8601 timestamp",
+            ) from exc
+    async with pool.acquire() as conn:
+        if cutoff is not None and cutoff > await conn.fetchval("select now()"):
+            raise HTTPException(
+                status.HTTP_422_UNPROCESSABLE_CONTENT,
+                "knowledge_cutoff must be at or before the database clock",
+            )
+        if not service_available:
+            raise HTTPException(
+                status.HTTP_503_SERVICE_UNAVAILABLE,
+                "Ask Agent is unavailable. Ask an administrator to configure the analysis service, then retry.",
+            )
+        job_id = await enqueue_global_ask_job(
+            conn,
+            valkey,
+            requesting_account_id=account.user_account_id,
+            question_text=normalized_question,
+            verify_external_requested=verify_external,
+            knowledge_cutoff=cutoff,
+            corporate_entity_ids=account.corporate_entity_ids,
+            process_unit_ids=account.process_unit_ids,
+        )
+    return {"ask_job_id": job_id, "job_status_code": "queued"}
+
+
+async def read_global_ask_job(
+    *, pool: asyncpg.Pool, account: CurrentAccount, ask_job_id: UUID
+) -> dict[str, Any]:
+    """Read one owner's durable Global Ask status and persisted result."""
+    if not account.has_permission("post_read"):
+        raise HTTPException(status.HTTP_403_FORBIDDEN, "post_read permission required")
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow(
+            "select requesting_account_id, job_status_code, answer_payload,"
+            " failure_detail from global_ask_job where global_ask_job_id = $1",
+            ask_job_id,
+        )
+    if row is None or str(row["requesting_account_id"]) != account.user_account_id:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "ask job not found")
+    body: dict[str, Any] = {
+        "ask_job_id": str(ask_job_id),
+        "job_status_code": row["job_status_code"],
+    }
+    if row["job_status_code"] == "succeeded" and row["answer_payload"] is not None:
+        payload = row["answer_payload"]
+        body["answer"] = json.loads(payload) if isinstance(payload, str) else payload
+    if row["job_status_code"] == "failed":
+        body["failure_detail"] = row["failure_detail"]
+    return body