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

core/engine/src/api/tools.py
585 строк18 KB
Alexander Efanov
upd fix
31 июл 2026, 19:17
31 июл 2026, 19:17d146d86
"""API endpoints для доступа к Tool Registry и MCP protocol endpoint.

Реализует MCP specification 2024-11-05:
- initialize — handshake
- tools/list, tools/call — инструменты
- resources/list, resources/read — ресурсы
- prompts/list, prompts/get — промпты

Улучшения:
- Dispatch table для JSON-RPC методов (вместо if/elif)
- Опциональный API key auth на MCP endpoint (security, timing-safe)
- Confirmation для опасных tools (requires_confirmation)

Эндпоинты:
    - GET    /api/v1/tools                     — список tools
    - GET    /api/v1/tools/stats               — статистика registry
    - GET    /api/v1/tools/names               — имена tools
    - GET    /api/v1/tools/bundles             — список MCP bundles
    - GET    /api/v1/tools/{name}              — информация о tool
    - POST   /api/v1/tools/call                — вызвать tool
    - POST   /api/v1/tools/mcp                 — MCP JSON-RPC 2.0 endpoint
    - POST   /api/v1/tools/bundles/{n}/enable  — включить bundle
    - POST   /api/v1/tools/bundles/{n}/disable — отключить bundle
"""

from __future__ import annotations

import base64
import hmac
from collections.abc import Awaitable, Callable
from typing import Any

import structlog
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, Field

from src import __version__
from src.api.dependencies import get_user_ctx
from src.config import get_settings
from src.middleware.auth import UserContext
from src.tools import (
    call_tool,
    get_manager,
    list_tool_names,
    list_tools,
)
from src.tools import (
    stats as tools_stats,
)

logger = structlog.get_logger()

router = APIRouter(prefix="/api/v1/tools", tags=["tools"])


# ============================================================================
# REQUEST/RESPONSE MODELS
# ============================================================================


class ToolInfo(BaseModel):
    """Информация о tool."""

    name: str
    description: str
    tags: list[str]
    input_schema: dict[str, Any] | None
    is_read_only: bool
    requires_confirmation: bool


class ToolCallRequest(BaseModel):
    """Запрос на вызов tool."""

    name: str = Field(..., description="Имя tool")
    arguments: dict[str, Any] = Field(default_factory=dict, description="Параметры tool")
    confirm: bool = Field(
        default=False,
        description="Подтверждение для tools с requires_confirmation",
    )


class ToolCallResponse(BaseModel):
    """Ответ от вызова tool."""

    success: bool
    text: str
    metadata: dict[str, Any] | None = None
    error: str | None = None


class ToolsStatsResponse(BaseModel):
    """Статистика Tool Registry."""

    total_tools: int
    builtin_tools: int
    mcp_tools: int
    mcp_bundles: int
    mcp_bundles_enabled: int
    mcp_bundles_initialized: int
    initialized: bool
    bundles: list[dict[str, Any]]


class MCPJsonRpcRequest(BaseModel):
    """JSON-RPC 2.0 request для MCP endpoint."""

    jsonrpc: str = "2.0"
    id: int | str | None = None
    method: str
    params: dict[str, Any] | None = None


class MCPJsonRpcResponse(BaseModel):
    """JSON-RPC 2.0 response для MCP endpoint."""

    jsonrpc: str = "2.0"
    id: int | str | None
    result: Any | None = None
    error: dict[str, Any] | None = None


# ============================================================================
# AUTH HELPERS
# ============================================================================


def _check_mcp_auth(request: Request) -> None:
    """
    Проверить API key для MCP endpoint (если настроен).

    Если ``mcp_api_key`` не задан в settings — режим dev (без auth).
    В production обязательно настройте ``mcp_api_key``.

    Использует timing-safe сравнение (защита от timing-атак).
    """
    settings = get_settings()
    api_key = getattr(settings, "mcp_api_key", None)
    if not api_key:
        logger.warning("mcp.auth.disabled", reason="mcp_api_key not configured")
        return

    # SecretStr → расшифровка для сравнения
    api_key_value = (
        api_key.get_secret_value() if hasattr(api_key, "get_secret_value") else str(api_key)
    )

    provided = request.headers.get("X-API-Key", "")
    if not provided:
        auth_header = request.headers.get("Authorization", "")
        if auth_header.startswith("Bearer "):
            provided = auth_header[7:]

    # Timing-safe сравнение
    if not provided or not hmac.compare_digest(provided, api_key_value):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or missing API key",
        )


