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)