src/mcp_sourcecontrol/config.py - master - pickling/mcp-sourcecontrol-python

src/mcp_sourcecontrol/config.py
172 строки6 KB
pickling-21
Python-порт mcp-sourcecontrol: 20 инструментов, stdio и Streamable HTTP
29 июл 2026, 02:18
29 июл 2026, 02:18bc8bde4
"""Разбор ENV-переменных, загрузка tenants.json, построение API URL, разбор HTTP-заголовков.

Порт src/config.ts. Логика (включая fallback'и) сохранена; единственное отличие —
исправлен краевой случай fallback-ветки buildApiUrl для доменов, уже начинающихся с "api.".
"""

from __future__ import annotations

import json
import os
import re
from collections.abc import Mapping
from dataclasses import dataclass, replace
from pathlib import Path

from .log import LogLevel

DEFAULT_TENANT_ID = "167df516-b484-48d5-ac64-d8645d456db9"

_LOG_LEVELS = {"debug", "info", "warn", "error"}
_API_PATH_RE = re.compile(r"/(?:sc|sccd|scgs)/api/v3")


@dataclass(frozen=True)
class TenantEntry:
    name: str
    uuid: str
    instance: str
    context: str
    description: str | None = None


@dataclass(frozen=True)
class ResolvedConfig:
    api_url: str
    tenant_id: str
    token: str | None
    log_level: LogLevel
    tls_reject_unauthorized: bool


def _candidate_paths() -> list[Path]:
    """Где искать tenants.json: checkout репозитория, затем копия внутри установленного пакета.

    Файл ищется относительно расположения модуля, а не process cwd — MCP-клиент
    может запускать процесс из любого каталога.
    """
    package_dir = Path(__file__).resolve().parent
    return [
        package_dir.parents[1] / "config" / "tenants.json",  # src-layout: корень репозитория
        package_dir / "config" / "tenants.json",  # копия, вложенная в wheel при сборке
    ]


def load_tenants() -> list[TenantEntry]:
    for path in _candidate_paths():
        if not path.is_file():
            continue
        try:
            raw = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            return []
        # Поддерживаются оба формата: {tenants: [...]} и просто [...]
        items = raw if isinstance(raw, list) else raw.get("tenants") if isinstance(raw, dict) else None
        if not isinstance(items, list):
            return []
        tenants: list[TenantEntry] = []
        for item in items:
            if not isinstance(item, dict) or not {"uuid", "instance", "context"} <= item.keys():
                continue
            tenants.append(
                TenantEntry(
                    name=str(item.get("name", "")),
                    uuid=str(item["uuid"]),
                    instance=str(item["instance"]),
                    context=str(item["context"]),
                    description=item.get("description"),
                )
            )
        return tenants
    return []


def _domain_of(url: str) -> str:
    return re.sub(r"^https?://", "", url).split("/")[0]


def resolve_tenant_id(base_url: str) -> str:
    """UUID тенанта для указанного base URL.

    Понимает и base-формат (sc-ci.sber.ru), и полный API URL (api.sc-ci.sber.ru/sc/api/v3).
    """
    tenants = load_tenants()
    if not tenants:
        return DEFAULT_TENANT_ID

    raw_domain = _domain_of(base_url)
    # instance тенантов хранится как api.* — нормализуем искомый домен
    lookup_domain = raw_domain if raw_domain.startswith("api.") else f"api.{raw_domain}"

    for tenant in tenants:
        if _domain_of(tenant.instance) == lookup_domain:
            return tenant.uuid

    return tenants[0].uuid


def build_api_url(base_url: str) -> str:
    """Полный API URL из base URL; уже полный URL возвращается без изменений.

    Примеры:
        https://sc-ci.sber.ru               → https://api.sc-ci.sber.ru/sc/api/v3
        https://sc-cd.sber.ru               → https://api.sc-cd.sber.ru/sccd/api/v3
        https://api.sc-ci.sber.ru/sc/api/v3 → без изменений
    """
    if _API_PATH_RE.search(base_url):
        return base_url

    clean = _domain_of(base_url)
    expected_instance = clean if clean.startswith("api.") else f"api.{clean}"

    for tenant in load_tenants():
        instance = _domain_of(tenant.instance)
        if instance == expected_instance:
            return f"https://{instance}{tenant.context}/api/v3"

    # Fallback: контекст выводится из префикса поддомена
    bare = clean[4:] if clean.startswith("api.") else clean
    parts = bare.split(".")
    prefix = parts[0] if parts and parts[0] else "sc"
    domain = ".".join(parts[1:])
    context = "/sc"
    if "cd" in prefix and "ci" not in prefix:
        context = "/sccd"
    elif "gs" in prefix:
        context = "/scgs"

    return f"https://api.{prefix}.{domain}{context}/api/v3"


def resolve_session_config(headers: Mapping[str, str], default_config: ResolvedConfig) -> ResolvedConfig:
    """Конфигурация сессии: заголовки X-SC-Url/X-SC-Token поверх дефолтной конфигурации."""
    lowered = {key.lower(): value for key, value in headers.items()}
    header_url = lowered.get("x-sc-url") or None
    header_token = lowered.get("x-sc-token") or None

    if not header_url and not header_token:
        return default_config

    base_url = header_url or default_config.api_url
    return replace(
        default_config,
        api_url=build_api_url(base_url),
        tenant_id=resolve_tenant_id(base_url),
        token=header_token or default_config.token,
    )


def resolve_config() -> ResolvedConfig:
    base_url = os.environ.get("SC_API_URL") or "https://sc-ci.sber.ru"
    token = os.environ.get("SC_TOKEN") or os.environ.get("BITBUCKET_TOKEN")
    raw_level = os.environ.get("SC_LOG_LEVEL") or "info"
    log_level = raw_level if raw_level in _LOG_LEVELS else "info"
    tls_reject_unauthorized = (os.environ.get("SC_TLS_REJECT_UNAUTHORIZED") or "false") != "false"

    return ResolvedConfig(
        api_url=build_api_url(base_url),
        tenant_id=resolve_tenant_id(base_url),
        token=token,
        log_level=log_level,  # type: ignore[arg-type]
        tls_reject_unauthorized=tls_reject_unauthorized,
    )