DataOps-platform/app/core/mcp/bootstrap.py

"""Fail-closed production dependency wiring for DataOps MCP processes."""

from __future__ import annotations

import os

from sqlalchemy import create_engine

from app.core.mcp.canary import CanaryVerifier
from app.core.mcp.context import ContextService
from app.core.mcp.gateway import SchedulingGateway
from app.core.mcp.persistence import (
    PostgresAuditSink,
    PostgresContextRepository,
    PostgresSchedulingPlanStore,
)
from app.core.orchestration.engines.kestra import KestraAdapter
from app.runner.auth import TaskTokenIssuer


def _required(name):
    value = str(os.getenv(name) or "").strip()
    if not value:
        raise ValueError(f"{name} is required")
    return value


def _bounded_integer(name, *, default, minimum, maximum):
    source = str(os.getenv(name, default)).strip()
    try:
        value = int(source)
    except ValueError as exc:
        raise ValueError(f"{name} must be an integer") from exc
    if value < minimum or value > maximum:
        raise ValueError(f"{name} must be between {minimum} and {maximum}")
    return value


def _database_engine():
    database_url = _required("DATAOPS_MCP_DATABASE_URL")
    if not database_url.startswith(("postgresql://", "postgresql+psycopg2://")):
        raise ValueError("DATAOPS_MCP_DATABASE_URL must use PostgreSQL")
    return create_engine(
        database_url,
        pool_pre_ping=True,
        pool_size=5,
        max_overflow=5,
        pool_recycle=300,
    )


def _kestra_adapter():
    return KestraAdapter(
        base_url=_required("KESTRA_BASE_URL"),
        username=_required("KESTRA_USERNAME"),
        password=_required("KESTRA_PASSWORD"),
        tenant_id=str(os.getenv("KESTRA_TENANT_ID", "main")).strip() or "main",
        timeout=_bounded_integer(
            "KESTRA_HTTP_TIMEOUT_SECONDS",
            default=30,
            minimum=1,
            maximum=300,
        ),
    )


def build_scheduling_gateway():
    engine = _database_engine()
    plans = PostgresSchedulingPlanStore(engine)
    return SchedulingGateway(
        plans=plans,
        audit=PostgresAuditSink(engine),
        engine=_kestra_adapter(),
        token_issuer=TaskTokenIssuer(
            _required("RUNNER_TASK_TOKEN_SECRET"),
            ttl_seconds=_bounded_integer(
                "RUNNER_TASK_TOKEN_TTL_SECONDS",
                default=60,
                minimum=1,
                maximum=300,
            ),
        ),
    )


def build_context_service():
    engine = _database_engine()
    repository = PostgresContextRepository(
        engine,
        max_concurrency=_bounded_integer(
            "DATAOPS_MCP_CONTEXT_MAX_CONCURRENCY",
            default=10,
            minimum=1,
            maximum=1000,
        ),
    )
    return ContextService(repository)


def build_canary_verifier():
    engine = _database_engine()
    return CanaryVerifier(
        plans=PostgresSchedulingPlanStore(engine),
        engine=_kestra_adapter(),
    )