from __future__ import annotations from collections.abc import AsyncIterator, Callable from contextlib import AsyncExitStack, asynccontextmanager, suppress import logging from uuid import uuid4 import uvicorn from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from starlette.concurrency import run_in_threadpool from qonto_assistant import __version__ from qonto_assistant.audit import AuditLogger 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.flex_auth_client import FlexAuthCheckClient from qonto_assistant.key_cape_auth import KeyCapeTokenVerifier from qonto_assistant.live_authorization import LiveAuthorizationGate 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 from qonto_assistant.rate_limits import ConcurrencyLimiter, RateLimiter from qonto_assistant.security_watch import DenyEscalationTracker from qonto_assistant.service import CapabilityService from qonto_assistant.tenant_engine_client import TenantEngineClient def create_app( *, settings: Settings | None = None, service: CapabilityService | None = None, audit_logger: AuditLogger | None = None, rate_limiter: RateLimiter | None = None, concurrency_limiter: ConcurrencyLimiter | None = None, key_cape_verifier: KeyCapeTokenVerifier | None = None, ) -> FastAPI: settings = settings or Settings.from_env() audit_logger = audit_logger or AuditLogger() service = service or _build_service( settings=settings, audit_logger=audit_logger, rate_limiter=rate_limiter, concurrency_limiter=concurrency_limiter, ) if key_cape_verifier is None and settings.key_cape_jwks_url: key_cape_verifier = KeyCapeTokenVerifier( jwks_url=settings.key_cape_jwks_url, issuer=settings.key_cape_issuer, audience=settings.key_cape_audience, required=settings.key_cape_required, default_lane=settings.default_actor_lane, timeout_seconds=settings.key_cape_timeout_seconds, cache_seconds=settings.key_cape_cache_seconds, ) mcp_server = create_mcp_server(settings=settings, service=service, key_cape_verifier=key_cape_verifier) 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]: async with AsyncExitStack() as stack: await stack.enter_async_context(mcp_app.router.lifespan_context(mcp_app)) yield with suppress(Exception): service.client.close() app = FastAPI(title="qonto-assistant", version=__version__, lifespan=lifespan) app.state.settings = settings app.state.service = service app.mount("/mcp", mcp_app) @app.exception_handler(QontoAssistantError) def handle_qonto_error(_: Request, exc: QontoAssistantError) -> JSONResponse: body = {"error_code": exc.error_code, "detail": exc.message} if isinstance(exc, UpstreamError) and exc.upstream_status is not None: body["upstream_status"] = exc.upstream_status return JSONResponse(status_code=exc.status_code, content=body) @app.get("/v1/health") async def health() -> dict[str, str]: return { "status": "ok", "service": settings.service_name, "version": __version__, "policy_file": str(settings.policy_file), } @app.get("/v1/accounts") async def get_accounts(request: Request) -> JSONResponse: claims = actor_claims_from_request(request, settings, key_cape_verifier=key_cape_verifier) request_id = _request_id(request) payload = await run_in_threadpool(service.get_accounts, claims=claims, request_id=request_id) return JSONResponse(content=payload, headers={"X-Request-ID": request_id}) @app.get("/v1/transactions") async def get_transactions( request: Request, account_slug: str | None = None, page: int = 1, page_size: int = 50, window_days: int = 31, status: str | None = "completed", side: str | None = None, ) -> JSONResponse: claims = actor_claims_from_request(request, settings, key_cape_verifier=key_cape_verifier) request_id = _request_id(request) payload = await run_in_threadpool( service.list_transactions, claims=claims, request_id=request_id, account_slug=account_slug, page=page, page_size=page_size, window_days=window_days, status=status, side=side, ) return JSONResponse(content=payload, headers={"X-Request-ID": request_id}) @app.get("/v1/snapshot") async def get_snapshot( request: Request, window_days: int = 31, page_size: int = 50, ) -> JSONResponse: claims = actor_claims_from_request(request, settings, key_cape_verifier=key_cape_verifier) request_id = _request_id(request) payload = await run_in_threadpool( service.get_snapshot, claims=claims, request_id=request_id, window_days=window_days, page_size=page_size, ) return JSONResponse(content=payload, headers={"X-Request-ID": request_id}) return app def _request_id(request: Request) -> str: existing = getattr(request.state, "request_id", None) if existing: return existing request.state.request_id = request.headers.get("x-request-id", str(uuid4())) return request.state.request_id def _build_service( *, settings: Settings, audit_logger: AuditLogger, rate_limiter: RateLimiter | None, concurrency_limiter: ConcurrencyLimiter | None, ) -> CapabilityService: policy = PolicyEngine.from_file( settings.policy_file, required_scope=settings.required_scope, enforce_scope=settings.enforce_scope, ) if settings.qonto_fixture_dir is not None: client = FixtureQontoClient(fixture_dir=settings.qonto_fixture_dir) else: credential_provider = build_credential_provider(settings) client = QontoClient( base_url=settings.qonto_base_url, organization_path=settings.qonto_organization_path, transactions_path=settings.qonto_transactions_path, auth_mode=settings.qonto_auth_mode, timeout_seconds=settings.qonto_timeout_seconds, max_retries=settings.qonto_max_retries, credential_provider=credential_provider, ) return CapabilityService( client=client, policy=policy, audit_logger=audit_logger, rate_limiter=rate_limiter or RateLimiter( limit=settings.rate_limit_requests, window_seconds=settings.rate_limit_window_seconds, ), concurrency_limiter=concurrency_limiter or ConcurrencyLimiter(limit=settings.max_concurrency), deny_escalation_tracker=( DenyEscalationTracker( threshold=settings.deny_escalation_threshold, window_seconds=settings.deny_escalation_window_seconds, lockout_seconds=settings.deny_escalation_lockout_seconds, ) if settings.deny_escalation_enabled else None ), live_authorization_gate=_build_live_authorization_gate(settings), ) def _build_live_authorization_gate(settings: Settings) -> LiveAuthorizationGate | None: if not settings.flex_auth_base_url: return None tenant_engine_client = ( TenantEngineClient( base_url=settings.tenant_engine_base_url, timeout_seconds=settings.tenant_engine_timeout_seconds, ) if settings.tenant_engine_base_url else None ) return LiveAuthorizationGate( flex_auth_client=FlexAuthCheckClient( base_url=settings.flex_auth_base_url, timeout_seconds=settings.flex_auth_timeout_seconds, ), tenant_engine_client=tenant_engine_client, required_tenant_roles=settings.tenant_engine_required_roles, ) def main() -> None: settings = Settings.from_env() logging.basicConfig(level=logging.INFO, format="%(message)s") uvicorn.run( "qonto_assistant.app:create_app", factory=True, host=settings.host, port=settings.port, reload=False, )