list_tools.py

• Pfad: /var/www/mcp-servers/mcp-tasks/tools/task_tools/list_tools.py
• Namespace: -
• Zeilen: 102 | Größe: 3,578 Bytes
• Geändert: 2025-12-28 12:21:49 | Gescannt: 2025-12-31 10:22:15

Code Hygiene Score: 87

• Dependencies: 50 (25%)
• LOC: 100 (20%)
• Methods: 100 (20%)
• Secrets: 100 (15%)
• Classes: 100 (10%)
• Magic Numbers: 90 (10%)

Issues 1

Dependencies 10

• use json
• use time
• use sys
• use constants.MS_PER_SECOND
• use config.Config
• use base.get_repo
• use base.get_task_logger
• use base.validate_status
• use base.validate_type
• use base.log_tool_call

Funktionen 1

• register_list_tools() Zeile 15

Code

"""Task listing and statistics tools."""
import json
import time

import sys
sys.path.insert(0, "/var/www/mcp-servers/mcp_tasks")
sys.path.insert(0, "/var/www/mcp-servers/shared")

from constants import MS_PER_SECOND

from config import Config
from .base import get_repo, get_task_logger, validate_status, validate_type, log_tool_call


def register_list_tools(mcp):
    """Register task list and statistics tools."""

    repo = get_repo()
    logger = get_task_logger()

    @mcp.tool()
    def tasks_list(
        status: str | None = None,
        type: str | None = None,
        search: str | None = None,
        limit: int = 10,
        offset: int = 0,
        compact: bool = True,
    ) -> dict:
        """
        Listet Tasks mit optionalen Filtern auf.

        Args:
            status: Filter nach Status (pending, in_progress, completed, failed, cancelled)
            type: Filter nach Typ (human_task, ai_task, mixed)
            search: Volltextsuche in Titel/Beschreibung
            limit: Maximale Anzahl Ergebnisse (1-100, default: 10)
            offset: Offset für Pagination
            compact: True = nur id/title/status/type (Token-sparend), False = alle Felder

        Returns:
            Dict mit tasks, total, limit, offset
        """
        start = time.time()
        request_str = json.dumps({"status": status, "type": type, "search": search, "limit": limit})

        try:
            limit = max(1, min(limit, Config.MAX_RESULTS))

            valid, error = validate_status(status)
            if not valid:
                log_tool_call(logger, "tasks_list", request_str, "denied", error_message=error)
                return {"success": False, "error": error}

            valid, error = validate_type(type)
            if not valid:
                log_tool_call(logger, "tasks_list", request_str, "denied", error_message=error)
                return {"success": False, "error": error}

            tasks = repo.find_all(status=status, task_type=type, search=search, limit=limit, offset=offset)
            total = repo.count(status=status, task_type=type, search=search)

            duration = int((time.time() - start) * MS_PER_SECOND)
            log_tool_call(logger, "tasks_list", request_str, "success", duration)

            return {
                "success": True,
                "tasks": [t.to_dict_compact() if compact else t.to_dict() for t in tasks],
                "total": total,
                "limit": limit,
                "offset": offset,
                "compact": compact,
            }

        except Exception as e:
            duration = int((time.time() - start) * MS_PER_SECOND)
            log_tool_call(logger, "tasks_list", request_str, "error", duration, error_message=str(e))
            return {"success": False, "error": str(e)}

    @mcp.tool()
    def tasks_statistics() -> dict:
        """
        Holt Statistiken über alle Tasks.

        Returns:
            Statistiken: Tasks nach Status/Typ, Token-Verbrauch, Modell-Nutzung
        """
        start = time.time()

        try:
            stats = repo.get_statistics()

            duration = int((time.time() - start) * MS_PER_SECOND)
            log_tool_call(logger, "tasks_statistics", "", "success", duration)

            return {"success": True, "statistics": stats}

        except Exception as e:
            duration = int((time.time() - start) * MS_PER_SECOND)
            log_tool_call(logger, "tasks_statistics", "", "error", duration, error_message=str(e))
            return {"success": False, "error": str(e)}