app/mcp/resources.py

from __future__ import annotations

from typing import Any

from mcp.server.fastmcp import FastMCP

from app.mcp.registry import MCPRegistry


def register_resources(server: FastMCP[Any], registry: MCPRegistry) -> None:
    """Register structured media capability and runtime resources."""

    @server.resource(
        "media://operations",
        name="media_operations",
        description="All MCP media operations grouped by domain.",
        mime_type="application/json",
    )
    async def media_operations() -> dict[str, Any]:
        """Return registered media operations."""
        return await registry.safe_resource("operations", registry.supported_operations_data)

    @server.resource(
        "media://formats",
        name="media_formats",
        description="Supported media, transcription, and download formats.",
        mime_type="application/json",
    )
    async def media_formats() -> dict[str, Any]:
        """Return supported formats."""
        return await registry.safe_resource("formats", registry.supported_formats_data)

    @server.resource(
        "media://codecs",
        name="media_codecs",
        description="FFmpeg codec decode and encode capabilities.",
        mime_type="application/json",
    )
    async def media_codecs() -> dict[str, Any]:
        """Return installed FFmpeg codecs."""
        return await registry.safe_resource("codecs", registry.codecs_data)

    @server.resource(
        "media://health",
        name="media_health",
        description="Health of media processing dependencies.",
        mime_type="application/json",
    )
    async def media_health() -> dict[str, Any]:
        """Return dependency health."""
        return await registry.safe_resource("health", registry.health_data)

    @server.resource(
        "media://configuration",
        name="media_configuration",
        description="Non-secret media API runtime configuration.",
        mime_type="application/json",
    )
    async def media_configuration() -> dict[str, Any]:
        """Return safe configuration values."""
        return await registry.safe_resource("configuration", registry.configuration_data)

    @server.resource(
        "media://version",
        name="media_version",
        description="Application, MCP, yt-dlp, and Whisper package versions.",
        mime_type="application/json",
    )
    async def media_version() -> dict[str, Any]:
        """Return version information."""
        return await registry.safe_resource("version", registry.version_data)