core/engine/src/tools/base.py - main - alexefan136/flowstack

core/engine/src/tools/base.py
1 349 строк44 KB
Alexander Efanov
Обновление репозитория
15 июл 2026, 12:19
15 июл 2026, 12:1976704c6
"""
Базовые классы для инструментов (tools), ресурсов и промптов.

Реализует Model Context Protocol (MCP) спецификацию 2024-11-05:
- Tools: https://modelcontextprotocol.io/specification/2024-11-05/server/tools
- Resources: https://modelcontextprotocol.io/specification/2024-11-05/server/resources
- Prompts: https://modelcontextprotocol.io/specification/2024-11-05/server/prompts

Архитектурные принципы:
- Чистые функции для преобразований данных
- Type-safe через dataclasses и Pydantic
- Async-first дизайн
- Следование "You Might Not Need an Effect"
- Совместимость с OpenAI function calling и MCP протоколом

Пример использования:

    class CalculatorTool(Tool):
        name = "calculator"
        description = "Вычисляет математические выражения"
        input_schema = {
            "type": "object",
            "properties": {
                "expression": {"type": "string"},
            },
            "required": ["expression"],
        }

        async def execute(self, **kwargs: Any) -> ToolResult:
            expression = kwargs.get("expression", "")
            # ... вычисление
            return ToolResult.success_result([TextContent(text=str(result))])

"""
from __future__ import annotations

import base64
import json
import logging
from abc import ABC, abstractmethod
from collections.abc import Callable, Coroutine
from dataclasses import dataclass, field
from typing import Any, ClassVar, Literal

from pydantic import BaseModel, ConfigDict


logger = logging.getLogger(__name__)


# ============================================================================
# Exceptions
# ============================================================================


class ToolError(Exception):
    """Базовое исключение для tools."""
    pass


class ToolExecutionError(ToolError):
    """Исключение при ошибке выполнения tool."""
    pass


class ToolValidationError(ToolError):
    """Исключение при ошибке валидации параметров."""

    def __init__(self, message: str, errors: list[str] | None = None):
        super().__init__(message)
        self.errors = errors or []


class ToolNotFoundError(ToolError):
    """Tool не найден в реестре."""
    pass


class ToolPermissionError(ToolError):
    """Отказано в разрешении на выполнение tool."""
    pass


class ResourceNotFoundError(ToolError):
    """Resource не найден."""
    pass


class PromptNotFoundError(ToolError):
    """Prompt не найден."""
    pass


# ============================================================================
# MCP Content Types
# ============================================================================


@dataclass
class TextContent:
    """
    Текстовый контент (MCP text content).

    Используется в результатах tools, сообщениях prompts.
    """

    text: str
    type: Literal["text"] = "text"

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция — преобразовать в MCP-совместимый dict."""
        return {"type": "text", "text": self.text}

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> TextContent:
        """Чистая функция — создать из dict."""
        return cls(text=str(data.get("text", "")))


@dataclass
class ImageContent:
    """
    Изображение в base64 (MCP image content).

    Используется для мультимодальных взаимодействий.
    """

    data: str  # base64-encoded
    mime_type: str = "image/png"
    type: Literal["image"] = "image"

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция."""
        return {
            "type": "image",
            "data": self.data,
            "mimeType": self.mime_type,
        }

    @classmethod
    def from_bytes(
        cls, data: bytes, mime_type: str = "image/png"
    ) -> ImageContent:
        """Создать из бинарных данных. Чистая функция."""
        encoded = base64.b64encode(data).decode("ascii")
        return cls(data=encoded, mime_type=mime_type)

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> ImageContent:
        """Чистая функция."""
        return cls(
            data=str(data.get("data", "")),
            mime_type=str(data.get("mimeType", "image/png")),
        )