def _check_admin_access(user_ctx: UserContext) -> None:
    """
    Проверить права администратора (для управления bundles).

    TODO: реализовать систему ролей. Пока — проверка аутентификации.
    """
    if user_ctx is None or user_ctx.user_id is None:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin access required",
        )


# ============================================================================
# HELPER FUNCTIONS
# ============================================================================


def _build_tool_info(tool: Any) -> ToolInfo:
    """Собрать ToolInfo из объекта tool."""
    return ToolInfo(
        name=tool.name,
        description=tool.description,
        tags=getattr(tool, "tags", []),
        input_schema=tool.get_input_schema(),
        is_read_only=getattr(tool, "is_read_only", True),
        requires_confirmation=getattr(tool, "requires_confirmation", False),
    )


# ============================================================================
# TOOL REGISTRY ENDPOINTS
# ============================================================================


@router.get("", response_model=list[ToolInfo])
async def list_tools_endpoint(
    user_ctx: UserContext = Depends(get_user_ctx),
) -> list[ToolInfo]:
    """Список всех доступных tools (builtin + MCP)."""
    tools = list_tools()

    logger.info(
        "tools.listed",
        count=len(tools),
        user_id=str(user_ctx.user_id),
        workspace_id=user_ctx.workspace_id,
    )

    return [_build_tool_info(tool) for tool in tools]


@router.get("/stats", response_model=ToolsStatsResponse)
async def get_tools_stats(
    user_ctx: UserContext = Depends(get_user_ctx),
) -> ToolsStatsResponse:
    """Статистика Tool Registry."""
    stats = tools_stats()

    logger.info(
        "tools.stats",
        user_id=str(user_ctx.user_id),
        workspace_id=user_ctx.workspace_id,
    )

    return ToolsStatsResponse(
        total_tools=stats["total_tools"],
        builtin_tools=stats["builtin_tools"],
        mcp_tools=stats["mcp_tools"],
        mcp_bundles=stats["mcp_bundles"],
        mcp_bundles_enabled=stats["mcp_bundles_enabled"],
        mcp_bundles_initialized=stats["mcp_bundles_initialized"],
        initialized=stats["initialized"],
        bundles=stats["bundles"],
    )


@router.get("/names")
async def list_tool_names_endpoint(
    user_ctx: UserContext = Depends(get_user_ctx),
) -> dict[str, Any]:
    """Список имён всех tools (быстрый endpoint)."""
    names = list_tool_names()

    logger.info(
        "tools.names.listed",
        count=len(names),
        user_id=str(user_ctx.user_id),
        workspace_id=user_ctx.workspace_id,
    )

    return {"tools": names, "count": len(names)}


# ⚠️ ВАЖНО: /bundles (GET) должен быть ДО /{tool_name},
# иначе path "bundles" перехватится как tool_name
@router.get("/bundles")
async def list_bundles_endpoint(
    user_ctx: UserContext = Depends(get_user_ctx),
) -> list[dict[str, Any]]:
    """Список MCP bundles с их статусами."""
    manager = get_manager()

    logger.info(
        "mcp.bundles.listed",
        user_id=str(user_ctx.user_id),
        workspace_id=user_ctx.workspace_id,
    )

    return manager.list_bundles()


@router.get("/{tool_name}", response_model=ToolInfo)
async def get_tool_info(
    tool_name: str,
    user_ctx: UserContext = Depends(get_user_ctx),
) -> ToolInfo:
    """Детальная информация о конкретном tool."""
    manager = get_manager()
    tool = manager.get_tool(tool_name)

    if tool is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Tool '{tool_name}' not found",
        )

    logger.info(
        "tool.info",
        tool_name=tool_name,
        user_id=str(user_ctx.user_id),
        workspace_id=user_ctx.workspace_id,
    )

    return _build_tool_info(tool)


