/
alexefan136
/
flowstack
Обзор
Документация
Войти
/
alexefan136
/
flowstack
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
main
core/engine/src/tools/mcp/confluence.py
1 657 строк
56 KB
Alexander Efanov
Обновление репозитория
15 июл 2026, 12:19
15 июл 2026, 12:19
76704c6
Код
Авторство
О чём код?
""" Confluence MCP Server — интеграция с Atlassian Confluence через MCP протокол. Предоставляет tools для работы с Confluence: - Управление страницами (CRUD) - Поиск по контенту (CQL) - Работа с пространствами - Комментарии, метки, вложения - Иерархия страниц Использует Confluence REST API v2: https://developer.atlassian.com/cloud/confluence/rest/v2/ Аутентификация: - Basic Auth (email + API token) - OAuth 2.0 (Bearer token) Архитектурные принципы: - Чистые функции для преобразования данных (dataclass -> dict) - Явная обработка ошибок без лишних сайд-эффектов - Retry логика с exponential backoff - Rate limiting для защиты от 429 ошибок """ from __future__ import annotations import asyncio import base64 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 ConfluenceConfig: """ Конфигурация подключения к Confluence. Поддерживает два типа аутентификации: 1. Basic Auth: email + api_token 2. Bearer token (OAuth 2.0) Чистая структура данных — не содержит side effects. """ # Базовый URL (например, https://your-domain.atlassian.net/wiki) base_url: str = "" # Аутентификация auth_type: str = "basic" # basic | bearer email: str | None = None api_token: str | None = None bearer_token: str | None = None # Настройки запросов timeout_seconds: int = 30 max_retries: int = 3 retry_delay_seconds: float = 1.0 # Rate limiting requests_per_second: float = 5.0 def get_auth_headers(self) -> dict[str, str]: """ Получить headers для аутентификации. Чистая функция — не мутирует состояние, возвращает новый dict. """ headers = { "Content-Type": "application/json", "Accept": "application/json", } if self.auth_type == "basic" and self.email and self.api_token: credentials = f"{self.email}:{self.api_token}" encoded = base64.b64encode(credentials.encode()).decode() headers["Authorization"] = f"Basic {encoded}" elif self.auth_type == "bearer" and self.bearer_token: headers["Authorization"] = f"Bearer {self.bearer_token}" return headers def validate(self) -> list[str]: """ Валидировать конфигурацию. Чистая функция — возвращает список ошибок, не мутирует состояние. """ errors: list[str] = [] if not self.base_url: errors.append("base_url is required") if self.auth_type == "basic": if not self.email: errors.append("email is required for basic auth") if not self.api_token: errors.append("api_token is required for basic auth") elif self.auth_type == "bearer": if not self.bearer_token: errors.append("bearer_token is required for bearer auth") return errors @classmethod def from_env(cls) -> ConfluenceConfig: """Создать конфигурацию из переменных окружения.""" return cls( base_url=os.getenv("CONFLUENCE_BASE_URL", ""), auth_type=os.getenv("CONFLUENCE_AUTH_TYPE", "basic"), email=os.getenv("CONFLUENCE_EMAIL"), api_token=os.getenv("CONFLUENCE_API_TOKEN"), bearer_token=os.getenv("CONFLUENCE_BEARER_TOKEN"), timeout_seconds=int(os.getenv("CONFLUENCE_TIMEOUT", "30")), max_retries=int(os.getenv("CONFLUENCE_MAX_RETRIES", "3")), requests_per_second=float(os.getenv("CONFLUENCE_RPS", "5.0")), ) # ============================================================================ # Exceptions # ============================================================================ class ConfluenceError(Exception): """Базовое исключение для Confluence.""" pass class ConfluenceAuthError(ConfluenceError): """Ошибка аутентификации.""" pass class ConfluenceNotFoundError(ConfluenceError): """Ресурс не найден (404).""" pass class ConfluenceRateLimitError(ConfluenceError): """Превышен rate limit.""" pass # ============================================================================ # Response Models # ============================================================================ @dataclass class ConfluencePage: """ Страница Confluence. Иммутабельная структура данных — используется только для чтения. """ id: str title: str space_id: str space_key: str status: str = "current" body_storage: str = "" body_wiki: str = "" version: int = 1 author_id: str | None = None author_name: str | None = None created_at: datetime | None = None updated_at: datetime | None = None parent_id: str | None = None children_count: int = 0 labels: list[str] = field(default_factory=list) # Metadata links: dict[str, str] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: """Преобразовать в словарь. Чистая функция.""" return { "id": self.id, "title": self.title, "space_id": self.space_id, "space_key": self.space_key, "status": self.status, "body_storage": self.body_storage, "body_wiki": self.body_wiki, "version": self.version, "author_id": self.author_id, "author_name": self.author_name, "created_at": self.created_at.isoformat() if self.created_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None, "parent_id": self.parent_id, "children_count": self.children_count, "labels": list(self.labels), "links": dict(self.links), } @classmethod def from_api_response(cls, data: dict[str, Any]) -> ConfluencePage: """Создать из ответа API. Чистая функция преобразования.""" space_id = str(data.get("spaceId", "")) space_key = str(data.get("spaceKey", "")) # Извлечь body body = data.get("body") or {} body_storage = (body.get("storage") or {}).get("value", "") body_wiki = (body.get("wiki") or {}).get("value", "") # Извлечь version version_info = data.get("version") or {} version = int(version_info.get("number", 1)) # Извлечь автора author_id = version_info.get("authorId") if author_id is not None: author_id = str(author_id) # Извлечь даты created_at_str = data.get("createdAt") updated_at_str = version_info.get("createdAt") # Parent ID parent_raw = data.get("parentId") parent_id = str(parent_raw) if parent_raw is not None else None # Children count children_info = data.get("children") or {} page_children = children_info.get("page") or {} children_count = int(page_children.get("count", 0)) # Labels labels_data = data.get("labels") or {} labels_results = labels_data.get("results") or [] labels = [ str(label.get("name", "")) for label in labels_results if isinstance(label, dict) ] return cls( id=str(data.get("id", "")), title=str(data.get("title", "")), space_id=space_id, space_key=space_key, status=str(data.get("status", "current")), body_storage=body_storage, body_wiki=body_wiki, version=version, author_id=author_id, author_name=data.get("authorDisplayName"), created_at=_parse_datetime(created_at_str), updated_at=_parse_datetime(updated_at_str), parent_id=parent_id, children_count=children_count, labels=labels, links=data.get("_links") or {}, ) @dataclass class ConfluenceSpace: """Пространство Confluence.""" id: str key: str name: str type: str = "global" # global | personal description: str = "" status: str = "current" author_id: str | None = None created_at: datetime | None = None updated_at: datetime | None = None def to_dict(self) -> dict[str, Any]: """Преобразовать в словарь. Чистая функция.""" return { "id": self.id, "key": self.key, "name": self.name, "type": self.type, "description": self.description, "status": self.status, "author_id": self.author_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, } @classmethod def from_api_response(cls, data: dict[str, Any]) -> ConfluenceSpace: """Создать из ответа API. Чистая функция преобразования.""" author_id = data.get("authorId") if author_id is not None: author_id = str(author_id) return cls( id=str(data.get("id", "")), key=str(data.get("key", "")), name=str(data.get("name", "")), type=str(data.get("type", "global")), description=str(data.get("description", "")), status=str(data.get("status", "current")), author_id=author_id, created_at=_parse_datetime(data.get("createdAt")), updated_at=_parse_datetime(data.get("updatedAt")), ) @dataclass class ConfluenceSearchResult: """Результат поиска в Confluence.""" id: str title: str excerpt: str url: str content_type: str # page | blogpost | comment | attachment space_key: str = "" last_modified: datetime | None = None score: float = 0.0 def to_dict(self) -> dict[str, Any]: """Преобразовать в словарь. Чистая функция.""" return { "id": self.id, "title": self.title, "excerpt": self.excerpt, "url": self.url, "content_type": self.content_type, "space_key": self.space_key, "last_modified": ( self.last_modified.isoformat() if self.last_modified else None ), "score": self.score, } @dataclass class ConfluenceComment: """Комментарий в Confluence.""" id: str body: str author_id: str | None = None author_name: str | None = None created_at: datetime | None = None updated_at: datetime | None = None def to_dict(self) -> dict[str, Any]: """Преобразовать в словарь. Чистая функция.""" return { "id": self.id, "body": self.body, "author_id": self.author_id, "author_name": self.author_name, "created_at": self.created_at.isoformat() if self.created_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None, } @dataclass class ConfluenceAttachment: """Вложение в Confluence.""" id: str title: str file_size: int = 0 media_type: str = "" comment: str = "" version: int = 1 created_at: datetime | None = None def to_dict(self) -> dict[str, Any]: """Преобразовать в словарь. Чистая функция.""" return { "id": self.id, "title": self.title, "file_size": self.file_size, "media_type": self.media_type, "comment": self.comment, "version": self.version, "created_at": self.created_at.isoformat() if self.created_at else None, } # ============================================================================ # Helper Functions # ============================================================================ def _parse_datetime(value: Any) -> datetime | None: """ Безопасно парсить datetime из строки. Чистая функция — не мутирует вход, возвращает datetime или None. Обрабатывает ISO 8601 с 'Z' суффиксом. """ if not isinstance(value, str): return None try: normalized = value.replace("Z", "+00:00") return datetime.fromisoformat(normalized) except (ValueError, TypeError): return None # ============================================================================ # Confluence Client # ============================================================================ class ConfluenceClient: """ Асинхронный клиент для Confluence REST API v2. Автоматически обрабатывает: - Аутентификацию - Rate limiting - Retry логику - Ошибки API Следует принципу "You Might Not Need an Effect": - Чистое разделение между HTTP логикой и обработкой данных - Side effects изолированы в методах _request - Преобразования данных — чистые функции (from_api_response, to_dict) """ def __init__(self, config: ConfluenceConfig): 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) -> ConfluenceClient: """Async context manager entry.""" await self.connect() return self async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: """Async context manager exit.""" await self.close() async def connect(self) -> None: """Подключиться к Confluence.""" if self._session is None: import aiohttp # type: ignore[import-not-found,import-untyped] 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 _get_session(self) -> aiohttp.ClientSession: """ Получить активную сессию или выбросить ошибку. Разделяет проверку None от использования, чтобы типизатор корректно определял тип. """ if self._session is None: raise ConfluenceError( "Client is not connected. Use 'async with' or call connect() first." ) return self._session async def _request( self, method: str, endpoint: str, params: dict[str, Any] | None = None, json_data: Any = None, data: Any = None, headers: dict[str, str] | None = None, ) -> dict[str, Any]: """ Выполнить HTTP запрос с retry логикой. Args: method: HTTP метод (GET, POST, PUT, DELETE) endpoint: API endpoint (например, /api/v2/pages) params: Query параметры json_data: JSON body (может быть dict или list) data: Raw body (для multipart) headers: Дополнительные headers """ session = self._get_session() url = f"{self.config.base_url.rstrip('/')}{endpoint}" # Merge headers request_headers = self.config.get_auth_headers() if headers: request_headers.update(headers) # Retry loop last_error: Exception | None = None loop = asyncio.get_running_loop() for attempt in range(self.config.max_retries): # Rate limiting async with self._rate_limiter: # Подождать если нужно для соблюдения RPS 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, data=data, headers=request_headers, ) as response: self._last_request_time = loop.time() # Обработка ошибок if response.status == 401: raise ConfluenceAuthError("Authentication failed") elif response.status == 404: raise ConfluenceNotFoundError( f"Resource not found: {endpoint}" ) elif response.status == 429: raise ConfluenceRateLimitError("Rate limit exceeded") elif response.status >= 400: error_text = await response.text() raise ConfluenceError( f"API error {response.status}: {error_text}" ) # Parse response content_type = response.content_type or "" if "application/json" in content_type: return await response.json() else: return {"content": await response.text()} except (ConfluenceAuthError, ConfluenceNotFoundError) as e: # Не повторяем client ошибки raise e except (ConfluenceRateLimitError, ConfluenceError) 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: # Exponential backoff delay = self.config.retry_delay_seconds * (2 ** attempt) await asyncio.sleep(delay) 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: delay = self.config.retry_delay_seconds * (2 ** attempt) await asyncio.sleep(delay) # Все попытки исчерпаны raise ConfluenceError( f"Request failed after {self.config.max_retries} attempts: {last_error}" ) # ======================================================================== # Spaces # ======================================================================== async def list_spaces( self, limit: int = 25, cursor: str | None = None, ) -> dict[str, Any]: """Получить список пространств.""" params: dict[str, Any] = {"limit": limit} if cursor: params["cursor"] = cursor response = await self._request("GET", "/api/v2/spaces", params=params) spaces = [ ConfluenceSpace.from_api_response(s) for s in (response.get("results") or []) ] links = response.get("_links") or {} return { "spaces": [s.to_dict() for s in spaces], "next_cursor": links.get("next"), } async def get_space(self, space_id: str) -> ConfluenceSpace: """Получить пространство по ID.""" response = await self._request("GET", f"/api/v2/spaces/{space_id}") return ConfluenceSpace.from_api_response(response) async def get_space_by_key(self, space_key: str) -> ConfluenceSpace: """Получить пространство по key.""" cursor: str | None = None while True: result = await self.list_spaces(limit=100, cursor=cursor) for space_data in result["spaces"]: if space_data["key"] == space_key: return ConfluenceSpace( id=space_data["id"], key=space_data["key"], name=space_data["name"], type=space_data.get("type", "global"), description=space_data.get("description", ""), status=space_data.get("status", "current"), author_id=space_data.get("author_id"), created_at=( datetime.fromisoformat(space_data["created_at"]) if space_data.get("created_at") else None ), updated_at=( datetime.fromisoformat(space_data["updated_at"]) if space_data.get("updated_at") else None ), ) cursor = result.get("next_cursor") if not cursor: break raise ConfluenceNotFoundError(f"Space with key '{space_key}' not found") # ======================================================================== # Pages # ======================================================================== async def list_pages( self, space_id: str | None = None, parent_id: str | None = None, status: str = "current", limit: int = 25, cursor: str | None = None, ) -> dict[str, Any]: """Получить список страниц.""" params: dict[str, Any] = { "status": status, "limit": limit, } if space_id: params["space-id"] = space_id if parent_id: params["parent-id"] = parent_id if cursor: params["cursor"] = cursor response = await self._request("GET", "/api/v2/pages", params=params) pages = [ ConfluencePage.from_api_response(p) for p in (response.get("results") or []) ] links = response.get("_links") or {} return { "pages": [p.to_dict() for p in pages], "next_cursor": links.get("next"), } async def get_page( self, page_id: str, include_body: bool = True, body_format: str = "storage", ) -> ConfluencePage: """ Получить страницу по ID. Args: page_id: ID страницы include_body: Включать ли содержимое страницы body_format: Формат body (storage | wiki | atlas_doc_format) """ params: dict[str, Any] = {} if include_body: params["body-format"] = body_format response = await self._request( "GET", f"/api/v2/pages/{page_id}", params=params ) return ConfluencePage.from_api_response(response) async def get_page_by_title( self, space_id: str, title: str, include_body: bool = True, ) -> ConfluencePage: """Получить страницу по заголовку в пространстве.""" params: dict[str, Any] = { "space-id": space_id, "title": title, "status": "current", } if include_body: params["body-format"] = "storage" response = await self._request("GET", "/api/v2/pages", params=params) results = response.get("results") or [] if not results: raise ConfluenceNotFoundError( f"Page with title '{title}' not found in space {space_id}" ) return ConfluencePage.from_api_response(results[0]) async def create_page( self, space_id: str, title: str, body: str, body_format: str = "storage", parent_id: str | None = None, status: str = "current", ) -> ConfluencePage: """ Создать новую страницу. Args: space_id: ID пространства title: Заголовок страницы body: Содержимое страницы (HTML в storage format или wiki markup) body_format: Формат body (storage | wiki) parent_id: ID родительской страницы (опционально) status: Статус страницы (current | draft) """ payload: dict[str, Any] = { "spaceId": space_id, "status": status, "title": title, "body": { "representation": body_format, "value": body, }, } if parent_id: payload["parentId"] = parent_id response = await self._request( "POST", "/api/v2/pages", json_data=payload ) return ConfluencePage.from_api_response(response) async def update_page( self, page_id: str, title: str | None = None, body: str | None = None, body_format: str = "storage", version_message: str = "Updated via MCP", ) -> ConfluencePage: """ Обновить страницу. Автоматически инкрементирует version. """ # Получить текущую версию current_page = await self.get_page(page_id, include_body=False) new_version = current_page.version + 1 payload: dict[str, Any] = { "id": page_id, "status": "current", "title": title if title is not None else current_page.title, "version": { "number": new_version, "message": version_message, }, } if body is not None: payload["body"] = { "representation": body_format, "value": body, } response = await self._request( "PUT", f"/api/v2/pages/{page_id}", json_data=payload ) return ConfluencePage.from_api_response(response) async def delete_page(self, page_id: str) -> bool: """Удалить страницу.""" await self._request("DELETE", f"/api/v2/pages/{page_id}") return True async def get_page_children( self, page_id: str, limit: int = 25, cursor: str | None = None, ) -> dict[str, Any]: """Получить дочерние страницы.""" params: dict[str, Any] = {"limit": limit} if cursor: params["cursor"] = cursor response = await self._request( "GET", f"/api/v2/pages/{page_id}/children", params=params, ) pages = [ ConfluencePage.from_api_response(p) for p in (response.get("results") or []) ] links = response.get("_links") or {} return { "pages": [p.to_dict() for p in pages], "next_cursor": links.get("next"), } # ======================================================================== # Search (CQL) # ======================================================================== async def search( self, cql: str, limit: int = 25, cursor: str | None = None, ) -> dict[str, Any]: """ Поиск по Confluence используя CQL (Confluence Query Language). Примеры CQL: - `type = page AND space = "DEV"` - `title ~ "documentation"` - `lastModified > startOfDay(-7)` - `label = "important" AND type = page` Документация: https://developer.atlassian.com/cloud/confluence/advanced-searching-using-cql/ """ params: dict[str, Any] = { "cql": cql, "limit": limit, } if cursor: params["cursor"] = cursor response = await self._request("GET", "/api/v2/search", params=params) results: list[dict[str, Any]] = [] for item in response.get("results") or []: if not isinstance(item, dict): continue results.append({ "id": str(item.get("id", "")), "title": str(item.get("title", "")), "excerpt": str(item.get("excerpt", "")), "url": str(item.get("url", "")), "content_type": str(item.get("type", "page")), "space_key": str(item.get("spaceKey", "")), "last_modified": item.get("lastModified"), }) return { "results": results, "total": int(response.get("totalSize", len(results))), "next_cursor": (response.get("_links") or {}).get("next"), } # ======================================================================== # Comments # ======================================================================== async def get_page_comments( self, page_id: str, limit: int = 25, ) -> list[ConfluenceComment]: """Получить комментарии страницы.""" params: dict[str, Any] = {"limit": limit} response = await self._request( "GET", f"/api/v2/pages/{page_id}/inline-comments", params=params, ) comments: list[ConfluenceComment] = [] for item in response.get("results") or []: if not isinstance(item, dict): continue body_data = item.get("body") or {} storage = body_data.get("storage") or {} body = str(storage.get("value", "")) version_info = item.get("version") or {} comments.append(ConfluenceComment( id=str(item.get("id", "")), body=body, author_id=version_info.get("authorId"), created_at=_parse_datetime(version_info.get("createdAt")), )) return comments async def add_comment( self, page_id: str, body: str, body_format: str = "storage", ) -> ConfluenceComment: """Добавить комментарий к странице.""" payload: dict[str, Any] = { "body": { "representation": body_format, "value": body, }, } response = await self._request( "POST", f"/api/v2/pages/{page_id}/inline-comments", json_data=payload, ) version_info = response.get("version") or {} if isinstance(response, dict) else {} return ConfluenceComment( id=str(response.get("id", "")) if isinstance(response, dict) else "", body=body, author_id=version_info.get("authorId"), created_at=_parse_datetime(version_info.get("createdAt")) or datetime.utcnow(), ) # ======================================================================== # Labels # ======================================================================== async def get_page_labels(self, page_id: str) -> list[str]: """Получить метки страницы.""" response = await self._request( "GET", f"/api/v2/pages/{page_id}/labels" ) return [ str(label.get("name", "")) for label in (response.get("results") or []) if isinstance(label, dict) ] async def add_label(self, page_id: str, label: str) -> bool: """Добавить метку к странице.""" payload: list[dict[str, str]] = [{"name": label, "prefix": "global"}] await self._request( "POST", f"/api/v2/pages/{page_id}/labels", json_data=payload, ) return True async def remove_label(self, page_id: str, label: str) -> bool: """Удалить метку со страницы.""" await self._request( "DELETE", f"/api/v2/pages/{page_id}/labels/{label}" ) return True # ======================================================================== # Attachments # ======================================================================== async def get_attachments( self, page_id: str, limit: int = 25, ) -> list[ConfluenceAttachment]: """Получить вложения страницы.""" params: dict[str, Any] = {"limit": limit} response = await self._request( "GET", f"/api/v2/pages/{page_id}/attachments", params=params, ) attachments: list[ConfluenceAttachment] = [] for item in response.get("results") or []: if not isinstance(item, dict): continue version_info = item.get("version") or {} attachments.append(ConfluenceAttachment( id=str(item.get("id", "")), title=str(item.get("title", "")), file_size=int(item.get("fileSize", 0)), media_type=str(item.get("mediaType", "")), comment=str(item.get("comment", "")), version=int(version_info.get("number", 1)), created_at=_parse_datetime(version_info.get("createdAt")), )) return attachments # ============================================================================ # MCP Tools Definition # ============================================================================ CONFLUENCE_TOOLS: list[dict[str, Any]] = [ { "name": "confluence_list_spaces", "description": "Получить список пространств (spaces) в Confluence", "parameters": { "type": "object", "properties": { "limit": { "type": "integer", "description": "Максимальное количество пространств (по умолчанию 25)", "default": 25, }, }, }, }, { "name": "confluence_get_space", "description": "Получить информацию о пространстве по ID или key", "parameters": { "type": "object", "properties": { "space_id": {"type": "string", "description": "ID пространства"}, "space_key": { "type": "string", "description": "Key пространства (альтернатива ID)", }, }, }, }, { "name": "confluence_list_pages", "description": "Получить список страниц с фильтрами", "parameters": { "type": "object", "properties": { "space_id": {"type": "string", "description": "ID пространства"}, "parent_id": { "type": "string", "description": "ID родительской страницы", }, "status": { "type": "string", "enum": ["current", "draft", "archived"], "default": "current", }, "limit": {"type": "integer", "default": 25}, }, "required": ["space_id"], }, }, { "name": "confluence_get_page", "description": "Получить страницу по ID с содержимым", "parameters": { "type": "object", "properties": { "page_id": {"type": "string", "description": "ID страницы"}, "include_body": {"type": "boolean", "default": True}, "body_format": { "type": "string", "enum": ["storage", "wiki", "atlas_doc_format"], "default": "storage", }, }, "required": ["page_id"], }, }, { "name": "confluence_get_page_by_title", "description": "Получить страницу по заголовку в пространстве", "parameters": { "type": "object", "properties": { "space_id": {"type": "string", "description": "ID пространства"}, "title": {"type": "string", "description": "Заголовок страницы"}, }, "required": ["space_id", "title"], }, }, { "name": "confluence_create_page", "description": "Создать новую страницу в Confluence", "parameters": { "type": "object", "properties": { "space_id": {"type": "string", "description": "ID пространства"}, "title": {"type": "string", "description": "Заголовок страницы"}, "body": { "type": "string", "description": "Содержимое (HTML или wiki markup)", }, "body_format": { "type": "string", "enum": ["storage", "wiki"], "default": "storage", }, "parent_id": { "type": "string", "description": "ID родительской страницы", }, "status": { "type": "string", "enum": ["current", "draft"], "default": "current", }, }, "required": ["space_id", "title", "body"], }, }, { "name": "confluence_update_page", "description": "Обновить существующую страницу", "parameters": { "type": "object", "properties": { "page_id": {"type": "string", "description": "ID страницы"}, "title": {"type": "string", "description": "Новый заголовок"}, "body": {"type": "string", "description": "Новое содержимое"}, "body_format": { "type": "string", "enum": ["storage", "wiki"], "default": "storage", }, "version_message": { "type": "string", "default": "Updated via MCP", }, }, "required": ["page_id"], }, }, { "name": "confluence_delete_page", "description": "Удалить страницу", "parameters": { "type": "object", "properties": { "page_id": {"type": "string", "description": "ID страницы"}, }, "required": ["page_id"], }, }, { "name": "confluence_search", "description": "Поиск по Confluence используя CQL (Confluence Query Language)", "parameters": { "type": "object", "properties": { "cql": { "type": "string", "description": ( "CQL запрос. Примеры: 'type = page AND space = \"DEV\"', " "'title ~ \"documentation\"'" ), }, "limit": {"type": "integer", "default": 25}, }, "required": ["cql"], }, }, { "name": "confluence_get_page_children", "description": "Получить дочерние страницы", "parameters": { "type": "object", "properties": { "page_id": { "type": "string", "description": "ID родительской страницы", }, "limit": {"type": "integer", "default": 25}, }, "required": ["page_id"], }, }, { "name": "confluence_add_comment", "description": "Добавить комментарий к странице", "parameters": { "type": "object", "properties": { "page_id": {"type": "string", "description": "ID страницы"}, "body": {"type": "string", "description": "Текст комментария (HTML)"}, }, "required": ["page_id", "body"], }, }, { "name": "confluence_get_page_labels", "description": "Получить метки (labels) страницы", "parameters": { "type": "object", "properties": { "page_id": {"type": "string", "description": "ID страницы"}, }, "required": ["page_id"], }, }, { "name": "confluence_add_label", "description": "Добавить метку к странице", "parameters": { "type": "object", "properties": { "page_id": {"type": "string", "description": "ID страницы"}, "label": {"type": "string", "description": "Название метки"}, }, "required": ["page_id", "label"], }, }, { "name": "confluence_get_attachments", "description": "Получить вложения страницы", "parameters": { "type": "object", "properties": { "page_id": {"type": "string", "description": "ID страницы"}, "limit": {"type": "integer", "default": 25}, }, "required": ["page_id"], }, }, ] # ============================================================================ # Tool Handlers (чистые функции-делегаты) # ============================================================================ async def _handle_confluence_list_spaces( client: ConfluenceClient, params: dict[str, Any], ) -> dict[str, Any]: """Обработчик confluence_list_spaces.""" limit = int(params.get("limit", 25)) return await client.list_spaces(limit=limit) async def _handle_confluence_get_space( client: ConfluenceClient, params: dict[str, Any], ) -> dict[str, Any]: """Обработчик confluence_get_space.""" space_id = params.get("space_id") space_key = params.get("space_key") if space_id: space = await client.get_space(str(space_id)) elif space_key: space = await client.get_space_by_key(str(space_key)) else: raise ConfluenceError("Either space_id or space_key is required") return space.to_dict() async def _handle_confluence_list_pages( client: ConfluenceClient, params: dict[str, Any], ) -> dict[str, Any]: """Обработчик confluence_list_pages.""" return await client.list_pages( space_id=params.get("space_id"), parent_id=params.get("parent_id"), status=str(params.get("status", "current")), limit=int(params.get("limit", 25)), ) async def _handle_confluence_get_page( client: ConfluenceClient, params: dict[str, Any], ) -> dict[str, Any]: """Обработчик confluence_get_page.""" page = await client.get_page( page_id=str(params["page_id"]), include_body=bool(params.get("include_body", True)), body_format=str(params.get("body_format", "storage")), ) return page.to_dict() async def _handle_confluence_get_page_by_title( client: ConfluenceClient, params: dict[str, Any], ) -> dict[str, Any]: """Обработчик confluence_get_page_by_title.""" page = await client.get_page_by_title( space_id=str(params["space_id"]), title=str(params["title"]), ) return page.to_dict() async def _handle_confluence_create_page( client: ConfluenceClient, params: dict[str, Any], ) -> dict[str, Any]: """Обработчик confluence_create_page.""" page = await client.create_page( space_id=str(params["space_id"]), title=str(params["title"]), body=str(params["body"]), body_format=str(params.get("body_format", "storage")), parent_id=params.get("parent_id"), status=str(params.get("status", "current")), ) return page.to_dict() async def _handle_confluence_update_page( client: ConfluenceClient, params: dict[str, Any], ) -> dict[str, Any]: """Обработчик confluence_update_page.""" page = await client.update_page( page_id=str(params["page_id"]), title=params.get("title"), body=params.get("body"), body_format=str(params.get("body_format", "storage")), version_message=str(params.get("version_message", "Updated via MCP")), ) return page.to_dict() async def _handle_confluence_delete_page( client: ConfluenceClient, params: dict[str, Any], ) -> dict[str, Any]: """Обработчик confluence_delete_page.""" page_id = str(params["page_id"]) success = await client.delete_page(page_id) return {"deleted": success, "page_id": page_id} async def _handle_confluence_search( client: ConfluenceClient, params: dict[str, Any], ) -> dict[str, Any]: """Обработчик confluence_search.""" return await client.search( cql=str(params["cql"]), limit=int(params.get("limit", 25)), ) async def _handle_confluence_get_page_children( client: ConfluenceClient, params: dict[str, Any], ) -> dict[str, Any]: """Обработчик confluence_get_page_children.""" return await client.get_page_children( page_id=str(params["page_id"]), limit=int(params.get("limit", 25)), ) async def _handle_confluence_add_comment( client: ConfluenceClient, params: dict[str, Any], ) -> dict[str, Any]: """Обработчик confluence_add_comment.""" comment = await client.add_comment( page_id=str(params["page_id"]), body=str(params["body"]), ) return comment.to_dict() async def _handle_confluence_get_page_labels( client: ConfluenceClient, params: dict[str, Any], ) -> dict[str, Any]: """Обработчик confluence_get_page_labels.""" page_id = str(params["page_id"]) labels = await client.get_page_labels(page_id) return {"labels": labels, "page_id": page_id} async def _handle_confluence_add_label( client: ConfluenceClient, params: dict[str, Any], ) -> dict[str, Any]: """Обработчик confluence_add_label.""" page_id = str(params["page_id"]) label = str(params["label"]) success = await client.add_label(page_id=page_id, label=label) return {"success": success, "page_id": page_id, "label": label} async def _handle_confluence_get_attachments( client: ConfluenceClient, params: dict[str, Any], ) -> dict[str, Any]: """Обработчик confluence_get_attachments.""" page_id = str(params["page_id"]) attachments = await client.get_attachments( page_id=page_id, limit=int(params.get("limit", 25)), ) return { "attachments": [a.to_dict() for a in attachments], "page_id": page_id, } # Таблица соответствия tool_name -> handler _TOOL_HANDLERS: dict[str, Any] = { "confluence_list_spaces": _handle_confluence_list_spaces, "confluence_get_space": _handle_confluence_get_space, "confluence_list_pages": _handle_confluence_list_pages, "confluence_get_page": _handle_confluence_get_page, "confluence_get_page_by_title": _handle_confluence_get_page_by_title, "confluence_create_page": _handle_confluence_create_page, "confluence_update_page": _handle_confluence_update_page, "confluence_delete_page": _handle_confluence_delete_page, "confluence_search": _handle_confluence_search, "confluence_get_page_children": _handle_confluence_get_page_children, "confluence_add_comment": _handle_confluence_add_comment, "confluence_get_page_labels": _handle_confluence_get_page_labels, "confluence_add_label": _handle_confluence_add_label, "confluence_get_attachments": _handle_confluence_get_attachments, } # ============================================================================ # MCP Server Implementation # ============================================================================ class ConfluenceMCPServer: """ MCP Server для Confluence. Реализует MCP протокол для работы с Confluence через tools. Следует принципу "You Might Not Need an Effect": - Состояние клиента явно управляется через context manager - Обработчики делегируются в чистые функции - Dispatch по таблице вместо getattr для type safety """ def __init__(self, config: ConfluenceConfig | None = None): self.config = config or ConfluenceConfig.from_env() self._client: ConfluenceClient | None = None async def __aenter__(self) -> ConfluenceMCPServer: """Async context manager entry.""" errors = self.config.validate() if errors: raise ConfluenceError(f"Invalid configuration: {', '.join(errors)}") self._client = ConfluenceClient(self.config) await self._client.connect() return self async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: """Async context manager exit.""" if self._client is not None: await self._client.close() self._client = None def _require_client(self) -> ConfluenceClient: """ Получить активный клиент или выбросить ошибку. Изолирует проверку None для корректной работы типизатора. """ if self._client is None: raise ConfluenceError( "Client not connected. Use 'async with' or call __aenter__()." ) return self._client def get_tools(self) -> list[dict[str, Any]]: """Получить список доступных tools. Чистая функция.""" return list(CONFLUENCE_TOOLS) async def call_tool( self, tool_name: str, parameters: dict[str, Any], context: MCPContext | None = None, ) -> dict[str, Any]: """ Вызвать tool по имени. Args: tool_name: Имя tool parameters: Параметры вызова context: MCP контекст Returns: Результат выполнения tool в формате {"success", "result"|"error", "error_type"?} """ # Проверка подключения try: client = self._require_client() except ConfluenceError as e: return { "success": False, "error": str(e), "error_type": "not_connected", } # Dispatch по таблице — type-safe и без getattr 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 ConfluenceNotFoundError as e: return { "success": False, "error": f"Not found: {e}", "error_type": "not_found", } except ConfluenceAuthError as e: return { "success": False, "error": f"Authentication failed: {e}", "error_type": "auth_error", } except ConfluenceRateLimitError as e: return { "success": False, "error": f"Rate limit exceeded: {e}", "error_type": "rate_limit", } except ConfluenceError as e: return { "success": False, "error": str(e), "error_type": "confluence_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_confluence_mcp( base_url: str | None = None, email: str | None = None, api_token: str | None = None, bearer_token: str | None = None, ) -> AsyncIterator[ConfluenceMCPServer]: """ Создать Confluence MCP Server с автоматическим управлением подключением. Usage: async with create_confluence_mcp( base_url="https://example.atlassian.net/wiki", email="user@example.com", api_token="your-api-token", ) as mcp: result = await mcp.call_tool("confluence_list_spaces", {}) """ config = ConfluenceConfig( base_url=base_url or os.getenv("CONFLUENCE_BASE_URL", ""), auth_type="bearer" if bearer_token else "basic", email=email or os.getenv("CONFLUENCE_EMAIL"), api_token=api_token or os.getenv("CONFLUENCE_API_TOKEN"), bearer_token=bearer_token or os.getenv("CONFLUENCE_BEARER_TOKEN"), ) async with ConfluenceMCPServer(config) as server: yield server # ============================================================================ # Exports # ============================================================================ __all__ = [ # Config "ConfluenceConfig", # Exceptions "ConfluenceError", "ConfluenceAuthError", "ConfluenceNotFoundError", "ConfluenceRateLimitError", # Models "ConfluencePage", "ConfluenceSpace", "ConfluenceSearchResult", "ConfluenceComment", "ConfluenceAttachment", # Client & Server "ConfluenceClient", "ConfluenceMCPServer", "CONFLUENCE_TOOLS", # Helpers "create_confluence_mcp", ]