QONTO-WP-0003-T03: MCP client auth + shared multi-harness config snippet

Gate /mcp with a shared-secret bearer token (QONTO_ASSISTANT_MCP_TOKEN,
mcp_auth.py::BearerTokenAuthMiddleware, constant-time compare, REST
untouched) since no OIDC issuer exists in this fleet yet -- pointing
FastMCP's OAuth Protected Resource flow at a non-existent issuer would be
worse than not having it. This token is a service credential, never a bank
credential; per-actor identity stays the existing X-Actor-* convention.

Add docs/mcp-integration.md: tool catalog, the two-layer auth model (workload
auth today vs. deferred OIDC target), and one shared {"mcpServers": {...}}
client config snippet (url + headers) usable across Claude Code, Claude
Desktop, Cursor, and Codex/Grok-style harnesses.

Verified live using only that snippet: unauthenticated and wrong-token
requests get 401 before reaching any tool; a request built from the
snippet's URL + headers lists tools and calls qonto_org_summary
successfully against the fixture-backed server.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
tegwick 2026-07-23 09:41:25 +02:00
parent e1e47ae304
commit d2ffd372b5
9 changed files with 259 additions and 5 deletions

View file

@ -16,6 +16,7 @@ from qonto_assistant.auth import actor_claims_from_request
from qonto_assistant.config import Settings
from qonto_assistant.credentials import build_credential_provider
from qonto_assistant.errors import QontoAssistantError, UpstreamError
from qonto_assistant.mcp_auth import BearerTokenAuthMiddleware
from qonto_assistant.mcp_server import create_mcp_server
from qonto_assistant.policy import PolicyEngine
from qonto_assistant.qonto_client import FixtureQontoClient, QontoClient
@ -42,6 +43,8 @@ def create_app(
mcp_server = create_mcp_server(settings=settings, service=service)
mcp_app = mcp_server.streamable_http_app()
if settings.mcp_auth_token:
mcp_app.add_middleware(BearerTokenAuthMiddleware, token=settings.mcp_auth_token)
@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:

View file

@ -32,6 +32,7 @@ class Settings:
openbao_path: str
openbao_command: str
openbao_timeout_seconds: float
mcp_auth_token: str | None
host: str
port: int
@ -69,6 +70,7 @@ class Settings:
openbao_path=os.getenv("QONTO_OPENBAO_PATH", "tenants/binky/qonto-api"),
openbao_command=os.getenv("QONTO_OPENBAO_COMMAND", "bao"),
openbao_timeout_seconds=float(os.getenv("QONTO_OPENBAO_TIMEOUT_SECONDS", "5")),
mcp_auth_token=os.getenv("QONTO_ASSISTANT_MCP_TOKEN") or None,
host=os.getenv("QONTO_ASSISTANT_HOST", "127.0.0.1"),
port=int(os.getenv("QONTO_ASSISTANT_PORT", "8080")),
)

View file

@ -0,0 +1,52 @@
from __future__ import annotations
import hmac
from starlette.requests import Headers
from starlette.responses import JSONResponse
from starlette.types import ASGIApp, Receive, Scope, Send
class BearerTokenAuthMiddleware:
"""Workload auth gate for the MCP endpoint: static shared-secret bearer token.
QONTO-WP-0003-T03 scope note: the blueprint's target is OIDC/workload
identity (specs/ArchitectureBlueprint.md §4.7), but no OIDC issuer exists
in this fleet yet. A shared-secret bearer token is the "or similar"
workload-auth primitive that's actually deployable today -- it keeps
unauthenticated callers off the MCP surface entirely, same custody
principle as REST: this token is a service credential, never a bank
credential, and per-actor identity still comes from X-Actor-* headers
checked by the policy kernel. Upgrading to real OIDC/workload identity
federation is fleet-level follow-on work, not something this repo can
stand up alone.
"""
def __init__(self, app: ASGIApp, *, token: str) -> None:
self.app = app
self.token = token
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] != "http":
await self.app(scope, receive, send)
return
headers = Headers(scope=scope)
if not _token_matches(headers.get("authorization", ""), self.token):
response = JSONResponse(
{"error_code": "unauthorized", "detail": "Missing or invalid bearer token"},
status_code=401,
headers={"WWW-Authenticate": "Bearer"},
)
await response(scope, receive, send)
return
await self.app(scope, receive, send)
def _token_matches(authorization_header: str, expected_token: str) -> bool:
prefix = "Bearer "
if not authorization_header.startswith(prefix):
return False
presented = authorization_header[len(prefix) :]
return hmac.compare_digest(presented, expected_token)