@router.post("/call", response_model=ToolCallResponse)
async def call_tool_endpoint(
    request: ToolCallRequest,
    user_ctx: UserContext = Depends(get_user_ctx),
) -> ToolCallResponse:
    """
    Вызвать tool по имени.

    Если tool имеет ``requires_confirmation=True`` и ``confirm=False`` —
    возвращает 409 (требуется подтверждение).
    """
    manager = get_manager()
    tool = manager.get_tool(request.name)

    if tool is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Tool '{request.name}' not found",
        )

    # Проверка подтверждения для опасных tools
    if getattr(tool, "requires_confirmation", False) and not request.confirm:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail={
                "error": f"Tool '{request.name}' requires confirmation",
                "hint": "Set confirm=true to proceed",
            },
        )

    logger.info(
        "tool.call.requested",
        tool_name=request.name,
        arguments_keys=list(request.arguments.keys()),
        confirmed=request.confirm,
        user_id=str(user_ctx.user_id),
        workspace_id=user_ctx.workspace_id,
    )

    result = await call_tool(request.name, **request.arguments)

    logger.info(
        "tool.call.completed",
        tool_name=request.name,
        success=result.is_success(),
        user_id=str(user_ctx.user_id),
        workspace_id=user_ctx.workspace_id,
    )

    return ToolCallResponse(
        success=result.is_success(),
        text=result.get_text(),
        metadata=result.metadata,
        error=result.error,
    )


# ============================================================================
# MCP JSON-RPC HANDLERS (dispatch table)
# ============================================================================


async def _mcp_initialize(params: dict[str, Any], manager: Any) -> dict[str, Any]:
    """Обработчик initialize (handshake)."""
    return {
        "protocolVersion": "2024-11-05",
        "capabilities": {
            "tools": {"listChanged": False},
            "resources": {"subscribe": False, "listChanged": False},
            "prompts": {"listChanged": False},
        },
        "serverInfo": {
            "name": "FlowStack Engine",
            "version": __version__,
        },
    }


async def _mcp_tools_list(params: dict[str, Any], manager: Any) -> dict[str, Any]:
    """Обработчик tools/list."""
    tools_list = [
        {
            "name": tool.name,
            "description": tool.description,
            "inputSchema": tool.get_input_schema(),
        }
        for tool in list_tools()
    ]
    return {"tools": tools_list}


async def _mcp_tools_call(params: dict[str, Any], manager: Any) -> dict[str, Any]:
    """Обработчик tools/call."""
    name = params.get("name")
    if not name:
        raise ValueError("missing tool name")

    arguments = params.get("arguments", {})
    result = await call_tool(name, **arguments)

    return {
        "content": [c.to_dict() for c in result.content],
        "isError": result.is_error,
    }


async def _mcp_resources_list(params: dict[str, Any], manager: Any) -> dict[str, Any]:
    """Обработчик resources/list."""
    return {"resources": manager.registry.list_mcp_resources()}


async def _mcp_resources_read(params: dict[str, Any], manager: Any) -> dict[str, Any]:
    """Обработчик resources/read."""
    uri = params.get("uri")
    if not uri:
        raise ValueError("missing resource uri")

    content = await manager.registry.read_resource(uri)

    if isinstance(content, bytes):
        return {
            "contents": [
                {
                    "uri": uri,
                    "blob": base64.b64encode(content).decode("ascii"),
                    "mimeType": "application/octet-stream",
                }
            ]
        }
    return {
        "contents": [
            {
                "uri": uri,
                "text": content,
                "mimeType": "text/plain",
            }
        ]
    }


async def _mcp_prompts_list(params: dict[str, Any], manager: Any) -> dict[str, Any]:
    """Обработчик prompts/list."""
    return {"prompts": manager.registry.list_mcp_prompts()}


