app/mcp/tools/system.py

from __future__ import annotations

import importlib.metadata
from typing import Any

from mcp.server.fastmcp import FastMCP

from app.mcp.registry import MCPRegistry


def register_system_tools(server: FastMCP[Any], registry: MCPRegistry) -> None:
    """Register health, cleanup, capability, and runtime utility tools."""

    @server.tool(description="Return media API and dependency health.")
    async def health() -> dict[str, Any]:
        """Return service health without loading expensive models."""
        return await registry.run_metadata_tool("health", registry.health_data)

    @server.tool(description="Run the existing expiry cleanup pass immediately.")
    async def cleanup_temp() -> dict[str, Any]:
        """Remove expired, inactive request directories."""

        async def action() -> dict[str, Any]:
            removed = await registry.container.cleanup.cleanup_expired()
            return {"removed_directories": removed}

        return await registry.run_metadata_tool("cleanup_temp", action)

    @server.tool(description="Return disk capacity and usage for temp and output storage.")
    async def disk_usage() -> dict[str, Any]:
        """Return storage usage statistics."""
        return await registry.run_metadata_tool("disk_usage", registry.disk_usage_data)

    @server.tool(description="Return CPU, memory, Python, and platform information.")
    async def system_info() -> dict[str, Any]:
        """Return process and host runtime information."""
        return await registry.run_metadata_tool("system_info", registry.system_info_data)

    @server.tool(description="List all media MCP operations by domain.")
    async def supported_operations() -> dict[str, Any]:
        """Return registered operation names."""
        return await registry.run_metadata_tool(
            "supported_operations", registry.supported_operations_data
        )

    @server.tool(description="List supported output, transcription, and model formats.")
    async def supported_formats() -> dict[str, Any]:
        """Return supported formats."""
        return await registry.run_metadata_tool(
            "supported_formats", registry.supported_formats_data
        )

    @server.tool(description="Return the installed FFmpeg version banner.")
    async def ffmpeg_version() -> dict[str, Any]:
        """Return FFmpeg version information."""

        async def action() -> dict[str, Any]:
            return {"version": await registry.container.ffmpeg.version()}

        return await registry.run_metadata_tool("ffmpeg_version", action)

    @server.tool(description="List faster-whisper models accepted by this server.")
    async def whisper_models() -> dict[str, Any]:
        """Return allowed and default Whisper models."""

        async def action() -> dict[str, Any]:
            return {
                "models": sorted(registry.container.whisper.ALLOWED_MODELS),
                "default": registry.container.settings.whisper_model,
                "device": "cpu",
                "compute_type": "int8",
            }

        return await registry.run_metadata_tool("whisper_models", action)

    @server.tool(description="Return the installed yt-dlp package version.")
    async def yt_dlp_version() -> dict[str, Any]:
        """Return yt-dlp version information."""

        async def action() -> dict[str, Any]:
            try:
                version = importlib.metadata.version("yt-dlp")
            except importlib.metadata.PackageNotFoundError:
                version = None
            return {"version": version}

        return await registry.run_metadata_tool("yt_dlp_version", action)