/
alexefan136
/
flowstack
Обзор
Документация
Войти
/
alexefan136
/
flowstack
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
main
core/engine/src/tools/mcp/github.py
1 976 строк
65 KB
Alexander Efanov
Обновление репозитория
15 июл 2026, 12:19
15 июл 2026, 12:19
76704c6
Код
Авторство
О чём код?
""" GitHub MCP Server — интеграция с GitHub REST API v3 через MCP протокол. Предоставляет tools для работы с GitHub: - Репозитории (list, get, search) - Issues (CRUD, поиск, комментарии) - Pull Requests (list, get, create, review) - Commits, branches, files - Code search - Users и organizations Использует GitHub REST API v3: https://docs.github.com/en/rest Аутентификация: - Personal Access Token (Bearer) - GitHub App JWT (опционально) Архитектурные принципы: - Чистые функции для преобразования данных - Type-safe dispatch через таблицу handlers - Явная обработка ошибок и rate limiting - Async-first с context managers """ 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 GitHubConfig: """ Конфигурация подключения к GitHub. Аутентификация через Personal Access Token или GitHub App. """ base_url: str = "https://api.github.com" token: str | None = None accept_header: str = "application/vnd.github+json" api_version: str = "2022-11-28" # 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 = { "Accept": self.accept_header, "X-GitHub-Api-Version": self.api_version, "User-Agent": "flowstack-mcp-github/1.0.0", } if self.token: headers["Authorization"] = f"Bearer {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 GitHub API") return errors @classmethod def from_env(cls) -> GitHubConfig: """Создать из переменных окружения.""" return cls( base_url=os.getenv("GITHUB_API_URL", "https://api.github.com"), token=os.getenv("GITHUB_TOKEN") or os.getenv("GITHUB_PERSONAL_ACCESS_TOKEN"), timeout_seconds=float(os.getenv("GITHUB_TIMEOUT", "30")), max_retries=int(os.getenv("GITHUB_MAX_RETRIES", "3")), requests_per_second=float(os.getenv("GITHUB_RPS", "5.0")), ) # ============================================================================ # Exceptions # ============================================================================ class GitHubError(Exception): """Базовое исключение GitHub.""" pass class GitHubAuthError(GitHubError): """Ошибка аутентификации.""" pass class GitHubNotFoundError(GitHubError): """Ресурс не найден (404).""" pass class GitHubRateLimitError(GitHubError): """Превышен rate limit.""" def __init__(self, message: str, reset_at: datetime | None = None): super().__init__(message) self.reset_at = reset_at class GitHubValidationError(GitHubError): """Ошибка валидации (422).""" def __init__(self, message: str, errors: list[dict[str, Any]] | None = None): super().__init__(message) self.errors = errors or [] # ============================================================================ # Response Models # ============================================================================ @dataclass class GitHubUser: """Пользователь GitHub.""" id: int login: str avatar_url: str = "" html_url: str = "" type: str = "User" # User | Organization | Bot def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "login": self.login, "avatar_url": self.avatar_url, "html_url": self.html_url, "type": self.type, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitHubUser: """Чистая функция преобразования.""" return cls( id=int(data.get("id", 0)), login=str(data.get("login", "")), avatar_url=str(data.get("avatar_url", "")), html_url=str(data.get("html_url", "")), type=str(data.get("type", "User")), ) @dataclass class GitHubLabel: """Label на issue/PR.""" 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]) -> GitHubLabel: """Чистая функция.""" return cls( id=int(data.get("id", 0)), name=str(data.get("name", "")), color=str(data.get("color", "")), description=str(data.get("description", "")), ) @dataclass class GitHubRepository: """Репозиторий GitHub.""" id: int name: str full_name: str owner: GitHubUser description: str = "" private: bool = False fork: bool = False html_url: str = "" clone_url: str = "" default_branch: str = "main" language: str | None = None stargazers_count: int = 0 forks_count: int = 0 open_issues_count: int = 0 created_at: datetime | None = None updated_at: datetime | None = None def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "name": self.name, "full_name": self.full_name, "owner": self.owner.to_dict(), "description": self.description, "private": self.private, "fork": self.fork, "html_url": self.html_url, "clone_url": self.clone_url, "default_branch": self.default_branch, "language": self.language, "stargazers_count": self.stargazers_count, "forks_count": self.forks_count, "open_issues_count": self.open_issues_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, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitHubRepository: """Чистая функция.""" owner_data = data.get("owner") or {} return cls( id=int(data.get("id", 0)), name=str(data.get("name", "")), full_name=str(data.get("full_name", "")), owner=GitHubUser.from_api_response(owner_data if isinstance(owner_data, dict) else {}), description=str(data.get("description", "") or ""), private=bool(data.get("private", False)), fork=bool(data.get("fork", False)), html_url=str(data.get("html_url", "")), clone_url=str(data.get("clone_url", "")), default_branch=str(data.get("default_branch", "main")), language=data.get("language"), stargazers_count=int(data.get("stargazers_count", 0)), forks_count=int(data.get("forks_count", 0)), open_issues_count=int(data.get("open_issues_count", 0)), created_at=_parse_datetime(data.get("created_at")), updated_at=_parse_datetime(data.get("updated_at")), ) @dataclass class GitHubIssue: """Issue на GitHub.""" id: int number: int title: str body: str = "" state: str = "open" # open | closed user: GitHubUser | None = None labels: list[GitHubLabel] = field(default_factory=list) assignees: list[GitHubUser] = field(default_factory=list) html_url: str = "" comments: int = 0 locked: bool = False 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, "number": self.number, "title": self.title, "body": self.body, "state": self.state, "user": self.user.to_dict() if self.user else None, "labels": [l.to_dict() for l in self.labels], "assignees": [a.to_dict() for a in self.assignees], "html_url": self.html_url, "comments": self.comments, "locked": self.locked, "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]) -> GitHubIssue: """Чистая функция.""" user_data = data.get("user") user = ( GitHubUser.from_api_response(user_data) if isinstance(user_data, dict) and user_data else None ) labels_raw = data.get("labels") or [] labels = [ GitHubLabel.from_api_response(l) for l in labels_raw if isinstance(l, dict) ] assignees_raw = data.get("assignees") or [] assignees = [ GitHubUser.from_api_response(a) for a in assignees_raw if isinstance(a, dict) ] return cls( id=int(data.get("id", 0)), number=int(data.get("number", 0)), title=str(data.get("title", "")), body=str(data.get("body", "") or ""), state=str(data.get("state", "open")), user=user, labels=labels, assignees=assignees, html_url=str(data.get("html_url", "")), comments=int(data.get("comments", 0)), locked=bool(data.get("locked", False)), 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 GitHubPullRequest: """Pull Request на GitHub.""" id: int number: int title: str body: str = "" state: str = "open" # open | closed merged: bool = False user: GitHubUser | None = None head_ref: str = "" base_ref: str = "" html_url: str = "" diff_url: str = "" additions: int = 0 deletions: int = 0 changed_files: 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, "number": self.number, "title": self.title, "body": self.body, "state": self.state, "merged": self.merged, "user": self.user.to_dict() if self.user else None, "head_ref": self.head_ref, "base_ref": self.base_ref, "html_url": self.html_url, "diff_url": self.diff_url, "additions": self.additions, "deletions": self.deletions, "changed_files": self.changed_files, "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]) -> GitHubPullRequest: """Чистая функция.""" user_data = data.get("user") user = ( GitHubUser.from_api_response(user_data) if isinstance(user_data, dict) and user_data else None ) head_data = data.get("head") or {} base_data = data.get("base") or {} return cls( id=int(data.get("id", 0)), number=int(data.get("number", 0)), title=str(data.get("title", "")), body=str(data.get("body", "") or ""), state=str(data.get("state", "open")), merged=bool(data.get("merged", False)), user=user, head_ref=str((head_data if isinstance(head_data, dict) else {}).get("ref", "")), base_ref=str((base_data if isinstance(base_data, dict) else {}).get("ref", "")), html_url=str(data.get("html_url", "")), diff_url=str(data.get("diff_url", "")), additions=int(data.get("additions", 0)), deletions=int(data.get("deletions", 0)), changed_files=int(data.get("changed_files", 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 GitHubComment: """Комментарий к issue/PR.""" id: int body: str user: GitHubUser | None = None html_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, "user": self.user.to_dict() if self.user else None, "html_url": self.html_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]) -> GitHubComment: """Чистая функция.""" user_data = data.get("user") user = ( GitHubUser.from_api_response(user_data) if isinstance(user_data, dict) and user_data else None ) return cls( id=int(data.get("id", 0)), body=str(data.get("body", "")), user=user, html_url=str(data.get("html_url", "")), created_at=_parse_datetime(data.get("created_at")), updated_at=_parse_datetime(data.get("updated_at")), ) @dataclass class GitHubCommit: """Commit на GitHub.""" sha: str message: str author_name: str = "" author_email: str = "" author_login: str | None = None html_url: str = "" committed_at: datetime | None = None def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "sha": self.sha, "message": self.message, "author_name": self.author_name, "author_email": self.author_email, "author_login": self.author_login, "html_url": self.html_url, "committed_at": self.committed_at.isoformat() if self.committed_at else None, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitHubCommit: """Чистая функция.""" commit_data = data.get("commit") or {} author_data = commit_data.get("author") or {} author_info = data.get("author") return cls( sha=str(data.get("sha", "")), message=str(commit_data.get("message", "")), author_name=str(author_data.get("name", "")), author_email=str(author_data.get("email", "")), author_login=( author_info.get("login") if isinstance(author_info, dict) and author_info else None ), html_url=str(data.get("html_url", "")), committed_at=_parse_datetime(author_data.get("date")), ) @dataclass class GitHubBranch: """Ветка репозитория.""" name: str protected: bool = False commit_sha: str = "" def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "name": self.name, "protected": self.protected, "commit_sha": self.commit_sha, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitHubBranch: """Чистая функция.""" commit_data = data.get("commit") or {} return cls( name=str(data.get("name", "")), protected=bool(data.get("protected", False)), commit_sha=str(commit_data.get("sha", "") if isinstance(commit_data, dict) else ""), ) @dataclass class GitHubFile: """Содержимое файла в репозитории.""" name: str path: str sha: str size: int type: str = "file" # file | dir | symlink | submodule content: str = "" # base64 encoded encoding: str = "" html_url: str = "" download_url: str | None = None 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 { "name": self.name, "path": self.path, "sha": self.sha, "size": self.size, "type": self.type, "content": self.get_decoded_content(), "encoding": self.encoding, "html_url": self.html_url, "download_url": self.download_url, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitHubFile: """Чистая функция.""" return cls( name=str(data.get("name", "")), path=str(data.get("path", "")), sha=str(data.get("sha", "")), size=int(data.get("size", 0)), type=str(data.get("type", "file")), content=str(data.get("content", "") or ""), encoding=str(data.get("encoding", "") or ""), html_url=str(data.get("html_url", "")), download_url=data.get("download_url"), ) @dataclass class GitHubSearchResult: """Результат поиска по коду.""" repository: str path: str name: str sha: str html_url: str = "" score: float = 0.0 text_matches: list[dict[str, Any]] = field(default_factory=list) def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "repository": self.repository, "path": self.path, "name": self.name, "sha": self.sha, "html_url": self.html_url, "score": self.score, "text_matches": self.text_matches, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> GitHubSearchResult: """Чистая функция.""" repo_data = data.get("repository") or {} text_matches_raw = data.get("text_matches") or [] text_matches: list[dict[str, Any]] = [ m for m in text_matches_raw if isinstance(m, dict) ] return cls( repository=str((repo_data if isinstance(repo_data, dict) else {}).get("full_name", "")), path=str(data.get("path", "")), name=str(data.get("name", "")), sha=str(data.get("sha", "")), html_url=str(data.get("html_url", "")), score=float(data.get("score", 0.0)), text_matches=text_matches, ) # ============================================================================ # 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 # ============================================================================ # GitHub Client # ============================================================================ class GitHubClient: """ Асинхронный клиент для GitHub REST API v3. Автоматически обрабатывает: - Аутентификацию через Bearer token - Rate limiting (читает X-RateLimit-* headers) - Retry логику с exponential backoff - Пагинацию через Link header """ def __init__(self, config: GitHubConfig): 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) -> GitHubClient: 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 GitHubError( "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 GitHubError("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 запрос к GitHub API с retry логикой. Обрабатывает rate limiting через X-RateLimit-Remaining headers. """ session = self._require_session() url = f"{self.config.base_url.rstrip('/')}{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() # Rate limit headers remaining = response.headers.get("X-RateLimit-Remaining") if response.status == 401: raise GitHubAuthError("Authentication failed") elif response.status == 403: # Может быть rate limit if remaining == "0": reset_str = response.headers.get("X-RateLimit-Reset") reset_at = ( datetime.fromtimestamp(int(reset_str)) if reset_str and reset_str.isdigit() else None ) raise GitHubRateLimitError( "Rate limit exceeded", reset_at=reset_at, ) text = await response.text() raise GitHubError(f"Forbidden: {text}") elif response.status == 404: raise GitHubNotFoundError( f"Resource not found: {endpoint}" ) elif response.status == 422: error_data = await response.json() errors = ( error_data.get("errors", []) if isinstance(error_data, dict) else [] ) message = ( error_data.get("message", "Validation failed") if isinstance(error_data, dict) else "Validation failed" ) raise GitHubValidationError(message, errors=errors) elif response.status == 429: raise GitHubRateLimitError("Too many requests") elif response.status >= 400: text = await response.text() raise GitHubError( f"GitHub API error {response.status}: {text}" ) # Успешный ответ if response.status == 204: # No content return None content_type = response.content_type or "" if "application/json" in content_type: return await response.json() else: return await response.text() except (GitHubAuthError, GitHubNotFoundError, GitHubValidationError) as e: # Client ошибки не повторяем raise e except GitHubRateLimitError as e: # Rate limit — можно подождать 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 GitHubError( f"Request failed after {self.config.max_retries} attempts: {last_error}" ) # ==================================================================== # Repositories # ==================================================================== async def list_user_repos( self, username: str | None = None, per_page: int = 30, page: int = 1, ) -> list[GitHubRepository]: """Список репозиториев пользователя (или аутентифицированного).""" if username: endpoint = f"/users/{username}/repos" else: endpoint = "/user/repos" params = {"per_page": per_page, "page": page} response = await self._request("GET", endpoint, params=params) return [ GitHubRepository.from_api_response(r) for r in (response if isinstance(response, list) else []) if isinstance(r, dict) ] async def get_repo(self, owner: str, repo: str) -> GitHubRepository: """Получить информацию о репозитории.""" response = await self._request("GET", f"/repos/{owner}/{repo}") if not isinstance(response, dict): raise GitHubError("Invalid response format") return GitHubRepository.from_api_response(response) async def search_repos( self, query: str, per_page: int = 30, page: int = 1, ) -> dict[str, Any]: """Поиск репозиториев.""" params = {"q": query, "per_page": per_page, "page": page} response = await self._request("GET", "/search/repositories", params=params) if not isinstance(response, dict): return {"items": [], "total": 0} items_raw = response.get("items") or [] items = [ GitHubRepository.from_api_response(r).to_dict() for r in items_raw if isinstance(r, dict) ] return { "items": items, "total": int(response.get("total_count", 0)), } # ==================================================================== # Issues # ==================================================================== async def list_issues( self, owner: str, repo: str, state: str = "open", labels: str | None = None, per_page: int = 30, page: int = 1, ) -> list[GitHubIssue]: """Список issues репозитория.""" params: dict[str, Any] = { "state": state, "per_page": per_page, "page": page, } if labels: params["labels"] = labels response = await self._request( "GET", f"/repos/{owner}/{repo}/issues", params=params ) # Фильтруем PR (они тоже возвращаются как issues) issues: list[GitHubIssue] = [] for item in response if isinstance(response, list) else []: if not isinstance(item, dict): continue if "pull_request" in item: continue # Это PR, пропускаем issues.append(GitHubIssue.from_api_response(item)) return issues async def get_issue(self, owner: str, repo: str, issue_number: int) -> GitHubIssue: """Получить issue по номеру.""" response = await self._request( "GET", f"/repos/{owner}/{repo}/issues/{issue_number}" ) if not isinstance(response, dict): raise GitHubError("Invalid response format") return GitHubIssue.from_api_response(response) async def create_issue( self, owner: str, repo: str, title: str, body: str | None = None, labels: list[str] | None = None, assignees: list[str] | None = None, ) -> GitHubIssue: """Создать новый issue.""" payload: dict[str, Any] = {"title": title} if body is not None: payload["body"] = body if labels: payload["labels"] = labels if assignees: payload["assignees"] = assignees response = await self._request( "POST", f"/repos/{owner}/{repo}/issues", json_data=payload ) if not isinstance(response, dict): raise GitHubError("Invalid response format") return GitHubIssue.from_api_response(response) async def update_issue( self, owner: str, repo: str, issue_number: int, title: str | None = None, body: str | None = None, state: str | None = None, labels: list[str] | None = None, assignees: list[str] | None = None, ) -> GitHubIssue: """Обновить issue.""" payload: dict[str, Any] = {} if title is not None: payload["title"] = title if body is not None: payload["body"] = body if state is not None: payload["state"] = state if labels is not None: payload["labels"] = labels if assignees is not None: payload["assignees"] = assignees response = await self._request( "PATCH", f"/repos/{owner}/{repo}/issues/{issue_number}", json_data=payload, ) if not isinstance(response, dict): raise GitHubError("Invalid response format") return GitHubIssue.from_api_response(response) async def close_issue( self, owner: str, repo: str, issue_number: int ) -> GitHubIssue: """Закрыть issue.""" return await self.update_issue(owner, repo, issue_number, state="closed") async def add_issue_comment( self, owner: str, repo: str, issue_number: int, body: str ) -> GitHubComment: """Добавить комментарий к issue.""" response = await self._request( "POST", f"/repos/{owner}/{repo}/issues/{issue_number}/comments", json_data={"body": body}, ) if not isinstance(response, dict): raise GitHubError("Invalid response format") return GitHubComment.from_api_response(response) async def list_issue_comments( self, owner: str, repo: str, issue_number: int, per_page: int = 30, ) -> list[GitHubComment]: """Список комментариев к issue.""" params = {"per_page": per_page} response = await self._request( "GET", f"/repos/{owner}/{repo}/issues/{issue_number}/comments", params=params, ) return [ GitHubComment.from_api_response(c) for c in (response if isinstance(response, list) else []) if isinstance(c, dict) ] # ==================================================================== # Pull Requests # ==================================================================== async def list_pull_requests( self, owner: str, repo: str, state: str = "open", per_page: int = 30, page: int = 1, ) -> list[GitHubPullRequest]: """Список pull requests.""" params = {"state": state, "per_page": per_page, "page": page} response = await self._request( "GET", f"/repos/{owner}/{repo}/pulls", params=params ) return [ GitHubPullRequest.from_api_response(pr) for pr in (response if isinstance(response, list) else []) if isinstance(pr, dict) ] async def get_pull_request( self, owner: str, repo: str, pr_number: int ) -> GitHubPullRequest: """Получить pull request по номеру.""" response = await self._request( "GET", f"/repos/{owner}/{repo}/pulls/{pr_number}" ) if not isinstance(response, dict): raise GitHubError("Invalid response format") return GitHubPullRequest.from_api_response(response) async def create_pull_request( self, owner: str, repo: str, title: str, head: str, base: str, body: str | None = None, draft: bool = False, ) -> GitHubPullRequest: """Создать pull request.""" payload: dict[str, Any] = { "title": title, "head": head, "base": base, } if body is not None: payload["body"] = body if draft: payload["draft"] = True response = await self._request( "POST", f"/repos/{owner}/{repo}/pulls", json_data=payload ) if not isinstance(response, dict): raise GitHubError("Invalid response format") return GitHubPullRequest.from_api_response(response) # ==================================================================== # Commits & Branches # ==================================================================== async def list_commits( self, owner: str, repo: str, sha: str | None = None, path: str | None = None, per_page: int = 30, ) -> list[GitHubCommit]: """Список commits.""" params: dict[str, Any] = {"per_page": per_page} if sha: params["sha"] = sha if path: params["path"] = path response = await self._request( "GET", f"/repos/{owner}/{repo}/commits", params=params ) return [ GitHubCommit.from_api_response(c) for c in (response if isinstance(response, list) else []) if isinstance(c, dict) ] async def list_branches( self, owner: str, repo: str, per_page: int = 30 ) -> list[GitHubBranch]: """Список веток репозитория.""" params = {"per_page": per_page} response = await self._request( "GET", f"/repos/{owner}/{repo}/branches", params=params ) return [ GitHubBranch.from_api_response(b) for b in (response if isinstance(response, list) else []) if isinstance(b, dict) ] # ==================================================================== # Files & Contents # ==================================================================== async def get_file_contents( self, owner: str, repo: str, path: str, ref: str | None = None ) -> GitHubFile: """Получить содержимое файла.""" params: dict[str, Any] = {} if ref: params["ref"] = ref response = await self._request( "GET", f"/repos/{owner}/{repo}/contents/{path}", params=params ) if not isinstance(response, dict): raise GitHubError("Invalid response format") return GitHubFile.from_api_response(response) async def list_directory( self, owner: str, repo: str, path: str = "", ref: str | None = None ) -> list[GitHubFile]: """Список файлов в директории.""" params: dict[str, Any] = {} if ref: params["ref"] = ref endpoint = f"/repos/{owner}/{repo}/contents" if path: endpoint += f"/{path}" response = await self._request("GET", endpoint, params=params) return [ GitHubFile.from_api_response(f) for f in (response if isinstance(response, list) else []) if isinstance(f, dict) ] # ==================================================================== # Search # ==================================================================== async def search_code( self, query: str, per_page: int = 30 ) -> dict[str, Any]: """Поиск по коду в GitHub.""" params = {"q": query, "per_page": per_page} response = await self._request( "GET", "/search/code", params=params ) if not isinstance(response, dict): return {"items": [], "total": 0} items_raw = response.get("items") or [] items = [ GitHubSearchResult.from_api_response(item).to_dict() for item in items_raw if isinstance(item, dict) ] return { "items": items, "total": int(response.get("total_count", 0)), } async def search_issues( self, query: str, per_page: int = 30 ) -> dict[str, Any]: """Поиск по issues.""" params = {"q": query, "per_page": per_page} response = await self._request( "GET", "/search/issues", params=params ) if not isinstance(response, dict): return {"items": [], "total": 0} items_raw = response.get("items") or [] items = [ GitHubIssue.from_api_response(item).to_dict() for item in items_raw if isinstance(item, dict) ] return { "items": items, "total": int(response.get("total_count", 0)), } # ============================================================================ # MCP Tools Definition # ============================================================================ GITHUB_TOOLS: list[dict[str, Any]] = [ { "name": "github_list_user_repos", "description": "Получить список репозиториев пользователя (или аутентифицированного пользователя)", "parameters": { "type": "object", "properties": { "username": { "type": "string", "description": "Имя пользователя (опционально, по умолчанию - текущий)", }, "per_page": {"type": "integer", "default": 30, "maximum": 100}, "page": {"type": "integer", "default": 1}, }, }, }, { "name": "github_get_repo", "description": "Получить информацию о репозитории", "parameters": { "type": "object", "properties": { "owner": {"type": "string", "description": "Владелец репозитория"}, "repo": {"type": "string", "description": "Имя репозитория"}, }, "required": ["owner", "repo"], }, }, { "name": "github_search_repos", "description": "Поиск репозиториев на GitHub", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Поисковый запрос"}, "per_page": {"type": "integer", "default": 30}, "page": {"type": "integer", "default": 1}, }, "required": ["query"], }, }, { "name": "github_list_issues", "description": "Получить список issues репозитория", "parameters": { "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "state": { "type": "string", "enum": ["open", "closed", "all"], "default": "open", }, "labels": { "type": "string", "description": "Список labels через запятую", }, "per_page": {"type": "integer", "default": 30}, "page": {"type": "integer", "default": 1}, }, "required": ["owner", "repo"], }, }, { "name": "github_get_issue", "description": "Получить issue по номеру", "parameters": { "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "issue_number": {"type": "integer"}, }, "required": ["owner", "repo", "issue_number"], }, }, { "name": "github_create_issue", "description": "Создать новый issue", "parameters": { "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "title": {"type": "string"}, "body": {"type": "string"}, "labels": { "type": "array", "items": {"type": "string"}, }, "assignees": { "type": "array", "items": {"type": "string"}, }, }, "required": ["owner", "repo", "title"], }, }, { "name": "github_update_issue", "description": "Обновить существующий issue", "parameters": { "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "issue_number": {"type": "integer"}, "title": {"type": "string"}, "body": {"type": "string"}, "state": {"type": "string", "enum": ["open", "closed"]}, "labels": {"type": "array", "items": {"type": "string"}}, "assignees": {"type": "array", "items": {"type": "string"}}, }, "required": ["owner", "repo", "issue_number"], }, }, { "name": "github_close_issue", "description": "Закрыть issue", "parameters": { "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "issue_number": {"type": "integer"}, }, "required": ["owner", "repo", "issue_number"], }, }, { "name": "github_add_issue_comment", "description": "Добавить комментарий к issue", "parameters": { "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "issue_number": {"type": "integer"}, "body": {"type": "string", "description": "Текст комментария"}, }, "required": ["owner", "repo", "issue_number", "body"], }, }, { "name": "github_list_issue_comments", "description": "Получить комментарии к issue", "parameters": { "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "issue_number": {"type": "integer"}, "per_page": {"type": "integer", "default": 30}, }, "required": ["owner", "repo", "issue_number"], }, }, { "name": "github_list_pull_requests", "description": "Получить список pull requests", "parameters": { "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "state": { "type": "string", "enum": ["open", "closed", "all"], "default": "open", }, "per_page": {"type": "integer", "default": 30}, "page": {"type": "integer", "default": 1}, }, "required": ["owner", "repo"], }, }, { "name": "github_get_pull_request", "description": "Получить pull request по номеру", "parameters": { "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "pr_number": {"type": "integer"}, }, "required": ["owner", "repo", "pr_number"], }, }, { "name": "github_create_pull_request", "description": "Создать новый pull request", "parameters": { "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "title": {"type": "string"}, "head": {"type": "string", "description": "Ветка с изменениями"}, "base": {"type": "string", "description": "Целевая ветка"}, "body": {"type": "string"}, "draft": {"type": "boolean", "default": False}, }, "required": ["owner", "repo", "title", "head", "base"], }, }, { "name": "github_list_commits", "description": "Получить список commits", "parameters": { "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "sha": {"type": "string", "description": "Branch или SHA"}, "path": {"type": "string", "description": "Фильтр по пути файла"}, "per_page": {"type": "integer", "default": 30}, }, "required": ["owner", "repo"], }, }, { "name": "github_list_branches", "description": "Получить список веток репозитория", "parameters": { "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "per_page": {"type": "integer", "default": 30}, }, "required": ["owner", "repo"], }, }, { "name": "github_get_file_contents", "description": "Получить содержимое файла", "parameters": { "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "path": {"type": "string", "description": "Путь к файлу"}, "ref": {"type": "string", "description": "Branch, tag или SHA"}, }, "required": ["owner", "repo", "path"], }, }, { "name": "github_list_directory", "description": "Получить список файлов в директории", "parameters": { "type": "object", "properties": { "owner": {"type": "string"}, "repo": {"type": "string"}, "path": {"type": "string", "default": ""}, "ref": {"type": "string"}, }, "required": ["owner", "repo"], }, }, { "name": "github_search_code", "description": "Поиск по коду в GitHub", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Поисковый запрос (например: 'function repo:owner/repo')", }, "per_page": {"type": "integer", "default": 30}, }, "required": ["query"], }, }, { "name": "github_search_issues", "description": "Поиск по issues на GitHub", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Поисковый запрос (например: 'bug is:open repo:owner/repo')", }, "per_page": {"type": "integer", "default": 30}, }, "required": ["query"], }, }, ] # ============================================================================ # Tool Handlers # ============================================================================ async def _handle_list_user_repos( client: GitHubClient, params: dict[str, Any] ) -> list[dict[str, Any]]: repos = await client.list_user_repos( username=params.get("username"), per_page=int(params.get("per_page", 30)), page=int(params.get("page", 1)), ) return [r.to_dict() for r in repos] async def _handle_get_repo( client: GitHubClient, params: dict[str, Any] ) -> dict[str, Any]: repo = await client.get_repo(str(params["owner"]), str(params["repo"])) return repo.to_dict() async def _handle_search_repos( client: GitHubClient, params: dict[str, Any] ) -> dict[str, Any]: return await client.search_repos( query=str(params["query"]), per_page=int(params.get("per_page", 30)), page=int(params.get("page", 1)), ) async def _handle_list_issues( client: GitHubClient, params: dict[str, Any] ) -> list[dict[str, Any]]: issues = await client.list_issues( owner=str(params["owner"]), repo=str(params["repo"]), state=str(params.get("state", "open")), labels=params.get("labels"), per_page=int(params.get("per_page", 30)), page=int(params.get("page", 1)), ) return [i.to_dict() for i in issues] async def _handle_get_issue( client: GitHubClient, params: dict[str, Any] ) -> dict[str, Any]: issue = await client.get_issue( owner=str(params["owner"]), repo=str(params["repo"]), issue_number=int(params["issue_number"]), ) return issue.to_dict() async def _handle_create_issue( client: GitHubClient, params: dict[str, Any] ) -> dict[str, Any]: issue = await client.create_issue( owner=str(params["owner"]), repo=str(params["repo"]), title=str(params["title"]), body=params.get("body"), labels=params.get("labels"), assignees=params.get("assignees"), ) return issue.to_dict() async def _handle_update_issue( client: GitHubClient, params: dict[str, Any] ) -> dict[str, Any]: issue = await client.update_issue( owner=str(params["owner"]), repo=str(params["repo"]), issue_number=int(params["issue_number"]), title=params.get("title"), body=params.get("body"), state=params.get("state"), labels=params.get("labels"), assignees=params.get("assignees"), ) return issue.to_dict() async def _handle_close_issue( client: GitHubClient, params: dict[str, Any] ) -> dict[str, Any]: issue = await client.close_issue( owner=str(params["owner"]), repo=str(params["repo"]), issue_number=int(params["issue_number"]), ) return issue.to_dict() async def _handle_add_issue_comment( client: GitHubClient, params: dict[str, Any] ) -> dict[str, Any]: comment = await client.add_issue_comment( owner=str(params["owner"]), repo=str(params["repo"]), issue_number=int(params["issue_number"]), body=str(params["body"]), ) return comment.to_dict() async def _handle_list_issue_comments( client: GitHubClient, params: dict[str, Any] ) -> list[dict[str, Any]]: comments = await client.list_issue_comments( owner=str(params["owner"]), repo=str(params["repo"]), issue_number=int(params["issue_number"]), per_page=int(params.get("per_page", 30)), ) return [c.to_dict() for c in comments] async def _handle_list_pull_requests( client: GitHubClient, params: dict[str, Any] ) -> list[dict[str, Any]]: prs = await client.list_pull_requests( owner=str(params["owner"]), repo=str(params["repo"]), state=str(params.get("state", "open")), per_page=int(params.get("per_page", 30)), page=int(params.get("page", 1)), ) return [pr.to_dict() for pr in prs] async def _handle_get_pull_request( client: GitHubClient, params: dict[str, Any] ) -> dict[str, Any]: pr = await client.get_pull_request( owner=str(params["owner"]), repo=str(params["repo"]), pr_number=int(params["pr_number"]), ) return pr.to_dict() async def _handle_create_pull_request( client: GitHubClient, params: dict[str, Any] ) -> dict[str, Any]: pr = await client.create_pull_request( owner=str(params["owner"]), repo=str(params["repo"]), title=str(params["title"]), head=str(params["head"]), base=str(params["base"]), body=params.get("body"), draft=bool(params.get("draft", False)), ) return pr.to_dict() async def _handle_list_commits( client: GitHubClient, params: dict[str, Any] ) -> list[dict[str, Any]]: commits = await client.list_commits( owner=str(params["owner"]), repo=str(params["repo"]), sha=params.get("sha"), path=params.get("path"), per_page=int(params.get("per_page", 30)), ) return [c.to_dict() for c in commits] async def _handle_list_branches( client: GitHubClient, params: dict[str, Any] ) -> list[dict[str, Any]]: branches = await client.list_branches( owner=str(params["owner"]), repo=str(params["repo"]), per_page=int(params.get("per_page", 30)), ) return [b.to_dict() for b in branches] async def _handle_get_file_contents( client: GitHubClient, params: dict[str, Any] ) -> dict[str, Any]: file = await client.get_file_contents( owner=str(params["owner"]), repo=str(params["repo"]), path=str(params["path"]), ref=params.get("ref"), ) return file.to_dict() async def _handle_list_directory( client: GitHubClient, params: dict[str, Any] ) -> list[dict[str, Any]]: files = await client.list_directory( owner=str(params["owner"]), repo=str(params["repo"]), path=str(params.get("path", "")), ref=params.get("ref"), ) return [f.to_dict() for f in files] async def _handle_search_code( client: GitHubClient, params: dict[str, Any] ) -> dict[str, Any]: return await client.search_code( query=str(params["query"]), per_page=int(params.get("per_page", 30)), ) async def _handle_search_issues( client: GitHubClient, params: dict[str, Any] ) -> dict[str, Any]: return await client.search_issues( query=str(params["query"]), per_page=int(params.get("per_page", 30)), ) # Таблица dispatch _TOOL_HANDLERS: dict[str, Any] = { "github_list_user_repos": _handle_list_user_repos, "github_get_repo": _handle_get_repo, "github_search_repos": _handle_search_repos, "github_list_issues": _handle_list_issues, "github_get_issue": _handle_get_issue, "github_create_issue": _handle_create_issue, "github_update_issue": _handle_update_issue, "github_close_issue": _handle_close_issue, "github_add_issue_comment": _handle_add_issue_comment, "github_list_issue_comments": _handle_list_issue_comments, "github_list_pull_requests": _handle_list_pull_requests, "github_get_pull_request": _handle_get_pull_request, "github_create_pull_request": _handle_create_pull_request, "github_list_commits": _handle_list_commits, "github_list_branches": _handle_list_branches, "github_get_file_contents": _handle_get_file_contents, "github_list_directory": _handle_list_directory, "github_search_code": _handle_search_code, "github_search_issues": _handle_search_issues, } # ============================================================================ # GitHub MCP Server # ============================================================================ class GitHubMCPServer: """ MCP Server для GitHub. Следует принципу "You Might Not Need an Effect": - Состояние клиента управляется через context manager - Dispatch по таблице вместо runtime reflection - Все преобразования — чистые функции """ def __init__(self, config: GitHubConfig | None = None): self.config = config or GitHubConfig.from_env() self._client: GitHubClient | None = None async def __aenter__(self) -> GitHubMCPServer: errors = self.config.validate() if errors: raise GitHubError(f"Invalid configuration: {', '.join(errors)}") self._client = GitHubClient(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) -> GitHubClient: """Получить активный клиент или выбросить ошибку.""" if self._client is None: raise GitHubError( "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(GITHUB_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 GitHubError 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 GitHubNotFoundError as e: return { "success": False, "error": f"Not found: {e}", "error_type": "not_found", } except GitHubAuthError as e: return { "success": False, "error": f"Authentication failed: {e}", "error_type": "auth_error", } except GitHubRateLimitError 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 GitHubValidationError as e: return { "success": False, "error": f"Validation error: {e}", "error_type": "validation_error", "errors": e.errors, } except GitHubError as e: return { "success": False, "error": str(e), "error_type": "github_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_github_mcp( token: str | None = None, base_url: str = "https://api.github.com", timeout_seconds: float = 30.0, ) -> AsyncIterator[GitHubMCPServer]: """ Создать GitHub MCP Server с автоматическим управлением подключением. Usage: async with create_github_mcp(token="ghp_...") as mcp: tools = mcp.get_tools() result = await mcp.call_tool( "github_create_issue", {"owner": "my-org", "repo": "my-repo", "title": "Bug"}, ) """ config = GitHubConfig( base_url=base_url, token=token or os.getenv("GITHUB_TOKEN") or os.getenv("GITHUB_PERSONAL_ACCESS_TOKEN"), timeout_seconds=timeout_seconds, ) async with GitHubMCPServer(config) as server: yield server # ============================================================================ # Exports # ============================================================================ __all__ = [ # Config "GitHubConfig", # Exceptions "GitHubError", "GitHubAuthError", "GitHubNotFoundError", "GitHubRateLimitError", "GitHubValidationError", # Models "GitHubUser", "GitHubLabel", "GitHubRepository", "GitHubIssue", "GitHubPullRequest", "GitHubComment", "GitHubCommit", "GitHubBranch", "GitHubFile", "GitHubSearchResult", # Client & Server "GitHubClient", "GitHubMCPServer", "GITHUB_TOOLS", # Helpers "create_github_mcp", ]