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

src/mcp_sourcecontrol/client.py
363 строки14 KB
pickling-21
Python-порт mcp-sourcecontrol: 20 инструментов, stdio и Streamable HTTP
29 июл 2026, 02:18
29 июл 2026, 02:18bc8bde4
"""HTTP-клиент SourceControl REST API. Порт src/client/sc-client.ts.

Пути, тела запросов и кодирование параметров повторяют TS-версию байт-в-байт,
чтобы обе реализации были взаимозаменяемы против одного API.
"""

from __future__ import annotations

import json
from typing import Any, Literal
from urllib.parse import quote, urlencode

import httpx

from .log import SimpleLogger

LineType = Literal["ADDED", "REMOVED", "CONTEXT"]

_UNSET: Any = object()

# Тайм-аут, которого у TS-версии не было (fetch ждёт бесконечно) — 30 секунд разумнее.
DEFAULT_TIMEOUT_SECONDS = 30.0


class ScError(Exception):
    """Базовая ошибка клиента SourceControl."""


class ScApiError(ScError):
    """API ответил не-2xx статусом. Сообщение — 'HTTP {status}: {текст ответа}'."""

    def __init__(self, status: int, body: str) -> None:
        super().__init__(f"HTTP {status}: {body}")
        self.status = status
        self.body = body


class ScNetworkError(ScError):
    """До API не удалось достучаться. code — имя класса исходного исключения httpx."""

    def __init__(self, message: str, code: str) -> None:
        super().__init__(f"Network error: {message}")
        self.code = code


def _encode_path(file_path: str) -> str:
    """Кодирует каждый сегмент пути отдельно, сохраняя '/' (как в TS-версии)."""
    return "/".join(quote(segment, safe="") for segment in file_path.split("/") if segment)