@dataclass
class ResourceContent:
    """
    Embedded resource (MCP embedded resource).

    Ссылка на server-side resource, встроенная в сообщение.
    """

    uri: str
    mime_type: str = "text/plain"
    text: str | None = None
    blob: str | None = None  # base64-encoded binary
    type: Literal["resource"] = "resource"

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция."""
        resource_dict: dict[str, Any] = {
            "uri": self.uri,
            "mimeType": self.mime_type,
        }
        if self.text is not None:
            resource_dict["text"] = self.text
        if self.blob is not None:
            resource_dict["blob"] = self.blob
        return {
            "type": "resource",
            "resource": resource_dict,
        }


# Type alias для любого MCP content
Content = TextContent | ImageContent | ResourceContent


def content_to_dict(content: Content) -> dict[str, Any]:
    """Преобразовать любой content в dict. Чистая функция."""
    return content.to_dict()


def content_list_to_dict_list(
    contents: list[Content],
) -> list[dict[str, Any]]:
    """Преобразовать список content в список dict. Чистая функция."""
    return [c.to_dict() for c in contents]


# ============================================================================
# Helper: нормализация content
# ============================================================================


def _normalize_content(
    content: list[Content] | Content | str | None,
) -> list[Content]:
    """
    Привести content к списку MCP content items.

    Чистая функция — не мутирует входные данные.
    """
    if content is None:
        return []
    if isinstance(content, str):
        return [TextContent(text=content)]
    if isinstance(content, (TextContent, ImageContent, ResourceContent)):
        return [content]
    if isinstance(content, list):
        return content
    # Fallback — сериализуем как строку
    return [TextContent(text=str(content))]


# ============================================================================
# Tool Result
# ============================================================================


class ToolResult(BaseModel):
    """
    Результат выполнения tool.

    Совместим с MCP Tool Result форматом:
    - content: список MCP content items (text, image, resource)
    - isError: флаг ошибки

    Usage:
        # MCP-совместимый формат (предпочтительно)
        return ToolResult.success_result([
            TextContent(text="Результат: 42"),
        ])

        # С ошибкой
        return ToolResult.failure("Деление на ноль")

        # С произвольными данными
        return ToolResult.from_data({"answer": 42})
    """

    model_config = ConfigDict(
        extra="allow",
        arbitrary_types_allowed=True,
    )

    # MCP-совместимые поля
    content: list[Content] = field(default_factory=list)
    is_error: bool = False
    error: str | None = None
    # Дополнительные метаданные (не часть MCP spec, но полезно)
    metadata: dict[str, Any] | None = None

    # ====================================================================
    # Properties
    # ====================================================================

    @property
    def success(self) -> bool:
        """
        Проверить, успешен ли результат.

        Read-only property — вычисляется из is_error.
        """
        return not self.is_error

    # ====================================================================
    # MCP Factory Methods
    # ====================================================================

    @classmethod
    def success_result(
        cls,
        content: list[Content] | Content | str | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> ToolResult:
        """
        Создать успешный результат.

        Args:
            content: MCP content (TextContent, ImageContent, ResourceContent),
                     список content, или строка (автоматически обернётся в TextContent)
            metadata: Дополнительные метаданные
        """
        return cls(
            content=_normalize_content(content),
            is_error=False,
            metadata=metadata,
        )

    @classmethod
    def failure(
        cls,
        error: str,
        content: list[Content] | Content | str | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> ToolResult:
        """
        Создать результат с ошибкой.

        Args:
            error: Сообщение об ошибке
            content: Опциональный контент (например, partial result)
            metadata: Дополнительные метаданные
        """
        from typing import cast
        
        final_content: list[Content] = _normalize_content(content)
        if not final_content:
            # Явно приводим к list[Content] для type safety
            final_content = cast(list[Content], [TextContent(text=error)])

        return cls(
            content=final_content,
            is_error=True,
            metadata=metadata,
        )

    @classmethod
    def from_data(
        cls,
        data: Any,
        metadata: dict[str, Any] | None = None,
    ) -> ToolResult:
        """
        Создать успешный результат из произвольных данных.

        Данные сериализуются в текст (JSON если возможно).
        """
        if data is None:
            text = ""
        elif isinstance(data, str):
            text = data
        else:
            try:
                text = json.dumps(data, ensure_ascii=False, indent=2)
            except (TypeError, ValueError):
                text = str(data)

        return cls(
            content=[TextContent(text=text)],
            is_error=False,
            metadata=metadata,
        )

    # ====================================================================
    # Helpers
    # ====================================================================

    def is_success(self) -> bool:
        """Проверить, что результат успешный."""
        return not self.is_error

    def is_failure(self) -> bool:
        """Проверить, что результат содержит ошибку."""
        return self.is_error

    def get_text(self) -> str:
        """Извлечь весь текстовый контент. Чистая функция."""
        texts: list[str] = []
        for item in self.content:
            if isinstance(item, TextContent):
                texts.append(item.text)
        return "\n".join(texts)

    def get_images(self) -> list[ImageContent]:
        """Извлечь все изображения. Чистая функция."""
        return [item for item in self.content if isinstance(item, ImageContent)]

    def get_resources(self) -> list[ResourceContent]:
        """Извлечь все embedded resources. Чистая функция."""
        return [
            item for item in self.content if isinstance(item, ResourceContent)
        ]

    def get_data(self, default: Any = None) -> Any:
        """
        Извлечь данные из текстового контента.

        Пытается распарсить текст как JSON. Если не получается — возвращает
        строку. Если строка пустая — возвращает default.
        """
        text = self.get_text()
        if not text:
            return default
        try:
            return json.loads(text)
        except (json.JSONDecodeError, ValueError):
            return text

    def raise_on_error(self) -> None:
        """Выбросить исключение, если результат содержит ошибку."""
        if self.is_error:
            raise ToolExecutionError(self.get_text() or "Unknown error")

    def to_mcp_dict(self) -> dict[str, Any]:
        """
        Преобразовать в MCP Tool Result формат.

        Чистая функция — не мутирует состояние.
        """
        result: dict[str, Any] = {
            "content": [c.to_dict() for c in self.content],
            "isError": self.is_error,
        }
        if self.metadata:
            result["_metadata"] = self.metadata
        return result

    def to_dict(self) -> dict[str, Any]:
        """Преобразовать в dict (для сериализации). Чистая функция."""
        return {
            "content": [c.to_dict() for c in self.content],
            "is_error": self.is_error,
            "success": self.success,
            "metadata": self.metadata,
        }


# ============================================================================
# Tool (Base Class)
# ============================================================================


class Tool(ABC):
    """
    Базовый абстрактный класс для инструментов.

    Реализует MCP Tool specification:
    https://modelcontextprotocol.io/specification/2024-11-05/server/tools

    Каждый tool должен определить:
    - `name` (str): уникальное имя
    - `description` (str): описание для LLM
    - `execute(**kwargs)` (async): метод выполнения

    Опционально:
    - `input_schema` (dict): JSON Schema параметров
    - `examples` (list): примеры использования
    - `tags` (list): теги для категоризации
    - `requires_confirmation` (bool): требует ли подтверждения от пользователя
    """

    # ====================================================================
    # Class-level attributes (переопределяются в наследниках)
    # ====================================================================

    name: ClassVar[str] = ""
    description: ClassVar[str] = ""

    # JSON Schema параметров (MCP inputSchema)
    input_schema: ClassVar[dict[str, Any] | None] = None

    # Backward compatibility alias
    parameters_schema: ClassVar[dict[str, Any] | None] = None

    # Примеры использования
    examples: ClassVar[list[dict[str, Any]]] = []

    # Теги для категоризации
    tags: ClassVar[list[str]] = []

    # Требует ли подтверждения от пользователя (для security)
    requires_confirmation: ClassVar[bool] = False

    # Read-only tool (не изменяет состояние)
    is_read_only: ClassVar[bool] = False

    # ====================================================================
    # Abstract methods
    # ====================================================================

    @abstractmethod
    async def execute(self, **kwargs: Any) -> ToolResult:
        """
        Выполнить инструмент с переданными параметрами.

        Args:
            **kwargs: Параметры, определённые в input_schema.

        Returns:
            ToolResult с контентом или ошибкой.

        Raises:
            ToolExecutionError: При критической ошибке.
        """
        ...

    # ====================================================================
    # Schema Methods (MCP & OpenAI compatible)
    # ====================================================================

    def get_input_schema(self) -> dict[str, Any]:
        """
        Получить JSON Schema параметров.

        Чистая функция — возвращает default schema если не задан.
        """
        schema = self.input_schema or self.parameters_schema
        if schema is not None:
            return schema

        # Default — принимаем любой объект
        return {
            "type": "object",
            "properties": {},
            "additionalProperties": True,
        }

    def to_mcp_tool(self) -> dict[str, Any]:
        """
        Вернуть tool в MCP формате.

        Чистая функция.

        MCP Tool format:
        {
            "name": "...",
            "description": "...",
            "inputSchema": { ... JSON Schema ... }
        }
        """
        return {
            "name": self.name,
            "description": self.description,
            "inputSchema": self.get_input_schema(),
        }

    def to_openai_tool(self) -> dict[str, Any]:
        """
        Вернуть tool в OpenAI function calling формате.

        Чистая функция.

        OpenAI format:
        {
            "type": "function",
            "function": {
                "name": "...",
                "description": "...",
                "parameters": { ... JSON Schema ... }
            }
        }
        """
        schema: dict[str, Any] = {
            "name": self.name,
            "description": self.description,
            "parameters": self.get_input_schema(),
        }

        # Добавляем additionalProperties: false для strict mode
        if schema["parameters"].get("additionalProperties") is None:
            schema["parameters"]["additionalProperties"] = False

        return {
            "type": "function",
            "function": schema,
        }

    def to_schema(self) -> dict[str, Any]:
        """
        Backward-compatible alias для получения function schema.

        Возвращает только function часть (без type: "function").
        """
        return self.to_openai_tool()["function"]

    # ====================================================================
    # Validation
    # ====================================================================

    def validate_input(
        self, kwargs: dict[str, Any]
    ) -> tuple[bool, str | None]:
        """
        Базовая валидация входных параметров.

        По умолчанию проверяет только required поля из input_schema.
        Переопределите в наследниках для строгой валидации.

        Чистая функция — не мутирует kwargs.

        Returns:
            Tuple из (is_valid, error_message)
        """
        schema = self.input_schema or self.parameters_schema
        if schema is None:
            return True, None

        required = schema.get("required") or []
        if not isinstance(required, list):
            return True, None

        missing = [field_name for field_name in required if field_name not in kwargs]
        if missing:
            return False, f"Missing required fields: {', '.join(missing)}"

        return True, None

    def validate_input_strict(
        self, kwargs: dict[str, Any]
    ) -> tuple[bool, list[str]]:
        """
        Строгая валидация с проверкой типов (если доступен jsonschema).

        Чистая функция.

        Returns:
            Tuple из (is_valid, list_of_errors)
        """
        errors: list[str] = []

        # Базовая проверка required
        is_valid, error_msg = self.validate_input(kwargs)
        if not is_valid and error_msg:
            errors.append(error_msg)

        # Попробуем использовать jsonschema если доступен
        schema = self.input_schema or self.parameters_schema
        if schema is not None:
            try:
                import jsonschema  # type: ignore[import-not-found]

                validator = jsonschema.Draft7Validator(schema)
                for error in validator.iter_errors(kwargs):
                    path = ".".join(str(p) for p in error.path) or "root"
                    errors.append(f"{path}: {error.message}")
            except ImportError:
                # jsonschema не установлен — пропускаем
                pass
            except Exception as e:
                logger.warning(f"JSON schema validation failed: {e}")

        return len(errors) == 0, errors

    # ====================================================================
    # Execution
    # ====================================================================

    async def safe_execute(self, **kwargs: Any) -> ToolResult:
        """
        Безопасное выполнение с обработкой исключений.

        Оборачивает execute() в try/except и возвращает ToolResult.failure
        при любом необработанном исключении.
        """
        # Валидация
        is_valid, error_msg = self.validate_input(kwargs)
        if not is_valid:
            return ToolResult.failure(
                f"Invalid input for tool '{self.name}': {error_msg}"
            )

        # Выполнение
        try:
            return await self.execute(**kwargs)
        except ToolExecutionError as e:
            return ToolResult.failure(str(e))
        except ToolPermissionError as e:
            return ToolResult.failure(
                f"Permission denied: {e}",
                metadata={"error_type": "permission_denied"},
            )
        except ToolValidationError as e:
            return ToolResult.failure(
                f"Validation error: {e}",
                metadata={"error_type": "validation", "errors": e.errors},
            )
        except Exception as e:
            logger.exception(f"Unexpected error in tool '{self.name}'")
            return ToolResult.failure(
                f"Unexpected error in tool '{self.name}': {type(e).__name__}: {e}",
                metadata={"error_type": "unexpected"},
            )

    async def __call__(self, **kwargs: Any) -> ToolResult:
        """
        Сделать tool callable.

        Позволяет использовать tool как функцию:
            result = await my_tool(param1="value1")
        """
        return await self.safe_execute(**kwargs)

    # ====================================================================
    # Lifecycle Hooks (опционально переопределяются)
    # ====================================================================

    async def on_before_execute(self, **kwargs: Any) -> None:
        """Hook перед выполнением. Для логирования/метрик."""
        pass

    async def on_after_execute(
        self, result: ToolResult, **kwargs: Any
    ) -> None:
        """Hook после выполнения. Для логирования/метрик."""
        pass

    async def on_error(self, error: Exception, **kwargs: Any) -> None:
        """Hook при ошибке. Для alerting/logging."""
        pass

    # ====================================================================
    # Magic methods
    # ====================================================================

    def __repr__(self) -> str:
        return f"<Tool name={self.name!r} description={self.description!r}>"

    def __str__(self) -> str:
        return self.name or self.__class__.__name__


# ============================================================================
# Resource (MCP Resource)
# ============================================================================


@dataclass
class ResourceDefinition:
    """
    Определение MCP Resource.

    https://modelcontextprotocol.io/specification/2024-11-05/server/resources

    Resources — это server-side данные, которые могут быть прочитаны клиентами.
    """

    uri: str
    name: str
    description: str = ""
    mime_type: str | None = None

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция."""
        result: dict[str, Any] = {
            "uri": self.uri,
            "name": self.name,
        }
        if self.description:
            result["description"] = self.description
        if self.mime_type:
            result["mimeType"] = self.mime_type
        return result


