app/mcp/tools/audio.py

from __future__ import annotations

from typing import Any

from mcp.server.fastmcp import FastMCP

from app.mcp.registry import MCPRegistry, MediaInput
from app.operations.compress import normalize_audio as normalize_audio_operation
from app.operations.convert import convert_audio as convert_audio_operation
from app.operations.extract_audio import remove_silence as remove_silence_operation
from app.operations.merge import merge_audio as merge_audio_operation
from app.operations.trim import trim_audio as trim_audio_operation


def register_audio_tools(server: FastMCP[Any], registry: MCPRegistry) -> None:
    """Register audio MCP tools backed by existing operations."""

    @server.tool(description="Convert audio to MP3, WAV, AAC, M4A, FLAC, OGG, or Opus.")
    async def convert_audio(input: MediaInput, format: str = "mp3") -> dict[str, Any]:
        """Convert one audio input."""
        return await registry.run_operation(
            "convert_audio", [input], {"format": format}, convert_audio_operation
        )

    @server.tool(description="Normalize audio loudness to a target LUFS value.")
    async def normalize_audio(input: MediaInput, target_lufs: float = -16) -> dict[str, Any]:
        """Normalize one audio input."""
        return await registry.run_operation(
            "normalize_audio",
            [input],
            {"target_lufs": target_lufs},
            normalize_audio_operation,
        )

    @server.tool(description="Trim audio by start, end, or duration.")
    async def trim_audio(
        input: MediaInput,
        start: float = 0,
        duration: float | None = None,
        end: float | None = None,
    ) -> dict[str, Any]:
        """Trim one audio input."""
        params = {"start": start}
        if duration is not None:
            params["duration"] = duration
        if end is not None:
            params["end"] = end
        return await registry.run_operation("trim_audio", [input], params, trim_audio_operation)

    @server.tool(description="Normalize and merge multiple audio files in sequence.")
    async def merge_audio(inputs: list[MediaInput]) -> dict[str, Any]:
        """Merge multiple audio inputs."""
        return await registry.run_operation("merge_audio", inputs, {}, merge_audio_operation)

    @server.tool(description="Remove leading, trailing, and internal silence from audio.")
    async def remove_silence(input: MediaInput, threshold: str = "-45dB") -> dict[str, Any]:
        """Remove silence through the existing audio filter operation."""
        return await registry.run_operation(
            "remove_silence",
            [input],
            {"threshold": threshold},
            remove_silence_operation,
        )