Spaces:
Running
Running
File size: 3,649 Bytes
fba6023 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | 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)
|