class Resource(ABC):
    """
    Базовый класс для MCP Resources.

    Resources предоставляют read-only доступ к данным:
    - Файлы, документы
    - API responses
    - Database records
    - Configuration data
    """

    uri: ClassVar[str] = ""
    name: ClassVar[str] = ""
    description: ClassVar[str] = ""
    mime_type: ClassVar[str | None] = None

    @abstractmethod
    async def read(self) -> str | bytes:
        """
        Прочитать содержимое resource.

        Returns:
            str для текстовых ресурсов, bytes для бинарных.
        """
        ...

    def to_definition(self) -> ResourceDefinition:
        """Чистая функция — получить определение."""
        return ResourceDefinition(
            uri=self.uri,
            name=self.name,
            description=self.description,
            mime_type=self.mime_type,
        )

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция."""
        return self.to_definition().to_dict()


# ============================================================================
# Prompt (MCP Prompt)
# ============================================================================


@dataclass
class PromptArgument:
    """Аргумент MCP Prompt."""

    name: str
    description: str = ""
    required: bool = False

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция."""
        result: dict[str, Any] = {"name": self.name}
        if self.description:
            result["description"] = self.description
        if self.required:
            result["required"] = True
        return result


@dataclass
class PromptMessage:
    """
    Сообщение в MCP Prompt.

    Может содержать:
    - role: "user" | "assistant"
    - content: TextContent | ImageContent | ResourceContent
    """

    role: Literal["user", "assistant"]
    content: Content

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция."""
        return {
            "role": self.role,
            "content": self.content.to_dict(),
        }


@dataclass
class PromptDefinition:
    """Определение MCP Prompt."""

    name: str
    description: str = ""
    arguments: list[PromptArgument] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция."""
        result: dict[str, Any] = {"name": self.name}
        if self.description:
            result["description"] = self.description
        if self.arguments:
            result["arguments"] = [a.to_dict() for a in self.arguments]
        return result


