Spaces:
Running
Running
File size: 1,792 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 | 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")
|