docs_reader.py

Code Hygiene Score: 88

Keine Issues gefunden.

Dependencies 10

Funktionen 2

Code

"""Read operations for documentation management."""

import json
import time
from datetime import datetime
from typing import Optional

from config import Config
from infrastructure.docs_repository import get_repository
from shared.domain import LogEntry
from shared.infrastructure import get_logger

from .constants import DEFAULT_LIMIT, MS_PER_SECOND


def _log_operation(
    logger,
    tool_name: str,
    request_data: dict,
    start_time: float,
    status: str = "success",
    error: Optional[str] = None,
) -> None:
    """Log an operation with timing information."""
    logger.log(
        LogEntry(
            timestamp=datetime.now(),
            client_name="mcp-docs",
            tool_name=tool_name,
            request=json.dumps(request_data),
            status=status,
            error_message=error,
            duration_ms=int((time.time() - start_time) * MS_PER_SECOND),
        )
    )


def register_reader_tools(mcp) -> None:
    """Register read-only documentation tools with the MCP server."""

    @mcp.tool()
    def docs_list(
        status: Optional[str] = None,
        parent_id: Optional[int] = None,
        search: Optional[str] = None,
        compact: bool = True,
        limit: int = DEFAULT_LIMIT,
    ) -> dict:
        """List all documents from the database."""
        start_time = time.time()
        logger = get_logger("mcp-docs", Config)
        repo = get_repository()

        try:
            docs = repo.find_all(
                status=status,
                parent_id=parent_id,
                search=search,
                limit=limit,
            )
            total = repo.count(status=status, parent_id=parent_id, search=search)

            result = {
                "success": True,
                "docs": [d.to_dict_compact() if compact else d.to_dict() for d in docs],
                "total": total,
                "limit": limit,
                "compact": compact,
            }

            _log_operation(
                logger,
                "docs_list",
                {"status": status, "parent_id": parent_id, "search": search, "limit": limit},
                start_time,
            )
            return result

        except Exception as e:
            _log_operation(
                logger,
                "docs_list",
                {"status": status, "search": search},
                start_time,
                "error",
                str(e),
            )
            return {"success": False, "error": str(e)}

    @mcp.tool()
    def docs_get(
        id: Optional[int] = None,
        path: Optional[str] = None,
        include_children: bool = False,
        include_breadcrumb: bool = False,
    ) -> dict:
        """Retrieve a document by ID or path."""
        start_time = time.time()
        logger = get_logger("mcp-docs", Config)
        repo = get_repository()

        try:
            if id is None and path is None:
                return {
                    "success": False,
                    "error": "Either id or path must be specified",
                }

            doc = None
            if id is not None:
                doc = repo.find_by_id(id)
            elif path is not None:
                doc = repo.find_by_path(path)

            if not doc:
                return {"success": False, "error": "Document not found"}

            result = {"success": True, "doc": doc.to_dict()}

            if include_children and doc.id:
                children = repo.find_children(doc.id)
                result["children"] = [c.to_dict_compact() for c in children]

            if include_breadcrumb and doc.id:
                result["breadcrumb"] = repo.get_breadcrumb(doc.id)

            _log_operation(logger, "docs_get", {"id": id, "path": path}, start_time)
            return result

        except Exception as e:
            _log_operation(
                logger,
                "docs_get",
                {"id": id, "path": path},
                start_time,
                "error",
                str(e),
            )
            return {"success": False, "error": str(e)}

    @mcp.tool()
    def docs_search(query: str, limit: int = DEFAULT_LIMIT) -> dict:
        """Full-text search in all documents."""
        start_time = time.time()
        logger = get_logger("mcp-docs", Config)
        repo = get_repository()

        try:
            docs = repo.find_all(search=query, limit=limit)

            _log_operation(
                logger, "docs_search", {"query": query, "limit": limit}, start_time
            )
            return {
                "success": True,
                "query": query,
                "results": [d.to_dict_compact() for d in docs],
                "count": len(docs),
            }

        except Exception as e:
            _log_operation(
                logger, "docs_search", {"query": query}, start_time, "error", str(e)
            )
            return {"success": False, "error": str(e)}

    @mcp.tool()
    def docs_statistics() -> dict:
        """Retrieve statistics about all documents."""
        start_time = time.time()
        logger = get_logger("mcp-docs", Config)
        repo = get_repository()

        try:
            stats = repo.get_statistics()

            _log_operation(logger, "docs_statistics", {}, start_time)
            return {"success": True, "statistics": stats}

        except Exception as e:
            _log_operation(
                logger, "docs_statistics", {}, start_time, "error", str(e)
            )
            return {"success": False, "error": str(e)}
← Übersicht