class Prompt(ABC):
    """
    Базовый класс для MCP Prompts.

    Prompts — это шаблоны сообщений, которые пользователи могут
    явно выбирать в UI (например, как slash commands).
    """

    name: ClassVar[str] = ""
    description: ClassVar[str] = ""
    arguments: ClassVar[list[PromptArgument]] = []

    @abstractmethod
    async def get_messages(self, **kwargs: Any) -> list[PromptMessage]:
        """
        Получить список сообщений prompt с подставленными аргументами.

        Args:
            **kwargs: Аргументы prompt.

        Returns:
            Список PromptMessage.
        """
        ...

    def to_definition(self) -> PromptDefinition:
        """Чистая функция — получить определение."""
        return PromptDefinition(
            name=self.name,
            description=self.description,
            arguments=self.arguments,
        )

    def to_dict(self) -> dict[str, Any]:
        """Чистая функция."""
        return self.to_definition().to_dict()


# ============================================================================
# Tool Registry
# ============================================================================


class ToolRegistry:
    """
    Реестр для управления tools, resources и prompts.

    Thread-safe (в контексте asyncio — все операции в одном event loop).

    Usage:
        registry = ToolRegistry()
        registry.register_tool(CalculatorTool())
        registry.register_resource(FileResource())

        # Получить schemas для MCP
        tools = registry.list_mcp_tools()
        resources = registry.list_mcp_resources()

        # Вызвать tool по имени
        result = await registry.call_tool("calculator", {"expression": "2+2"})
    """

    def __init__(self) -> None:
        self._tools: dict[str, Tool] = {}
        self._resources: dict[str, Resource] = {}
        self._prompts: dict[str, Prompt] = {}

    # ====================================================================
    # Tools
    # ====================================================================

    def register_tool(self, tool_instance: Tool) -> None:
        """Зарегистрировать tool."""
        if not tool_instance.name:
            raise ToolError(
                f"Tool {tool_instance.__class__.__name__} has no name"
            )
        if tool_instance.name in self._tools:
            logger.warning(
                f"Tool '{tool_instance.name}' already registered, replacing"
            )
        self._tools[tool_instance.name] = tool_instance

    def unregister_tool(self, name: str) -> bool:
        """Удалить tool из реестра."""
        if name in self._tools:
            del self._tools[name]
            return True
        return False

    def get_tool(self, name: str) -> Tool:
        """Получить tool по имени."""
        tool_instance = self._tools.get(name)
        if tool_instance is None:
            raise ToolNotFoundError(f"Tool '{name}' not found")
        return tool_instance

    def has_tool(self, name: str) -> bool:
        """Проверить наличие tool."""
        return name in self._tools

    def list_tools(self) -> list[Tool]:
        """Получить список всех tools."""
        return list(self._tools.values())

    def list_tool_names(self) -> list[str]:
        """Получить список имён tools."""
        return list(self._tools.keys())

    def list_mcp_tools(self) -> list[dict[str, Any]]:
        """Получить tools в MCP формате. Чистая функция."""
        return [t.to_mcp_tool() for t in self._tools.values()]

    def list_openai_tools(self) -> list[dict[str, Any]]:
        """Получить tools в OpenAI формате. Чистая функция."""
        return [t.to_openai_tool() for t in self._tools.values()]

    async def call_tool(
        self,
        name: str,
        arguments: dict[str, Any] | None = None,
    ) -> ToolResult:
        """
        Вызвать tool по имени с аргументами.

        Args:
            name: Имя tool
            arguments: Параметры вызова

        Returns:
            ToolResult с результатом или ошибкой.
        """
        try:
            tool_instance = self.get_tool(name)
        except ToolNotFoundError as e:
            return ToolResult.failure(str(e))

        return await tool_instance.safe_execute(**(arguments or {}))

    # ====================================================================
    # Resources
    # ====================================================================

    def register_resource(self, resource_instance: Resource) -> None:
        """Зарегистрировать resource."""
        if not resource_instance.uri:
            raise ToolError(
                f"Resource {resource_instance.__class__.__name__} has no URI"
            )
        if resource_instance.uri in self._resources:
            logger.warning(
                f"Resource '{resource_instance.uri}' already registered"
            )
        self._resources[resource_instance.uri] = resource_instance

    def unregister_resource(self, uri: str) -> bool:
        """Удалить resource."""
        if uri in self._resources:
            del self._resources[uri]
            return True
        return False

    def get_resource(self, uri: str) -> Resource:
        """Получить resource по URI."""
        resource_instance = self._resources.get(uri)
        if resource_instance is None:
            raise ResourceNotFoundError(f"Resource '{uri}' not found")
        return resource_instance

    def has_resource(self, uri: str) -> bool:
        """Проверить наличие resource."""
        return uri in self._resources

    def list_resources(self) -> list[Resource]:
        """Получить список всех resources."""
        return list(self._resources.values())

    def list_mcp_resources(self) -> list[dict[str, Any]]:
        """Получить resources в MCP формате. Чистая функция."""
        return [r.to_dict() for r in self._resources.values()]

    async def read_resource(self, uri: str) -> str | bytes:
        """Прочитать resource по URI."""
        resource_instance = self.get_resource(uri)
        return await resource_instance.read()

    # ====================================================================
    # Prompts
    # ====================================================================

    def register_prompt(self, prompt_instance: Prompt) -> None:
        """Зарегистрировать prompt."""
        if not prompt_instance.name:
            raise ToolError(
                f"Prompt {prompt_instance.__class__.__name__} has no name"
            )
        if prompt_instance.name in self._prompts:
            logger.warning(
                f"Prompt '{prompt_instance.name}' already registered"
            )
        self._prompts[prompt_instance.name] = prompt_instance

    def unregister_prompt(self, name: str) -> bool:
        """Удалить prompt."""
        if name in self._prompts:
            del self._prompts[name]
            return True
        return False

    def get_prompt(self, name: str) -> Prompt:
        """Получить prompt по имени."""
        prompt_instance = self._prompts.get(name)
        if prompt_instance is None:
            raise PromptNotFoundError(f"Prompt '{name}' not found")
        return prompt_instance

    def has_prompt(self, name: str) -> bool:
        """Проверить наличие prompt."""
        return name in self._prompts

    def list_prompts(self) -> list[Prompt]:
        """Получить список всех prompts."""
        return list(self._prompts.values())

    def list_mcp_prompts(self) -> list[dict[str, Any]]:
        """Получить prompts в MCP формате. Чистая функция."""
        return [p.to_dict() for p in self._prompts.values()]

    async def get_prompt_messages(
        self, name: str, arguments: dict[str, Any] | None = None
    ) -> list[PromptMessage]:
        """Получить сообщения prompt с аргументами."""
        prompt_instance = self.get_prompt(name)
        return await prompt_instance.get_messages(**(arguments or {}))

    # ====================================================================
    # Bulk Operations
    # ====================================================================

    def clear(self) -> None:
        """Очистить все реестры."""
        self._tools.clear()
        self._resources.clear()
        self._prompts.clear()

    def stats(self) -> dict[str, int]:
        """Получить статистику реестра. Чистая функция."""
        return {
            "tools": len(self._tools),
            "resources": len(self._resources),
            "prompts": len(self._prompts),
        }


