/
alexefan136
/
flowstack
Обзор
Документация
Войти
/
alexefan136
/
flowstack
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
main
core/engine/src/tools/mcp/gitlab.py
2 555 строк
85 KB
Alexander Efanov
Обновление репозитория
15 июл 2026, 12:19
15 июл 2026, 12:19
76704c6
Код
Авторство
О чём код?
""" GitLab MCP Server — интеграция с GitLab REST API v4 через MCP протокол. Предоставляет tools для работы с GitLab: - Projects (list, get, search, members) - Issues (CRUD, labels, comments, assignees) - Merge Requests (list, get, create, approve, merge) - Commits, branches, tags - Files, repositories - Users, groups - Pipelines, jobs Использует GitLab REST API v4: https://docs.gitlab.com/ee/api/ Аутентификация: - Personal Access Token (PRIVATE-TOKEN header) - OAuth2 Bearer token Архитектурные принципы: - Чистые функции для преобразования данных (from_api_response, to_dict) - Type-safe dispatch через таблицу handlers - Явная обработка ошибок и rate limiting - Async-first с context managers - Следование "You Might Not Need an Effect" """ from __future__ import annotations import asyncio import logging import os from collections.abc import AsyncIterator from contextlib import asynccontextmanager from dataclasses import dataclass, field from datetime import datetime from typing import Any, TYPE_CHECKING from src.primitives.context import MCPContext if TYPE_CHECKING: import aiohttp # type: ignore[import-not-found,import-untyped] logger = logging.getLogger(__name__) # ============================================================================ # Configuration # ============================================================================ @dataclass class GitLabConfig: """ Конфигурация подключения к GitLab. Поддерживает self-hosted GitLab и gitlab.com. """ base_url: str = "https://gitlab.com" token: str | None = None token_type: str = "private" # private | bearer # Rate limiting timeout_seconds: float = 30.0 max_retries: int = 3 retry_delay_seconds: float = 1.0 requests_per_second: float = 5.0 def get_headers(self) -> dict[str, str]: """Получить headers для аутентификации. Чистая функция.""" headers = { "Content-Type": "application/json", "Accept": "application/json", } if self.token: if self.token_type == "bearer": headers["Authorization"] = f"Bearer {self.token}" else: headers["PRIVATE-TOKEN"] = self.token return headers def validate(self) -> list[str]: """Валидировать конфигурацию. Чистая функция.""" errors: list[str] = [] if not self.base_url: errors.append("base_url is required") if not self.token: errors.append("token is required for GitLab API") if self.token_type not in ("private", "bearer"): errors.append("token_type must be 'private' or 'bearer'") return errors @classmethod def from_env(cls) -> GitLabConfig: """Создать из переменных окружения.""" token = ( os.getenv("GITLAB_TOKEN") or os.getenv("GITLAB_PRIVATE_TOKEN") or os.getenv("GITLAB_ACCESS_TOKEN") ) return cls( base_url=os.getenv("GITLAB_URL", "https://gitlab.com"), token=token, token_type=os.getenv("GITLAB_TOKEN_TYPE", "private"), timeout_seconds=float(os.getenv("GITLAB_TIMEOUT", "30")), max_retries=int(os.getenv("GITLAB_MAX_RETRIES", "3")), requests_per_second=float(os.getenv("GITLAB_RPS", "5.0")), ) # ============================================================================ # Exceptions # ============================================================================ class GitLabError(Exception): """Базовое исключение GitLab.""" pass class GitLabAuthError(GitLabError): """Ошибка аутентификации.""" pass class GitLabNotFoundError(GitLabError): """Ресурс не найден (404).""" pass class GitLabRateLimitError(GitLabError): """Превышен rate limit.""" def __init__(self, message: str, reset_at: datetime | None = None): super().__init__(message) self.reset_at = reset_at class GitLabValidationError(GitLabError): """Ошибка валидации.""" pass class GitLabForbiddenError(GitLabError): """Доступ запрещен.""" pass # ============================================================================ # Response Models # ============================================================================ @dataclass class GitLabUser: """Пользователь GitLab.""" id: int username: str name: str = "" email: str | None = None avatar_url: str = "" web_url: str = "" state: str = "active" # active | blocked | deactivated def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "username": self.username, "name": self.name, "email": self.email, "avatar_url": self.avatar_url, "web_url": self.web_url, "state": self.state, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitLabUser: """Чистая функция преобразования.""" return cls( id=int(data.get("id", 0)), username=str(data.get("username", "")), name=str(data.get("name", "")), email=data.get("email"), avatar_url=str(data.get("avatar_url", "")), web_url=str(data.get("web_url", "")), state=str(data.get("state", "active")), ) @dataclass class GitLabLabel: """Label на issue/MR.""" id: int name: str color: str = "" description: str = "" def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "name": self.name, "color": self.color, "description": self.description, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitLabLabel: """Чистая функция.""" return cls( id=int(data.get("id", 0)), name=str(data.get("name", "")), color=str(data.get("color", "")), description=str(data.get("description", "") or ""), ) @dataclass class GitLabNamespace: """Namespace (группа или пользователь) в GitLab.""" id: int name: str path: str kind: str = "group" # group | user full_path: str = "" web_url: str = "" def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "name": self.name, "path": self.path, "kind": self.kind, "full_path": self.full_path, "web_url": self.web_url, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitLabNamespace: """Чистая функция.""" return cls( id=int(data.get("id", 0)), name=str(data.get("name", "")), path=str(data.get("path", "")), kind=str(data.get("kind", "group")), full_path=str(data.get("full_path", "")), web_url=str(data.get("web_url", "")), ) @dataclass class GitLabProject: """Проект GitLab.""" id: int name: str path: str path_with_namespace: str description: str = "" default_branch: str = "main" visibility: str = "private" # private | internal | public web_url: str = "" http_url: str = "" ssh_url: str = "" archived: bool = False fork: bool = False star_count: int = 0 forks_count: int = 0 open_issues_count: int = 0 namespace: GitLabNamespace | None = None created_at: datetime | None = None last_activity_at: datetime | None = None def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "name": self.name, "path": self.path, "path_with_namespace": self.path_with_namespace, "description": self.description, "default_branch": self.default_branch, "visibility": self.visibility, "web_url": self.web_url, "http_url": self.http_url, "ssh_url": self.ssh_url, "archived": self.archived, "fork": self.fork, "star_count": self.star_count, "forks_count": self.forks_count, "open_issues_count": self.open_issues_count, "namespace": self.namespace.to_dict() if self.namespace else None, "created_at": self.created_at.isoformat() if self.created_at else None, "last_activity_at": ( self.last_activity_at.isoformat() if self.last_activity_at else None ), } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitLabProject: """Чистая функция.""" namespace_data = data.get("namespace") namespace = ( GitLabNamespace.from_api_response(namespace_data) if isinstance(namespace_data, dict) else None ) return cls( id=int(data.get("id", 0)), name=str(data.get("name", "")), path=str(data.get("path", "")), path_with_namespace=str(data.get("path_with_namespace", "")), description=str(data.get("description", "") or ""), default_branch=str(data.get("default_branch", "main")), visibility=str(data.get("visibility", "private")), web_url=str(data.get("web_url", "")), http_url=str(data.get("http_url_to_repo", "")), ssh_url=str(data.get("ssh_url_to_repo", "")), archived=bool(data.get("archived", False)), fork=bool(data.get("forked_from_project") is not None), star_count=int(data.get("star_count", 0)), forks_count=int(data.get("forks_count", 0)), open_issues_count=int(data.get("open_issues_count", 0)), namespace=namespace, created_at=_parse_datetime(data.get("created_at")), last_activity_at=_parse_datetime(data.get("last_activity_at")), ) @dataclass class GitLabIssue: """Issue в GitLab.""" id: int iid: int project_id: int title: str description: str = "" state: str = "opened" # opened | closed labels: list[str] = field(default_factory=list) author: GitLabUser | None = None assignees: list[GitLabUser] = field(default_factory=list) milestone: dict[str, Any] | None = None web_url: str = "" weight: int | None = None confidential: bool = False due_date: str | None = None created_at: datetime | None = None updated_at: datetime | None = None closed_at: datetime | None = None def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "iid": self.iid, "project_id": self.project_id, "title": self.title, "description": self.description, "state": self.state, "labels": self.labels, "author": self.author.to_dict() if self.author else None, "assignees": [a.to_dict() for a in self.assignees], "milestone": self.milestone, "web_url": self.web_url, "weight": self.weight, "confidential": self.confidential, "due_date": self.due_date, "created_at": self.created_at.isoformat() if self.created_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None, "closed_at": self.closed_at.isoformat() if self.closed_at else None, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitLabIssue: """Чистая функция.""" author_data = data.get("author") author = ( GitLabUser.from_api_response(author_data) if isinstance(author_data, dict) else None ) assignees_raw = data.get("assignees") or [] assignees = [ GitLabUser.from_api_response(a) for a in assignees_raw if isinstance(a, dict) ] labels_raw = data.get("labels") or [] labels: list[str] = [str(l) for l in labels_raw] return cls( id=int(data.get("id", 0)), iid=int(data.get("iid", 0)), project_id=int(data.get("project_id") or data.get("web_url", "0")), title=str(data.get("title", "")), description=str(data.get("description", "") or ""), state=str(data.get("state", "opened")), labels=labels, author=author, assignees=assignees, milestone=data.get("milestone"), web_url=str(data.get("web_url", "")), weight=data.get("weight"), confidential=bool(data.get("confidential", False)), due_date=data.get("due_date"), created_at=_parse_datetime(data.get("created_at")), updated_at=_parse_datetime(data.get("updated_at")), closed_at=_parse_datetime(data.get("closed_at")), ) @dataclass class GitLabMergeRequest: """Merge Request в GitLab.""" id: int iid: int project_id: int title: str description: str = "" state: str = "opened" # opened | closed | locked | merged merged: bool = False source_branch: str = "" target_branch: str = "" author: GitLabUser | None = None assignees: list[GitLabUser] = field(default_factory=list) reviewers: list[GitLabUser] = field(default_factory=list) labels: list[str] = field(default_factory=list) web_url: str = "" draft: bool = False merge_status: str = "" # can_be_merged | unchecked | checking | cannot_be_merged has_conflicts: bool = False changes_count: int = 0 user_notes_count: int = 0 created_at: datetime | None = None updated_at: datetime | None = None merged_at: datetime | None = None closed_at: datetime | None = None def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "iid": self.iid, "project_id": self.project_id, "title": self.title, "description": self.description, "state": self.state, "merged": self.merged, "source_branch": self.source_branch, "target_branch": self.target_branch, "author": self.author.to_dict() if self.author else None, "assignees": [a.to_dict() for a in self.assignees], "reviewers": [r.to_dict() for r in self.reviewers], "labels": self.labels, "web_url": self.web_url, "draft": self.draft, "merge_status": self.merge_status, "has_conflicts": self.has_conflicts, "changes_count": self.changes_count, "user_notes_count": self.user_notes_count, "created_at": self.created_at.isoformat() if self.created_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None, "merged_at": self.merged_at.isoformat() if self.merged_at else None, "closed_at": self.closed_at.isoformat() if self.closed_at else None, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitLabMergeRequest: """Чистая функция.""" author_data = data.get("author") author = ( GitLabUser.from_api_response(author_data) if isinstance(author_data, dict) else None ) assignees = [ GitLabUser.from_api_response(a) for a in (data.get("assignees") or []) if isinstance(a, dict) ] reviewers = [ GitLabUser.from_api_response(r) for r in (data.get("reviewers") or []) if isinstance(r, dict) ] labels_raw = data.get("labels") or [] labels = [str(l) for l in labels_raw] return cls( id=int(data.get("id", 0)), iid=int(data.get("iid", 0)), project_id=int(data.get("project_id") or 0), title=str(data.get("title", "")), description=str(data.get("description", "") or ""), state=str(data.get("state", "opened")), merged=bool(data.get("merged", False)), source_branch=str(data.get("source_branch", "")), target_branch=str(data.get("target_branch", "")), author=author, assignees=assignees, reviewers=reviewers, labels=labels, web_url=str(data.get("web_url", "")), draft=bool(data.get("draft", False)), merge_status=str(data.get("merge_status", "")), has_conflicts=bool(data.get("has_conflicts", False)), changes_count=int(data.get("changes_count", 0)), user_notes_count=int(data.get("user_notes_count", 0)), created_at=_parse_datetime(data.get("created_at")), updated_at=_parse_datetime(data.get("updated_at")), merged_at=_parse_datetime(data.get("merged_at")), closed_at=_parse_datetime(data.get("closed_at")), ) @dataclass class GitLabNote: """Комментарий (Note) в GitLab.""" id: int body: str author: GitLabUser | None = None noteable_type: str = "" # Issue | MergeRequest | Commit | Snippet noteable_id: int = 0 system: bool = False resolvable: bool = False resolved: bool = False web_url: str = "" created_at: datetime | None = None updated_at: datetime | None = None def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "body": self.body, "author": self.author.to_dict() if self.author else None, "noteable_type": self.noteable_type, "noteable_id": self.noteable_id, "system": self.system, "resolvable": self.resolvable, "resolved": self.resolved, "web_url": self.web_url, "created_at": self.created_at.isoformat() if self.created_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitLabNote: """Чистая функция.""" author_data = data.get("author") author = ( GitLabUser.from_api_response(author_data) if isinstance(author_data, dict) else None ) return cls( id=int(data.get("id", 0)), body=str(data.get("body", "")), author=author, noteable_type=str(data.get("noteable_type", "")), noteable_id=int(data.get("noteable_id", 0)), system=bool(data.get("system", False)), resolvable=bool(data.get("resolvable", False)), resolved=bool(data.get("resolved", False)), web_url=str(data.get("web_url") or data.get("noteable_iid") or ""), created_at=_parse_datetime(data.get("created_at")), updated_at=_parse_datetime(data.get("updated_at")), ) @dataclass class GitLabCommit: """Commit в GitLab.""" id: str # SHA short_id: str title: str message: str = "" author_name: str = "" author_email: str = "" web_url: str = "" created_at: datetime | None = None def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "short_id": self.short_id, "title": self.title, "message": self.message, "author_name": self.author_name, "author_email": self.author_email, "web_url": self.web_url, "created_at": self.created_at.isoformat() if self.created_at else None, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitLabCommit: """Чистая функция.""" return cls( id=str(data.get("id", "")), short_id=str(data.get("short_id", "")), title=str(data.get("title", "")), message=str(data.get("message", "")), author_name=str(data.get("author_name", "")), author_email=str(data.get("author_email", "")), web_url=str(data.get("web_url", "")), created_at=_parse_datetime(data.get("created_at")), ) @dataclass class GitLabBranch: """Ветка репозитория в GitLab.""" name: str protected: bool = False default: bool = False merged: bool = False commit_sha: str = "" web_url: str = "" def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "name": self.name, "protected": self.protected, "default": self.default, "merged": self.merged, "commit_sha": self.commit_sha, "web_url": self.web_url, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitLabBranch: """Чистая функция.""" commit_data = data.get("commit") or {} return cls( name=str(data.get("name", "")), protected=bool(data.get("protected", False)), default=bool(data.get("default", False)), merged=bool(data.get("merged", False)), commit_sha=str(commit_data.get("id", "") if isinstance(commit_data, dict) else ""), web_url=str(data.get("web_url", "")), ) @dataclass class GitLabTag: """Тег в GitLab.""" name: str message: str = "" commit_sha: str = "" web_url: str = "" def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "name": self.name, "message": self.message, "commit_sha": self.commit_sha, "web_url": self.web_url, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitLabTag: """Чистая функция.""" commit_data = data.get("commit") or {} return cls( name=str(data.get("name", "")), message=str(data.get("message", "") or ""), commit_sha=str(commit_data.get("id", "") if isinstance(commit_data, dict) else ""), web_url=str(data.get("web_url") or ""), ) @dataclass class GitLabFile: """Файл в репозитории GitLab.""" file_name: str file_path: str size: int = 0 encoding: str = "base64" content: str = "" # base64 encoded (если запрашивался content) ref: str = "" blob_id: str = "" commit_id: str = "" last_commit_id: str = "" def get_decoded_content(self) -> str: """Декодировать base64 контент. Чистая функция.""" if not self.content or self.encoding != "base64": return "" import base64 try: return base64.b64decode(self.content).decode("utf-8") except (ValueError, UnicodeDecodeError): return "" def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "file_name": self.file_name, "file_path": self.file_path, "size": self.size, "encoding": self.encoding, "content": self.get_decoded_content(), "ref": self.ref, "blob_id": self.blob_id, "commit_id": self.commit_id, "last_commit_id": self.last_commit_id, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitLabFile: """Чистая функция.""" return cls( file_name=str(data.get("file_name", "")), file_path=str(data.get("file_path", "")), size=int(data.get("size", 0)), encoding=str(data.get("encoding", "base64")), content=str(data.get("content", "") or ""), ref=str(data.get("ref", "")), blob_id=str(data.get("blob_id", "")), commit_id=str(data.get("commit_id", "")), last_commit_id=str(data.get("last_commit_id", "")), ) @dataclass class GitLabTreeItem: """Элемент дерева файлов в GitLab.""" id: str name: str type: str = "blob" # blob | tree | commit path: str = "" mode: str = "" def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "name": self.name, "type": self.type, "path": self.path, "mode": self.mode, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitLabTreeItem: """Чистая функция.""" return cls( id=str(data.get("id", "")), name=str(data.get("name", "")), type=str(data.get("type", "blob")), path=str(data.get("path", "")), mode=str(data.get("mode", "")), ) @dataclass class GitLabPipeline: """Pipeline в GitLab.""" id: int status: str # pending | running | success | failed | canceled | skipped ref: str = "" sha: str = "" source: str = "" web_url: str = "" created_at: datetime | None = None updated_at: datetime | None = None def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "status": self.status, "ref": self.ref, "sha": self.sha, "source": self.source, "web_url": self.web_url, "created_at": self.created_at.isoformat() if self.created_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitLabPipeline: """Чистая функция.""" return cls( id=int(data.get("id", 0)), status=str(data.get("status", "")), ref=str(data.get("ref", "")), sha=str(data.get("sha", "")), source=str(data.get("source", "")), web_url=str(data.get("web_url", "")), created_at=_parse_datetime(data.get("created_at")), updated_at=_parse_datetime(data.get("updated_at")), ) @dataclass class GitLabJob: """Job в GitLab CI.""" id: int name: str stage: str = "" status: str = "" ref: str = "" web_url: str = "" duration: float | None = None created_at: datetime | None = None started_at: datetime | None = None finished_at: datetime | None = None def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "name": self.name, "stage": self.stage, "status": self.status, "ref": self.ref, "web_url": self.web_url, "duration": self.duration, "created_at": self.created_at.isoformat() if self.created_at else None, "started_at": self.started_at.isoformat() if self.started_at else None, "finished_at": self.finished_at.isoformat() if self.finished_at else None, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitLabJob: """Чистая функция.""" return cls( id=int(data.get("id", 0)), name=str(data.get("name", "")), stage=str(data.get("stage", "")), status=str(data.get("status", "")), ref=str(data.get("ref", "")), web_url=str(data.get("web_url", "")), duration=data.get("duration"), created_at=_parse_datetime(data.get("created_at")), started_at=_parse_datetime(data.get("started_at")), finished_at=_parse_datetime(data.get("finished_at")), ) @dataclass class GitLabGroup: """Группа в GitLab.""" id: int name: str path: str full_path: str = "" description: str = "" visibility: str = "private" web_url: str = "" parent_id: int | None = None created_at: datetime | None = None def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "name": self.name, "path": self.path, "full_path": self.full_path, "description": self.description, "visibility": self.visibility, "web_url": self.web_url, "parent_id": self.parent_id, "created_at": self.created_at.isoformat() if self.created_at else None, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitLabGroup: """Чистая функция.""" return cls( id=int(data.get("id", 0)), name=str(data.get("name", "")), path=str(data.get("path", "")), full_path=str(data.get("full_path", "")), description=str(data.get("description", "") or ""), visibility=str(data.get("visibility", "private")), web_url=str(data.get("web_url", "")), parent_id=data.get("parent_id"), created_at=_parse_datetime(data.get("created_at")), ) # ============================================================================ # Helper Functions # ============================================================================ def _parse_datetime(value: Any) -> datetime | None: """Безопасно парсить datetime из ISO 8601 строки.""" if not isinstance(value, str): return None try: normalized = value.replace("Z", "+00:00") return datetime.fromisoformat(normalized) except (ValueError, TypeError): return None def _encode_project_id(project_id: str | int) -> str: """ Кодирует project_id для URL (path with namespace требует URL encode). Пример: 'my-group/my-project' → 'my-group%2Fmy-project' """ if isinstance(project_id, int): return str(project_id) import urllib.parse return urllib.parse.quote(str(project_id), safe="") # ============================================================================ # GitLab Client # ============================================================================ class GitLabClient: """ Асинхронный клиент для GitLab REST API v4. Автоматически обрабатывает: - Аутентификацию (PRIVATE-TOKEN / Bearer) - Rate limiting (читает RateLimit-* headers) - Retry логику с exponential backoff - URL encoding для path with namespace """ def __init__(self, config: GitLabConfig): self.config = config self._session: aiohttp.ClientSession | None = None self._rate_limiter = asyncio.Semaphore(int(config.requests_per_second)) self._last_request_time: float = 0.0 async def __aenter__(self) -> GitLabClient: await self.connect() return self async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: await self.close() async def connect(self) -> None: """Установить HTTP соединение.""" if self._session is not None: return try: import aiohttp # type: ignore[import-not-found,import-untyped] except ImportError as e: raise GitLabError( "aiohttp is required. Install with: pip install aiohttp" ) from e timeout = aiohttp.ClientTimeout(total=self.config.timeout_seconds) self._session = aiohttp.ClientSession( timeout=timeout, headers=self.config.get_headers(), ) async def close(self) -> None: """Закрыть соединение.""" if self._session is not None: await self._session.close() self._session = None def _require_session(self) -> aiohttp.ClientSession: """Получить активную сессию или выбросить ошибку.""" if self._session is None: raise GitLabError("Client is not connected") return self._session async def _request( self, method: str, endpoint: str, params: dict[str, Any] | None = None, json_data: Any = None, ) -> Any: """ Выполнить HTTP запрос к GitLab API с retry логикой. Обрабатывает rate limiting через RateLimit-* headers. """ session = self._require_session() url = f"{self.config.base_url.rstrip('/')}/api/v4{endpoint}" last_error: Exception | None = None loop = asyncio.get_running_loop() for attempt in range(self.config.max_retries): async with self._rate_limiter: now = loop.time() min_interval = 1.0 / self.config.requests_per_second elapsed = now - self._last_request_time if elapsed < min_interval: await asyncio.sleep(min_interval - elapsed) try: async with session.request( method, url, params=params, json=json_data, ) as response: self._last_request_time = loop.time() if response.status == 401: raise GitLabAuthError("Authentication failed") elif response.status == 403: raise GitLabForbiddenError("Access forbidden") elif response.status == 404: raise GitLabNotFoundError( f"Resource not found: {endpoint}" ) elif response.status == 422: text = await response.text() raise GitLabValidationError(f"Validation failed: {text}") elif response.status == 429: reset_str = response.headers.get("RateLimit-Reset") reset_at = ( datetime.fromtimestamp(int(reset_str)) if reset_str and reset_str.isdigit() else None ) raise GitLabRateLimitError( "Rate limit exceeded", reset_at=reset_at, ) elif response.status >= 400: text = await response.text() raise GitLabError( f"GitLab API error {response.status}: {text}" ) if response.status == 204: return None content_type = response.content_type or "" if "application/json" in content_type: return await response.json() else: return await response.text() except ( GitLabAuthError, GitLabNotFoundError, GitLabForbiddenError, GitLabValidationError, ) as e: raise e except GitLabRateLimitError as e: last_error = e if attempt < self.config.max_retries - 1: wait_time = self.config.retry_delay_seconds * (2 ** attempt) logger.warning(f"Rate limited, waiting {wait_time}s") await asyncio.sleep(wait_time) else: raise except Exception as e: last_error = e logger.warning( f"Request failed (attempt {attempt + 1}/" f"{self.config.max_retries}): {e}" ) if attempt < self.config.max_retries - 1: wait_time = self.config.retry_delay_seconds * (2 ** attempt) await asyncio.sleep(wait_time) raise GitLabError( f"Request failed after {self.config.max_retries} attempts: {last_error}" ) # ==================================================================== # Current User # ==================================================================== async def get_current_user(self) -> GitLabUser: """Получить информацию об аутентифицированном пользователе.""" response = await self._request("GET", "/user") if not isinstance(response, dict): raise GitLabError("Invalid response format") return GitLabUser.from_api_response(response) # ==================================================================== # Projects # ==================================================================== async def list_projects( self, membership: bool = False, owned: bool = False, search: str | None = None, per_page: int = 20, page: int = 1, ) -> list[GitLabProject]: """Список проектов.""" params: dict[str, Any] = {"per_page": per_page, "page": page} if membership: params["membership"] = "true" if owned: params["owned"] = "true" if search: params["search"] = search response = await self._request("GET", "/projects", params=params) return [ GitLabProject.from_api_response(p) for p in (response if isinstance(response, list) else []) if isinstance(p, dict) ] async def get_project(self, project_id: str | int) -> GitLabProject: """Получить проект по ID или path with namespace.""" encoded_id = _encode_project_id(project_id) response = await self._request("GET", f"/projects/{encoded_id}") if not isinstance(response, dict): raise GitLabError("Invalid response format") return GitLabProject.from_api_response(response) async def list_user_projects( self, user_id: str | int, per_page: int = 20, page: int = 1, ) -> list[GitLabProject]: """Список проектов пользователя.""" params = {"per_page": per_page, "page": page} response = await self._request( "GET", f"/users/{user_id}/projects", params=params ) return [ GitLabProject.from_api_response(p) for p in (response if isinstance(response, list) else []) if isinstance(p, dict) ] async def list_group_projects( self, group_id: str | int, include_subgroups: bool = True, per_page: int = 20, page: int = 1, ) -> list[GitLabProject]: """Список проектов группы.""" params: dict[str, Any] = { "per_page": per_page, "page": page, "include_subgroups": "true" if include_subgroups else "false", } encoded_id = _encode_project_id(group_id) response = await self._request( "GET", f"/groups/{encoded_id}/projects", params=params ) return [ GitLabProject.from_api_response(p) for p in (response if isinstance(response, list) else []) if isinstance(p, dict) ] # ==================================================================== # Issues # ==================================================================== async def list_project_issues( self, project_id: str | int, state: str = "opened", labels: str | None = None, per_page: int = 20, page: int = 1, ) -> list[GitLabIssue]: """Список issues проекта.""" params: dict[str, Any] = { "state": state, "per_page": per_page, "page": page, } if labels: params["labels"] = labels encoded_id = _encode_project_id(project_id) response = await self._request( "GET", f"/projects/{encoded_id}/issues", params=params ) return [ GitLabIssue.from_api_response(i) for i in (response if isinstance(response, list) else []) if isinstance(i, dict) ] async def get_issue( self, project_id: str | int, issue_iid: int ) -> GitLabIssue: """Получить issue по IID.""" encoded_id = _encode_project_id(project_id) response = await self._request( "GET", f"/projects/{encoded_id}/issues/{issue_iid}" ) if not isinstance(response, dict): raise GitLabError("Invalid response format") return GitLabIssue.from_api_response(response) async def create_issue( self, project_id: str | int, title: str, description: str | None = None, labels: list[str] | None = None, assignee_ids: list[int] | None = None, confidential: bool = False, due_date: str | None = None, ) -> GitLabIssue: """Создать новый issue.""" payload: dict[str, Any] = { "title": title, "confidential": confidential, } if description is not None: payload["description"] = description if labels: payload["labels"] = ",".join(labels) if assignee_ids: payload["assignee_ids"] = assignee_ids if due_date: payload["due_date"] = due_date encoded_id = _encode_project_id(project_id) response = await self._request( "POST", f"/projects/{encoded_id}/issues", json_data=payload ) if not isinstance(response, dict): raise GitLabError("Invalid response format") return GitLabIssue.from_api_response(response) async def update_issue( self, project_id: str | int, issue_iid: int, title: str | None = None, description: str | None = None, state_event: str | None = None, labels: list[str] | None = None, assignee_ids: list[int] | None = None, ) -> GitLabIssue: """Обновить issue.""" payload: dict[str, Any] = {} if title is not None: payload["title"] = title if description is not None: payload["description"] = description if state_event is not None: payload["state_event"] = state_event if labels is not None: payload["labels"] = ",".join(labels) if assignee_ids is not None: payload["assignee_ids"] = assignee_ids encoded_id = _encode_project_id(project_id) response = await self._request( "PUT", f"/projects/{encoded_id}/issues/{issue_iid}", json_data=payload, ) if not isinstance(response, dict): raise GitLabError("Invalid response format") return GitLabIssue.from_api_response(response) async def close_issue( self, project_id: str | int, issue_iid: int ) -> GitLabIssue: """Закрыть issue.""" return await self.update_issue( project_id, issue_iid, state_event="close" ) async def list_issue_notes( self, project_id: str | int, issue_iid: int, per_page: int = 20, ) -> list[GitLabNote]: """Список комментариев к issue.""" params = {"per_page": per_page} encoded_id = _encode_project_id(project_id) response = await self._request( "GET", f"/projects/{encoded_id}/issues/{issue_iid}/notes", params=params, ) return [ GitLabNote.from_api_response(n) for n in (response if isinstance(response, list) else []) if isinstance(n, dict) ] async def create_issue_note( self, project_id: str | int, issue_iid: int, body: str ) -> GitLabNote: """Добавить комментарий к issue.""" encoded_id = _encode_project_id(project_id) response = await self._request( "POST", f"/projects/{encoded_id}/issues/{issue_iid}/notes", json_data={"body": body}, ) if not isinstance(response, dict): raise GitLabError("Invalid response format") return GitLabNote.from_api_response(response) # ==================================================================== # Merge Requests # ==================================================================== async def list_merge_requests( self, project_id: str | int, state: str = "opened", per_page: int = 20, page: int = 1, ) -> list[GitLabMergeRequest]: """Список merge requests проекта.""" params = {"state": state, "per_page": per_page, "page": page} encoded_id = _encode_project_id(project_id) response = await self._request( "GET", f"/projects/{encoded_id}/merge_requests", params=params ) return [ GitLabMergeRequest.from_api_response(mr) for mr in (response if isinstance(response, list) else []) if isinstance(mr, dict) ] async def get_merge_request( self, project_id: str | int, mr_iid: int ) -> GitLabMergeRequest: """Получить MR по IID.""" encoded_id = _encode_project_id(project_id) response = await self._request( "GET", f"/projects/{encoded_id}/merge_requests/{mr_iid}" ) if not isinstance(response, dict): raise GitLabError("Invalid response format") return GitLabMergeRequest.from_api_response(response) async def create_merge_request( self, project_id: str | int, source_branch: str, target_branch: str, title: str, description: str | None = None, labels: list[str] | None = None, assignee_ids: list[int] | None = None, reviewer_ids: list[int] | None = None, draft: bool = False, ) -> GitLabMergeRequest: """Создать merge request.""" payload: dict[str, Any] = { "source_branch": source_branch, "target_branch": target_branch, "title": title, } if description is not None: payload["description"] = description if labels: payload["labels"] = ",".join(labels) if assignee_ids: payload["assignee_ids"] = assignee_ids if reviewer_ids: payload["reviewer_ids"] = reviewer_ids if draft and not title.startswith("Draft:"): payload["title"] = f"Draft: {title}" encoded_id = _encode_project_id(project_id) response = await self._request( "POST", f"/projects/{encoded_id}/merge_requests", json_data=payload, ) if not isinstance(response, dict): raise GitLabError("Invalid response format") return GitLabMergeRequest.from_api_response(response) async def merge_merge_request( self, project_id: str | int, mr_iid: int, merge_commit_message: str | None = None, squash: bool = False, ) -> GitLabMergeRequest: """Смержить MR.""" payload: dict[str, Any] = {"squash": squash} if merge_commit_message is not None: payload["merge_commit_message"] = merge_commit_message encoded_id = _encode_project_id(project_id) response = await self._request( "PUT", f"/projects/{encoded_id}/merge_requests/{mr_iid}/merge", json_data=payload, ) if not isinstance(response, dict): raise GitLabError("Invalid response format") return GitLabMergeRequest.from_api_response(response) async def approve_merge_request( self, project_id: str | int, mr_iid: int ) -> dict[str, Any]: """Одобрить MR.""" encoded_id = _encode_project_id(project_id) response = await self._request( "POST", f"/projects/{encoded_id}/merge_requests/{mr_iid}/approve", ) return response if isinstance(response, dict) else {} # ==================================================================== # Commits & Branches # ==================================================================== async def list_commits( self, project_id: str | int, ref_name: str | None = None, path: str | None = None, per_page: int = 20, ) -> list[GitLabCommit]: """Список commits.""" params: dict[str, Any] = {"per_page": per_page} if ref_name: params["ref_name"] = ref_name if path: params["path"] = path encoded_id = _encode_project_id(project_id) response = await self._request( "GET", f"/projects/{encoded_id}/repository/commits", params=params ) return [ GitLabCommit.from_api_response(c) for c in (response if isinstance(response, list) else []) if isinstance(c, dict) ] async def list_branches( self, project_id: str | int, per_page: int = 20 ) -> list[GitLabBranch]: """Список веток.""" params = {"per_page": per_page} encoded_id = _encode_project_id(project_id) response = await self._request( "GET", f"/projects/{encoded_id}/repository/branches", params=params ) return [ GitLabBranch.from_api_response(b) for b in (response if isinstance(response, list) else []) if isinstance(b, dict) ] async def list_tags( self, project_id: str | int, per_page: int = 20 ) -> list[GitLabTag]: """Список тегов.""" params = {"per_page": per_page} encoded_id = _encode_project_id(project_id) response = await self._request( "GET", f"/projects/{encoded_id}/repository/tags", params=params ) return [ GitLabTag.from_api_response(t) for t in (response if isinstance(response, list) else []) if isinstance(t, dict) ] # ==================================================================== # Files # ==================================================================== async def get_file( self, project_id: str | int, file_path: str, ref: str = "main", ) -> GitLabFile: """Получить файл (с содержимым).""" import urllib.parse encoded_file_path = urllib.parse.quote(file_path, safe="") encoded_id = _encode_project_id(project_id) response = await self._request( "GET", f"/projects/{encoded_id}/repository/files/{encoded_file_path}", params={"ref": ref}, ) if not isinstance(response, dict): raise GitLabError("Invalid response format") return GitLabFile.from_api_response(response) async def list_tree( self, project_id: str | int, path: str = "", ref: str = "main", recursive: bool = False, per_page: int = 100, ) -> list[GitLabTreeItem]: """Список файлов в репозитории.""" params: dict[str, Any] = { "ref": ref, "recursive": "true" if recursive else "false", "per_page": per_page, } if path: params["path"] = path encoded_id = _encode_project_id(project_id) response = await self._request( "GET", f"/projects/{encoded_id}/repository/tree", params=params ) return [ GitLabTreeItem.from_api_response(item) for item in (response if isinstance(response, list) else []) if isinstance(item, dict) ] # ==================================================================== # Groups # ==================================================================== async def list_groups( self, search: str | None = None, owned: bool = False, per_page: int = 20, ) -> list[GitLabGroup]: """Список групп.""" params: dict[str, Any] = {"per_page": per_page} if search: params["search"] = search if owned: params["owned"] = "true" response = await self._request("GET", "/groups", params=params) return [ GitLabGroup.from_api_response(g) for g in (response if isinstance(response, list) else []) if isinstance(g, dict) ] async def get_group(self, group_id: str | int) -> GitLabGroup: """Получить группу по ID или full_path.""" encoded_id = _encode_project_id(group_id) response = await self._request("GET", f"/groups/{encoded_id}") if not isinstance(response, dict): raise GitLabError("Invalid response format") return GitLabGroup.from_api_response(response) # ==================================================================== # Pipelines & Jobs # ==================================================================== async def list_pipelines( self, project_id: str | int, status: str | None = None, ref: str | None = None, per_page: int = 20, ) -> list[GitLabPipeline]: """Список pipelines.""" params: dict[str, Any] = {"per_page": per_page} if status: params["status"] = status if ref: params["ref"] = ref encoded_id = _encode_project_id(project_id) response = await self._request( "GET", f"/projects/{encoded_id}/pipelines", params=params ) return [ GitLabPipeline.from_api_response(p) for p in (response if isinstance(response, list) else []) if isinstance(p, dict) ] async def list_pipeline_jobs( self, project_id: str | int, pipeline_id: int ) -> list[GitLabJob]: """Список jobs в pipeline.""" encoded_id = _encode_project_id(project_id) response = await self._request( "GET", f"/projects/{encoded_id}/pipelines/{pipeline_id}/jobs" ) return [ GitLabJob.from_api_response(j) for j in (response if isinstance(response, list) else []) if isinstance(j, dict) ] # ============================================================================ # MCP Tools Definition # ============================================================================ GITLAB_TOOLS: list[dict[str, Any]] = [ { "name": "gitlab_get_current_user", "description": "Получить информацию об аутентифицированном пользователе", "parameters": {"type": "object", "properties": {}}, }, { "name": "gitlab_list_projects", "description": "Получить список проектов", "parameters": { "type": "object", "properties": { "membership": { "type": "boolean", "description": "Только проекты, где пользователь участник", }, "owned": { "type": "boolean", "description": "Только проекты во владении пользователя", }, "search": {"type": "string", "description": "Поиск по имени"}, "per_page": {"type": "integer", "default": 20}, "page": {"type": "integer", "default": 1}, }, }, }, { "name": "gitlab_get_project", "description": "Получить информацию о проекте", "parameters": { "type": "object", "properties": { "project_id": { "type": ["string", "integer"], "description": "ID проекта или path with namespace (например, 'my-group/my-project')", }, }, "required": ["project_id"], }, }, { "name": "gitlab_list_user_projects", "description": "Получить список проектов пользователя", "parameters": { "type": "object", "properties": { "user_id": { "type": ["string", "integer"], "description": "ID или username пользователя", }, "per_page": {"type": "integer", "default": 20}, "page": {"type": "integer", "default": 1}, }, "required": ["user_id"], }, }, { "name": "gitlab_list_group_projects", "description": "Получить список проектов группы", "parameters": { "type": "object", "properties": { "group_id": { "type": ["string", "integer"], "description": "ID или full_path группы", }, "include_subgroups": {"type": "boolean", "default": True}, "per_page": {"type": "integer", "default": 20}, "page": {"type": "integer", "default": 1}, }, "required": ["group_id"], }, }, { "name": "gitlab_list_project_issues", "description": "Получить список issues проекта", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "state": { "type": "string", "enum": ["opened", "closed", "all"], "default": "opened", }, "labels": { "type": "string", "description": "Список labels через запятую", }, "per_page": {"type": "integer", "default": 20}, "page": {"type": "integer", "default": 1}, }, "required": ["project_id"], }, }, { "name": "gitlab_get_issue", "description": "Получить issue по IID", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "issue_iid": {"type": "integer"}, }, "required": ["project_id", "issue_iid"], }, }, { "name": "gitlab_create_issue", "description": "Создать новый issue", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "title": {"type": "string"}, "description": {"type": "string"}, "labels": {"type": "array", "items": {"type": "string"}}, "assignee_ids": {"type": "array", "items": {"type": "integer"}}, "confidential": {"type": "boolean", "default": False}, "due_date": {"type": "string", "description": "YYYY-MM-DD"}, }, "required": ["project_id", "title"], }, }, { "name": "gitlab_update_issue", "description": "Обновить существующий issue", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "issue_iid": {"type": "integer"}, "title": {"type": "string"}, "description": {"type": "string"}, "state_event": { "type": "string", "enum": ["close", "reopen"], }, "labels": {"type": "array", "items": {"type": "string"}}, "assignee_ids": {"type": "array", "items": {"type": "integer"}}, }, "required": ["project_id", "issue_iid"], }, }, { "name": "gitlab_close_issue", "description": "Закрыть issue", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "issue_iid": {"type": "integer"}, }, "required": ["project_id", "issue_iid"], }, }, { "name": "gitlab_list_issue_notes", "description": "Получить комментарии к issue", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "issue_iid": {"type": "integer"}, "per_page": {"type": "integer", "default": 20}, }, "required": ["project_id", "issue_iid"], }, }, { "name": "gitlab_create_issue_note", "description": "Добавить комментарий к issue", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "issue_iid": {"type": "integer"}, "body": {"type": "string"}, }, "required": ["project_id", "issue_iid", "body"], }, }, { "name": "gitlab_list_merge_requests", "description": "Получить список merge requests", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "state": { "type": "string", "enum": ["opened", "closed", "merged", "all"], "default": "opened", }, "per_page": {"type": "integer", "default": 20}, "page": {"type": "integer", "default": 1}, }, "required": ["project_id"], }, }, { "name": "gitlab_get_merge_request", "description": "Получить MR по IID", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "mr_iid": {"type": "integer"}, }, "required": ["project_id", "mr_iid"], }, }, { "name": "gitlab_create_merge_request", "description": "Создать merge request", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "source_branch": {"type": "string"}, "target_branch": {"type": "string"}, "title": {"type": "string"}, "description": {"type": "string"}, "labels": {"type": "array", "items": {"type": "string"}}, "assignee_ids": {"type": "array", "items": {"type": "integer"}}, "reviewer_ids": {"type": "array", "items": {"type": "integer"}}, "draft": {"type": "boolean", "default": False}, }, "required": ["project_id", "source_branch", "target_branch", "title"], }, }, { "name": "gitlab_merge_merge_request", "description": "Смержить MR", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "mr_iid": {"type": "integer"}, "merge_commit_message": {"type": "string"}, "squash": {"type": "boolean", "default": False}, }, "required": ["project_id", "mr_iid"], }, }, { "name": "gitlab_approve_merge_request", "description": "Одобрить MR", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "mr_iid": {"type": "integer"}, }, "required": ["project_id", "mr_iid"], }, }, { "name": "gitlab_list_commits", "description": "Получить список commits", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "ref_name": {"type": "string", "description": "Branch или tag"}, "path": {"type": "string", "description": "Фильтр по пути файла"}, "per_page": {"type": "integer", "default": 20}, }, "required": ["project_id"], }, }, { "name": "gitlab_list_branches", "description": "Получить список веток", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "per_page": {"type": "integer", "default": 20}, }, "required": ["project_id"], }, }, { "name": "gitlab_list_tags", "description": "Получить список тегов", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "per_page": {"type": "integer", "default": 20}, }, "required": ["project_id"], }, }, { "name": "gitlab_get_file", "description": "Получить файл из репозитория", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "file_path": {"type": "string", "description": "Путь к файлу"}, "ref": {"type": "string", "default": "main"}, }, "required": ["project_id", "file_path"], }, }, { "name": "gitlab_list_tree", "description": "Получить список файлов в репозитории", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "path": {"type": "string", "default": ""}, "ref": {"type": "string", "default": "main"}, "recursive": {"type": "boolean", "default": False}, "per_page": {"type": "integer", "default": 100}, }, "required": ["project_id"], }, }, { "name": "gitlab_list_groups", "description": "Получить список групп", "parameters": { "type": "object", "properties": { "search": {"type": "string"}, "owned": {"type": "boolean", "default": False}, "per_page": {"type": "integer", "default": 20}, }, }, }, { "name": "gitlab_get_group", "description": "Получить информацию о группе", "parameters": { "type": "object", "properties": { "group_id": {"type": ["string", "integer"]}, }, "required": ["group_id"], }, }, { "name": "gitlab_list_pipelines", "description": "Получить список pipelines", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "status": { "type": "string", "enum": [ "running", "pending", "success", "failed", "canceled", "skipped", ], }, "ref": {"type": "string"}, "per_page": {"type": "integer", "default": 20}, }, "required": ["project_id"], }, }, { "name": "gitlab_list_pipeline_jobs", "description": "Получить список jobs в pipeline", "parameters": { "type": "object", "properties": { "project_id": {"type": ["string", "integer"]}, "pipeline_id": {"type": "integer"}, }, "required": ["project_id", "pipeline_id"], }, }, ] # ============================================================================ # Tool Handlers # ============================================================================ async def _handle_get_current_user( client: GitLabClient, params: dict[str, Any] ) -> dict[str, Any]: user = await client.get_current_user() return user.to_dict() async def _handle_list_projects( client: GitLabClient, params: dict[str, Any] ) -> list[dict[str, Any]]: projects = await client.list_projects( membership=bool(params.get("membership", False)), owned=bool(params.get("owned", False)), search=params.get("search"), per_page=int(params.get("per_page", 20)), page=int(params.get("page", 1)), ) return [p.to_dict() for p in projects] async def _handle_get_project( client: GitLabClient, params: dict[str, Any] ) -> dict[str, Any]: project = await client.get_project(params["project_id"]) return project.to_dict() async def _handle_list_user_projects( client: GitLabClient, params: dict[str, Any] ) -> list[dict[str, Any]]: projects = await client.list_user_projects( user_id=params["user_id"], per_page=int(params.get("per_page", 20)), page=int(params.get("page", 1)), ) return [p.to_dict() for p in projects] async def _handle_list_group_projects( client: GitLabClient, params: dict[str, Any] ) -> list[dict[str, Any]]: projects = await client.list_group_projects( group_id=params["group_id"], include_subgroups=bool(params.get("include_subgroups", True)), per_page=int(params.get("per_page", 20)), page=int(params.get("page", 1)), ) return [p.to_dict() for p in projects] async def _handle_list_project_issues( client: GitLabClient, params: dict[str, Any] ) -> list[dict[str, Any]]: issues = await client.list_project_issues( project_id=params["project_id"], state=str(params.get("state", "opened")), labels=params.get("labels"), per_page=int(params.get("per_page", 20)), page=int(params.get("page", 1)), ) return [i.to_dict() for i in issues] async def _handle_get_issue( client: GitLabClient, params: dict[str, Any] ) -> dict[str, Any]: issue = await client.get_issue( project_id=params["project_id"], issue_iid=int(params["issue_iid"]), ) return issue.to_dict() async def _handle_create_issue( client: GitLabClient, params: dict[str, Any] ) -> dict[str, Any]: issue = await client.create_issue( project_id=params["project_id"], title=str(params["title"]), description=params.get("description"), labels=params.get("labels"), assignee_ids=params.get("assignee_ids"), confidential=bool(params.get("confidential", False)), due_date=params.get("due_date"), ) return issue.to_dict() async def _handle_update_issue( client: GitLabClient, params: dict[str, Any] ) -> dict[str, Any]: issue = await client.update_issue( project_id=params["project_id"], issue_iid=int(params["issue_iid"]), title=params.get("title"), description=params.get("description"), state_event=params.get("state_event"), labels=params.get("labels"), assignee_ids=params.get("assignee_ids"), ) return issue.to_dict() async def _handle_close_issue( client: GitLabClient, params: dict[str, Any] ) -> dict[str, Any]: issue = await client.close_issue( project_id=params["project_id"], issue_iid=int(params["issue_iid"]), ) return issue.to_dict() async def _handle_list_issue_notes( client: GitLabClient, params: dict[str, Any] ) -> list[dict[str, Any]]: notes = await client.list_issue_notes( project_id=params["project_id"], issue_iid=int(params["issue_iid"]), per_page=int(params.get("per_page", 20)), ) return [n.to_dict() for n in notes] async def _handle_create_issue_note( client: GitLabClient, params: dict[str, Any] ) -> dict[str, Any]: note = await client.create_issue_note( project_id=params["project_id"], issue_iid=int(params["issue_iid"]), body=str(params["body"]), ) return note.to_dict() async def _handle_list_merge_requests( client: GitLabClient, params: dict[str, Any] ) -> list[dict[str, Any]]: mrs = await client.list_merge_requests( project_id=params["project_id"], state=str(params.get("state", "opened")), per_page=int(params.get("per_page", 20)), page=int(params.get("page", 1)), ) return [mr.to_dict() for mr in mrs] async def _handle_get_merge_request( client: GitLabClient, params: dict[str, Any] ) -> dict[str, Any]: mr = await client.get_merge_request( project_id=params["project_id"], mr_iid=int(params["mr_iid"]), ) return mr.to_dict() async def _handle_create_merge_request( client: GitLabClient, params: dict[str, Any] ) -> dict[str, Any]: mr = await client.create_merge_request( project_id=params["project_id"], source_branch=str(params["source_branch"]), target_branch=str(params["target_branch"]), title=str(params["title"]), description=params.get("description"), labels=params.get("labels"), assignee_ids=params.get("assignee_ids"), reviewer_ids=params.get("reviewer_ids"), draft=bool(params.get("draft", False)), ) return mr.to_dict() async def _handle_merge_merge_request( client: GitLabClient, params: dict[str, Any] ) -> dict[str, Any]: mr = await client.merge_merge_request( project_id=params["project_id"], mr_iid=int(params["mr_iid"]), merge_commit_message=params.get("merge_commit_message"), squash=bool(params.get("squash", False)), ) return mr.to_dict() async def _handle_approve_merge_request( client: GitLabClient, params: dict[str, Any] ) -> dict[str, Any]: return await client.approve_merge_request( project_id=params["project_id"], mr_iid=int(params["mr_iid"]), ) async def _handle_list_commits( client: GitLabClient, params: dict[str, Any] ) -> list[dict[str, Any]]: commits = await client.list_commits( project_id=params["project_id"], ref_name=params.get("ref_name"), path=params.get("path"), per_page=int(params.get("per_page", 20)), ) return [c.to_dict() for c in commits] async def _handle_list_branches( client: GitLabClient, params: dict[str, Any] ) -> list[dict[str, Any]]: branches = await client.list_branches( project_id=params["project_id"], per_page=int(params.get("per_page", 20)), ) return [b.to_dict() for b in branches] async def _handle_list_tags( client: GitLabClient, params: dict[str, Any] ) -> list[dict[str, Any]]: tags = await client.list_tags( project_id=params["project_id"], per_page=int(params.get("per_page", 20)), ) return [t.to_dict() for t in tags] async def _handle_get_file( client: GitLabClient, params: dict[str, Any] ) -> dict[str, Any]: file = await client.get_file( project_id=params["project_id"], file_path=str(params["file_path"]), ref=str(params.get("ref", "main")), ) return file.to_dict() async def _handle_list_tree( client: GitLabClient, params: dict[str, Any] ) -> list[dict[str, Any]]: items = await client.list_tree( project_id=params["project_id"], path=str(params.get("path", "")), ref=str(params.get("ref", "main")), recursive=bool(params.get("recursive", False)), per_page=int(params.get("per_page", 100)), ) return [item.to_dict() for item in items] async def _handle_list_groups( client: GitLabClient, params: dict[str, Any] ) -> list[dict[str, Any]]: groups = await client.list_groups( search=params.get("search"), owned=bool(params.get("owned", False)), per_page=int(params.get("per_page", 20)), ) return [g.to_dict() for g in groups] async def _handle_get_group( client: GitLabClient, params: dict[str, Any] ) -> dict[str, Any]: group = await client.get_group(params["group_id"]) return group.to_dict() async def _handle_list_pipelines( client: GitLabClient, params: dict[str, Any] ) -> list[dict[str, Any]]: pipelines = await client.list_pipelines( project_id=params["project_id"], status=params.get("status"), ref=params.get("ref"), per_page=int(params.get("per_page", 20)), ) return [p.to_dict() for p in pipelines] async def _handle_list_pipeline_jobs( client: GitLabClient, params: dict[str, Any] ) -> list[dict[str, Any]]: jobs = await client.list_pipeline_jobs( project_id=params["project_id"], pipeline_id=int(params["pipeline_id"]), ) return [j.to_dict() for j in jobs] # Таблица dispatch _TOOL_HANDLERS: dict[str, Any] = { "gitlab_get_current_user": _handle_get_current_user, "gitlab_list_projects": _handle_list_projects, "gitlab_get_project": _handle_get_project, "gitlab_list_user_projects": _handle_list_user_projects, "gitlab_list_group_projects": _handle_list_group_projects, "gitlab_list_project_issues": _handle_list_project_issues, "gitlab_get_issue": _handle_get_issue, "gitlab_create_issue": _handle_create_issue, "gitlab_update_issue": _handle_update_issue, "gitlab_close_issue": _handle_close_issue, "gitlab_list_issue_notes": _handle_list_issue_notes, "gitlab_create_issue_note": _handle_create_issue_note, "gitlab_list_merge_requests": _handle_list_merge_requests, "gitlab_get_merge_request": _handle_get_merge_request, "gitlab_create_merge_request": _handle_create_merge_request, "gitlab_merge_merge_request": _handle_merge_merge_request, "gitlab_approve_merge_request": _handle_approve_merge_request, "gitlab_list_commits": _handle_list_commits, "gitlab_list_branches": _handle_list_branches, "gitlab_list_tags": _handle_list_tags, "gitlab_get_file": _handle_get_file, "gitlab_list_tree": _handle_list_tree, "gitlab_list_groups": _handle_list_groups, "gitlab_get_group": _handle_get_group, "gitlab_list_pipelines": _handle_list_pipelines, "gitlab_list_pipeline_jobs": _handle_list_pipeline_jobs, } # ============================================================================ # GitLab MCP Server # ============================================================================ class GitLabMCPServer: """ MCP Server для GitLab. Следует принципу "You Might Not Need an Effect": - Состояние клиента управляется через context manager - Dispatch по таблице вместо runtime reflection - Все преобразования — чистые функции - Явная обработка ошибок с типизированными исключениями """ def __init__(self, config: GitLabConfig | None = None): self.config = config or GitLabConfig.from_env() self._client: GitLabClient | None = None async def __aenter__(self) -> GitLabMCPServer: errors = self.config.validate() if errors: raise GitLabError(f"Invalid configuration: {', '.join(errors)}") self._client = GitLabClient(self.config) await self._client.connect() return self async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: if self._client is not None: await self._client.close() self._client = None def _require_client(self) -> GitLabClient: """Получить активный клиент или выбросить ошибку.""" if self._client is None: raise GitLabError( "Client is not connected. Use 'async with' to manage connection." ) return self._client def get_tools(self) -> list[dict[str, Any]]: """Получить список MCP tools. Чистая функция.""" return list(GITLAB_TOOLS) async def call_tool( self, tool_name: str, parameters: dict[str, Any], context: MCPContext | None = None, ) -> dict[str, Any]: """ Вызвать tool по имени. Returns: {"success": bool, "result"|"error": ..., "error_type"?} """ try: client = self._require_client() except GitLabError as e: return { "success": False, "error": str(e), "error_type": "not_connected", } handler = _TOOL_HANDLERS.get(tool_name) if handler is None: return { "success": False, "error": f"Unknown tool: {tool_name}", "error_type": "unknown_tool", } try: result = await handler(client, parameters) if context is not None: context.add_tool_call() return { "success": True, "result": result, } except GitLabNotFoundError as e: return { "success": False, "error": f"Not found: {e}", "error_type": "not_found", } except GitLabAuthError as e: return { "success": False, "error": f"Authentication failed: {e}", "error_type": "auth_error", } except GitLabForbiddenError as e: return { "success": False, "error": f"Forbidden: {e}", "error_type": "forbidden", } except GitLabRateLimitError as e: return { "success": False, "error": f"Rate limit exceeded: {e}", "error_type": "rate_limit", "reset_at": e.reset_at.isoformat() if e.reset_at else None, } except GitLabValidationError as e: return { "success": False, "error": f"Validation error: {e}", "error_type": "validation_error", } except GitLabError as e: return { "success": False, "error": str(e), "error_type": "gitlab_error", } except Exception as e: logger.exception(f"Unexpected error in tool {tool_name}") return { "success": False, "error": f"Unexpected error: {e}", "error_type": "unexpected", } # ============================================================================ # Helper Functions (public API) # ============================================================================ @asynccontextmanager async def create_gitlab_mcp( token: str | None = None, base_url: str = "https://gitlab.com", token_type: str = "private", timeout_seconds: float = 30.0, ) -> AsyncIterator[GitLabMCPServer]: """ Создать GitLab MCP Server с автоматическим управлением подключением. Usage: async with create_gitlab_mcp(token="glpat-...") as mcp: tools = mcp.get_tools() result = await mcp.call_tool( "gitlab_list_project_issues", {"project_id": "my-group/my-project", "state": "opened"}, ) """ resolved_token = ( token or os.getenv("GITLAB_TOKEN") or os.getenv("GITLAB_PRIVATE_TOKEN") or os.getenv("GITLAB_ACCESS_TOKEN") ) config = GitLabConfig( base_url=base_url, token=resolved_token, token_type=token_type, timeout_seconds=timeout_seconds, ) async with GitLabMCPServer(config) as server: yield server # ============================================================================ # Exports # ============================================================================ __all__ = [ # Config "GitLabConfig", # Exceptions "GitLabError", "GitLabAuthError", "GitLabNotFoundError", "GitLabRateLimitError", "GitLabValidationError", "GitLabForbiddenError", # Models "GitLabUser", "GitLabLabel", "GitLabNamespace", "GitLabProject", "GitLabIssue", "GitLabMergeRequest", "GitLabNote", "GitLabCommit", "GitLabBranch", "GitLabTag", "GitLabFile", "GitLabTreeItem", "GitLabPipeline", "GitLabJob", "GitLabGroup", # Client & Server "GitLabClient", "GitLabMCPServer", "GITLAB_TOOLS", # Helpers "create_gitlab_mcp", ]