class SimpleScClient:
    """Клиент SourceControl API: авторизация, обработка ошибок, все 19 методов."""

    def __init__(
        self,
        base_url: str,
        tenant_id: str,
        token: str | None = None,
        logger: SimpleLogger | None = None,
        *,
        verify: bool = False,
        transport: httpx.AsyncBaseTransport | None = None,
        timeout: float = DEFAULT_TIMEOUT_SECONDS,
    ) -> None:
        self.base_url = base_url
        self.tenant_id = tenant_id
        self.token = token
        self.logger = logger
        self.verify = verify
        self.transport = transport
        self.timeout = timeout

    async def _do_fetch(self, method: str, url_path: str, *, body: Any = _UNSET, text: bool = False) -> Any:
        url = f"{self.base_url}{url_path}"
        headers = {
            "Content-Type": "application/json",
            "User-Agent": "mcp-sourcecontrol/1.0",
        }
        if text:
            headers["Accept"] = "text/plain"
        if self.token:
            headers["Authorization"] = f"Bearer {self.token}"

        content = None if body is _UNSET else json.dumps(body, ensure_ascii=False, separators=(",", ":"))

        if self.logger:
            self.logger.debug("API request", {"method": method, "url": url, "hasToken": bool(self.token)})

        try:
            async with httpx.AsyncClient(
                verify=self.verify,
                transport=self.transport,
                timeout=self.timeout,
                follow_redirects=True,
            ) as client:
                response = await client.request(method, url, headers=headers, content=content)
            if self.logger:
                self.logger.debug("API response", {"status": response.status_code})
        except httpx.HTTPError as error:
            if self.logger:
                self.logger.error("API fetch error", {"method": method, "url": url, "error": str(error)})
            raise ScNetworkError(str(error) or "fetch failed", code=type(error).__name__) from error

        if not response.is_success:
            error_text = response.text
            if self.logger:
                self.logger.error(
                    "API request failed",
                    {"method": method, "url": url, "status": response.status_code, "errorText": error_text},
                )
            raise ScApiError(response.status_code, error_text)

        if response.status_code == 204:
            return None

        return response.text if text else response.json()

    def _repo_path(self, project: str, repository: str) -> str:
        return f"/repos/{self.tenant_id}/{project}/{repository}"

    # --- Ветки ---------------------------------------------------------------

    async def list_branches(self, project: str, repository: str, page: int = 1, limit: int = 30) -> Any:
        return await self._do_fetch("GET", f"{self._repo_path(project, repository)}/branches?page={page}&limit={limit}")

    async def create_branch(self, project: str, repository: str, branch_name: str, parent_commit_sha: str) -> Any:
        return await self._do_fetch(
            "POST",
            f"{self._repo_path(project, repository)}/branches",
            body={"new_branch": branch_name, "parent_commit_SHA": parent_commit_sha},
        )

    async def get_branch(self, project: str, repository: str, branch_name: str) -> Any:
        return await self._do_fetch(
            "GET", f"{self._repo_path(project, repository)}/branches/{quote(branch_name, safe='')}"
        )

    async def delete_branch(self, project: str, repository: str, branch_name: str) -> None:
        await self._do_fetch(
            "DELETE", f"{self._repo_path(project, repository)}/branches/{quote(branch_name, safe='')}"
        )

    # --- Коммиты -------------------------------------------------------------

    async def list_commits(
        self,
        project: str,
        repository: str,
        branch: str | None = None,
        since: str | None = None,
        until: str | None = None,
        file_path: str | None = None,
        page: int = 1,
        limit: int = 30,
    ) -> Any:
        params: list[tuple[str, str]] = [("page", str(page)), ("limit", str(limit))]
        if branch:
            params.append(("branch", branch))
        if since:
            params.append(("since", since))
        if until:
            params.append(("until", until))
        if file_path:
            params.append(("path", file_path))
        return await self._do_fetch("GET", f"{self._repo_path(project, repository)}/commits?{urlencode(params)}")

    # --- Pull requests ---------------------------------------------------------

    async def list_pull_requests(
        self,
        project: str,
        repository: str,
        state: str | None = None,
        sort: str | None = None,
        milestone: str | None = None,
        labels: str | None = None,
        target_branch: str | None = None,
        author: str | None = None,
        page: int = 1,
        limit: int = 30,
    ) -> Any:
        params: list[tuple[str, str]] = [("page", str(page)), ("limit", str(limit))]
        if state:
            params.append(("state", state))
        if sort:
            params.append(("sort", sort))
        if milestone:
            params.append(("milestone", milestone))
        if labels:
            params.append(("labels", labels))
        if target_branch:
            params.append(("target_branch", target_branch))
        if author:
            params.append(("author", author))
        return await self._do_fetch("GET", f"{self._repo_path(project, repository)}/pulls?{urlencode(params)}")

    async def create_pull_request(
        self,
        project: str,
        repository: str,
        head: str,
        base: str,
        title: str,
        body: str | None = None,
    ) -> Any:
        payload: dict[str, Any] = {"title": title, "head": head, "base": base}
        if body is not None:
            payload["body"] = body
        return await self._do_fetch("POST", f"{self._repo_path(project, repository)}/pulls", body=payload)

    async def get_pull_request(self, project: str, repository: str, pull_request_index: int) -> Any:
        return await self._do_fetch("GET", f"{self._repo_path(project, repository)}/pulls/{pull_request_index}")

    async def merge_pull_request(
        self, project: str, repository: str, pull_request_index: int, merge_method: str
    ) -> Any:
        return await self._do_fetch(
            "POST",
            f"{self._repo_path(project, repository)}/pulls/{pull_request_index}/merge",
            body={"merge_method": merge_method},
        )

    async def decline_pull_request(
        self, project: str, repository: str, pull_request_index: int, message: str | None = None
    ) -> Any:
        payload: dict[str, Any] = {}
        if message:
            payload["message"] = message
        return await self._do_fetch(
            "POST", f"{self._repo_path(project, repository)}/pulls/{pull_request_index}/decline", body=payload
        )

    async def update_pull_request(
        self,
        project: str,
        repository: str,
        pull_request_index: int,
        title: str | None = None,
        body: str | None = None,
        assignees: list[Any] | None = None,
        reviewers: list[Any] | None = None,
        labels: list[str] | None = None,
        milestone: str | None = None,
        due_date: str | None = None,
        target_branch: str | None = None,
    ) -> Any:
        payload: dict[str, Any] = {}
        if title is not None:
            payload["title"] = title
        if body is not None:
            payload["body"] = body
        if assignees is not None:
            payload["assignees"] = assignees
        if reviewers is not None:
            payload["reviewers"] = reviewers
        if labels is not None:
            payload["labels"] = labels
        if milestone is not None:
            payload["milestone"] = milestone
        if due_date is not None:
            payload["due_date"] = due_date
        if target_branch is not None:
            payload["toRef"] = {"id": f"refs/heads/{target_branch}"}
        return await self._do_fetch(
            "PATCH", f"{self._repo_path(project, repository)}/pulls/{pull_request_index}", body=payload
        )

    async def get_pull_request_diff(
        self, project: str, repository: str, pull_request_index: int, file_path: str | None = None
    ) -> str:
        url_path = f"{self._repo_path(project, repository)}/pulls/{pull_request_index}/diff"
        if file_path:
            url_path += f"/{_encode_path(file_path)}"
        return await self._do_fetch("GET", url_path, text=True)

    async def list_pull_request_files(
        self, project: str, repository: str, pull_request_index: int, page: int = 1, limit: int = 30
    ) -> Any:
        params = urlencode([("page", str(page)), ("limit", str(limit))])
        return await self._do_fetch(
            "GET", f"{self._repo_path(project, repository)}/pulls/{pull_request_index}/files?{params}"
        )

    async def list_pull_request_commits(
        self, project: str, repository: str, pull_request_index: int, page: int = 1, limit: int = 30
    ) -> Any:
        params = urlencode([("page", str(page)), ("limit", str(limit))])
        return await self._do_fetch(
            "GET", f"{self._repo_path(project, repository)}/pulls/{pull_request_index}/commits?{params}"
        )

    async def list_pull_request_comments(
        self, project: str, repository: str, pull_request_index: int, page: int = 1, limit: int = 30
    ) -> Any:
        params = urlencode([("page", str(page)), ("limit", str(limit))])
        return await self._do_fetch(
            "GET", f"{self._repo_path(project, repository)}/pulls/{pull_request_index}/comments?{params}"
        )

    async def add_pull_request_comment(
        self,
        project: str,
        repository: str,
        pull_request_index: int,
        body: str,
        file_path: str | None = None,
        line: int | None = None,
        line_type: LineType | None = None,
    ) -> Any:
        payload: dict[str, Any] = {"body": body}
        if file_path:
            anchor: dict[str, Any] = {
                "filepath": file_path,
                "file_variant": "FROM" if line_type == "REMOVED" else "TO",
                "diff_type": "EFFECTIVE",
            }
            if line is not None:
                anchor["line"] = line
                if line_type is not None:
                    anchor["line_type"] = line_type
            payload["anchor"] = anchor
        return await self._do_fetch(
            "POST", f"{self._repo_path(project, repository)}/pulls/{pull_request_index}/comments", body=payload
        )

    # --- Файлы ---------------------------------------------------------------

    async def get_file_content(self, project: str, repository: str, file_path: str, ref: str | None = None) -> str:
        url_path = f"{self._repo_path(project, repository)}/raw/{_encode_path(file_path)}"
        if ref:
            url_path += f"?ref={quote(ref, safe='')}"
        return await self._do_fetch("GET", url_path, text=True)

    async def list_files(
        self,
        project: str,
        repository: str,
        file_path: str | None = None,
        ref: str | None = None,
        page: int | None = None,
        limit: int | None = None,
        sort: str | None = None,
    ) -> Any:
        url_path = f"{self._repo_path(project, repository)}/contents"
        if file_path and file_path != ".":
            url_path += f"/{_encode_path(file_path)}"
        params: list[tuple[str, str]] = []
        if ref:
            params.append(("ref", ref))
        if page is not None:
            params.append(("page", str(page)))
        if limit is not None:
            params.append(("limit", str(limit)))
        if sort:
            params.append(("sort", sort))
        if params:
            url_path += f"?{urlencode(params)}"
        return await self._do_fetch("GET", url_path)

    # --- Пользователи ----------------------------------------------------------

    async def get_user(self, username: str) -> Any:
        return await self._do_fetch("GET", f"/users/{quote(username, safe='')}")