/
alexefan136
/
flowstack
Обзор
Документация
Войти
/
alexefan136
/
flowstack
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
main
core/engine/src/tools/mcp/linear.py
2 806 строк
92 KB
Alexander Efanov
Обновление репозитория
15 июл 2026, 12:19
15 июл 2026, 12:19
76704c6
Код
Авторство
О чём код?
""" Linear MCP Server — интеграция с Linear через MCP протокол. Предоставляет tools для работы с Linear (issue tracker для команд): - Users (текущий пользователь, список) - Teams (команды) - Issues (CRUD, поиск через filter) - Projects (проекты) - Cycles (итерации/спринты) - Labels (метки) - Comments (комментарии) - Workflow States (статусы) Используемые API: - Linear GraphQL API: https://api.linear.app/graphql - Linear REST API: https://api.linear.app (для некоторых операций) Документация: - https://developers.linear.app/docs/graphql/overview - https://developers.linear.app/docs/graphql/working-with-the-graphql-api Аутентификация: - Personal API Key (Authorization: <api_key> header) - OAuth2 Bearer token Архитектурные принципы: - Чистые функции для преобразования данных (from_api_response, to_dict) - Type-safe dispatch через таблицу handlers - Явная обработка ошибок - Async-first с context managers - Следование "You Might Not Need an Effect" - GraphQL-first подход """ 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 LinearConfig: """ Конфигурация подключения к Linear. Linear использует GraphQL API на https://api.linear.app/graphql. """ # Базовый URL base_url: str = "https://api.linear.app" graphql_endpoint: str = "/graphql" # Аутентификация api_key: str | None = None oauth_token: str | None = None # 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_auth_headers(self) -> dict[str, str]: """Получить headers для аутентификации. Чистая функция.""" headers = { "Content-Type": "application/json", "Accept": "application/json", } if self.api_key: # Linear использует plain API key в Authorization header headers["Authorization"] = self.api_key elif self.oauth_token: headers["Authorization"] = f"Bearer {self.oauth_token}" return headers def get_graphql_url(self) -> str: """Получить полный URL GraphQL endpoint.""" return f"{self.base_url.rstrip('/')}{self.graphql_endpoint}" def validate(self) -> list[str]: """Валидировать конфигурацию. Чистая функция.""" errors: list[str] = [] if not self.base_url: errors.append("base_url is required") if not self.api_key and not self.oauth_token: errors.append("Either api_key or oauth_token is required") return errors @classmethod def from_env(cls) -> LinearConfig: """Создать из переменных окружения.""" return cls( base_url=os.getenv("LINEAR_BASE_URL", "https://api.linear.app"), api_key=( os.getenv("LINEAR_API_KEY") or os.getenv("LINEAR_PERSONAL_API_KEY") ), oauth_token=os.getenv("LINEAR_OAUTH_TOKEN"), timeout_seconds=float(os.getenv("LINEAR_TIMEOUT", "30")), max_retries=int(os.getenv("LINEAR_MAX_RETRIES", "3")), requests_per_second=float(os.getenv("LINEAR_RPS", "5.0")), ) # ============================================================================ # Exceptions # ============================================================================ class LinearError(Exception): """Базовое исключение Linear.""" pass class LinearAuthError(LinearError): """Ошибка аутентификации.""" pass class LinearNotFoundError(LinearError): """Ресурс не найден.""" pass class LinearRateLimitError(LinearError): """Превышен rate limit.""" def __init__(self, message: str, retry_after: int | None = None): super().__init__(message) self.retry_after = retry_after class LinearForbiddenError(LinearError): """Доступ запрещен.""" pass class LinearValidationError(LinearError): """Ошибка валидации запроса.""" def __init__(self, message: str, errors: list[dict[str, Any]] | None = None): super().__init__(message) self.errors = errors or [] class LinearGraphQLError(LinearError): """Ошибка GraphQL (из поля errors в ответе).""" def __init__( self, message: str, extensions: dict[str, Any] | None = None, ): super().__init__(message) self.extensions = extensions or {} # ============================================================================ # Response Models # ============================================================================ @dataclass class LinearUser: """Пользователь Linear.""" id: str name: str display_name: str = "" email: str = "" avatar_url: str = "" active: bool = True admin: bool = False guest: bool = False timezone: str = "" url: str = "" def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "name": self.name, "display_name": self.display_name, "email": self.email, "avatar_url": self.avatar_url, "active": self.active, "admin": self.admin, "guest": self.guest, "timezone": self.timezone, "url": self.url, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> LinearUser: """Чистая функция.""" return cls( id=str(data.get("id", "")), name=str(data.get("name", "")), display_name=str(data.get("displayName", "") or data.get("name", "")), email=str(data.get("email", "") or ""), avatar_url=str(data.get("avatarUrl", "") or ""), active=bool(data.get("active", True)), admin=bool(data.get("admin", False)), guest=bool(data.get("guest", False)), timezone=str(data.get("timezone", "") or ""), url=str(data.get("url", "") or ""), ) @dataclass class LinearOrganization: """Организация (workspace) в Linear.""" id: str name: str url_key: str = "" logo_url: str = "" created_at: datetime | None = None def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "name": self.name, "url_key": self.url_key, "logo_url": self.logo_url, "created_at": ( self.created_at.isoformat() if self.created_at else None ), } @classmethod def from_api_response(cls, data: dict[str, Any]) -> LinearOrganization: """Чистая функция.""" return cls( id=str(data.get("id", "")), name=str(data.get("name", "")), url_key=str(data.get("urlKey", "") or ""), logo_url=str(data.get("logoUrl", "") or ""), created_at=_parse_datetime(data.get("createdAt")), ) @dataclass class LinearTeam: """Команда в Linear.""" id: str key: str name: str description: str = "" private: bool = False icon: str = "" color: str = "" organization_id: str = "" def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "key": self.key, "name": self.name, "description": self.description, "private": self.private, "icon": self.icon, "color": self.color, "organization_id": self.organization_id, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> LinearTeam: """Чистая функция.""" org_data = data.get("organization") org_id = "" if isinstance(org_data, dict): oid = org_data.get("id") if isinstance(oid, str): org_id = oid return cls( id=str(data.get("id", "")), key=str(data.get("key", "")), name=str(data.get("name", "")), description=str(data.get("description", "") or ""), private=bool(data.get("private", False)), icon=str(data.get("icon", "") or ""), color=str(data.get("color", "") or ""), organization_id=org_id, ) @dataclass class LinearWorkflowState: """Workflow state (статус issue).""" id: str name: str description: str = "" type: str = "" # triage | backlog | unstarted | started | completed | canceled color: str = "" position: float = 0.0 team_id: str = "" def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "name": self.name, "description": self.description, "type": self.type, "color": self.color, "position": self.position, "team_id": self.team_id, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> LinearWorkflowState: """Чистая функция.""" team_data = data.get("team") team_id = "" if isinstance(team_data, dict): tid = team_data.get("id") if isinstance(tid, str): team_id = tid position_raw = data.get("position") position = 0.0 if isinstance(position_raw, (int, float)): position = float(position_raw) return cls( id=str(data.get("id", "")), name=str(data.get("name", "")), description=str(data.get("description", "") or ""), type=str(data.get("type", "")), color=str(data.get("color", "") or ""), position=position, team_id=team_id, ) @dataclass class LinearLabel: """Label (метка) в Linear.""" id: str name: str description: str = "" color: str = "" team_id: str | None = None is_global: bool = False def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "name": self.name, "description": self.description, "color": self.color, "team_id": self.team_id, "is_global": self.is_global, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> LinearLabel: """Чистая функция.""" team_data = data.get("team") team_id: str | None = None if isinstance(team_data, dict): tid = team_data.get("id") if isinstance(tid, str): team_id = tid return cls( id=str(data.get("id", "")), name=str(data.get("name", "")), description=str(data.get("description", "") or ""), color=str(data.get("color", "") or ""), team_id=team_id, is_global=bool(data.get("isGlobal", team_id is None)), ) @dataclass class LinearProject: """Проект в Linear.""" id: str name: str description: str = "" slug_id: str = "" state: str = "" # planned | started | paused | completed | canceled icon: str = "" color: str = "" url: str = "" start_date: str | None = None target_date: str | None = None created_at: datetime | None = None updated_at: datetime | None = None def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "name": self.name, "description": self.description, "slug_id": self.slug_id, "state": self.state, "icon": self.icon, "color": self.color, "url": self.url, "start_date": self.start_date, "target_date": self.target_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 ), } @classmethod def from_api_response(cls, data: dict[str, Any]) -> LinearProject: """Чистая функция.""" return cls( id=str(data.get("id", "")), name=str(data.get("name", "")), description=str(data.get("description", "") or ""), slug_id=str(data.get("slugId", "") or ""), state=str(data.get("state", "") or ""), icon=str(data.get("icon", "") or ""), color=str(data.get("color", "") or ""), url=str(data.get("url", "") or ""), start_date=data.get("startDate") if isinstance(data.get("startDate"), str) else None, target_date=data.get("targetDate") if isinstance(data.get("targetDate"), str) else None, created_at=_parse_datetime(data.get("createdAt")), updated_at=_parse_datetime(data.get("updatedAt")), ) @dataclass class LinearCycle: """Cycle (итерация/спринт) в Linear.""" id: str number: int = 0 name: str = "" description: str = "" starts_at: datetime | None = None ends_at: datetime | None = None completed_at: datetime | None = None team_id: str = "" is_current: bool = False is_future: bool = False is_past: bool = False def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "number": self.number, "name": self.name, "description": self.description, "starts_at": self.starts_at.isoformat() if self.starts_at else None, "ends_at": self.ends_at.isoformat() if self.ends_at else None, "completed_at": ( self.completed_at.isoformat() if self.completed_at else None ), "team_id": self.team_id, "is_current": self.is_current, "is_future": self.is_future, "is_past": self.is_past, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> LinearCycle: """Чистая функция.""" team_data = data.get("team") team_id = "" if isinstance(team_data, dict): tid = team_data.get("id") if isinstance(tid, str): team_id = tid number_raw = data.get("number") number = int(number_raw) if isinstance(number_raw, int) else 0 return cls( id=str(data.get("id", "")), number=number, name=str(data.get("name", "") or ""), description=str(data.get("description", "") or ""), starts_at=_parse_datetime(data.get("startsAt")), ends_at=_parse_datetime(data.get("endsAt")), completed_at=_parse_datetime(data.get("completedAt")), team_id=team_id, is_current=bool(data.get("isCurrent", False)), is_future=bool(data.get("isFuture", False)), is_past=bool(data.get("isPast", False)), ) @dataclass class LinearIssue: """Issue в Linear.""" id: str identifier: str # например, "ENG-123" title: str description: str = "" priority: int = 0 # 0=No priority, 1=Urgent, 2=High, 3=Medium, 4=Low priority_label: str = "No priority" estimate: float | None = None url: str = "" branch_name: str = "" number: int = 0 team_id: str = "" team_key: str = "" team_name: str = "" assignee: LinearUser | None = None creator: LinearUser | None = None state: LinearWorkflowState | None = None project: LinearProject | None = None cycle: LinearCycle | None = None parent: LinearIssue | None = None labels: list[LinearLabel] = field(default_factory=list) created_at: datetime | None = None updated_at: datetime | None = None due_date: str | None = None started_at: datetime | None = None completed_at: datetime | None = None canceled_at: datetime | None = None archived_at: datetime | None = None def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "identifier": self.identifier, "title": self.title, "description": self.description, "priority": self.priority, "priority_label": self.priority_label, "estimate": self.estimate, "url": self.url, "branch_name": self.branch_name, "number": self.number, "team_id": self.team_id, "team_key": self.team_key, "team_name": self.team_name, "assignee": self.assignee.to_dict() if self.assignee else None, "creator": self.creator.to_dict() if self.creator else None, "state": self.state.to_dict() if self.state else None, "project": self.project.to_dict() if self.project else None, "cycle": self.cycle.to_dict() if self.cycle else None, "parent": self.parent.to_dict() if self.parent else None, "labels": [l.to_dict() for l in self.labels], "created_at": ( self.created_at.isoformat() if self.created_at else None ), "updated_at": ( self.updated_at.isoformat() if self.updated_at else None ), "due_date": self.due_date, "started_at": ( self.started_at.isoformat() if self.started_at else None ), "completed_at": ( self.completed_at.isoformat() if self.completed_at else None ), "canceled_at": ( self.canceled_at.isoformat() if self.canceled_at else None ), "archived_at": ( self.archived_at.isoformat() if self.archived_at else None ), } @classmethod def from_api_response(cls, data: dict[str, Any]) -> LinearIssue: """Чистая функция.""" # Team team_data = data.get("team") team_id = "" team_key = "" team_name = "" if isinstance(team_data, dict): tid = team_data.get("id") tk = team_data.get("key") tn = team_data.get("name") if isinstance(tid, str): team_id = tid if isinstance(tk, str): team_key = tk if isinstance(tn, str): team_name = tn # Users assignee_data = data.get("assignee") assignee = ( LinearUser.from_api_response(assignee_data) if isinstance(assignee_data, dict) else None ) creator_data = data.get("creator") creator = ( LinearUser.from_api_response(creator_data) if isinstance(creator_data, dict) else None ) # State state_data = data.get("state") state = ( LinearWorkflowState.from_api_response(state_data) if isinstance(state_data, dict) else None ) # Project project_data = data.get("project") project = ( LinearProject.from_api_response(project_data) if isinstance(project_data, dict) else None ) # Cycle cycle_data = data.get("cycle") cycle = ( LinearCycle.from_api_response(cycle_data) if isinstance(cycle_data, dict) else None ) # Parent parent_data = data.get("parent") parent = ( LinearIssue.from_api_response(parent_data) if isinstance(parent_data, dict) else None ) # Labels labels_data = data.get("labels") labels_nodes: list[Any] = [] if isinstance(labels_data, dict): nodes = labels_data.get("nodes") if isinstance(nodes, list): labels_nodes = nodes elif isinstance(labels_data, list): labels_nodes = labels_data labels = [ LinearLabel.from_api_response(l) for l in labels_nodes if isinstance(l, dict) ] # Priority priority_raw = data.get("priority") priority = 0 if isinstance(priority_raw, int): priority = priority_raw priority_label_map = { 0: "No priority", 1: "Urgent", 2: "High", 3: "Medium", 4: "Low", } priority_label = priority_label_map.get(priority, "No priority") # Estimate estimate_raw = data.get("estimate") estimate: float | None = None if isinstance(estimate_raw, (int, float)): estimate = float(estimate_raw) # Number number_raw = data.get("number") number = int(number_raw) if isinstance(number_raw, int) else 0 # Due date (ISO строка YYYY-MM-DD) due_date_raw = data.get("dueDate") due_date: str | None = None if isinstance(due_date_raw, str): due_date = due_date_raw return cls( id=str(data.get("id", "")), identifier=str(data.get("identifier", "")), title=str(data.get("title", "")), description=str(data.get("description", "") or ""), priority=priority, priority_label=priority_label, estimate=estimate, url=str(data.get("url", "") or ""), branch_name=str(data.get("branchName", "") or ""), number=number, team_id=team_id, team_key=team_key, team_name=team_name, assignee=assignee, creator=creator, state=state, project=project, cycle=cycle, parent=parent, labels=labels, created_at=_parse_datetime(data.get("createdAt")), updated_at=_parse_datetime(data.get("updatedAt")), due_date=due_date, started_at=_parse_datetime(data.get("startedAt")), completed_at=_parse_datetime(data.get("completedAt")), canceled_at=_parse_datetime(data.get("canceledAt")), archived_at=_parse_datetime(data.get("archivedAt")), ) @dataclass class LinearComment: """Комментарий к issue.""" id: str body: str user: LinearUser | None = None issue_id: str = "" parent_id: str | None = None created_at: datetime | None = None updated_at: datetime | None = None url: str = "" def to_dict(self) -> dict[str, Any]: """Чистая функция.""" return { "id": self.id, "body": self.body, "user": self.user.to_dict() if self.user else None, "issue_id": self.issue_id, "parent_id": self.parent_id, "created_at": ( self.created_at.isoformat() if self.created_at else None ), "updated_at": ( self.updated_at.isoformat() if self.updated_at else None ), "url": self.url, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> LinearComment: """Чистая функция.""" user_data = data.get("user") user = ( LinearUser.from_api_response(user_data) if isinstance(user_data, dict) else None ) issue_data = data.get("issue") issue_id = "" if isinstance(issue_data, dict): iid = issue_data.get("id") if isinstance(iid, str): issue_id = iid parent_data = data.get("parent") parent_id: str | None = None if isinstance(parent_data, dict): pid = parent_data.get("id") if isinstance(pid, str): parent_id = pid return cls( id=str(data.get("id", "")), body=str(data.get("body", "") or ""), user=user, issue_id=issue_id, parent_id=parent_id, created_at=_parse_datetime(data.get("createdAt")), updated_at=_parse_datetime(data.get("editedAt") or data.get("updatedAt")), url=str(data.get("url", "") or ""), ) # ============================================================================ # 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 # ============================================================================ # Linear Client (GraphQL) # ============================================================================ class LinearClient: """ Асинхронный клиент для Linear GraphQL API. Все операции выполняются через единый GraphQL endpoint: https://api.linear.app/graphql Автоматически обрабатывает: - API Key / OAuth аутентификацию - Rate limiting через Retry-After header - Retry логику с exponential backoff - GraphQL ошибки (в поле errors ответа) """ def __init__(self, config: LinearConfig): 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) -> LinearClient: 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 LinearError( "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_auth_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 LinearError("Client is not connected") return self._session async def graphql( self, query: str, variables: dict[str, Any] | None = None, ) -> dict[str, Any]: """ Выполнить GraphQL запрос. Args: query: GraphQL query/mutation variables: Переменные запроса Returns: Поле data из ответа Raises: LinearGraphQLError: если в ответе есть errors LinearAuthError, LinearRateLimitError, etc. """ session = self._require_session() url = self.config.get_graphql_url() payload: dict[str, Any] = {"query": query} if variables: payload["variables"] = variables 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.post(url, json=payload) as response: self._last_request_time = loop.time() if response.status == 400: # GraphQL обычно возвращает 200 с errors, # но 400 — это invalid request text = await response.text() raise LinearValidationError(f"Bad request: {text}") elif response.status == 401: raise LinearAuthError("Authentication failed") elif response.status == 403: raise LinearForbiddenError("Access forbidden") elif response.status == 404: raise LinearNotFoundError("Endpoint not found") elif response.status == 429: retry_after_str = response.headers.get("Retry-After") retry_after: int | None = None if retry_after_str and retry_after_str.isdigit(): retry_after = int(retry_after_str) raise LinearRateLimitError( "Rate limit exceeded", retry_after=retry_after, ) elif response.status >= 400: text = await response.text() raise LinearError( f"Linear API error {response.status}: {text}" ) content_type = response.content_type or "" if "application/json" not in content_type: text = await response.text() raise LinearError(f"Unexpected content type: {text}") result = await response.json() if not isinstance(result, dict): raise LinearError("Invalid GraphQL response format") # Проверяем GraphQL errors errors_raw = result.get("errors") if isinstance(errors_raw, list) and errors_raw: first_error = errors_raw[0] if isinstance(first_error, dict): message = str(first_error.get("message", "Unknown GraphQL error")) extensions_raw = first_error.get("extensions") extensions: dict[str, Any] = ( extensions_raw if isinstance(extensions_raw, dict) else {} ) # Определяем тип ошибки по extensions.code code = extensions.get("code") if code == "AUTHENTICATION_ERROR": raise LinearAuthError(message) elif code in ("FORBIDDEN", "FORBIDDEN_RESOURCE"): raise LinearForbiddenError(message) elif code == "NOT_FOUND": raise LinearNotFoundError(message) elif code == "BAD_USER_INPUT": raise LinearValidationError(message) else: raise LinearGraphQLError( message, extensions=extensions ) data = result.get("data") if not isinstance(data, dict): raise LinearError("GraphQL response missing 'data'") return data except ( LinearAuthError, LinearNotFoundError, LinearForbiddenError, LinearValidationError, LinearGraphQLError, ) as e: raise e except LinearRateLimitError as e: last_error = e if attempt < self.config.max_retries - 1: wait_time = ( e.retry_after if e.retry_after is not None else 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 LinearError( f"Request failed after {self.config.max_retries} attempts: {last_error}" ) # ==================================================================== # Users # ==================================================================== async def get_current_user(self) -> LinearUser: """Получить информацию об аутентифицированном пользователе.""" query = """ query { viewer { id name displayName email avatarUrl active admin guest timezone url } } """ data = await self.graphql(query) viewer = data.get("viewer") if not isinstance(viewer, dict): raise LinearError("Invalid viewer response") return LinearUser.from_api_response(viewer) async def get_organization(self) -> LinearOrganization: """Получить информацию об организации.""" query = """ query { organization { id name urlKey logoUrl createdAt } } """ data = await self.graphql(query) org = data.get("organization") if not isinstance(org, dict): raise LinearError("Invalid organization response") return LinearOrganization.from_api_response(org) async def list_users(self, include_disabled: bool = False) -> list[LinearUser]: """Список пользователей организации.""" query = """ query ListUsers($includeDisabled: Boolean) { users(includeDisabled: $includeDisabled) { nodes { id name displayName email avatarUrl active admin guest timezone url } } } """ data = await self.graphql( query, {"includeDisabled": include_disabled} ) users_data = data.get("users") if not isinstance(users_data, dict): return [] nodes = users_data.get("nodes") if not isinstance(nodes, list): return [] return [ LinearUser.from_api_response(u) for u in nodes if isinstance(u, dict) ] # ==================================================================== # Teams # ==================================================================== async def list_teams(self) -> list[LinearTeam]: """Список команд.""" query = """ query { teams { nodes { id key name description private icon color organization { id } } } } """ data = await self.graphql(query) teams_data = data.get("teams") if not isinstance(teams_data, dict): return [] nodes = teams_data.get("nodes") if not isinstance(nodes, list): return [] return [ LinearTeam.from_api_response(t) for t in nodes if isinstance(t, dict) ] async def get_team(self, team_id: str) -> LinearTeam: """Получить команду по ID.""" query = """ query GetTeam($id: String!) { team(id: $id) { id key name description private icon color organization { id } } } """ data = await self.graphql(query, {"id": team_id}) team = data.get("team") if not isinstance(team, dict): raise LinearNotFoundError(f"Team {team_id} not found") return LinearTeam.from_api_response(team) # ==================================================================== # Issues # ==================================================================== async def list_issues( self, team_id: str | None = None, assignee_id: str | None = None, project_id: str | None = None, cycle_id: str | None = None, state_id: str | None = None, limit: int = 50, ) -> list[LinearIssue]: """ Список issues с фильтрами. Args: team_id: Фильтр по team assignee_id: Фильтр по assignee project_id: Фильтр по project cycle_id: Фильтр по cycle state_id: Фильтр по state limit: Максимальное количество (до 250) """ filter_obj: dict[str, Any] = {} if team_id: filter_obj["team"] = {"id": {"eq": team_id}} if assignee_id: filter_obj["assignee"] = {"id": {"eq": assignee_id}} if project_id: filter_obj["project"] = {"id": {"eq": project_id}} if cycle_id: filter_obj["cycle"] = {"id": {"eq": cycle_id}} if state_id: filter_obj["state"] = {"id": {"eq": state_id}} variables: dict[str, Any] = {"first": min(limit, 250)} if filter_obj: variables["filter"] = filter_obj query = """ query ListIssues($first: Int, $filter: IssueFilter) { issues(first: $first, filter: $filter) { nodes { id identifier title description priority estimate url branchName number dueDate createdAt updatedAt startedAt completedAt canceledAt archivedAt team { id key name } assignee { id name displayName email avatarUrl active admin guest timezone url } creator { id name displayName email avatarUrl active admin guest timezone url } state { id name description type color position team { id } } project { id name description slugId state icon color url startDate targetDate createdAt updatedAt } cycle { id number name description startsAt endsAt completedAt team { id } isCurrent isFuture isPast } parent { id identifier title } labels { nodes { id name description color isGlobal team { id } } } } } } """ data = await self.graphql(query, variables) issues_data = data.get("issues") if not isinstance(issues_data, dict): return [] nodes = issues_data.get("nodes") if not isinstance(nodes, list): return [] return [ LinearIssue.from_api_response(i) for i in nodes if isinstance(i, dict) ] async def get_issue(self, issue_id: str) -> LinearIssue: """Получить issue по ID.""" query = """ query GetIssue($id: String!) { issue(id: $id) { id identifier title description priority estimate url branchName number dueDate createdAt updatedAt startedAt completedAt canceledAt archivedAt team { id key name } assignee { id name displayName email avatarUrl active admin guest timezone url } creator { id name displayName email avatarUrl active admin guest timezone url } state { id name description type color position team { id } } project { id name description slugId state icon color url startDate targetDate createdAt updatedAt } cycle { id number name description startsAt endsAt completedAt team { id } isCurrent isFuture isPast } parent { id identifier title } labels { nodes { id name description color isGlobal team { id } } } } } """ data = await self.graphql(query, {"id": issue_id}) issue = data.get("issue") if not isinstance(issue, dict): raise LinearNotFoundError(f"Issue {issue_id} not found") return LinearIssue.from_api_response(issue) async def search_issues( self, query_text: str, team_id: str | None = None, limit: int = 50, ) -> list[LinearIssue]: """ Поиск issues по тексту (использует filter). Linear не предоставляет отдельный search API для issues, поэтому используется filter с contains по title/description. """ filter_obj: dict[str, Any] = { "or": [ {"title": {"containsIgnoreCase": query_text}}, {"description": {"containsIgnoreCase": query_text}}, ] } if team_id: filter_obj["team"] = {"id": {"eq": team_id}} query = """ query SearchIssues($first: Int, $filter: IssueFilter) { issues(first: $first, filter: $filter, orderBy: updatedAt) { nodes { id identifier title description priority estimate url branchName number dueDate createdAt updatedAt team { id key name } assignee { id name displayName email avatarUrl active admin guest timezone url } state { id name type } labels { nodes { id name color } } } } } """ data = await self.graphql( query, {"first": min(limit, 250), "filter": filter_obj} ) issues_data = data.get("issues") if not isinstance(issues_data, dict): return [] nodes = issues_data.get("nodes") if not isinstance(nodes, list): return [] return [ LinearIssue.from_api_response(i) for i in nodes if isinstance(i, dict) ] async def create_issue( self, team_id: str, title: str, description: str | None = None, priority: int | None = None, assignee_id: str | None = None, state_id: str | None = None, project_id: str | None = None, cycle_id: str | None = None, label_ids: list[str] | None = None, estimate: float | None = None, due_date: str | None = None, parent_id: str | None = None, ) -> LinearIssue: """Создать новый issue.""" variables: dict[str, Any] = { "input": { "teamId": team_id, "title": title, } } input_obj: dict[str, Any] = variables["input"] if description is not None: input_obj["description"] = description if priority is not None: input_obj["priority"] = priority if assignee_id is not None: input_obj["assigneeId"] = assignee_id if state_id is not None: input_obj["stateId"] = state_id if project_id is not None: input_obj["projectId"] = project_id if cycle_id is not None: input_obj["cycleId"] = cycle_id if label_ids: input_obj["labelIds"] = label_ids if estimate is not None: input_obj["estimate"] = estimate if due_date is not None: input_obj["dueDate"] = due_date if parent_id is not None: input_obj["parentId"] = parent_id mutation = """ mutation CreateIssue($input: IssueCreateInput!) { issueCreate(input: $input) { success issue { id identifier title description priority estimate url branchName number dueDate createdAt updatedAt team { id key name } assignee { id name displayName email avatarUrl } creator { id name displayName email avatarUrl } state { id name type color } project { id name } cycle { id number name } parent { id identifier title } labels { nodes { id name color } } } } } """ data = await self.graphql(mutation, variables) result = data.get("issueCreate") if not isinstance(result, dict): raise LinearError("Invalid issueCreate response") if not result.get("success"): raise LinearError("Failed to create issue") issue_data = result.get("issue") if not isinstance(issue_data, dict): raise LinearError("No issue returned") return LinearIssue.from_api_response(issue_data) async def update_issue( self, issue_id: str, title: str | None = None, description: str | None = None, priority: int | None = None, assignee_id: str | None = None, state_id: str | None = None, project_id: str | None = None, cycle_id: str | None = None, label_ids: list[str] | None = None, estimate: float | None = None, due_date: str | None = None, ) -> LinearIssue: """Обновить issue.""" input_obj: dict[str, Any] = {} if title is not None: input_obj["title"] = title if description is not None: input_obj["description"] = description if priority is not None: input_obj["priority"] = priority if assignee_id is not None: input_obj["assigneeId"] = assignee_id if state_id is not None: input_obj["stateId"] = state_id if project_id is not None: input_obj["projectId"] = project_id if cycle_id is not None: input_obj["cycleId"] = cycle_id if label_ids is not None: input_obj["labelIds"] = label_ids if estimate is not None: input_obj["estimate"] = estimate if due_date is not None: input_obj["dueDate"] = due_date mutation = """ mutation UpdateIssue($id: String!, $input: IssueUpdateInput!) { issueUpdate(id: $id, input: $input) { success issue { id identifier title description priority estimate url branchName number dueDate createdAt updatedAt team { id key name } assignee { id name displayName email avatarUrl } state { id name type color } project { id name } cycle { id number name } labels { nodes { id name color } } } } } """ data = await self.graphql( mutation, {"id": issue_id, "input": input_obj} ) result = data.get("issueUpdate") if not isinstance(result, dict): raise LinearError("Invalid issueUpdate response") if not result.get("success"): raise LinearError("Failed to update issue") issue_data = result.get("issue") if not isinstance(issue_data, dict): raise LinearError("No issue returned") return LinearIssue.from_api_response(issue_data) async def delete_issue(self, issue_id: str) -> bool: """Удалить issue.""" mutation = """ mutation DeleteIssue($id: String!) { issueDelete(id: $id) { success } } """ data = await self.graphql(mutation, {"id": issue_id}) result = data.get("issueDelete") if not isinstance(result, dict): raise LinearError("Invalid issueDelete response") return bool(result.get("success", False)) # ==================================================================== # Projects # ==================================================================== async def list_projects( self, team_id: str | None = None, limit: int = 50, ) -> list[LinearProject]: """Список проектов.""" filter_obj: dict[str, Any] = {} if team_id: filter_obj["accessibleTeams"] = { "and": [{"id": {"eq": team_id}}] } variables: dict[str, Any] = {"first": min(limit, 250)} if filter_obj: variables["filter"] = filter_obj query = """ query ListProjects($first: Int, $filter: ProjectFilter) { projects(first: $first, filter: $filter) { nodes { id name description slugId state icon color url startDate targetDate createdAt updatedAt } } } """ data = await self.graphql(query, variables) projects_data = data.get("projects") if not isinstance(projects_data, dict): return [] nodes = projects_data.get("nodes") if not isinstance(nodes, list): return [] return [ LinearProject.from_api_response(p) for p in nodes if isinstance(p, dict) ] async def get_project(self, project_id: str) -> LinearProject: """Получить проект по ID.""" query = """ query GetProject($id: String!) { project(id: $id) { id name description slugId state icon color url startDate targetDate createdAt updatedAt } } """ data = await self.graphql(query, {"id": project_id}) project = data.get("project") if not isinstance(project, dict): raise LinearNotFoundError(f"Project {project_id} not found") return LinearProject.from_api_response(project) async def create_project( self, name: str, team_ids: list[str], description: str | None = None, state: str = "planned", start_date: str | None = None, target_date: str | None = None, ) -> LinearProject: """Создать новый проект.""" input_obj: dict[str, Any] = { "name": name, "teamIds": team_ids, "state": state, } if description is not None: input_obj["description"] = description if start_date is not None: input_obj["startDate"] = start_date if target_date is not None: input_obj["targetDate"] = target_date mutation = """ mutation CreateProject($input: ProjectCreateInput!) { projectCreate(input: $input) { success project { id name description slugId state icon color url startDate targetDate createdAt updatedAt } } } """ data = await self.graphql(mutation, {"input": input_obj}) result = data.get("projectCreate") if not isinstance(result, dict): raise LinearError("Invalid projectCreate response") if not result.get("success"): raise LinearError("Failed to create project") project_data = result.get("project") if not isinstance(project_data, dict): raise LinearError("No project returned") return LinearProject.from_api_response(project_data) # ==================================================================== # Cycles # ==================================================================== async def list_cycles( self, team_id: str, limit: int = 50, ) -> list[LinearCycle]: """Список cycles (итераций) команды.""" query = """ query ListCycles($teamId: String!, $first: Int) { team(id: $teamId) { cycles(first: $first) { nodes { id number name description startsAt endsAt completedAt team { id } isCurrent isFuture isPast } } } } """ data = await self.graphql( query, {"teamId": team_id, "first": min(limit, 250)} ) team_data = data.get("team") if not isinstance(team_data, dict): return [] cycles_data = team_data.get("cycles") if not isinstance(cycles_data, dict): return [] nodes = cycles_data.get("nodes") if not isinstance(nodes, list): return [] return [ LinearCycle.from_api_response(c) for c in nodes if isinstance(c, dict) ] async def get_cycle(self, cycle_id: str) -> LinearCycle: """Получить cycle по ID.""" query = """ query GetCycle($id: String!) { cycle(id: $id) { id number name description startsAt endsAt completedAt team { id } isCurrent isFuture isPast } } """ data = await self.graphql(query, {"id": cycle_id}) cycle = data.get("cycle") if not isinstance(cycle, dict): raise LinearNotFoundError(f"Cycle {cycle_id} not found") return LinearCycle.from_api_response(cycle) # ==================================================================== # Labels # ==================================================================== async def list_labels( self, team_id: str | None = None, include_global: bool = True, ) -> list[LinearLabel]: """Список labels.""" if team_id: query = """ query ListTeamLabels($teamId: String!) { team(id: $teamId) { labels { nodes { id name description color isGlobal team { id } } } } } """ data = await self.graphql(query, {"teamId": team_id}) team_data = data.get("team") if not isinstance(team_data, dict): return [] labels_data = team_data.get("labels") else: query = """ query { issueLabels { nodes { id name description color isGlobal team { id } } } } """ data = await self.graphql(query) labels_data = data.get("issueLabels") if not isinstance(labels_data, dict): return [] nodes = labels_data.get("nodes") if not isinstance(nodes, list): return [] labels = [ LinearLabel.from_api_response(l) for l in nodes if isinstance(l, dict) ] if not include_global: labels = [l for l in labels if not l.is_global] return labels # ==================================================================== # Workflow States # ==================================================================== async def list_workflow_states( self, team_id: str | None = None ) -> list[LinearWorkflowState]: """Список workflow states.""" if team_id: query = """ query ListTeamStates($teamId: String!) { team(id: $teamId) { states { nodes { id name description type color position team { id } } } } } """ data = await self.graphql(query, {"teamId": team_id}) team_data = data.get("team") if not isinstance(team_data, dict): return [] states_data = team_data.get("states") else: query = """ query { workflowStates { nodes { id name description type color position team { id } } } } """ data = await self.graphql(query) states_data = data.get("workflowStates") if not isinstance(states_data, dict): return [] nodes = states_data.get("nodes") if not isinstance(nodes, list): return [] return [ LinearWorkflowState.from_api_response(s) for s in nodes if isinstance(s, dict) ] # ==================================================================== # Comments # ==================================================================== async def list_comments( self, issue_id: str, limit: int = 50 ) -> list[LinearComment]: """Список комментариев к issue.""" query = """ query ListComments($issueId: String!, $first: Int) { issue(id: $issueId) { comments(first: $first) { nodes { id body createdAt editedAt url user { id name displayName email avatarUrl active admin guest timezone url } issue { id } parent { id } } } } } """ data = await self.graphql( query, {"issueId": issue_id, "first": min(limit, 250)} ) issue_data = data.get("issue") if not isinstance(issue_data, dict): return [] comments_data = issue_data.get("comments") if not isinstance(comments_data, dict): return [] nodes = comments_data.get("nodes") if not isinstance(nodes, list): return [] return [ LinearComment.from_api_response(c) for c in nodes if isinstance(c, dict) ] async def add_comment( self, issue_id: str, body: str, parent_id: str | None = None ) -> LinearComment: """Добавить комментарий к issue.""" input_obj: dict[str, Any] = { "issueId": issue_id, "body": body, } if parent_id is not None: input_obj["parentId"] = parent_id mutation = """ mutation AddComment($input: CommentCreateInput!) { commentCreate(input: $input) { success comment { id body createdAt editedAt url user { id name displayName email avatarUrl } issue { id } parent { id } } } } """ data = await self.graphql(mutation, {"input": input_obj}) result = data.get("commentCreate") if not isinstance(result, dict): raise LinearError("Invalid commentCreate response") if not result.get("success"): raise LinearError("Failed to add comment") comment_data = result.get("comment") if not isinstance(comment_data, dict): raise LinearError("No comment returned") return LinearComment.from_api_response(comment_data) # ============================================================================ # MCP Tools Definition # ============================================================================ LINEAR_TOOLS: list[dict[str, Any]] = [ # ==================================================================== # Users & Organization (3) # ==================================================================== { "name": "linear_get_current_user", "description": "Получить информацию об аутентифицированном пользователе", "parameters": {"type": "object", "properties": {}}, }, { "name": "linear_get_organization", "description": "Получить информацию об организации (workspace)", "parameters": {"type": "object", "properties": {}}, }, { "name": "linear_list_users", "description": "Получить список пользователей организации", "parameters": { "type": "object", "properties": { "include_disabled": { "type": "boolean", "default": False, "description": "Включать ли деактивированных пользователей", }, }, }, }, # ==================================================================== # Teams (2) # ==================================================================== { "name": "linear_list_teams", "description": "Получить список команд (teams)", "parameters": {"type": "object", "properties": {}}, }, { "name": "linear_get_team", "description": "Получить информацию о команде по ID", "parameters": { "type": "object", "properties": { "team_id": {"type": "string", "description": "ID команды"}, }, "required": ["team_id"], }, }, # ==================================================================== # Issues (6) # ==================================================================== { "name": "linear_list_issues", "description": "Получить список issues с фильтрами", "parameters": { "type": "object", "properties": { "team_id": { "type": "string", "description": "Фильтр по team ID", }, "assignee_id": { "type": "string", "description": "Фильтр по assignee ID", }, "project_id": { "type": "string", "description": "Фильтр по project ID", }, "cycle_id": { "type": "string", "description": "Фильтр по cycle ID", }, "state_id": { "type": "string", "description": "Фильтр по state ID", }, "limit": {"type": "integer", "default": 50, "maximum": 250}, }, }, }, { "name": "linear_get_issue", "description": "Получить issue по ID (например, 'ENG-123' или UUID)", "parameters": { "type": "object", "properties": { "issue_id": {"type": "string"}, }, "required": ["issue_id"], }, }, { "name": "linear_search_issues", "description": "Поиск issues по тексту в title и description", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Текст для поиска"}, "team_id": {"type": "string", "description": "Опциональный фильтр по team"}, "limit": {"type": "integer", "default": 50, "maximum": 250}, }, "required": ["query"], }, }, { "name": "linear_create_issue", "description": "Создать новый issue", "parameters": { "type": "object", "properties": { "team_id": {"type": "string", "description": "ID команды (обязательно)"}, "title": {"type": "string", "description": "Заголовок issue"}, "description": {"type": "string", "description": "Описание (Markdown)"}, "priority": { "type": "integer", "description": "Приоритет: 0=No priority, 1=Urgent, 2=High, 3=Medium, 4=Low", "minimum": 0, "maximum": 4, }, "assignee_id": {"type": "string", "description": "ID пользователя-исполнителя"}, "state_id": {"type": "string", "description": "ID статуса (workflow state)"}, "project_id": {"type": "string", "description": "ID проекта"}, "cycle_id": {"type": "string", "description": "ID цикла (спринта)"}, "label_ids": { "type": "array", "items": {"type": "string"}, "description": "Список ID меток", }, "estimate": { "type": "number", "description": "Оценка сложности (если включено в команде)", }, "due_date": { "type": "string", "description": "Дедлайн в формате YYYY-MM-DD", }, "parent_id": { "type": "string", "description": "ID родительского issue (для sub-issues)", }, }, "required": ["team_id", "title"], }, }, { "name": "linear_update_issue", "description": "Обновить существующий issue", "parameters": { "type": "object", "properties": { "issue_id": {"type": "string", "description": "ID issue"}, "title": {"type": "string"}, "description": {"type": "string"}, "priority": { "type": "integer", "minimum": 0, "maximum": 4, }, "assignee_id": {"type": "string"}, "state_id": {"type": "string", "description": "Изменить статус"}, "project_id": {"type": "string"}, "cycle_id": {"type": "string"}, "label_ids": {"type": "array", "items": {"type": "string"}}, "estimate": {"type": "number"}, "due_date": {"type": "string"}, }, "required": ["issue_id"], }, }, { "name": "linear_delete_issue", "description": "Удалить issue", "parameters": { "type": "object", "properties": { "issue_id": {"type": "string"}, }, "required": ["issue_id"], }, }, # ==================================================================== # Projects (3) # ==================================================================== { "name": "linear_list_projects", "description": "Получить список проектов", "parameters": { "type": "object", "properties": { "team_id": {"type": "string", "description": "Фильтр по team ID"}, "limit": {"type": "integer", "default": 50, "maximum": 250}, }, }, }, { "name": "linear_get_project", "description": "Получить проект по ID", "parameters": { "type": "object", "properties": { "project_id": {"type": "string"}, }, "required": ["project_id"], }, }, { "name": "linear_create_project", "description": "Создать новый проект", "parameters": { "type": "object", "properties": { "name": {"type": "string"}, "team_ids": { "type": "array", "items": {"type": "string"}, "description": "Список ID команд, участвующих в проекте", }, "description": {"type": "string"}, "state": { "type": "string", "enum": ["planned", "started", "paused", "completed", "canceled"], "default": "planned", }, "start_date": {"type": "string", "description": "YYYY-MM-DD"}, "target_date": {"type": "string", "description": "YYYY-MM-DD"}, }, "required": ["name", "team_ids"], }, }, # ==================================================================== # Cycles (2) # ==================================================================== { "name": "linear_list_cycles", "description": "Получить список циклов (итераций/спринтов) команды", "parameters": { "type": "object", "properties": { "team_id": {"type": "string"}, "limit": {"type": "integer", "default": 50, "maximum": 250}, }, "required": ["team_id"], }, }, { "name": "linear_get_cycle", "description": "Получить цикл по ID", "parameters": { "type": "object", "properties": { "cycle_id": {"type": "string"}, }, "required": ["cycle_id"], }, }, # ==================================================================== # Labels (1) # ==================================================================== { "name": "linear_list_labels", "description": "Получить список меток (labels)", "parameters": { "type": "object", "properties": { "team_id": { "type": "string", "description": "Фильтр по team (если не указан — все labels)", }, "include_global": { "type": "boolean", "default": True, "description": "Включать ли глобальные labels", }, }, }, }, # ==================================================================== # Workflow States (1) # ==================================================================== { "name": "linear_list_workflow_states", "description": "Получить список статусов (workflow states)", "parameters": { "type": "object", "properties": { "team_id": { "type": "string", "description": "Фильтр по team (если не указан — все states)", }, }, }, }, # ==================================================================== # Comments (2) # ==================================================================== { "name": "linear_list_comments", "description": "Получить комментарии к issue", "parameters": { "type": "object", "properties": { "issue_id": {"type": "string"}, "limit": {"type": "integer", "default": 50, "maximum": 250}, }, "required": ["issue_id"], }, }, { "name": "linear_add_comment", "description": "Добавить комментарий к issue", "parameters": { "type": "object", "properties": { "issue_id": {"type": "string"}, "body": {"type": "string", "description": "Текст комментария (Markdown)"}, "parent_id": { "type": "string", "description": "ID родительского комментария (для ответов)", }, }, "required": ["issue_id", "body"], }, }, ] # ============================================================================ # Tool Handlers # ============================================================================ # Users & Organization async def _handle_get_current_user( client: LinearClient, params: dict[str, Any] ) -> dict[str, Any]: user = await client.get_current_user() return user.to_dict() async def _handle_get_organization( client: LinearClient, params: dict[str, Any] ) -> dict[str, Any]: org = await client.get_organization() return org.to_dict() async def _handle_list_users( client: LinearClient, params: dict[str, Any] ) -> list[dict[str, Any]]: users = await client.list_users( include_disabled=bool(params.get("include_disabled", False)) ) return [u.to_dict() for u in users] # Teams async def _handle_list_teams( client: LinearClient, params: dict[str, Any] ) -> list[dict[str, Any]]: teams = await client.list_teams() return [t.to_dict() for t in teams] async def _handle_get_team( client: LinearClient, params: dict[str, Any] ) -> dict[str, Any]: team = await client.get_team(str(params["team_id"])) return team.to_dict() # Issues async def _handle_list_issues( client: LinearClient, params: dict[str, Any] ) -> list[dict[str, Any]]: issues = await client.list_issues( team_id=params.get("team_id"), assignee_id=params.get("assignee_id"), project_id=params.get("project_id"), cycle_id=params.get("cycle_id"), state_id=params.get("state_id"), limit=int(params.get("limit", 50)), ) return [i.to_dict() for i in issues] async def _handle_get_issue( client: LinearClient, params: dict[str, Any] ) -> dict[str, Any]: issue = await client.get_issue(str(params["issue_id"])) return issue.to_dict() async def _handle_search_issues( client: LinearClient, params: dict[str, Any] ) -> list[dict[str, Any]]: issues = await client.search_issues( query_text=str(params["query"]), team_id=params.get("team_id"), limit=int(params.get("limit", 50)), ) return [i.to_dict() for i in issues] async def _handle_create_issue( client: LinearClient, params: dict[str, Any] ) -> dict[str, Any]: label_ids_raw = params.get("label_ids") label_ids: list[str] | None = None if isinstance(label_ids_raw, list): label_ids = [str(l) for l in label_ids_raw if isinstance(l, str)] estimate_raw = params.get("estimate") estimate: float | None = None if isinstance(estimate_raw, (int, float)): estimate = float(estimate_raw) priority_raw = params.get("priority") priority: int | None = None if isinstance(priority_raw, int): priority = priority_raw issue = await client.create_issue( team_id=str(params["team_id"]), title=str(params["title"]), description=params.get("description"), priority=priority, assignee_id=params.get("assignee_id"), state_id=params.get("state_id"), project_id=params.get("project_id"), cycle_id=params.get("cycle_id"), label_ids=label_ids, estimate=estimate, due_date=params.get("due_date"), parent_id=params.get("parent_id"), ) return issue.to_dict() async def _handle_update_issue( client: LinearClient, params: dict[str, Any] ) -> dict[str, Any]: label_ids_raw = params.get("label_ids") label_ids: list[str] | None = None if isinstance(label_ids_raw, list): label_ids = [str(l) for l in label_ids_raw if isinstance(l, str)] estimate_raw = params.get("estimate") estimate: float | None = None if isinstance(estimate_raw, (int, float)): estimate = float(estimate_raw) priority_raw = params.get("priority") priority: int | None = None if isinstance(priority_raw, int): priority = priority_raw issue = await client.update_issue( issue_id=str(params["issue_id"]), title=params.get("title"), description=params.get("description"), priority=priority, assignee_id=params.get("assignee_id"), state_id=params.get("state_id"), project_id=params.get("project_id"), cycle_id=params.get("cycle_id"), label_ids=label_ids, estimate=estimate, due_date=params.get("due_date"), ) return issue.to_dict() async def _handle_delete_issue( client: LinearClient, params: dict[str, Any] ) -> dict[str, Any]: deleted = await client.delete_issue(str(params["issue_id"])) return {"deleted": deleted, "issue_id": str(params["issue_id"])} # Projects async def _handle_list_projects( client: LinearClient, params: dict[str, Any] ) -> list[dict[str, Any]]: projects = await client.list_projects( team_id=params.get("team_id"), limit=int(params.get("limit", 50)), ) return [p.to_dict() for p in projects] async def _handle_get_project( client: LinearClient, params: dict[str, Any] ) -> dict[str, Any]: project = await client.get_project(str(params["project_id"])) return project.to_dict() async def _handle_create_project( client: LinearClient, params: dict[str, Any] ) -> dict[str, Any]: team_ids_raw = params.get("team_ids") or [] team_ids: list[str] = [ str(t) for t in team_ids_raw if isinstance(t, str) ] project = await client.create_project( name=str(params["name"]), team_ids=team_ids, description=params.get("description"), state=str(params.get("state", "planned")), start_date=params.get("start_date"), target_date=params.get("target_date"), ) return project.to_dict() # Cycles async def _handle_list_cycles( client: LinearClient, params: dict[str, Any] ) -> list[dict[str, Any]]: cycles = await client.list_cycles( team_id=str(params["team_id"]), limit=int(params.get("limit", 50)), ) return [c.to_dict() for c in cycles] async def _handle_get_cycle( client: LinearClient, params: dict[str, Any] ) -> dict[str, Any]: cycle = await client.get_cycle(str(params["cycle_id"])) return cycle.to_dict() # Labels async def _handle_list_labels( client: LinearClient, params: dict[str, Any] ) -> list[dict[str, Any]]: labels = await client.list_labels( team_id=params.get("team_id"), include_global=bool(params.get("include_global", True)), ) return [l.to_dict() for l in labels] # Workflow States async def _handle_list_workflow_states( client: LinearClient, params: dict[str, Any] ) -> list[dict[str, Any]]: states = await client.list_workflow_states( team_id=params.get("team_id") ) return [s.to_dict() for s in states] # Comments async def _handle_list_comments( client: LinearClient, params: dict[str, Any] ) -> list[dict[str, Any]]: comments = await client.list_comments( issue_id=str(params["issue_id"]), limit=int(params.get("limit", 50)), ) return [c.to_dict() for c in comments] async def _handle_add_comment( client: LinearClient, params: dict[str, Any] ) -> dict[str, Any]: comment = await client.add_comment( issue_id=str(params["issue_id"]), body=str(params["body"]), parent_id=params.get("parent_id"), ) return comment.to_dict() # Таблица dispatch _TOOL_HANDLERS: dict[str, Any] = { # Users & Organization "linear_get_current_user": _handle_get_current_user, "linear_get_organization": _handle_get_organization, "linear_list_users": _handle_list_users, # Teams "linear_list_teams": _handle_list_teams, "linear_get_team": _handle_get_team, # Issues "linear_list_issues": _handle_list_issues, "linear_get_issue": _handle_get_issue, "linear_search_issues": _handle_search_issues, "linear_create_issue": _handle_create_issue, "linear_update_issue": _handle_update_issue, "linear_delete_issue": _handle_delete_issue, # Projects "linear_list_projects": _handle_list_projects, "linear_get_project": _handle_get_project, "linear_create_project": _handle_create_project, # Cycles "linear_list_cycles": _handle_list_cycles, "linear_get_cycle": _handle_get_cycle, # Labels "linear_list_labels": _handle_list_labels, # Workflow States "linear_list_workflow_states": _handle_list_workflow_states, # Comments "linear_list_comments": _handle_list_comments, "linear_add_comment": _handle_add_comment, } # ============================================================================ # Linear MCP Server # ============================================================================ class LinearMCPServer: """ MCP Server для Linear. Следует принципу "You Might Not Need an Effect": - Состояние клиента управляется через context manager - Dispatch по таблице вместо runtime reflection - Все преобразования — чистые функции - Явная обработка ошибок с типизированными исключениями - GraphQL-first подход """ def __init__(self, config: LinearConfig | None = None): self.config = config or LinearConfig.from_env() self._client: LinearClient | None = None async def __aenter__(self) -> LinearMCPServer: errors = self.config.validate() if errors: raise LinearError(f"Invalid configuration: {', '.join(errors)}") self._client = LinearClient(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) -> LinearClient: """Получить активный клиент или выбросить ошибку.""" if self._client is None: raise LinearError( "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(LINEAR_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 LinearError 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 LinearNotFoundError as e: return { "success": False, "error": f"Not found: {e}", "error_type": "not_found", } except LinearAuthError as e: return { "success": False, "error": f"Authentication failed: {e}", "error_type": "auth_error", } except LinearForbiddenError as e: return { "success": False, "error": f"Forbidden: {e}", "error_type": "forbidden", } except LinearRateLimitError as e: return { "success": False, "error": f"Rate limit exceeded: {e}", "error_type": "rate_limit", "retry_after": e.retry_after, } except LinearValidationError as e: return { "success": False, "error": f"Validation error: {e}", "error_type": "validation_error", "errors": e.errors, } except LinearGraphQLError as e: return { "success": False, "error": f"GraphQL error: {e}", "error_type": "graphql_error", "extensions": e.extensions, } except LinearError as e: return { "success": False, "error": str(e), "error_type": "linear_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_linear_mcp( api_key: str | None = None, oauth_token: str | None = None, base_url: str = "https://api.linear.app", timeout_seconds: float = 30.0, ) -> AsyncIterator[LinearMCPServer]: """ Создать Linear MCP Server с автоматическим управлением подключением. Usage: async with create_linear_mcp(api_key="lin_api_...") as mcp: tools = mcp.get_tools() result = await mcp.call_tool( "linear_list_issues", {"team_id": "team_uuid", "limit": 20}, ) """ resolved_api_key = ( api_key or os.getenv("LINEAR_API_KEY") or os.getenv("LINEAR_PERSONAL_API_KEY") ) resolved_oauth = oauth_token or os.getenv("LINEAR_OAUTH_TOKEN") config = LinearConfig( base_url=base_url, api_key=resolved_api_key, oauth_token=resolved_oauth, timeout_seconds=timeout_seconds, ) async with LinearMCPServer(config) as server: yield server # ============================================================================ # Exports # ============================================================================ __all__ = [ # Config "LinearConfig", # Exceptions "LinearError", "LinearAuthError", "LinearNotFoundError", "LinearRateLimitError", "LinearForbiddenError", "LinearValidationError", "LinearGraphQLError", # Models "LinearUser", "LinearOrganization", "LinearTeam", "LinearWorkflowState", "LinearLabel", "LinearProject", "LinearCycle", "LinearIssue", "LinearComment", # Client & Server "LinearClient", "LinearMCPServer", "LINEAR_TOOLS", # Helpers "create_linear_mcp", ]