# ============================================================================
# Global Registry (singleton)
# ============================================================================


_global_registry: ToolRegistry | None = None


def get_global_registry() -> ToolRegistry:
    """Получить глобальный реестр (singleton)."""
    global _global_registry
    if _global_registry is None:
        _global_registry = ToolRegistry()
    return _global_registry


# ============================================================================
# Decorators
# ============================================================================


def tool(
    name: str,
    description: str,
    input_schema: dict[str, Any] | None = None,
    tags: list[str] | None = None,
    requires_confirmation: bool = False,
    is_read_only: bool = False,
) -> Callable[
    [Callable[..., Coroutine[Any, Any, ToolResult | Any]]],
    type[Tool],
]:
    """
    Декоратор для создания tool из асинхронной функции.

    Example:
        @tool(
            name="add",
            description="Складывает два числа",
            input_schema={
                "type": "object",
                "properties": {
                    "a": {"type": "number"},
                    "b": {"type": "number"},
                },
                "required": ["a", "b"],
            },
        )
        async def add(a: float, b: float) -> ToolResult:
            return ToolResult.success_result([TextContent(text=str(a + b))])
    """

    def decorator(
        func: Callable[..., Coroutine[Any, Any, ToolResult | Any]],
    ) -> type[Tool]:
        # Создаём класс динамически с реализованным execute
        async def execute_impl(self: Tool, **kwargs: Any) -> ToolResult:
            result = await func(**kwargs)
            # Автоматически оборачиваем примитивные типы в ToolResult
            if isinstance(result, ToolResult):
                return result
            if isinstance(result, str):
                return ToolResult.success_result([TextContent(text=result)])
            if isinstance(
                result, (TextContent, ImageContent, ResourceContent)
            ):
                return ToolResult.success_result([result])
            # Сериализуем как JSON
            try:
                text = json.dumps(result, ensure_ascii=False, indent=2)
            except (TypeError, ValueError):
                text = str(result)
            return ToolResult.success_result([TextContent(text=text)])

        # Создаём класс через type() с реализованным execute
        # Это обходит проблему абстрактного класса
        attrs: dict[str, Any] = {
            "name": name,
            "description": description,
            "execute": execute_impl,
            "requires_confirmation": requires_confirmation,
            "is_read_only": is_read_only,
        }
        if input_schema is not None:
            attrs["input_schema"] = input_schema
        if tags is not None:
            attrs["tags"] = tags

        function_tool_cls = type(
            f"{name}_tool",
            (Tool,),
            attrs,
        )

        return function_tool_cls  # type: ignore[return-value]

    return decorator