async def _mcp_prompts_get(params: dict[str, Any], manager: Any) -> dict[str, Any]:
    """Обработчик prompts/get."""
    name = params.get("name")
    if not name:
        raise ValueError("missing prompt name")

    arguments = params.get("arguments", {})
    messages = await manager.registry.get_prompt_messages(name, arguments)
    return {"messages": [m.to_dict() for m in messages]}


# Dispatch table: method → handler
_MCP_HANDLERS: dict[str, Callable[[dict[str, Any], Any], Awaitable[dict[str, Any]]]] = {
    "initialize": _mcp_initialize,
    "tools/list": _mcp_tools_list,
    "tools/call": _mcp_tools_call,
    "resources/list": _mcp_resources_list,
    "resources/read": _mcp_resources_read,
    "prompts/list": _mcp_prompts_list,
    "prompts/get": _mcp_prompts_get,
}


# ============================================================================
# MCP PROTOCOL ENDPOINT (JSON-RPC 2.0)
# ============================================================================


@router.post("/mcp", response_model=MCPJsonRpcResponse)
async def mcp_endpoint(
    request: MCPJsonRpcRequest,
    http_request: Request,
) -> MCPJsonRpcResponse:
    """
    Unified MCP JSON-RPC 2.0 endpoint (spec 2024-11-05).

    Аутентификация: через ``X-API-Key`` или ``Authorization: Bearer`` header
    (если настроен ``mcp_api_key`` в settings).

    Example request:
    ```json
    {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tools/list",
        "params": {}
    }
    ```
    """
    # Проверка аутентификации (если настроена)
    _check_mcp_auth(http_request)

    manager = get_manager()

    logger.debug("mcp.endpoint.called", method=request.method)

    # Dispatch по методу
    handler = _MCP_HANDLERS.get(request.method)
    if handler is None:
        return MCPJsonRpcResponse(
            id=request.id,
            error={
                "code": -32601,
                "message": f"Method not found: {request.method}",
            },
        )

    try:
        result = await handler(request.params or {}, manager)
        return MCPJsonRpcResponse(id=request.id, result=result)
    except ValueError as e:
        # Invalid params
        return MCPJsonRpcResponse(
            id=request.id,
            error={"code": -32602, "message": f"Invalid params: {e}"},
        )
    except Exception as e:
        logger.exception("mcp.endpoint.error", method=request.method)
        return MCPJsonRpcResponse(
            id=request.id,
            error={"code": -32603, "message": f"Internal error: {e}"},
        )


# ============================================================================
# BUNDLES MANAGEMENT ENDPOINTS (enable/disable)
# ============================================================================


@router.post("/bundles/{name}/enable")
async def enable_bundle_endpoint(
    name: str,
    user_ctx: UserContext = Depends(get_user_ctx),
) -> dict[str, Any]:
    """Включить MCP bundle (требует прав администратора)."""
    _check_admin_access(user_ctx)

    manager = get_manager()
    success = manager.enable_mcp_bundle(name)

    if success:
        await manager.initialize()

    logger.info(
        "mcp.bundle.enabled",
        bundle=name,
        success=success,
        user_id=str(user_ctx.user_id),
        workspace_id=user_ctx.workspace_id,
    )

    return {
        "success": success,
        "bundle": name,
        "message": f"Bundle '{name}' enabled" if success else f"Bundle '{name}' not found",
    }


@router.post("/bundles/{name}/disable")
async def disable_bundle_endpoint(
    name: str,
    user_ctx: UserContext = Depends(get_user_ctx),
) -> dict[str, Any]:
    """Отключить MCP bundle (требует прав администратора)."""
    _check_admin_access(user_ctx)

    manager = get_manager()
    success = manager.disable_mcp_bundle(name)

    logger.info(
        "mcp.bundle.disabled",
        bundle=name,
        success=success,
        user_id=str(user_ctx.user_id),
        workspace_id=user_ctx.workspace_id,
    )

    return {
        "success": success,
        "bundle": name,
        "message": f"Bundle '{name}' disabled" if success else f"Bundle '{name}' not found",
    }