app/mcp/tools/probe.py

from __future__ import annotations

from typing import Any

from mcp.server.fastmcp import FastMCP

from app.mcp.registry import MCPRegistry, MediaInput


def register_probe_tools(server: FastMCP[Any], registry: MCPRegistry) -> None:
    """Register FFprobe metadata views through MediaProcessor.probe."""

    @server.tool(description="Return complete normalized FFprobe metadata.")
    async def probe_media(input: MediaInput) -> dict[str, Any]:
        """Probe all media streams and container metadata."""
        return await registry.run_probe("probe_media", input, "all")

    @server.tool(
        name="probe_video_metadata",
        description="Return FFprobe video stream, resolution, FPS, and rotation metadata.",
    )
    async def ffprobe_video_metadata(input: MediaInput) -> dict[str, Any]:
        """Return the video-specific FFprobe metadata view."""
        return await registry.run_probe("probe_video_metadata", input, "video")

    @server.tool(description="Return FFprobe audio stream metadata.")
    async def audio_metadata(input: MediaInput) -> dict[str, Any]:
        """Return audio streams, duration, and bitrate."""
        return await registry.run_probe("audio_metadata", input, "audio")

    @server.tool(description="Return all video, audio, and subtitle stream summaries.")
    async def stream_info(input: MediaInput) -> dict[str, Any]:
        """Return normalized stream information."""
        return await registry.run_probe("stream_info", input, "streams")

    @server.tool(description="Return container, tags, creation date, size, and duration.")
    async def container_info(input: MediaInput) -> dict[str, Any]:
        """Return normalized container information."""
        return await registry.run_probe("container_info", input, "container")