def register_in_global(
    tool_instance: Tool | type[Tool] | None = None,
    *,
    resource: Resource | None = None,
    prompt: Prompt | None = None,
) -> Tool | Resource | Prompt | None:
    """
    Зарегистрировать tool/resource/prompt в глобальном реестре.

    Можно использовать как декоратор или функцию:

        # Как декоратор
        @register_in_global
        class MyTool(Tool):
            ...

        # Как функцию
        register_in_global(my_tool_instance)
        register_in_global(resource=my_resource)
    """
    registry = get_global_registry()

    if tool_instance is not None:
        # Если передан класс — инстанцируем
        actual_tool: Tool
        if isinstance(tool_instance, type):
            actual_tool = tool_instance()
        else:
            actual_tool = tool_instance
        registry.register_tool(actual_tool)
        return actual_tool

    if resource is not None:
        registry.register_resource(resource)
        return resource

    if prompt is not None:
        registry.register_prompt(prompt)
        return prompt

    return None


# ============================================================================
# Helper: Create simple text tool
# ============================================================================


def create_text_tool(
    name: str,
    description: str,
    handler: Callable[..., Coroutine[Any, Any, str]],
    input_schema: dict[str, Any] | None = None,
) -> Tool:
    """
    Создать простой tool, возвращающий текст.

    Useful для быстрого прототипирования.

    Example:
        tool = create_text_tool(
            name="echo",
            description="Эхо",
            handler=lambda text: text,
            input_schema={
                "type": "object",
                "properties": {"text": {"type": "string"}},
            },
        )
    """

    async def execute_impl(self: Tool, **kwargs: Any) -> ToolResult:
        try:
            text = await handler(**kwargs)
            return ToolResult.success_result([TextContent(text=str(text))])
        except Exception as e:
            return ToolResult.failure(f"Error in {name}: {e}")

    # Создаём класс через type() с реализованным execute
    # Это обходит проблему абстрактного класса
    attrs: dict[str, Any] = {
        "name": name,
        "description": description,
        "execute": execute_impl,
    }
    if input_schema is not None:
        attrs["input_schema"] = input_schema

    simple_tool_cls = type(
        f"{name}_tool",
        (Tool,),
        attrs,
    )

    return simple_tool_cls()  # type: ignore[return-value]


# ============================================================================
# Exports
# ============================================================================


__all__ = [
    # Exceptions
    "ToolError",
    "ToolExecutionError",
    "ToolValidationError",
    "ToolNotFoundError",
    "ToolPermissionError",
    "ResourceNotFoundError",
    "PromptNotFoundError",
    # MCP Content Types
    "TextContent",
    "ImageContent",
    "ResourceContent",
    "Content",
    "content_to_dict",
    "content_list_to_dict_list",
    # Tool Result
    "ToolResult",
    # Base Classes
    "Tool",
    "Resource",
    "ResourceDefinition",
    "Prompt",
    "PromptDefinition",
    "PromptArgument",
    "PromptMessage",
    # Registry
    "ToolRegistry",
    "get_global_registry",
    # Decorators & Helpers
    "tool",
    "register_in_global",
    "create_text_tool",
]