src/mcp_sourcecontrol/tools/__init__.py - master - pickling/mcp-sourcecontrol-python

src/mcp_sourcecontrol/tools/__init__.py
108 строк4 KB
pickling-21
Python-порт mcp-sourcecontrol: 20 инструментов, stdio и Streamable HTTP
29 июл 2026, 02:18
29 июл 2026, 02:18bc8bde4
"""Реестр MCP-инструментов и их регистрация. Порт src/tools/index.ts."""

from __future__ import annotations

import functools
import inspect

from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.exceptions import ToolError

from ..log import logger
from . import (  # noqa: E402  (импорты инструментов после базовых)
    add_pull_request_comment,
    create_branch,
    create_pull_request,
    decline_pull_request,
    delete_branch,
    get_branch,
    get_file_content,
    get_pull_request,
    get_pull_request_diff,
    get_user,
    list_branches,
    list_commits,
    list_files,
    list_pull_request_comments,
    list_pull_request_commits,
    list_pull_request_files,
    list_pull_requests,
    merge_pull_request,
    ping,
    update_pull_request,
)
from .define import ToolDefinition
from .errors import to_tool_error_message

# Единый источник правды обо всех регистрируемых MCP-инструментах.
TOOL_REGISTRY: list[ToolDefinition] = [
    # Диагностика
    ping.TOOL,
    # Ветки
    list_branches.TOOL,
    create_branch.TOOL,
    get_branch.TOOL,
    delete_branch.TOOL,
    # Коммиты
    list_commits.TOOL,
    # Pull requests
    list_pull_requests.TOOL,
    create_pull_request.TOOL,
    get_pull_request.TOOL,
    merge_pull_request.TOOL,
    decline_pull_request.TOOL,
    update_pull_request.TOOL,
    get_pull_request_diff.TOOL,
    list_pull_request_files.TOOL,
    list_pull_request_commits.TOOL,
    list_pull_request_comments.TOOL,
    add_pull_request_comment.TOOL,
    # Файлы
    get_file_content.TOOL,
    list_files.TOOL,
    # Пользователи
    get_user.TOOL,
]


def _wrap_handler(tool: ToolDefinition):
    """Централизованная обработка ошибок: любое исключение → MCP tool error."""

    @functools.wraps(tool.handler)
    async def wrapper(*args, **kwargs):
        try:
            return await tool.handler(*args, **kwargs)
        except ToolError:
            raise
        except Exception as error:
            logger.error("tool handler error", {"tool": tool.name, "error": str(error)})
            raise ToolError(to_tool_error_message(error)) from error

    # FastMCP строит схему по сигнатуре — сохраняем её явно;
    # __name__ задаёт заголовок схемы аргументов (вместо «_handlerArguments»)
    wrapper.__signature__ = inspect.signature(tool.handler)  # type: ignore[attr-defined]
    wrapper.__name__ = tool.name
    return wrapper


# structured_output появился не во всех версиях SDK — подключаем, когда доступен,
# чтобы схема инструментов не обрастала outputSchema (как в TS-версии).
_EXTRA_TOOL_KWARGS = (
    {"structured_output": False}
    if "structured_output" in inspect.signature(FastMCP.tool).parameters
    else {}
)


def register_tools(server: FastMCP) -> None:
    """Регистрирует все инструменты из TOOL_REGISTRY на переданном сервере."""
    for tool in TOOL_REGISTRY:
        try:
            server.tool(
                name=tool.name,
                description=tool.description,
                annotations=tool.annotations,
                **_EXTRA_TOOL_KWARGS,
            )(_wrap_handler(tool))
        except Exception as error:
            logger.warn("tool registration failed", {"tool": tool.name, "error": str(error)})