Upload 142 files

.dockerignore

@@ -0,0 +1,16 @@
+.git
+.venv
+.github
+.env
+.pytest_cache
+.ruff_cache
+__pycache__
+*.py[cod]
+*.log
+*.zip
+outputs/*
+!outputs/.gitkeep
+temp/*
+!temp/.gitkeep
+tests
+requirements-dev.txt

.env.example

@@ -0,0 +1,16 @@
+TEMP_DIR=./temp
+OUTPUT_DIR=./outputs
+TEMPLATE_DIR=./app/templates/categories
+MAX_UPLOAD_SIZE=1073741824
+WHISPER_MODEL=small
+CLEANUP_MINUTES=60
+CLEANUP_INTERVAL_SECONDS=60
+MAX_WORKERS=2
+LOG_LEVEL=INFO
+MAX_DURATION_SECONDS=21600
+MAX_RESOLUTION_PIXELS=33177600
+DOWNLOAD_TIMEOUT_SECONDS=300
+ALLOW_PRIVATE_URLS=false
+BASE_URL=
+FFMPEG_BINARY=ffmpeg
+FFPROBE_BINARY=ffprobe

.gitignore

@@ -1,5 +1,16 @@
+.env
+.venv/
+venv/
 __pycache__/
 *.py[cod]
+*.log
 .pytest_cache/
-temp/
-.env
+.ruff_cache/
+.mypy_cache/
+.coverage
+htmlcov/
+outputs/*
+!outputs/.gitkeep
+temp/*
+!temp/.gitkeep
+*.zip

Dockerfile

@@ -0,0 +1,43 @@
+FROM python:3.10-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PIP_NO_CACHE_DIR=1 \
+    PIP_DISABLE_PIP_VERSION_CHECK=1 \
+    HF_HOME=/home/user/.cache/huggingface \
+    TEMP_DIR=/app/temp \
+    OUTPUT_DIR=/app/outputs \
+    PORT=7860
+
+RUN apt-get update \
+    && apt-get install --no-install-recommends -y \
+        ca-certificates \
+        ffmpeg \
+        imagemagick \
+        libglib2.0-0 \
+        libgomp1 \
+        libsndfile1 \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+COPY requirements.txt ./
+RUN python -m pip install --upgrade pip \
+    && python -m pip install -r requirements.txt
+
+RUN useradd --create-home --uid 1000 user \
+    && mkdir -p /app/temp /app/outputs /home/user/.cache/huggingface \
+    && chown -R user:user /app /home/user
+
+COPY --chown=user:user app ./app
+COPY --chown=user:user api services operations workers core models ./
+COPY --chown=user:user main.py README.md ./
+
+USER user
+
+EXPOSE 7860
+
+HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 \
+    CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:7860/health', timeout=4)" || exit 1
+
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860", "--workers", "1", "--proxy-headers", "--forwarded-allow-ips", "*"]

README.md

@@ -0,0 +1,995 @@
+---
+title: Enterprise Media Processing API
+emoji: 🎬
+colorFrom: blue
+colorTo: purple
+sdk: docker
+app_port: 7860
+pinned: false
+---
+
+# Enterprise Media Processing API
+
+A production-oriented, CPU-optimized REST, Model Context Protocol (MCP), and YAML workflow API for FFmpeg, FFprobe, yt-dlp, and faster-whisper. It is designed to run unchanged as a Hugging Face Docker Space and to act as a reusable media backend for AI clients, n8n, and other automation systems.
+
+The service accepts multipart uploads, JSON URLs, JSON Base64, n8n binary objects, and streamed raw request bodies. Every input becomes an `InputMedia` before it enters the operation layer, so operations never need to know how media arrived.
+
+## Architecture
+
+```text
+REST /v1/*                       MCP stdio or /mcp/
+     │                                  │
+     ├───────────────┬──────────────────┘
+                     ▼
+            request-ID + JSON logging
+                     │
+                InputResolver
+          ├── streamed multipart/raw upload
+          ├── Base64/n8n decoding
+          ├── HTTP(S) streaming downloader + SSRF guard
+          └── automatic yt-dlp extractor selection
+                    │
+                    ▼
+             normalized InputMedia
+                    │
+          FFprobe validation/metadata
+                    │
+      optional versioned Template Engine
+                    │
+          operation (video/audio/image)
+                    │
+       concurrency-limited FFmpeg/Whisper
+                    │
+       published output + streaming URL
+                    │
+          background expiry cleanup
+```
+
+REST routes and MCP tools contain transport concerns only. Both use the same dependency container, `InputResolver`, `MediaProcessor`, operation functions, and infrastructure services. `MediaProcessor` orchestrates resolution, probing, validation, publishing, and metrics; operation modules build safe FFmpeg argument arrays; infrastructure services own subprocesses, downloads, model loading, and storage. No command is executed through a shell.
+
+## Project layout
+
+```text
+media-api/
+├── app/
+│   ├── api/              # versioned, thin FastAPI routes
+│   ├── core/             # settings, errors, logging, response models
+│   ├── mcp/              # MCP server, registry, tools, resources, prompts
+│   ├── models/           # InputMedia and request/result models
+│   ├── operations/       # reusable FFmpeg operation functions
+│   ├── services/         # FFmpeg, FFprobe, resolver, downloads, Whisper
+│   ├── templates/        # YAML schemas, loader, registry, executor, categories
+│   ├── workers/          # asynchronous expiry cleanup worker
+│   └── container.py      # dependency construction
+├── api/                   # requested top-level import compatibility
+├── services/              # requested top-level import compatibility
+├── operations/            # requested top-level import compatibility
+├── workers/               # requested top-level import compatibility
+├── core/                  # requested top-level import compatibility
+├── models/                # requested top-level import compatibility
+├── outputs/              # published, expiring request outputs
+├── temp/                 # request/{uploads,outputs,logs}
+├── tests/
+├── Dockerfile
+├── requirements.txt
+├── requirements-dev.txt
+├── .env.example
+└── main.py
+```
+
+The top-level `api/`, `services/`, `operations/`, `workers/`, `core/`, and `models/` packages mirror the canonical `app/` modules as import-compatible entry points for integrations that use the requested layout. Runtime composition uses the single implementation under `app/`, so business logic is not duplicated.
+
+Each request creates `TEMP_DIR/<uuid>/{uploads,outputs,logs}`. Completed files are atomically moved to `OUTPUT_DIR/<uuid>` so they can be streamed. Both locations expire after `CLEANUP_MINUTES`; active requests are protected from the cleanup worker.
+
+## Run locally
+
+FFmpeg, FFprobe, and ImageMagick must be installed on the host.
+
+```bash
+python3.10 -m venv .venv
+. .venv/bin/activate
+pip install -r requirements.txt
+cp .env.example .env
+uvicorn main:app --host 0.0.0.0 --port 7860
+```
+
+Open `http://localhost:7860/docs` for OpenAPI/Swagger or `http://localhost:7860/redoc` for ReDoc.
+
+### Docker
+
+```bash
+docker build -t media-api .
+docker run --rm -p 7860:7860 --env-file .env media-api
+```
+
+The image uses `python:3.10-slim`, installs FFmpeg, FFprobe, ImageMagick and the system libraries needed by CTranslate2/faster-whisper, clears apt and pip caches, runs as UID 1000, exposes port 7860, and includes a container health check.
+
+## Deploy to Hugging Face Spaces
+
+1. Create a new Space and select **Docker** as the SDK.
+2. Push the contents of this directory to the Space repository. Keep the YAML block at the top of this README; `app_port` is already `7860`.
+3. Add environment variables under **Settings → Variables and secrets** if the defaults need changing. Do not store secrets in `.env` in Git.
+4. Wait for the Docker build. The first Whisper call downloads the selected model to the Hugging Face cache. Persistent storage is optional, but avoids downloading models again after a cold rebuild.
+5. Check `https://<owner>-<space>.hf.space/health`, then use `/docs` or call `/v1/*` from n8n.
+
+If using the supplied `media-api-huggingface.zip`, extract it first and push the extracted files so `Dockerfile` and this `README.md` are at the Space repository root. Do not commit the ZIP as the only repository file.
+
+Use one Uvicorn process in a CPU Space. FFmpeg and Whisper concurrency is managed in-process by `MAX_WORKERS`; multiple Uvicorn workers duplicate Whisper models and memory.
+
+## Model Context Protocol (MCP)
+
+MCP is an open protocol that lets an AI client discover and call typed tools, read application resources, and use reusable prompts. This project uses the official Python MCP SDK and exposes the same processing implementation through both interfaces:
+
+```text
+FastAPI REST routes ─┐
+                     ├─ InputResolver → MediaProcessor → shared services/operations
+MCP tools ───────────┘                         ├─ FFmpeg / FFprobe
+                                               ├─ yt-dlp
+                                               └─ faster-whisper
+```
+
+No FFmpeg command, download implementation, transcription implementation, input parser, logger, or cleanup implementation is duplicated in `app/mcp`. The MCP registry creates a UUID for each tool call and reuses the existing request workspace, validation, structured logging, publication, and expiry cleanup behavior. MCP has no authentication, as requested; use a private Space or an external gateway if access control or quotas are required.
+
+### MCP transports
+
+The normal Docker command starts FastAPI on port `7860` and serves both REST and MCP:
+
+- REST and OpenAPI: `https://<owner>-<space>.hf.space/v1/*` and `/docs`
+- Streamable HTTP MCP: `https://<owner>-<space>.hf.space/mcp/`
+
+The trailing slash on `/mcp/` is recommended. Streamable HTTP is stateless and returns JSON responses while remaining compliant with the MCP transport. The application owns the MCP session-manager lifespan, so it works correctly even though the MCP ASGI application is mounted under FastAPI.
+
+For a local stdio client, run from the project directory:
+
+```bash
+python -m app.mcp.server
+```
+
+Structured logs are sent to stderr in stdio mode so JSON-RPC messages on stdout are never corrupted. A standalone MCP-only HTTP process is also available for development:
+
+```bash
+python -m app.mcp.server --transport streamable-http
+```
+
+Use the normal `uvicorn main:app ...` command in Hugging Face because it exposes both interfaces together.
+
+### MCP client configuration
+
+Claude Desktop and other clients that accept the conventional `mcpServers` JSON can launch stdio directly. Replace the paths with absolute paths on the client machine:
+
+```json
+{
+  "mcpServers": {
+    "enterprise-media": {
+      "command": "/absolute/path/media-api/.venv/bin/python",
+      "args": ["-m", "app.mcp.server"],
+      "cwd": "/absolute/path/media-api",
+      "env": {
+        "TEMP_DIR": "/absolute/path/media-api/temp",
+        "OUTPUT_DIR": "/absolute/path/media-api/outputs",
+        "WHISPER_MODEL": "small",
+        "MAX_WORKERS": "2"
+      }
+    }
+  }
+}
+```
+
+Clients that support remote Streamable HTTP can use:
+
+```json
+{
+  "mcpServers": {
+    "enterprise-media": {
+      "url": "https://OWNER-SPACE.hf.space/mcp/"
+    }
+  }
+}
+```
+
+Common client locations and connection choices are:
+
+| Client | Configuration |
+|---|---|
+| Claude Desktop | Add the stdio `mcpServers` entry to `claude_desktop_config.json`, then restart Claude. |
+| ChatGPT | Add a remote custom connector in developer/connector settings with the Space `/mcp/` URL. The Space must be reachable from ChatGPT. |
+| Cursor | Put the remote `mcpServers` entry in `.cursor/mcp.json`, or use the stdio entry for local files. |
+| VS Code | Add an HTTP server under `servers` in `.vscode/mcp.json`: `{"type":"http","url":"https://OWNER-SPACE.hf.space/mcp/"}`. |
+| Continue | Add an MCP server in Continue configuration using the Streamable HTTP URL or the stdio command above. |
+| Cline | Open **MCP Servers → Configure** and add the `mcpServers` JSON entry. |
+| Windsurf | Add the same entry in Windsurf MCP settings (`mcp_config.json`). |
+
+Client configuration keys can vary between releases; select **Streamable HTTP**, not legacy SSE, when a client asks for the transport. No authorization header is needed.
+
+For a Docker-based stdio client, override the image command and keep stdin open:
+
+```bash
+docker run --rm -i \
+  -v "$PWD/temp:/app/temp" \
+  -v "$PWD/outputs:/app/outputs" \
+  media-api python -m app.mcp.server
+```
+
+In stdio mode, consume the returned `output_file` locally. In Streamable HTTP mode, set `BASE_URL=https://<owner>-<space>.hf.space` so `download_url` is absolute.
+
+### MCP media input
+
+Every MCP media tool delegates to the existing `InputResolver`. A `MediaInput` accepts exactly one source:
+
+```json
+{"url": "https://cdn.example.com/video.mp4"}
+```
+
+```json
+{
+  "base64": "data:audio/wav;base64,UklGR...",
+  "filename": "speech.wav",
+  "mime_type": "audio/wav"
+}
+```
+
+```json
+{
+  "binary": {
+    "data": "AAAAHGZ0eXBpc29t...",
+    "fileName": "clip.mp4",
+    "mimeType": "video/mp4"
+  }
+}
+```
+
+```json
+{"temp_path": "/app/outputs/<request-uuid>/converted.mp4"}
+```
+
+The `binary` object accepts n8n properties such as `data`, `file`, `video`, or `audio`. `temp_path` is copied into the new request workspace and is accepted only when it resolves below configured `TEMP_DIR` or `OUTPUT_DIR`; arbitrary host paths and traversal are rejected. URLs use the same SSRF protection and automatic yt-dlp detection as REST.
+
+MCP tool arguments are JSON by protocol, so a client file attachment must be represented as a URL, Base64/data URI, n8n binary object, or a managed `temp_path`. Native `multipart/form-data` and streamed `application/octet-stream` remain available on every corresponding REST endpoint and enter the same resolver and service layer.
+
+Single-input tools use an `input` argument. Multi-input tools use `inputs`; composition tools use descriptive arguments such as `video`, `audio`, `watermark`, `overlay`, or `subtitles`. A successful file-producing call returns both the managed server `output_file` and normal REST `download_url`:
+
+```json
+{
+  "success": true,
+  "request_id": "43b32630-9b32-4b27-b038-a39fbfd4b803",
+  "processing_time": 1.52,
+  "output_file": "/app/outputs/43b32630-9b32-4b27-b038-a39fbfd4b803/compressed.mp4",
+  "download_url": "https://OWNER-SPACE.hf.space/v1/media/43b32630-4b27-b038-a39fbfd4b803/compressed.mp4",
+  "metadata": {}
+}
+```
+
+Errors use the same safe application codes and never expose raw Python exceptions:
+
+```json
+{
+  "success": false,
+  "request_id": "43b32630-9b32-4b27-b038-a39fbfd4b803",
+  "processing_time": 0.01,
+  "error": {
+    "code": "INVALID_INPUT",
+    "message": "The managed temporary file does not exist",
+    "details": null
+  }
+}
+```
+
+### Available MCP tools
+
+All tools return the structured envelope above. Defaults shown here match the callable schemas advertised to MCP clients.
+
+| Video tool | Purpose and principal options |
+|---|---|
+| `compress_video` | CPU-optimized compression; `crf=28`, `preset=medium`, `format=mp4`, optional `max_width` and `bitrate`. |
+| `resize_video` | Resize using `width`, `height`, `fit` (`contain`, `cover`, `fill`), and `format`. |
+| `crop_video` | Crop using required `width`/`height` and optional `x`/`y`. |
+| `trim_video` | Trim with `start` plus `duration` or `end`. |
+| `convert_video` | Convert to a supported video `format`. |
+| `merge_videos` | Normalize and merge `inputs`; optional output `width`/`height`. |
+| `concat_videos` | Concatenate `inputs`; optional `stream_copy` for already-compatible streams. |
+| `watermark_video` | Watermark `video` with an image; `position`, `opacity`, and `watermark_scale`. |
+| `overlay_video` | Overlay media on `video`; `position`, `opacity`, and `watermark_scale`. |
+| `extract_audio` | Extract a video audio stream to `format` (default MP3). |
+| `replace_audio` | Replace the audio in `video` with the supplied `audio`. |
+| `remove_audio` | Remove all audio streams. |
+| `generate_thumbnail` | Create a JPEG at `timestamp`; optional FFmpeg `quality`. |
+| `extract_frames` | Extract at `fps`, optionally bounded by `max_frames`, and return a ZIP. |
+| `burn_subtitles` | Burn SRT, VTT, ASS, or SSA `subtitles` into `video`; optional ASS `style`. |
+
+| Audio tool | Purpose and principal options |
+|---|---|
+| `convert_audio` | Convert to MP3, WAV, AAC, M4A, FLAC, OGG, or Opus. |
+| `normalize_audio` | EBU loudness normalization with `target_lufs=-16`. |
+| `trim_audio` | Trim with `start` plus `duration` or `end`. |
+| `merge_audio` | Normalize and merge multiple `inputs`. |
+| `remove_silence` | Remove silence using a threshold such as `-45dB`. |
+
+| Image tool | Purpose and principal options |
+|---|---|
+| `image_to_video` | Turn one still image into H.264 video; `duration` and `fps`. |
+| `slideshow` | Create a video from `inputs`; `duration_per_image`, `width`, `height`, and `fps`. |
+| `watermark_image` | Watermark an image; `position`, `opacity`, `watermark_scale`, and output `format`. |
+| `resize_image` | Resize with `width`, `height`, `fit`, and output `format`. |
+
+| Whisper tool | Purpose and principal options |
+|---|---|
+| `transcribe` | Transcribe speech; optional `model`, `language`, `output_format`, `beam_size`, and `vad_filter`. |
+| `translate` | Translate speech to English with the same options. |
+| `detect_language` | Detect spoken language using an optional `model`. |
+| `generate_srt` | Generate SRT with optional `model`, `language`, and `task`. |
+| `generate_vtt` | Generate WebVTT with optional `model`, `language`, and `task`. |
+| `generate_json` | Generate JSON transcript segments with optional `model`, `language`, and `task`. |
+
+| yt-dlp tool | Purpose and principal options |
+|---|---|
+| `download_video` | Download best video; optional yt-dlp `format_selector`. |
+| `download_audio` | Download/convert audio; `audio_format=mp3` and optional `format_selector`. |
+| `video_metadata` | Extract platform metadata without downloading media. |
+| `playlist_metadata` | Extract flat playlist metadata, bounded by `max_entries` (1–1000). |
+| `list_formats` | List normalized available audio/video formats. |
+
+| FFprobe tool | Purpose |
+|---|---|
+| `probe_media` | Complete normalized metadata. |
+| `probe_video_metadata` | Video codec, dimensions, FPS, rotation, duration, and streams. |
+| `audio_metadata` | Audio streams, duration, and bitrate. |
+| `stream_info` | Video, audio, and subtitle stream summaries. |
+| `container_info` | Container, tags, creation date, size, bitrate, and duration. |
+
+MCP tool names must be unique. The yt-dlp metadata tool retains the requested `video_metadata` name; the FFprobe view is exposed as `probe_video_metadata` to disambiguate it.
+
+| Utility tool | Purpose |
+|---|---|
+| `health` | Application and dependency health without loading Whisper. |
+| `cleanup_temp` | Run the existing expired-workspace cleanup pass. |
+| `disk_usage` | Capacity and use for temp and output storage. |
+| `system_info` | CPU, memory, Python, platform, and process details. |
+| `supported_operations` | Tool names grouped by domain. |
+| `supported_formats` | Media, transcription, model, and yt-dlp formats. |
+| `ffmpeg_version` | Installed FFmpeg version banner. |
+| `whisper_models` | Allowed/default models and CPU compute configuration. |
+| `yt_dlp_version` | Installed yt-dlp package version. |
+
+| Template tool | Purpose |
+|---|---|
+| `list_templates` | Discover all dynamically loaded template versions, optionally filtered by category. |
+| `template_details` | Return complete metadata and pipeline documentation for `id`, `id@version`, or `id@latest`. |
+| `run_template` | Resolve one or more media inputs and execute a versioned YAML workflow. |
+| `template_categories` | Return categories discovered from loaded YAML templates. |
+
+### Available MCP resources
+
+Resources return structured JSON, including a safe `success` indicator:
+
+| Resource URI | Contents |
+|---|---|
+| `media://operations` | Registered operations grouped by video, audio, image, Whisper, yt-dlp, probe, templates, and system. |
+| `media://formats` | Supported containers, transcription formats, models, and audio download formats. |
+| `media://codecs` | Installed FFmpeg codecs and encode/decode capabilities. |
+| `media://health` | Application version and dependency availability. |
+| `media://configuration` | Non-secret runtime limits and paths. |
+| `media://version` | Application, MCP SDK, yt-dlp, and faster-whisper versions. |
+
+### Available MCP prompts
+
+| Prompt | Workflow |
+|---|---|
+| `compress_for_social_media` | Probe, resize as appropriate, and call `compress_video`. |
+| `youtube_to_mp3` | Call `download_audio` as MP3. |
+| `download_and_transcribe` | Download audio, then call `transcribe` on its managed output. |
+| `generate_subtitles` | Call `generate_srt` or `generate_vtt`. |
+| `extract_audio` | Call the `extract_audio` tool in the requested format. |
+| `make_thumbnail` | Call `generate_thumbnail` at a timestamp. |
+| `probe_media` | Call `probe_media` and summarize streams/container data. |
+| `instagram_reel` | Probe, resize to 1080×1920, then compress. |
+| `tiktok_video` | Resize vertically and compress for TikTok. |
+| `podcast_audio` | Normalize to -16 LUFS and optionally convert. |
+
+### Example MCP calls
+
+AI clients construct the JSON-RPC envelope automatically. The tool argument payloads are:
+
+Compress a remote video:
+
+```json
+{
+  "name": "compress_video",
+  "arguments": {
+    "input": {"url": "https://cdn.example.com/input.mp4"},
+    "crf": 28,
+    "preset": "veryfast",
+    "format": "mp4",
+    "max_width": 1280
+  }
+}
+```
+
+Transcribe Base64 audio:
+
+```json
+{
+  "name": "transcribe",
+  "arguments": {
+    "input": {
+      "base64": "data:audio/mpeg;base64,SUQz...",
+      "filename": "meeting.mp3",
+      "mime_type": "audio/mpeg"
+    },
+    "model": "small",
+    "output_format": "json"
+  }
+}
+```
+
+Download a YouTube video:
+
+```json
+{
+  "name": "download_video",
+  "arguments": {
+    "url": "https://www.youtube.com/watch?v=VIDEO_ID"
+  }
+}
+```
+
+Generate subtitles:
+
+```json
+{
+  "name": "generate_srt",
+  "arguments": {
+    "input": {"url": "https://cdn.example.com/interview.mp4"},
+    "language": "en",
+    "task": "transcribe"
+  }
+}
+```
+
+Probe media:
+
+```json
+{
+  "name": "probe_media",
+  "arguments": {
+    "input": {"url": "https://cdn.example.com/input.mp4"}
+  }
+}
+```
+
+## Enterprise Template Engine
+
+The Template Engine is a versioned orchestration layer above the existing media operations. At startup, `TemplateLoader` recursively scans `TEMPLATE_DIR` for `.yaml` and `.yml` files, parses them with `yaml.safe_load_all`, validates every definition, and builds an immutable `TemplateRegistry`. REST and MCP resolve media through the same `InputResolver`, then `TemplateExecutor` chains allow-listed operation functions through `OperationExecutor`:
+
+```text
+REST /v1/templates/run ─┐
+                       ├─ InputResolver → Template Registry/Validator
+MCP run_template ──────┘                         │
+                                                ▼
+                                 parameter substitution + pipeline
+                                                │
+                                                ▼
+                              existing operation functions and services
+```
+
+Intermediate artifacts stay inside `TEMP_DIR/<request-id>/outputs`; only the final result is published. Every generated media artifact is FFprobed before the next step when applicable. Logs include the template ID/version, resolved parameters, request ID, operations, wall and CPU time, memory, and output bytes. Existing FFmpeg logging still records commands and bounded stdout/stderr.
+
+### Template folders and built-ins
+
+```text
+app/templates/
+├── __init__.py
+├── loader.py
+├── registry.py
+├── executor.py
+├── schema.py
+├── validator.py
+├── models.py
+└── categories/
+    ├── social/
+    ├── faceless/
+    ├── motivation/
+    ├── lyrics/
+    ├── podcast/
+    ├── subtitles/
+    ├── youtube/
+    ├── conversion/
+    ├── branding/
+    ├── utility/
+    └── custom/
+```
+
+The distribution contains 71 versioned workflows:
+
+| Category | Templates |
+|---|---|
+| Social | `youtube_shorts`, `tiktok_hd`, `facebook_reel`, `instagram_reel`, `linkedin_video`, `twitter_video`, `whatsapp_status` |
+| Faceless | `reddit_story`, `ai_story`, `movie_recap`, `history_short`, `true_crime`, `did_you_know`, `top10_video`, `facts_video`, `book_summary`, `finance_short`, `crypto_news`, `tech_news` |
+| Motivation | `motivational_video`, `morning_motivation`, `business_motivation`, `gym_motivation`, `success_quotes`, `stoic_quotes`, `daily_quotes`, `affirmations` |
+| Lyrics | `lyrics_basic`, `karaoke`, `spotify_style`, `cinematic_lyrics`, `neon_lyrics`, `music_video` |
+| Podcast | `podcast_video`, `podcast_short`, `audiogram`, `waveform_video` |
+| Subtitles/AI | `auto_subtitles`, `translate_video`, `transcribe`, `transcribe_srt`, `transcribe_vtt`, `transcribe_json`, `youtube_to_transcript` |
+| YouTube | `youtube_to_mp3`, `youtube_to_audio`, `youtube_to_shorts`, `youtube_to_podcast`, `download_only` |
+| Branding | `company_branding`, `creator_branding`, `watermark`, `intro_outro`, `logo_animation` |
+| Conversion | `mp4`, `mov`, `avi`, `webm`, `gif`, `mp3`, `wav`, `aac`, `flac` |
+| Utility | `thumbnail_pack`, `extract_frames`, `extract_audio`, `merge_videos`, `concat_videos`, `compress_max`, `compress_balanced`, `compress_mobile` |
+
+Faceless and motivation templates format already assembled source media; they do not pretend to generate narration, images, or copyrighted source content. Lyrics templates consume supplied timed subtitle files. `waveform_video` consumes a pre-rendered waveform artwork image plus audio, reusing the existing image/video and audio replacement operations.
+
+### Template schema and parameters
+
+Each template includes the required metadata (`id`, `name`, `description`, `category`, `author`, `version`, `tags`, `estimated_runtime`, `supported_inputs`, and `supported_outputs`), parameter documentation, at least one pipeline operation, an output contract, and examples:
+
+```yaml
+id: instagram_reel_custom
+name: Instagram Reel Custom
+category: custom
+description: Resize and compress video for a vertical Instagram Reel.
+author: Your Team
+version: 1
+tags: [instagram, vertical]
+estimated_runtime: medium
+supported_inputs: [video, url, ytdlp]
+supported_outputs: [mp4]
+
+parameters:
+  crf:
+    type: integer
+    default: 23
+    minimum: 0
+    maximum: 51
+  width:
+    type: integer
+    default: 1080
+    minimum: 2
+
+pipeline:
+  - operation: resize
+    width: "{{ width }}"
+    height: 1920
+    fit: cover
+  - operation: fps
+    value: 30
+  - operation: compress
+    crf: "{{ crf }}"
+    preset: veryfast
+    format: mp4
+
+output:
+  format: mp4
+
+examples:
+  - input: {url: "https://example.com/input.mp4"}
+    parameters: {crf: 23, width: 1080}
+```
+
+Parameter types are `string`, `integer`, `number`, `boolean`, `array`, and `object`. Definitions can use `required`, `default`, `enum`, `minimum`, `maximum`, `min_length`, and `max_length`. Unknown parameters and type coercion are rejected: the string `"23"` is not accepted for an integer parameter.
+
+Variables use only `{{ parameter_name }}` syntax—expressions and executable template code are not supported. When a YAML value consists only of a variable, substitution preserves the declared type, so an integer remains an integer and a boolean can control `when`:
+
+```yaml
+parameters:
+  add_logo: {type: boolean, default: false}
+pipeline:
+  - operation: watermark_video
+    when: "{{ add_logo }}"
+    inputs: [current, original:1]
+```
+
+Multiple-input and multi-artifact workflows can select `current`, `original`, `original:N`, `originals`, or `artifact:name`. A step can save its result for a later step:
+
+```yaml
+pipeline:
+  - operation: transcribe
+    output_format: srt
+    save_as: captions
+  - operation: burn_subtitles
+    inputs: [original:0, artifact:captions]
+```
+
+Supported YAML operation names are allow-listed aliases over existing implementations. They cover video compression/resizing/cropping/trimming/rotation/merging/concatenation/conversion/overlays/subtitles/effects, audio extraction/conversion/normalization/trimming/merging/filters, image conversion/composition, Whisper `transcribe`/`translate`, and resolver-backed `download`. No YAML value is used as a command name or shell string.
+
+### Validation and versioning
+
+The application refuses to start when a configured YAML file has invalid syntax, missing metadata, duplicate ID/version, invalid defaults, undeclared variables, unsafe selectors, unsupported operations, or an output format outside `supported_outputs`. Runtime parameters are validated again before any operation runs.
+
+References support stable version selection:
+
+- `youtube_shorts` resolves the highest installed version.
+- `youtube_shorts@latest` explicitly resolves the highest installed version.
+- `youtube_shorts@1` remains pinned to version 1 when version 2 is added.
+
+Keep old YAML documents when introducing a new version so existing automations remain reproducible. A YAML file may contain one template, multiple `---` documents, or a top-level `templates` list.
+
+### Template REST API
+
+| Method | Endpoint | Purpose |
+|---|---|---|
+| `GET` | `/v1/templates` | List all versions with parameters, metadata, and examples; optional `?category=social`. |
+| `GET` | `/v1/templates/categories` | List categories discovered from YAML. |
+| `GET` | `/v1/templates/{reference}` | Get full details and pipeline for an ID/version reference. |
+| `POST` | `/v1/templates/run` | Resolve media and execute a template. |
+
+List and inspect:
+
+```bash
+curl http://localhost:7860/v1/templates
+curl http://localhost:7860/v1/templates/instagram_reel@1
+```
+
+Run with a direct or yt-dlp-supported URL:
+
+```bash
+curl -X POST http://localhost:7860/v1/templates/run \
+  -H 'Content-Type: application/json' \
+  -d '{
+    "template":"instagram_reel@latest",
+    "input":{"url":"https://example.com/input.mp4"},
+    "parameters":{"crf":21}
+  }'
+```
+
+Run with multipart media; `parameters` is a JSON form field:
+
+```bash
+curl -X POST http://localhost:7860/v1/templates/run \
+  -F 'file=@input.mp4' \
+  -F 'template=youtube_shorts@1' \
+  -F 'parameters={"crf":23,"max_duration":60}'
+```
+
+Multiple-input templates collect uploaded files in multipart order or use an `inputs` JSON array. For `company_branding`, input 0 is video and input 1 is the logo; for `intro_outro`, send intro, main, and outro in that order.
+
+### Template MCP examples
+
+Discover templates:
+
+```json
+{"name":"list_templates","arguments":{"category":"social"}}
+```
+
+Inspect a pinned version:
+
+```json
+{"name":"template_details","arguments":{"template":"youtube_shorts@1"}}
+```
+
+Execute a URL template:
+
+```json
+{
+  "name": "run_template",
+  "arguments": {
+    "template": "youtube_to_mp3@latest",
+    "input": {"url": "https://www.youtube.com/watch?v=VIDEO_ID"},
+    "parameters": {}
+  }
+}
+```
+
+For multiple inputs, omit `input` and provide `inputs`. MCP returns the same structured request ID, `output_file`, `download_url`, metadata, and safe error contract as the other tools.
+
+### Template n8n examples
+
+For an n8n URL workflow, configure an **HTTP Request** node with `POST`, JSON body, and `/v1/templates/run`:
+
+```javascript
+{
+  "template": "compress_mobile@1",
+  "input": {"url": "{{$json.media_url}}"},
+  "parameters": {"crf": 28}
+}
+```
+
+For n8n binary JSON input:
+
+```javascript
+{
+  "template": "mp3@1",
+  "input": {
+    "binary": {
+      "data": "{{$binary.audio.data}}",
+      "fileName": "{{$binary.audio.fileName}}",
+      "mimeType": "{{$binary.audio.mimeType}}"
+    }
+  },
+  "parameters": {}
+}
+```
+
+For large n8n files, send the binary property as multipart under any file field, plus text fields `template` and JSON `parameters`. The existing resolver supports `binary.data`, `binary.file`, `binary.video`, `binary.audio`, Base64/data URIs, raw bytes, direct HTTP(S), and automatic yt-dlp selection for templates exactly as it does for normal operations.
+
+### Adding custom templates
+
+1. Add a `.yaml` or `.yml` file below `app/templates/categories/custom/` or a mounted `TEMPLATE_DIR`.
+2. Choose a unique `id` and positive integer `version`; never replace an old version used by automation.
+3. Declare strict parameter types/defaults and complete metadata.
+4. Compose allow-listed existing operations. Use `save_as` and input selectors for branched workflows.
+5. Restart the process. Startup scanning automatically validates, registers, exposes, and documents the template through REST and MCP—no Python registration change is required.
+6. If a genuinely new media primitive is needed, implement it once in `app/operations`, validate its arguments, add it to `OPERATION_BINDINGS`, and then reference it from any number of YAML workflows.
+
+Set `TEMPLATE_DIR` to an external mounted directory to operate a private workflow catalog without modifying application code. Invalid catalogs fail closed and are never partially executed.
+
+## Input contract
+
+All media-processing and probe endpoints use the same resolver. Operation parameters can be top-level JSON fields, multipart text fields, an `options` JSON object, or query parameters. Body/form values take precedence over query parameters.
+
+### Multipart file
+
+```bash
+curl -X POST http://localhost:7860/v1/video/compress \
+  -F 'file=@input.mp4' \
+  -F 'crf=28' \
+  -F 'preset=veryfast'
+```
+
+Multiple-input operations accept repeated or differently named file fields; every uploaded file part is collected in form order.
+
+```bash
+curl -X POST http://localhost:7860/v1/video/watermark \
+  -F 'video=@input.mp4' \
+  -F 'watermark=@logo.png' \
+  -F 'position=bottom-right' \
+  -F 'opacity=0.7'
+```
+
+### JSON URL
+
+```bash
+curl -X POST http://localhost:7860/v1/video/resize \
+  -H 'Content-Type: application/json' \
+  -d '{"url":"https://cdn.example.com/video.mp4","width":1280,"height":720,"fit":"contain"}'
+```
+
+URLs supported by a non-generic yt-dlp extractor automatically use yt-dlp. This covers YouTube, TikTok, Instagram, Facebook, X/Twitter, Reddit, Vimeo, SoundCloud, and the other sites supported by the installed yt-dlp release. Ordinary direct HTTP(S) media is downloaded in bounded chunks. Redirect targets are revalidated, credentials in URLs are rejected, and private/loopback/link-local addresses are blocked unless `ALLOW_PRIVATE_URLS=true`.
+
+For multiple remote inputs:
+
+```json
+{
+  "inputs": [
+    {"url": "https://cdn.example.com/part-1.mp4"},
+    {"url": "https://cdn.example.com/part-2.mp4"}
+  ],
+  "width": 1280,
+  "height": 720
+}
+```
+
+### JSON Base64 and n8n binary
+
+```json
+{
+  "base64": "data:audio/wav;base64,UklGR...",
+  "filename": "speech.wav",
+  "format": "mp3"
+}
+```
+
+The resolver recognizes n8n properties named `binary.data`, `binary.file`, `binary.video`, `binary.audio`, and any other key under `binary`:
+
+```json
+{
+  "binary": {
+    "video": {
+      "data": "AAAAHGZ0eXBpc29t...",
+      "fileName": "clip.mp4",
+      "mimeType": "video/mp4"
+    }
+  },
+  "crf": 26
+}
+```
+
+### Raw streamed bytes
+
+```bash
+curl -X POST 'http://localhost:7860/v1/probe' \
+  -H 'Content-Type: application/octet-stream' \
+  -H 'X-Filename: input.mp4' \
+  --data-binary '@input.mp4'
+```
+
+Raw and multipart uploads are written to disk in 1 MiB chunks. Downloads are also streamed; Base64 is size-checked before and after decoding.
+
+## Endpoints
+
+All processing routes use `POST`. Health and output downloads use `GET`.
+
+| Group | Endpoints |
+|---|---|
+| System | `/health`, `/v1/probe`, `/v1/media/{request_id}/{filename}` |
+| Video basics | `/v1/video/compress`, `resize`, `crop`, `trim`, `rotate`, `reverse`, `convert`, `merge`, `concat` |
+| Video composition | `/v1/video/overlay`, `watermark`, `replace-audio`, `subtitles/burn`, `subtitles/soft` |
+| Video outputs | `/v1/video/frames`, `gif`, `thumbnail`, `remove-audio`, `mute` |
+| Video timing/quality | `/v1/video/speed`, `speed-up`, `slow-motion`, `fps`, `bitrate`, `scale`, `pad`, `blur`, `sharpen`, `denoise`, `normalize` |
+| Audio | `/v1/audio/extract`, `convert`, `normalize`, `trim`, `merge`, `concat`, `fade`, `volume`, `remove-silence`, `noise-reduction` |
+| Image | `/v1/image/resize`, `crop`, `convert`, `slideshow`, `sequence`, `video`, `watermark`, `overlay` |
+| yt-dlp | `/v1/ytdlp/download` |
+| Whisper | `/v1/whisper/transcribe`, `subtitles`, `detect-language` |
+| Templates | `GET /v1/templates`, `GET /v1/templates/categories`, `GET /v1/templates/{reference}`, `POST /v1/templates/run` |
+
+Common video parameters include `format`, `width`, `height`, `fit`, `crf`, `preset`, `start`, `end`, `duration`, `fps`, `bitrate`, `factor`, `position`, and `opacity`. Swagger lists every route; the defaults and bounds are enforced by the operation that consumes each field.
+
+### Probe metadata
+
+`POST /v1/probe` returns normalized duration, resolution, FPS, bitrate, primary codec, video/audio/subtitle stream summaries, rotation, container, creation date, size, and tags. The same probe data is included under `metadata.inputs` for processing responses.
+
+```bash
+curl -X POST http://localhost:7860/v1/probe -F 'file=@input.mp4'
+```
+
+### yt-dlp
+
+`mode` is `video`, `audio`, `thumbnail`, or `metadata`. `format` accepts a yt-dlp format selector. Audio supports `mp3`, `m4a`, `wav`, `opus`, and `flac`.
+
+```bash
+curl -X POST http://localhost:7860/v1/ytdlp/download \
+  -H 'Content-Type: application/json' \
+  -d '{"url":"https://www.youtube.com/watch?v=VIDEO_ID","mode":"audio","audio_format":"mp3"}'
+```
+
+```bash
+curl -X POST http://localhost:7860/v1/ytdlp/download \
+  -H 'Content-Type: application/json' \
+  -d '{"url":"https://vimeo.com/VIDEO_ID","mode":"metadata"}'
+```
+
+### faster-whisper
+
+Models: `tiny`, `base`, `small`, `medium`, and `large-v3`. The default is `small`. Models load only on their first request and always use `device="cpu"` and `compute_type="int8"`. Tasks are `transcribe` and `translate`; omit `language` for automatic language detection. Formats are `txt`, `srt`, `vtt`, `json`, and `tsv`.
+
+```bash
+curl -X POST http://localhost:7860/v1/whisper/transcribe \
+  -F 'file=@meeting.mp3' \
+  -F 'model=small' \
+  -F 'task=transcribe' \
+  -F 'output_format=json' \
+  -F 'vad_filter=true'
+```
+
+```bash
+curl -X POST http://localhost:7860/v1/whisper/subtitles \
+  -F 'file=@interview.mp4' \
+  -F 'task=translate' \
+  -F 'output_format=vtt'
+```
+
+## Responses and downloads
+
+Processing responses are JSON:
+
+```json
+{
+  "success": true,
+  "request_id": "43b32630-9b32-4b27-b038-a39fbfd4b803",
+  "processing_time": 1.42,
+  "download_url": "/v1/media/43b32630-9b32-4b27-b038-a39fbfd4b803/compressed.mp4",
+  "metadata": {}
+}
+```
+
+Failures never include Python exceptions or tracebacks:
+
+```json
+{
+  "success": false,
+  "request_id": "43b32630-9b32-4b27-b038-a39fbfd4b803",
+  "error": {
+    "code": "INVALID_INPUT",
+    "message": "The uploaded media is empty",
+    "details": null
+  }
+}
+```
+
+Fetch `download_url` before `CLEANUP_MINUTES` expires. The download endpoint uses `StreamingResponse`, reads in 1 MiB chunks, and supplies `Content-Length` and `Content-Disposition`; output files are never loaded into RAM. Streaming file bodies are necessarily binary rather than JSON envelopes.
+
+Set `BASE_URL=https://<owner>-<space>.hf.space` when clients require absolute download URLs. If it is empty, URLs are relative, which works well in n8n when joined to the request host.
+
+## n8n recipes
+
+### Multipart upload
+
+In an **HTTP Request** node:
+
+- Method: `POST`
+- URL: `https://<space>.hf.space/v1/video/compress`
+- Send Body: on
+- Body Content Type: `Form-Data`
+- Add a **n8n Binary File** parameter named `file`, selecting the incoming binary property (usually `data`)
+- Add text parameters `crf=28` and `preset=veryfast`
+
+Use a second **HTTP Request** node with `{{$json.download_url}}`, enable **Download**, and store its response as binary.
+
+### JSON URL
+
+Set Body Content Type to JSON:
+
+```javascript
+{
+  "url": "{{$json.media_url}}",
+  "width": 1280,
+  "height": 720,
+  "fit": "contain"
+}
+```
+
+### n8n binary property as JSON Base64
+
+When the upstream binary property is `data`, use an expression body:
+
+```javascript
+{
+  "binary": {
+    "data": {
+      "data": "{{$binary.data.data}}",
+      "fileName": "{{$binary.data.fileName}}",
+      "mimeType": "{{$binary.data.mimeType}}"
+    }
+  },
+  "format": "mp3"
+}
+```
+
+For large files, prefer n8n's multipart binary option or raw binary body; JSON Base64 temporarily expands data by roughly 33%.
+
+## Configuration
+
+| Variable | Default | Purpose |
+|---|---:|---|
+| `TEMP_DIR` | `./temp` | Request workspaces and in-progress files |
+| `OUTPUT_DIR` | `./outputs` | Published files served by download URLs |
+| `TEMPLATE_DIR` | built-in `app/templates/categories` | Recursively scanned YAML workflow catalog |
+| `MAX_UPLOAD_SIZE` | `1073741824` | Maximum bytes for each upload/download |
+| `WHISPER_MODEL` | `small` | Default faster-whisper model |
+| `CLEANUP_MINUTES` | `60` | TTL for request workspaces and outputs |
+| `CLEANUP_INTERVAL_SECONDS` | `60` | Cleanup scan interval |
+| `MAX_WORKERS` | `2` | Shared per-service CPU process/task concurrency |
+| `LOG_LEVEL` | `INFO` | Structured log threshold |
+| `MAX_DURATION_SECONDS` | `21600` | Maximum probed media duration |
+| `MAX_RESOLUTION_PIXELS` | `33177600` | Maximum width × height (8K default) |
+| `DOWNLOAD_TIMEOUT_SECONDS` | `300` | Remote download timeout; FFmpeg gets 4× this value |
+| `ALLOW_PRIVATE_URLS` | `false` | Permit private/loopback URL downloads (normally unsafe) |
+| `BASE_URL` | empty | Optional public origin for absolute download URLs |
+| `FFMPEG_BINARY` | `ffmpeg` | FFmpeg executable name/path |
+| `FFPROBE_BINARY` | `ffprobe` | FFprobe executable name/path |
+
+## Logging and safety
+
+Logs are one JSON object per line. Request completion and operation records include UUID, operation, input/output bytes, wall time, CPU percentage, RSS memory, status, and errors. FFmpeg records include the exact argument array, bounded stdout/stderr, exit status, and duration.
+
+Safety controls include filename normalization, path containment checks, extension/MIME/size validation, duration and pixel limits, HTTP scheme and redirect validation, SSRF address filtering, bounded concurrency, bounded output capture, and subprocess arrays with `shell=False` semantics. User filenames never select output paths.
+
+## Testing and quality checks
+
+```bash
+pip install -r requirements.txt -r requirements-dev.txt
+pytest -q
+ruff check app tests main.py
+black --check app tests main.py
+```
+
+The suite covers health, Base64/n8n resolution, streamed URL download, real FFprobe metadata, a real FFmpeg conversion, lazy mocked Whisper transcription, MCP registration, YAML loading and versioning, parameter substitution, template execution, cleanup behavior, and safe error envelopes. Binary integration tests skip only when the respective system executable is absent.
+
+## Extending the API
+
+1. Add an async operation to the most relevant module in `app/operations`. Its inputs are `FFmpegService`, `Sequence[InputMedia]`, parsed parameters, and the request output directory; return an `OperationResult`.
+2. Validate every parameter before building arguments. Pass an argument list to `FFmpegService`; never use a shell or concatenate a command string.
+3. Register the operation in the appropriate API router with `operation_route`. The shared executor supplies input resolution, FFprobe validation, publication, metrics, errors, and cleanup.
+4. To expose it over MCP, add a thin typed function to the matching `app/mcp/tools/` registration module and call `MCPRegistry.run_operation` with the same operation function. Do not build FFmpeg arguments in the MCP module.
+5. Add unit coverage plus a small FFmpeg integration test when the operation changes media bytes, and update the expected MCP tool set when applicable.
+
+This contract keeps new operations independent of multipart, URLs, Base64, n8n, storage, and response handling.
+
+## Troubleshooting
+
+- **Space is out of memory:** use `tiny`, `base`, or `small`; set `MAX_WORKERS=1`; do not launch multiple Uvicorn workers. `medium` and `large-v3` require substantially more RAM even with int8.
+- **First transcription is slow:** the model is downloaded and initialized lazily. Use persistent Space storage or pre-warm with a short request after deployment.
+- **FFmpeg reports incompatible streams during concat:** leave `stream_copy` false (the default), which normalizes dimensions and re-encodes. Enable it only for files with identical stream layout, codec, time base, and parameters.
+- **Remote URL is rejected:** private and non-global addresses are blocked intentionally. Set `ALLOW_PRIVATE_URLS=true` only in a trusted network and never on a public Space.
+- **Download URL returns 404:** the request TTL expired, the Space restarted without persistent storage, or the filename/request ID was changed. Download results promptly.
+- **Upload receives 413/422:** increase `MAX_UPLOAD_SIZE` only after checking Space disk and RAM. Prefer multipart/raw streaming over Base64.
+- **ImageMagick policy error:** media operations use FFmpeg for image transforms; ImageMagick is installed for extension use but is not required by the built-in image routes.
+- **No audio after merging videos:** if any input lacks an audio stream, the merge deliberately emits video-only output instead of failing the whole request. Add silent audio before merging if a continuous audio track is required.
+- **MCP client cannot connect:** use the trailing-slash URL `/mcp/`, select Streamable HTTP rather than legacy SSE, and verify `/health` first. Stdio clients must launch the command with the project directory as their working directory.
+- **MCP output URL is relative:** set `BASE_URL` to the public Space origin. A managed `output_file` can also be passed directly into a later MCP call as `temp_path` before cleanup expires.

api/__init__.py

@@ -0,0 +1 @@
+"""Compatibility exports for the canonical :mod:`app.api` package."""

api/audio.py

@@ -0,0 +1 @@
+from app.api.audio import *  # noqa

api/health.py

@@ -0,0 +1 @@
+from app.api.health import *  # noqa

api/image.py

@@ -0,0 +1 @@
+from app.api.image import *  # noqa

api/media.py

@@ -0,0 +1 @@
+from app.api.media import *  # noqa

api/probe.py

@@ -0,0 +1 @@
+from app.api.probe import *  # noqa

api/video.py

@@ -0,0 +1 @@
+from app.api.video import *  # noqa

api/whisper.py

@@ -0,0 +1 @@
+from app.api.whisper import *  # noqa

api/ytdlp.py

@@ -0,0 +1 @@
+from app.api.ytdlp import *  # noqa

app/__init__.py

@@ -0,0 +1,3 @@
+"""Enterprise Media Processing API package."""
+
+__version__ = "1.0.0"

app/api/__init__.py

@@ -0,0 +1 @@
+"""HTTP API routers."""

app/api/audio.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Request
+
+from app.api.media import execute_operation
+from app.core.response import SuccessResponse
+from app.operations.compress import normalize_audio
+from app.operations.concat import concat_audio
+from app.operations.convert import convert_audio
+from app.operations.extract_audio import (
+    extract_audio,
+    fade_audio,
+    noise_reduction,
+    remove_silence,
+    set_volume,
+)
+from app.operations.merge import merge_audio
+from app.operations.trim import trim_audio
+from app.services.media_service import Operation
+
+router = APIRouter(prefix="/v1/audio", tags=["audio"])
+
+
+def operation_route(path: str, name: str, operation: Operation) -> None:
+    async def endpoint(request: Request) -> SuccessResponse:
+        return await execute_operation(request, name, operation)
+
+    endpoint.__name__ = name.replace(".", "_")
+    router.add_api_route(
+        path, endpoint, methods=["POST"], response_model=SuccessResponse, name=name
+    )
+
+
+operation_route("/extract", "audio.extract", extract_audio)
+operation_route("/convert", "audio.convert", convert_audio)
+operation_route("/normalize", "audio.normalize", normalize_audio)
+operation_route("/trim", "audio.trim", trim_audio)
+operation_route("/merge", "audio.merge", merge_audio)
+operation_route("/concat", "audio.concat", concat_audio)
+operation_route("/fade", "audio.fade", fade_audio)
+operation_route("/volume", "audio.volume", set_volume)
+operation_route("/remove-silence", "audio.remove_silence", remove_silence)
+operation_route("/noise-reduction", "audio.noise_reduction", noise_reduction)

app/api/health.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import importlib.util
+import shutil
+import time
+
+from fastapi import APIRouter, Request
+
+from app.core.response import SuccessResponse
+
+router = APIRouter(tags=["health"])
+
+
+async def _health(request: Request) -> SuccessResponse:
+    started = time.monotonic()
+    settings = request.app.state.container.settings
+    dependencies = {
+        "ffmpeg": shutil.which(settings.ffmpeg_binary) is not None,
+        "ffprobe": shutil.which(settings.ffprobe_binary) is not None,
+        "yt_dlp": importlib.util.find_spec("yt_dlp") is not None,
+        "faster_whisper": importlib.util.find_spec("faster_whisper") is not None,
+    }
+    return SuccessResponse(
+        request_id=request.state.request_id,
+        processing_time=round(time.monotonic() - started, 4),
+        metadata={
+            "status": "healthy",
+            "version": settings.app_version,
+            "dependencies": dependencies,
+        },
+    )
+
+
+router.add_api_route("/health", _health, methods=["GET"], response_model=SuccessResponse)
+router.add_api_route(
+    "/v1/health", _health, methods=["GET"], response_model=SuccessResponse, include_in_schema=False
+)

app/api/image.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Request
+
+from app.api.media import execute_operation
+from app.core.response import SuccessResponse
+from app.operations.concat import image_sequence, image_slideshow, image_to_video
+from app.operations.convert import convert_image
+from app.operations.crop import crop_image
+from app.operations.resize import resize_image
+from app.operations.watermark import overlay_image, watermark_image
+from app.services.media_service import Operation
+
+router = APIRouter(prefix="/v1/image", tags=["image"])
+
+
+def operation_route(path: str, name: str, operation: Operation) -> None:
+    async def endpoint(request: Request) -> SuccessResponse:
+        return await execute_operation(request, name, operation)
+
+    endpoint.__name__ = name.replace(".", "_")
+    router.add_api_route(
+        path, endpoint, methods=["POST"], response_model=SuccessResponse, name=name
+    )
+
+
+operation_route("/resize", "image.resize", resize_image)
+operation_route("/crop", "image.crop", crop_image)
+operation_route("/convert", "image.convert", convert_image)
+operation_route("/slideshow", "image.slideshow", image_slideshow)
+operation_route("/sequence", "image.sequence", image_sequence)
+operation_route("/video", "image.video", image_to_video)
+operation_route("/watermark", "image.watermark", watermark_image)
+operation_route("/overlay", "image.overlay", overlay_image)

app/api/media.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from pathlib import Path
+
+import aiofiles
+from fastapi import APIRouter, Request
+from fastapi.responses import StreamingResponse
+
+from app.core.response import SuccessResponse
+from app.services.media_service import MediaProcessor, Operation
+
+router = APIRouter(tags=["media"])
+
+
+def get_processor(request: Request) -> MediaProcessor:
+    return request.app.state.container.processor
+
+
+async def execute_operation(request: Request, name: str, operation: Operation) -> SuccessResponse:
+    request.state.operation = name
+    processor = get_processor(request)
+    resolved = await processor.resolver.resolve(request)
+    return await processor.run(resolved, name, operation)
+
+
+async def stream_file(path: Path) -> AsyncIterator[bytes]:
+    async with aiofiles.open(path, "rb") as media:
+        while chunk := await media.read(1024 * 1024):
+            yield chunk
+
+
+@router.get("/v1/media/{request_id}/{filename}", name="download_media")
+async def download_media(request: Request, request_id: str, filename: str) -> StreamingResponse:
+    request.state.operation = "media.download"
+    container = request.app.state.container
+    path = container.cleanup.resolve_download(request_id, filename)
+    media_type = container.validator.infer_mime(path)
+    headers = {
+        "Content-Disposition": f'attachment; filename="{path.name}"',
+        "Content-Length": str(path.stat().st_size),
+        "X-Request-ID": request_id,
+    }
+    return StreamingResponse(stream_file(path), media_type=media_type, headers=headers)

app/api/probe.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Request
+
+from app.api.media import get_processor
+from app.core.response import SuccessResponse
+
+router = APIRouter(prefix="/v1", tags=["probe"])
+
+
+@router.post("/probe", response_model=SuccessResponse)
+async def probe(request: Request) -> SuccessResponse:
+    request.state.operation = "probe"
+    processor = get_processor(request)
+    resolved = await processor.resolver.resolve(request)
+    return await processor.probe(resolved)

app/api/templates.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import time
+from typing import Any
+
+from fastapi import APIRouter, Query, Request
+
+from app.core.response import SuccessResponse
+
+router = APIRouter(prefix="/v1/templates", tags=["templates"])
+
+
+def _response(request: Request, started: float, metadata: dict[str, Any]) -> SuccessResponse:
+    return SuccessResponse(
+        request_id=request.state.request_id,
+        processing_time=round(time.monotonic() - started, 4),
+        metadata=metadata,
+    )
+
+
+@router.get("", response_model=SuccessResponse)
+async def list_templates(
+    request: Request, category: str | None = Query(default=None)
+) -> SuccessResponse:
+    """List metadata, parameters, and examples for every template version."""
+    started = time.monotonic()
+    request.state.operation = "templates.list"
+    templates = request.app.state.container.template_registry.list_templates(category)
+    return _response(
+        request,
+        started,
+        {"templates": templates, "count": len(templates), "category": category},
+    )
+
+
+@router.get("/categories", response_model=SuccessResponse)
+async def template_categories(request: Request) -> SuccessResponse:
+    """List categories discovered dynamically from YAML definitions."""
+    started = time.monotonic()
+    request.state.operation = "templates.categories"
+    categories = request.app.state.container.template_registry.categories()
+    return _response(request, started, {"categories": categories})
+
+
+@router.post("/run", response_model=SuccessResponse)
+async def run_template(request: Request) -> SuccessResponse:
+    """Resolve any supported media input and execute a versioned template."""
+    request.state.operation = "templates.run"
+    container = request.app.state.container
+    resolved = await container.resolver.resolve(request)
+    return await container.template_executor.execute_request(resolved)
+
+
+@router.get("/{template_id}", response_model=SuccessResponse)
+async def template_details(request: Request, template_id: str) -> SuccessResponse:
+    """Return complete metadata and pipeline details for a template reference."""
+    started = time.monotonic()
+    request.state.operation = "templates.details"
+    template = request.app.state.container.template_registry.template_details(template_id)
+    return _response(request, started, {"template": template})

app/api/video.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Request
+
+from app.api.media import execute_operation
+from app.core.response import SuccessResponse
+from app.operations.compress import compress_video, normalize_video
+from app.operations.concat import concat_video
+from app.operations.convert import convert_video
+from app.operations.crop import crop_video
+from app.operations.extract_audio import mute_video, remove_audio, replace_audio
+from app.operations.merge import merge_video
+from app.operations.resize import pad_video, resize_video, scale_video
+from app.operations.rotate import (
+    change_bitrate,
+    change_fps,
+    change_speed,
+    reverse_video,
+    rotate_video,
+    slow_motion,
+)
+from app.operations.subtitles import burn_subtitles, soft_subtitles
+from app.operations.thumbnails import (
+    blur_video,
+    denoise_video,
+    extract_frames,
+    generate_gif,
+    sharpen_video,
+    thumbnail,
+)
+from app.operations.trim import trim_video
+from app.operations.watermark import overlay_video, watermark_video
+from app.services.media_service import Operation
+
+router = APIRouter(prefix="/v1/video", tags=["video"])
+
+
+def operation_route(path: str, name: str, operation: Operation) -> None:
+    async def endpoint(request: Request) -> SuccessResponse:
+        return await execute_operation(request, name, operation)
+
+    endpoint.__name__ = name.replace(".", "_")
+    router.add_api_route(
+        path, endpoint, methods=["POST"], response_model=SuccessResponse, name=name
+    )
+
+
+operation_route("/compress", "video.compress", compress_video)
+operation_route("/resize", "video.resize", resize_video)
+operation_route("/convert", "video.convert", convert_video)
+operation_route("/crop", "video.crop", crop_video)
+operation_route("/merge", "video.merge", merge_video)
+operation_route("/trim", "video.trim", trim_video)
+operation_route("/concat", "video.concat", concat_video)
+operation_route("/rotate", "video.rotate", rotate_video)
+operation_route("/reverse", "video.reverse", reverse_video)
+operation_route("/overlay", "video.overlay", overlay_video)
+operation_route("/watermark", "video.watermark", watermark_video)
+operation_route("/frames", "video.extract_frames", extract_frames)
+operation_route("/gif", "video.gif", generate_gif)
+operation_route("/thumbnail", "video.thumbnail", thumbnail)
+operation_route("/replace-audio", "video.replace_audio", replace_audio)
+operation_route("/remove-audio", "video.remove_audio", remove_audio)
+operation_route("/mute", "video.mute", mute_video)
+operation_route("/speed", "video.speed", change_speed)
+operation_route("/speed-up", "video.speed_up", change_speed)
+operation_route("/slow-motion", "video.slow_motion", slow_motion)
+operation_route("/fps", "video.fps", change_fps)
+operation_route("/bitrate", "video.bitrate", change_bitrate)
+operation_route("/subtitles/burn", "video.subtitles.burn", burn_subtitles)
+operation_route("/subtitles/soft", "video.subtitles.soft", soft_subtitles)
+operation_route("/scale", "video.scale", scale_video)
+operation_route("/pad", "video.pad", pad_video)
+operation_route("/blur", "video.blur", blur_video)
+operation_route("/sharpen", "video.sharpen", sharpen_video)
+operation_route("/denoise", "video.denoise", denoise_video)
+operation_route("/normalize", "video.normalize", normalize_video)

app/api/whisper.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Request
+
+from app.api.media import get_processor
+from app.core.response import SuccessResponse
+
+router = APIRouter(prefix="/v1/whisper", tags=["whisper"])
+
+
+async def _transcribe(request: Request, default_format: str | None = None) -> SuccessResponse:
+    request.state.operation = "whisper.transcribe"
+    processor = get_processor(request)
+    resolved = await processor.resolver.resolve(request)
+    if default_format and "output_format" not in resolved.params:
+        resolved.params["output_format"] = default_format
+    return await processor.run_whisper(resolved)
+
+
+@router.post("/transcribe", response_model=SuccessResponse)
+async def transcribe(request: Request) -> SuccessResponse:
+    return await _transcribe(request)
+
+
+@router.post("/subtitles", response_model=SuccessResponse)
+async def subtitles(request: Request) -> SuccessResponse:
+    return await _transcribe(request, "srt")
+
+
+@router.post("/detect-language", response_model=SuccessResponse)
+async def detect_language(request: Request) -> SuccessResponse:
+    return await _transcribe(request, "json")

app/api/ytdlp.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Request
+
+from app.api.media import get_processor
+from app.core.response import SuccessResponse
+
+router = APIRouter(prefix="/v1/ytdlp", tags=["yt-dlp"])
+
+
+@router.post("/download", response_model=SuccessResponse)
+async def download(request: Request) -> SuccessResponse:
+    request.state.operation = "ytdlp.download"
+    processor = get_processor(request)
+    resolved = await processor.resolver.resolve(request)
+    return await processor.run_ytdlp(resolved)

app/container.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from app.core.config import Settings
+from app.services.cleanup import CleanupService
+from app.services.downloader import Downloader
+from app.services.ffmpeg_service import FFmpegService
+from app.services.ffprobe_service import FFprobeService
+from app.services.input_resolver import InputResolver
+from app.services.media_service import MediaProcessor
+from app.services.validator import MediaValidator
+from app.services.whisper_service import WhisperService
+from app.services.ytdlp_service import YTDLPService
+from app.templates.executor import OperationExecutor, TemplateExecutor
+from app.templates.loader import TemplateLoader
+from app.templates.registry import TemplateRegistry
+from app.templates.validator import TemplateValidator
+
+
+@dataclass(slots=True)
+class Container:
+    settings: Settings
+    cleanup: CleanupService
+    validator: MediaValidator
+    downloader: Downloader
+    ytdlp: YTDLPService
+    ffmpeg: FFmpegService
+    ffprobe: FFprobeService
+    whisper: WhisperService
+    resolver: InputResolver
+    processor: MediaProcessor
+    template_registry: TemplateRegistry
+    template_executor: TemplateExecutor
+
+
+def build_container(settings: Settings) -> Container:
+    cleanup = CleanupService(settings)
+    validator = MediaValidator(settings)
+    downloader = Downloader(settings, validator)
+    ytdlp = YTDLPService(settings, validator)
+    ffmpeg = FFmpegService(settings)
+    ffprobe = FFprobeService(settings)
+    whisper = WhisperService(settings)
+    resolver = InputResolver(settings, cleanup, downloader, ytdlp, validator)
+    processor = MediaProcessor(
+        settings, resolver, cleanup, validator, ffmpeg, ffprobe, ytdlp, whisper
+    )
+    operation_executor = OperationExecutor(processor)
+    template_validator = TemplateValidator(operation_executor.supported_operations)
+    template_registry = TemplateRegistry(
+        TemplateLoader(settings.template_dir, template_validator), template_validator
+    )
+    template_executor = TemplateExecutor(template_registry, operation_executor, processor)
+    return Container(
+        settings,
+        cleanup,
+        validator,
+        downloader,
+        ytdlp,
+        ffmpeg,
+        ffprobe,
+        whisper,
+        resolver,
+        processor,
+        template_registry,
+        template_executor,
+    )

app/core/__init__.py

@@ -0,0 +1 @@
+"""Core application infrastructure."""

app/core/config.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+from functools import lru_cache
+from pathlib import Path
+
+from pydantic import Field, field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+DEFAULT_TEMPLATE_DIR = Path(__file__).resolve().parents[1] / "templates" / "categories"
+
+
+class Settings(BaseSettings):
+    """Runtime configuration loaded from environment variables."""
+
+    model_config = SettingsConfigDict(
+        env_file=".env", env_file_encoding="utf-8", case_sensitive=False, extra="ignore"
+    )
+
+    app_name: str = "Enterprise Media Processing API"
+    app_version: str = "1.0.0"
+    host: str = "0.0.0.0"
+    port: int = 7860
+    temp_dir: Path = Path("./temp")
+    output_dir: Path = Path("./outputs")
+    template_dir: Path = DEFAULT_TEMPLATE_DIR
+    max_upload_size: int = Field(default=1_073_741_824, ge=1_048_576)
+    max_duration_seconds: float = Field(default=21_600.0, gt=0)
+    max_resolution_pixels: int = Field(default=33_177_600, ge=1)
+    whisper_model: str = "small"
+    cleanup_minutes: int = Field(default=60, ge=1)
+    cleanup_interval_seconds: int = Field(default=60, ge=5)
+    max_workers: int = Field(default=2, ge=1, le=32)
+    log_level: str = "INFO"
+    download_timeout_seconds: float = Field(default=300.0, gt=0)
+    allow_private_urls: bool = False
+    base_url: str = ""
+    ffmpeg_binary: str = "ffmpeg"
+    ffprobe_binary: str = "ffprobe"
+
+    @field_validator("whisper_model")
+    @classmethod
+    def validate_whisper_model(cls, value: str) -> str:
+        allowed = {"tiny", "base", "small", "medium", "large-v3"}
+        if value not in allowed:
+            raise ValueError(f"WHISPER_MODEL must be one of: {', '.join(sorted(allowed))}")
+        return value
+
+    @field_validator("log_level")
+    @classmethod
+    def normalize_log_level(cls, value: str) -> str:
+        normalized = value.upper()
+        allowed = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
+        if normalized not in allowed:
+            raise ValueError(f"LOG_LEVEL must be one of: {', '.join(sorted(allowed))}")
+        return normalized
+
+    def ensure_directories(self) -> None:
+        self.temp_dir.mkdir(parents=True, exist_ok=True)
+        self.output_dir.mkdir(parents=True, exist_ok=True)
+
+
+@lru_cache
+def get_settings() -> Settings:
+    settings = Settings()
+    settings.ensure_directories()
+    return settings

app/core/exceptions.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+class MediaAPIError(Exception):
+    """A safe, client-facing application error."""
+
+    code = "MEDIA_API_ERROR"
+    status_code = 400
+
+    def __init__(self, message: str, details: Any | None = None) -> None:
+        super().__init__(message, details)
+        self.message = message
+        self.details = details
+
+    def __str__(self) -> str:
+        return self.message
+
+
+class InputError(MediaAPIError):
+    code = "INVALID_INPUT"
+    status_code = 422
+
+
+class DownloadError(MediaAPIError):
+    code = "DOWNLOAD_FAILED"
+    status_code = 400
+
+
+class ProcessingError(MediaAPIError):
+    code = "PROCESSING_FAILED"
+    status_code = 422
+
+
+class NotFoundError(MediaAPIError):
+    code = "NOT_FOUND"
+    status_code = 404
+
+
+class TemplateNotFoundError(MediaAPIError):
+    """Raised when a requested template reference is unavailable."""
+
+    code = "TEMPLATE_NOT_FOUND"
+    status_code = 404
+
+
+class TemplateValidationError(MediaAPIError):
+    """Raised when a template definition or runtime parameter is invalid."""
+
+    code = "INVALID_TEMPLATE"
+    status_code = 422
+
+
+class TemplateExecutionError(MediaAPIError):
+    """Raised when a validated template cannot produce its declared output."""
+
+    code = "TEMPLATE_EXECUTION_FAILED"
+    status_code = 422

app/core/logger.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import contextvars
+import json
+import logging
+import sys
+from datetime import datetime, timezone
+from typing import Any, TextIO
+
+from app.core.config import get_settings
+
+request_id_context: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-")
+
+
+class JsonFormatter(logging.Formatter):
+    """One-line structured JSON logs suitable for container log collectors."""
+
+    _standard = set(logging.makeLogRecord({}).__dict__) | {"message", "asctime"}
+
+    def format(self, record: logging.LogRecord) -> str:
+        payload: dict[str, Any] = {
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+            "level": record.levelname,
+            "logger": record.name,
+            "message": record.getMessage(),
+            "request_id": getattr(record, "request_id", request_id_context.get()),
+        }
+        for key, value in record.__dict__.items():
+            if key not in self._standard and not key.startswith("_"):
+                payload[key] = self._json_safe(value)
+        if record.exc_info:
+            payload["exception"] = self.formatException(record.exc_info)
+        return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
+
+    @staticmethod
+    def _json_safe(value: Any) -> Any:
+        try:
+            json.dumps(value)
+            return value
+        except (TypeError, ValueError):
+            return str(value)
+
+
+def configure_logging(stream: TextIO | None = None) -> None:
+    """Configure structured logging, optionally targeting a stdio-safe stream."""
+    settings = get_settings()
+    handler = logging.StreamHandler(stream or sys.stdout)
+    handler.setFormatter(JsonFormatter())
+    root = logging.getLogger()
+    root.handlers.clear()
+    root.addHandler(handler)
+    root.setLevel(settings.log_level)
+    for name in ("uvicorn.access", "uvicorn.error"):
+        logging.getLogger(name).handlers.clear()
+        logging.getLogger(name).propagate = True
+
+
+def get_logger(name: str) -> logging.Logger:
+    return logging.getLogger(name)

app/core/response.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class ErrorBody(BaseModel):
+    code: str
+    message: str
+    details: Any | None = None
+
+
+class ErrorResponse(BaseModel):
+    success: bool = False
+    request_id: str
+    error: ErrorBody
+
+
+class SuccessResponse(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    success: bool = True
+    request_id: str
+    processing_time: float = Field(ge=0)
+    download_url: str | None = None
+    metadata: dict[str, Any] = Field(default_factory=dict)

app/mcp/__init__.py

@@ -0,0 +1 @@
+"""Model Context Protocol interface for the media processing service."""

app/mcp/prompts.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+from typing import Any
+
+from mcp.server.fastmcp import FastMCP
+
+
+def register_prompts(server: FastMCP[Any]) -> None:
+    """Register reusable prompt templates that direct clients to MCP tools."""
+
+    @server.prompt(description="Compress a video for a social media platform.")
+    def compress_for_social_media(source: str, platform: str = "instagram") -> str:
+        """Guide the model through social-media compression."""
+        return (
+            f"Use probe_media on {source!r}, then call compress_video for {platform}. "
+            "Choose MP4, H.264-compatible settings, CRF 28, veryfast preset, and a platform-appropriate max width. "
+            "Return the download_url and summarized output metadata."
+        )
+
+    @server.prompt(description="Download a YouTube URL as MP3 audio.")
+    def youtube_to_mp3(url: str) -> str:
+        """Guide the model through a YouTube-to-MP3 workflow."""
+        return f"Call download_audio with url={url!r} and audio_format='mp3'. Return the resulting download_url."
+
+    @server.prompt(description="Download remote media and transcribe it.")
+    def download_and_transcribe(url: str, model: str = "small") -> str:
+        """Guide the model through download followed by transcription."""
+        return (
+            f"First call download_audio with url={url!r}. Then call transcribe with the returned output_file as "
+            f"temp_path and model={model!r}. Return the transcript text, language, and download_url."
+        )
+
+    @server.prompt(description="Generate subtitles from media.")
+    def generate_subtitles(source: str, format: str = "srt", language: str = "auto") -> str:
+        """Guide the model through subtitle generation."""
+        tool = "generate_vtt" if format.lower() == "vtt" else "generate_srt"
+        language_instruction = (
+            "omit language for automatic detection"
+            if language == "auto"
+            else f"language={language!r}"
+        )
+        return f"Call {tool} for {source!r}; {language_instruction}. Return the subtitle download_url and detected language."
+
+    @server.prompt(description="Extract audio from a video.")
+    def extract_audio(source: str, format: str = "mp3") -> str:
+        """Guide the model through audio extraction."""
+        return f"Call extract_audio for {source!r} with format={format!r}, then return output metadata and download_url."
+
+    @server.prompt(description="Create a representative video thumbnail.")
+    def make_thumbnail(source: str, timestamp: float = 1.0) -> str:
+        """Guide the model through thumbnail generation."""
+        return f"Call generate_thumbnail for {source!r} at timestamp={timestamp}. Return the JPEG download_url."
+
+    @server.prompt(description="Inspect complete media metadata.")
+    def probe_media(source: str) -> str:
+        """Guide the model through FFprobe analysis."""
+        return f"Call probe_media for {source!r}. Summarize duration, resolution, FPS, codecs, streams, and container."
+
+    @server.prompt(description="Prepare an Instagram Reel from source media.")
+    def instagram_reel(source: str) -> str:
+        """Guide the model through Reel preparation."""
+        return (
+            f"Use probe_media on {source!r}, then resize_video to 1080x1920 with fit='cover', "
+            "and compress_video as MP4 with CRF 27 and preset='veryfast'. Return the final download_url."
+        )
+
+    @server.prompt(description="Prepare a vertical TikTok video.")
+    def tiktok_video(source: str) -> str:
+        """Guide the model through TikTok preparation."""
+        return (
+            f"Use resize_video on {source!r} at 1080x1920 with fit='cover', then compress_video "
+            "with MP4, CRF 26, and preset='veryfast'. Return the final download_url."
+        )
+
+    @server.prompt(description="Normalize and prepare podcast audio.")
+    def podcast_audio(source: str, format: str = "mp3") -> str:
+        """Guide the model through podcast audio preparation."""
+        return (
+            f"Call normalize_audio for {source!r} with target_lufs=-16. If format is not WAV, "
+            f"call convert_audio with format={format!r} on the normalized output_file. Return the final download_url."
+        )

app/mcp/registry.py

@@ -0,0 +1,505 @@
+from __future__ import annotations
+
+import importlib.metadata
+import importlib.util
+import os
+import platform
+import shutil
+import time
+from collections.abc import Awaitable, Callable, Sequence
+from pathlib import Path
+from typing import Any, Literal
+from urllib.parse import unquote, urlparse
+from uuid import uuid4
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+from app.container import Container
+from app.core.exceptions import InputError, MediaAPIError
+from app.core.logger import get_logger, request_id_context
+from app.core.response import SuccessResponse
+from app.models.media import ResolvedRequest
+from app.operations.common import AUDIO_FORMATS, IMAGE_FORMATS, VIDEO_FORMATS
+from app.services.media_service import Operation
+
+logger = get_logger(__name__)
+
+VIDEO_TOOLS = [
+    "compress_video",
+    "resize_video",
+    "crop_video",
+    "trim_video",
+    "convert_video",
+    "merge_videos",
+    "concat_videos",
+    "watermark_video",
+    "overlay_video",
+    "extract_audio",
+    "replace_audio",
+    "remove_audio",
+    "generate_thumbnail",
+    "extract_frames",
+    "burn_subtitles",
+]
+AUDIO_TOOLS = [
+    "convert_audio",
+    "normalize_audio",
+    "trim_audio",
+    "merge_audio",
+    "remove_silence",
+]
+IMAGE_TOOLS = ["image_to_video", "slideshow", "watermark_image", "resize_image"]
+WHISPER_TOOLS = [
+    "transcribe",
+    "translate",
+    "detect_language",
+    "generate_srt",
+    "generate_vtt",
+    "generate_json",
+]
+YTDLP_TOOLS = [
+    "download_video",
+    "download_audio",
+    "video_metadata",
+    "playlist_metadata",
+    "list_formats",
+]
+PROBE_TOOLS = [
+    "probe_media",
+    "probe_video_metadata",
+    "audio_metadata",
+    "stream_info",
+    "container_info",
+]
+SYSTEM_TOOLS = [
+    "health",
+    "cleanup_temp",
+    "disk_usage",
+    "system_info",
+    "supported_operations",
+    "supported_formats",
+    "ffmpeg_version",
+    "whisper_models",
+    "yt_dlp_version",
+]
+TEMPLATE_TOOLS = [
+    "list_templates",
+    "template_details",
+    "run_template",
+    "template_categories",
+]
+
+
+class MediaInput(BaseModel):
+    """Transport-neutral media input accepted by all MCP media tools."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    url: str | None = Field(default=None, description="HTTP(S) or yt-dlp-supported URL")
+    base64: str | None = Field(default=None, description="Raw Base64 or a Base64 data URI")
+    binary: dict[str, Any] | None = Field(
+        default=None, description="n8n-style binary property object"
+    )
+    temp_path: str | None = Field(
+        default=None,
+        description="Existing file below configured TEMP_DIR or OUTPUT_DIR",
+    )
+    filename: str | None = Field(default=None, description="Original filename")
+    mime_type: str | None = Field(default=None, description="Declared media MIME type")
+
+    @model_validator(mode="after")
+    def validate_source(self) -> MediaInput:
+        sources = [self.url, self.base64, self.binary, self.temp_path]
+        if sum(value is not None for value in sources) != 1:
+            raise ValueError("Exactly one of url, base64, binary, or temp_path is required")
+        return self
+
+    def descriptor(self) -> dict[str, Any]:
+        """Return the existing InputResolver descriptor representation."""
+        payload = self.model_dump(exclude_none=True)
+        if "mime_type" in payload:
+            payload["mimeType"] = payload.pop("mime_type")
+        return payload
+
+
+class ToolError(BaseModel):
+    code: str
+    message: str
+    details: Any | None = None
+
+
+class ToolResponse(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    success: bool
+    request_id: str
+    processing_time: float = Field(ge=0)
+    output_file: str | None = None
+    download_url: str | None = None
+    metadata: dict[str, Any] = Field(default_factory=dict)
+    error: ToolError | None = None
+
+
+Action = Callable[[str], Awaitable[SuccessResponse]]
+MetadataAction = Callable[[], Awaitable[dict[str, Any]]]
+
+
+class MCPRegistry:
+    """Shared MCP execution facade over the application's existing container."""
+
+    def __init__(self, container: Container) -> None:
+        self.container = container
+
+    async def run_operation(
+        self,
+        tool_name: str,
+        media: Sequence[MediaInput],
+        params: dict[str, Any],
+        operation: Operation,
+    ) -> dict[str, Any]:
+        """Resolve input and run an existing FFmpeg operation through MediaProcessor."""
+
+        async def action(request_id: str) -> SuccessResponse:
+            resolved = await self._resolve(media, params, request_id)
+            return await self.container.processor.run(resolved, tool_name, operation)
+
+        return await self._execute(tool_name, action)
+
+    async def run_whisper(
+        self,
+        tool_name: str,
+        media: MediaInput,
+        params: dict[str, Any],
+    ) -> dict[str, Any]:
+        """Resolve input and invoke the shared faster-whisper service."""
+
+        async def action(request_id: str) -> SuccessResponse:
+            resolved = await self._resolve([media], params, request_id)
+            return await self.container.processor.run_whisper(resolved)
+
+        return await self._execute(tool_name, action)
+
+    async def run_ytdlp(
+        self,
+        tool_name: str,
+        media: MediaInput,
+        params: dict[str, Any],
+    ) -> dict[str, Any]:
+        """Invoke yt-dlp through InputResolver and the shared MediaProcessor."""
+
+        async def action(request_id: str) -> SuccessResponse:
+            resolved = await self._resolve([media], params, request_id, ytdlp_options=params)
+            return await self.container.processor.run_ytdlp(resolved)
+
+        return await self._execute(tool_name, action)
+
+    async def run_probe(
+        self,
+        tool_name: str,
+        media: MediaInput,
+        view: Literal["all", "video", "audio", "streams", "container"] = "all",
+    ) -> dict[str, Any]:
+        """Probe media once and return the requested structured metadata view."""
+
+        async def action(request_id: str) -> SuccessResponse:
+            resolved = await self._resolve([media], {}, request_id)
+            response = await self.container.processor.probe(resolved)
+            primary = (response.metadata.get("inputs") or [{}])[0]
+            response.metadata = self._probe_view(primary, view)
+            return response
+
+        return await self._execute(tool_name, action)
+
+    async def run_template(
+        self,
+        template: str,
+        input_media: MediaInput | None,
+        input_collection: Sequence[MediaInput] | None,
+        parameters: dict[str, Any] | None,
+    ) -> dict[str, Any]:
+        """Resolve MCP media through InputResolver and execute a shared template."""
+
+        async def action(request_id: str) -> SuccessResponse:
+            if (input_media is None) == (input_collection is None):
+                raise InputError("Provide exactly one of input or inputs")
+            media = list(input_collection or [])
+            if input_media is not None:
+                media = [input_media]
+            if not media:
+                raise InputError("At least one template media input is required")
+            resolved = await self._resolve(media, {}, request_id)
+            return await self.container.template_executor.execute(resolved, template, parameters)
+
+        return await self._execute("run_template", action)
+
+    async def run_metadata_tool(self, tool_name: str, action: MetadataAction) -> dict[str, Any]:
+        """Run a non-media utility with the same logging and error contract."""
+
+        async def wrapped(request_id: str) -> SuccessResponse:
+            started = time.monotonic()
+            metadata = await action()
+            return SuccessResponse(
+                request_id=request_id,
+                processing_time=round(time.monotonic() - started, 4),
+                metadata=metadata,
+            )
+
+        return await self._execute(tool_name, wrapped)
+
+    async def health_data(self) -> dict[str, Any]:
+        """Return dependency health without loading the Whisper model."""
+        settings = self.container.settings
+        return {
+            "status": "healthy",
+            "version": settings.app_version,
+            "dependencies": {
+                "ffmpeg": shutil.which(settings.ffmpeg_binary) is not None,
+                "ffprobe": shutil.which(settings.ffprobe_binary) is not None,
+                "yt_dlp": importlib.util.find_spec("yt_dlp") is not None,
+                "faster_whisper": importlib.util.find_spec("faster_whisper") is not None,
+                "mcp": importlib.util.find_spec("mcp") is not None,
+            },
+        }
+
+    async def disk_usage_data(self) -> dict[str, Any]:
+        """Return disk usage for temporary and published output locations."""
+        return {
+            "temp": self._disk_usage(self.container.settings.temp_dir),
+            "outputs": self._disk_usage(self.container.settings.output_dir),
+        }
+
+    async def system_info_data(self) -> dict[str, Any]:
+        """Return CPU, memory, platform, and process information."""
+        data: dict[str, Any] = {
+            "platform": platform.platform(),
+            "python": platform.python_version(),
+            "cpu_count": os.cpu_count(),
+            "pid": os.getpid(),
+        }
+        try:
+            import psutil
+
+            process = psutil.Process(os.getpid())
+            data.update(
+                {
+                    "cpu_percent": psutil.cpu_percent(interval=None),
+                    "memory_bytes": process.memory_info().rss,
+                    "system_memory": psutil.virtual_memory()._asdict(),
+                }
+            )
+        except ImportError:
+            data.update({"cpu_percent": None, "memory_bytes": None})
+        return data
+
+    async def configuration_data(self) -> dict[str, Any]:
+        """Return non-secret runtime configuration."""
+        settings = self.container.settings
+        return {
+            "temp_dir": str(settings.temp_dir),
+            "output_dir": str(settings.output_dir),
+            "template_dir": str(settings.template_dir),
+            "max_upload_size": settings.max_upload_size,
+            "max_duration_seconds": settings.max_duration_seconds,
+            "max_resolution_pixels": settings.max_resolution_pixels,
+            "whisper_model": settings.whisper_model,
+            "cleanup_minutes": settings.cleanup_minutes,
+            "max_workers": settings.max_workers,
+            "allow_private_urls": settings.allow_private_urls,
+        }
+
+    async def supported_operations_data(self) -> dict[str, Any]:
+        """Return every registered operation grouped by domain."""
+        return {
+            "video": VIDEO_TOOLS,
+            "audio": AUDIO_TOOLS,
+            "image": IMAGE_TOOLS,
+            "whisper": WHISPER_TOOLS,
+            "ytdlp": YTDLP_TOOLS,
+            "probe": PROBE_TOOLS,
+            "system": SYSTEM_TOOLS,
+            "templates": TEMPLATE_TOOLS,
+        }
+
+    async def supported_formats_data(self) -> dict[str, Any]:
+        """Return supported output containers and transcription formats."""
+        return {
+            "video": sorted(VIDEO_FORMATS),
+            "audio": sorted(AUDIO_FORMATS),
+            "image": sorted(IMAGE_FORMATS),
+            "whisper": sorted(self.container.whisper.ALLOWED_FORMATS),
+            "whisper_models": sorted(self.container.whisper.ALLOWED_MODELS),
+            "yt_dlp_audio": ["mp3", "m4a", "wav", "opus", "flac"],
+        }
+
+    async def version_data(self) -> dict[str, Any]:
+        """Return application and protocol package versions."""
+        return {
+            "application": self.container.settings.app_version,
+            "mcp": self._package_version("mcp"),
+            "yt_dlp": self._package_version("yt-dlp"),
+            "faster_whisper": self._package_version("faster-whisper"),
+        }
+
+    async def codecs_data(self) -> dict[str, Any]:
+        """Return FFmpeg codec capabilities."""
+        codecs = await self.container.ffmpeg.codecs()
+        return {"count": len(codecs), "codecs": codecs}
+
+    async def safe_resource(self, name: str, action: MetadataAction) -> dict[str, Any]:
+        """Return resource data without exposing raw exceptions."""
+        response = await self.run_metadata_tool(f"resource.{name}", action)
+        if response["success"]:
+            return {"success": True, **response["metadata"]}
+        return {
+            "success": False,
+            "error": response["error"],
+            "request_id": response["request_id"],
+        }
+
+    async def _resolve(
+        self,
+        media: Sequence[MediaInput],
+        params: dict[str, Any],
+        request_id: str,
+        *,
+        ytdlp_options: dict[str, Any] | None = None,
+    ) -> ResolvedRequest:
+        payload: dict[str, Any] = {
+            "inputs": [item.descriptor() for item in media],
+            **params,
+        }
+        return await self.container.resolver.resolve_payload(
+            payload,
+            request_id,
+            ytdlp_options=ytdlp_options,
+        )
+
+    async def _execute(self, tool_name: str, action: Action) -> dict[str, Any]:
+        request_id = str(uuid4())
+        token = request_id_context.set(request_id)
+        started = time.monotonic()
+        cpu_started = time.process_time()
+        try:
+            response = await action(request_id)
+            result = self._tool_success(response)
+            logger.info(
+                "MCP tool completed",
+                extra={
+                    "tool": tool_name,
+                    "duration": round(time.monotonic() - started, 4),
+                    "cpu_time": round(time.process_time() - cpu_started, 6),
+                    "memory_bytes": self._memory_bytes(),
+                    "output_size": response.metadata.get("output_size", 0),
+                },
+            )
+            return result
+        except MediaAPIError as exc:
+            logger.warning(
+                "MCP tool failed",
+                extra={
+                    "tool": tool_name,
+                    "error_code": exc.code,
+                    "duration": round(time.monotonic() - started, 4),
+                    "cpu_time": round(time.process_time() - cpu_started, 6),
+                    "memory_bytes": self._memory_bytes(),
+                },
+            )
+            return self._tool_failure(request_id, started, exc.code, exc.message, exc.details)
+        except Exception:
+            logger.exception("unhandled MCP tool error", extra={"tool": tool_name})
+            return self._tool_failure(
+                request_id,
+                started,
+                "INTERNAL_ERROR",
+                "An unexpected internal error occurred",
+                None,
+            )
+        finally:
+            try:
+                await self.container.cleanup.complete(request_id)
+            except Exception:
+                logger.exception(
+                    "MCP request cleanup finalization failed",
+                    extra={"tool": tool_name},
+                )
+            finally:
+                request_id_context.reset(token)
+
+    def _tool_success(self, response: SuccessResponse) -> dict[str, Any]:
+        output_file: str | None = None
+        if response.download_url:
+            filename = Path(unquote(urlparse(response.download_url).path)).name
+            candidate = self.container.settings.output_dir / response.request_id / filename
+            if candidate.is_file():
+                output_file = str(candidate.resolve())
+        return ToolResponse(
+            success=True,
+            request_id=response.request_id,
+            processing_time=response.processing_time,
+            output_file=output_file,
+            download_url=response.download_url,
+            metadata=response.metadata,
+        ).model_dump(mode="json", exclude_none=True)
+
+    @staticmethod
+    def _tool_failure(
+        request_id: str,
+        started: float,
+        code: str,
+        message: str,
+        details: Any | None,
+    ) -> dict[str, Any]:
+        return ToolResponse(
+            success=False,
+            request_id=request_id,
+            processing_time=round(time.monotonic() - started, 4),
+            error=ToolError(code=code, message=message, details=details),
+        ).model_dump(mode="json", exclude_none=True)
+
+    @staticmethod
+    def _probe_view(primary: dict[str, Any], view: str) -> dict[str, Any]:
+        if view == "video":
+            keys = (
+                "filename",
+                "duration",
+                "resolution",
+                "fps",
+                "codec",
+                "rotation",
+                "video_streams",
+            )
+        elif view == "audio":
+            keys = ("filename", "duration", "bitrate", "audio_streams")
+        elif view == "streams":
+            keys = ("filename", "video_streams", "audio_streams", "subtitle_streams")
+        elif view == "container":
+            keys = ("filename", "duration", "bitrate", "container", "creation_date", "size", "tags")
+        else:
+            return {"media": primary}
+        return {key: primary.get(key) for key in keys}
+
+    @staticmethod
+    def _disk_usage(path: Path) -> dict[str, Any]:
+        usage = shutil.disk_usage(path)
+        return {
+            "path": str(path),
+            "total": usage.total,
+            "used": usage.used,
+            "free": usage.free,
+        }
+
+    @staticmethod
+    def _memory_bytes() -> int | None:
+        try:
+            import psutil
+
+            return psutil.Process(os.getpid()).memory_info().rss
+        except ImportError:
+            return None
+
+    @staticmethod
+    def _package_version(name: str) -> str | None:
+        try:
+            return importlib.metadata.version(name)
+        except importlib.metadata.PackageNotFoundError:
+            return None

app/mcp/resources.py

@@ -0,0 +1,71 @@
+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)

app/mcp/server.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+import argparse
+import asyncio
+import sys
+from typing import Any, Literal, cast
+
+from mcp.server.fastmcp import FastMCP
+
+from app.container import Container, build_container
+from app.core.config import get_settings
+from app.core.logger import configure_logging
+from app.mcp.prompts import register_prompts
+from app.mcp.registry import MCPRegistry
+from app.mcp.resources import register_resources
+from app.mcp.tools.audio import register_audio_tools
+from app.mcp.tools.image import register_image_tools
+from app.mcp.tools.probe import register_probe_tools
+from app.mcp.tools.system import register_system_tools
+from app.mcp.tools.templates import register_template_tools
+from app.mcp.tools.video import register_video_tools
+from app.mcp.tools.whisper import register_whisper_tools
+from app.mcp.tools.ytdlp import register_ytdlp_tools
+from app.workers.cleanup_worker import CleanupWorker
+
+MCPLogLevel = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
+
+
+def create_mcp_server(container: Container) -> FastMCP[Any]:
+    """Create a fully registered MCP server over an existing service container."""
+    settings = container.settings
+    server: FastMCP[Any] = FastMCP(
+        name="Enterprise Media Processing API",
+        instructions=(
+            "Use the registered tools for safe media processing. All media inputs accept URL, Base64, "
+            "n8n binary objects, or managed temp_path values. Read media://operations for capabilities."
+        ),
+        host=settings.host,
+        port=settings.port,
+        streamable_http_path="/",
+        json_response=True,
+        stateless_http=True,
+        log_level=cast(MCPLogLevel, settings.log_level),
+    )
+    registry = MCPRegistry(container)
+    register_video_tools(server, registry)
+    register_audio_tools(server, registry)
+    register_image_tools(server, registry)
+    register_whisper_tools(server, registry)
+    register_ytdlp_tools(server, registry)
+    register_probe_tools(server, registry)
+    register_system_tools(server, registry)
+    register_template_tools(server, registry)
+    register_resources(server, registry)
+    register_prompts(server)
+    return server
+
+
+async def run_server(transport: Literal["stdio", "streamable-http"] = "stdio") -> None:
+    """Run MCP as a standalone stdio or Streamable HTTP server."""
+    configure_logging(stream=sys.stderr if transport == "stdio" else None)
+    settings = get_settings()
+    container = build_container(settings)
+    worker = CleanupWorker(container.cleanup, settings.cleanup_interval_seconds)
+    server = create_mcp_server(container)
+    await worker.start()
+    try:
+        if transport == "stdio":
+            await server.run_stdio_async()
+        else:
+            await server.run_streamable_http_async()
+    finally:
+        await worker.stop()
+
+
+def main() -> None:
+    """CLI entry point for local MCP clients."""
+    parser = argparse.ArgumentParser(description="Enterprise Media API MCP server")
+    parser.add_argument(
+        "--transport",
+        choices=("stdio", "streamable-http"),
+        default="stdio",
+        help="MCP transport to run (default: stdio)",
+    )
+    arguments = parser.parse_args()
+    asyncio.run(run_server(arguments.transport))
+
+
+if __name__ == "__main__":
+    main()

app/mcp/tools/__init__.py

@@ -0,0 +1 @@
+"""MCP tool registration modules."""

app/mcp/tools/audio.py

@@ -0,0 +1,63 @@
+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,
+        )

app/mcp/tools/image.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+from typing import Any
+
+from mcp.server.fastmcp import FastMCP
+
+from app.mcp.registry import MCPRegistry, MediaInput
+from app.operations.concat import image_slideshow
+from app.operations.concat import image_to_video as image_to_video_operation
+from app.operations.resize import resize_image as resize_image_operation
+from app.operations.watermark import watermark_image as watermark_image_operation
+
+
+def register_image_tools(server: FastMCP[Any], registry: MCPRegistry) -> None:
+    """Register image MCP tools backed by existing operations."""
+
+    @server.tool(description="Turn one still image into an H.264 video.")
+    async def image_to_video(
+        input: MediaInput, duration: float = 5, fps: int = 30
+    ) -> dict[str, Any]:
+        """Create a video from one image."""
+        return await registry.run_operation(
+            "image_to_video",
+            [input],
+            {"duration": duration, "fps": fps},
+            image_to_video_operation,
+        )
+
+    @server.tool(description="Create a video slideshow from multiple images.")
+    async def slideshow(
+        inputs: list[MediaInput],
+        duration_per_image: float = 3,
+        width: int = 1280,
+        height: int = 720,
+        fps: int = 30,
+    ) -> dict[str, Any]:
+        """Create a slideshow through the shared image operation."""
+        return await registry.run_operation(
+            "slideshow",
+            inputs,
+            {
+                "duration_per_image": duration_per_image,
+                "width": width,
+                "height": height,
+                "fps": fps,
+            },
+            image_slideshow,
+        )
+
+    @server.tool(description="Apply an image watermark to another image.")
+    async def watermark_image(
+        image: MediaInput,
+        watermark: MediaInput,
+        position: str = "bottom-right",
+        opacity: float = 1.0,
+        watermark_scale: float = 1.0,
+        format: str = "png",
+    ) -> dict[str, Any]:
+        """Apply a watermark through the existing image operation."""
+        return await registry.run_operation(
+            "watermark_image",
+            [image, watermark],
+            {
+                "position": position,
+                "opacity": opacity,
+                "watermark_scale": watermark_scale,
+                "format": format,
+            },
+            watermark_image_operation,
+        )
+
+    @server.tool(description="Resize an image with contain, cover, or fill behavior.")
+    async def resize_image(
+        input: MediaInput,
+        width: int = 1280,
+        height: int = 720,
+        fit: str = "contain",
+        format: str = "png",
+    ) -> dict[str, Any]:
+        """Resize one image through the existing image operation."""
+        return await registry.run_operation(
+            "resize_image",
+            [input],
+            {"width": width, "height": height, "fit": fit, "format": format},
+            resize_image_operation,
+        )

app/mcp/tools/probe.py

@@ -0,0 +1,39 @@
+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")

app/mcp/tools/system.py

@@ -0,0 +1,87 @@
+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)

app/mcp/tools/templates.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from typing import Any
+
+from mcp.server.fastmcp import FastMCP
+
+from app.mcp.registry import MCPRegistry, MediaInput
+
+
+def register_template_tools(server: FastMCP[Any], registry: MCPRegistry) -> None:
+    """Register YAML template discovery and execution tools."""
+
+    @server.tool(description="List all dynamically loaded media template versions.")
+    async def list_templates(category: str | None = None) -> dict[str, Any]:
+        """Return template metadata, parameters, and examples."""
+
+        async def action() -> dict[str, Any]:
+            templates = registry.container.template_registry.list_templates(category)
+            return {"templates": templates, "count": len(templates)}
+
+        return await registry.run_metadata_tool("list_templates", action)
+
+    @server.tool(description="Return complete details for id, id@version, or id@latest.")
+    async def template_details(template: str) -> dict[str, Any]:
+        """Return one template's metadata and operation pipeline."""
+
+        async def action() -> dict[str, Any]:
+            return {"template": registry.container.template_registry.template_details(template)}
+
+        return await registry.run_metadata_tool("template_details", action)
+
+    @server.tool(description="Execute a versioned YAML template using the shared media resolver.")
+    async def run_template(
+        template: str,
+        input: MediaInput | None = None,
+        inputs: list[MediaInput] | None = None,
+        parameters: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        """Run a template with exactly one input object or an input list."""
+        return await registry.run_template(template, input, inputs, parameters)
+
+    @server.tool(description="List media template categories discovered from YAML.")
+    async def template_categories() -> dict[str, Any]:
+        """Return dynamic template category names."""
+
+        async def action() -> dict[str, Any]:
+            return {"categories": registry.container.template_registry.categories()}
+
+        return await registry.run_metadata_tool("template_categories", action)

app/mcp/tools/video.py

@@ -0,0 +1,223 @@
+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 compress_video as compress_video_operation
+from app.operations.concat import concat_video
+from app.operations.convert import convert_video as convert_video_operation
+from app.operations.crop import crop_video as crop_video_operation
+from app.operations.extract_audio import extract_audio as extract_audio_operation
+from app.operations.extract_audio import remove_audio as remove_audio_operation
+from app.operations.extract_audio import replace_audio as replace_audio_operation
+from app.operations.merge import merge_video
+from app.operations.resize import resize_video as resize_video_operation
+from app.operations.subtitles import burn_subtitles as burn_subtitles_operation
+from app.operations.thumbnails import extract_frames as extract_frames_operation
+from app.operations.thumbnails import thumbnail as thumbnail_operation
+from app.operations.trim import trim_video as trim_video_operation
+from app.operations.watermark import overlay_video as overlay_video_operation
+from app.operations.watermark import watermark_video as watermark_video_operation
+
+
+def register_video_tools(server: FastMCP[Any], registry: MCPRegistry) -> None:
+    """Register video MCP tools backed by existing operation functions."""
+
+    @server.tool(description="Compress a video with CPU-optimized FFmpeg settings.")
+    async def compress_video(
+        input: MediaInput,
+        crf: int = 28,
+        preset: str = "medium",
+        format: str = "mp4",
+        max_width: int | None = None,
+        bitrate: str | None = None,
+    ) -> dict[str, Any]:
+        """Compress one video and publish the output."""
+        params = _without_none(
+            {
+                "crf": crf,
+                "preset": preset,
+                "format": format,
+                "max_width": max_width,
+                "bitrate": bitrate,
+            }
+        )
+        return await registry.run_operation(
+            "compress_video", [input], params, compress_video_operation
+        )
+
+    @server.tool(description="Resize a video using contain, cover, or fill behavior.")
+    async def resize_video(
+        input: MediaInput,
+        width: int = 1280,
+        height: int = 720,
+        fit: str = "contain",
+        format: str = "mp4",
+    ) -> dict[str, Any]:
+        """Resize one video through the shared resize operation."""
+        return await registry.run_operation(
+            "resize_video",
+            [input],
+            {"width": width, "height": height, "fit": fit, "format": format},
+            resize_video_operation,
+        )
+
+    @server.tool(description="Crop a rectangular region from a video.")
+    async def crop_video(
+        input: MediaInput,
+        width: int,
+        height: int,
+        x: int = 0,
+        y: int = 0,
+    ) -> dict[str, Any]:
+        """Crop one video through the shared crop operation."""
+        return await registry.run_operation(
+            "crop_video",
+            [input],
+            {"width": width, "height": height, "x": x, "y": y},
+            crop_video_operation,
+        )
+
+    @server.tool(description="Trim a video by start, end, or duration.")
+    async def trim_video(
+        input: MediaInput,
+        start: float = 0,
+        duration: float | None = None,
+        end: float | None = None,
+    ) -> dict[str, Any]:
+        """Trim one video through the shared trim operation."""
+        return await registry.run_operation(
+            "trim_video",
+            [input],
+            _without_none({"start": start, "duration": duration, "end": end}),
+            trim_video_operation,
+        )
+
+    @server.tool(description="Convert a video to another supported container.")
+    async def convert_video(input: MediaInput, format: str = "mp4") -> dict[str, Any]:
+        """Convert one video through the shared conversion operation."""
+        return await registry.run_operation(
+            "convert_video", [input], {"format": format}, convert_video_operation
+        )
+
+    @server.tool(description="Normalize and merge multiple videos in sequence.")
+    async def merge_videos(
+        inputs: list[MediaInput], width: int = 1280, height: int = 720
+    ) -> dict[str, Any]:
+        """Merge videos with normalized dimensions and stream layout."""
+        return await registry.run_operation(
+            "merge_videos",
+            inputs,
+            {"width": width, "height": height},
+            merge_video,
+        )
+
+    @server.tool(description="Concatenate videos, optionally using stream copy.")
+    async def concat_videos(inputs: list[MediaInput], stream_copy: bool = False) -> dict[str, Any]:
+        """Concatenate multiple videos through the shared concat operation."""
+        return await registry.run_operation(
+            "concat_videos", inputs, {"stream_copy": stream_copy}, concat_video
+        )
+
+    @server.tool(description="Add an image watermark to a video.")
+    async def watermark_video(
+        video: MediaInput,
+        watermark: MediaInput,
+        position: str = "bottom-right",
+        opacity: float = 1.0,
+        watermark_scale: float = 1.0,
+    ) -> dict[str, Any]:
+        """Apply a watermark using the existing overlay operation."""
+        return await registry.run_operation(
+            "watermark_video",
+            [video, watermark],
+            {
+                "position": position,
+                "opacity": opacity,
+                "watermark_scale": watermark_scale,
+            },
+            watermark_video_operation,
+        )
+
+    @server.tool(description="Overlay an image or video layer on a video.")
+    async def overlay_video(
+        video: MediaInput,
+        overlay: MediaInput,
+        position: str = "center",
+        opacity: float = 1.0,
+        watermark_scale: float = 1.0,
+    ) -> dict[str, Any]:
+        """Overlay media through the shared video overlay operation."""
+        return await registry.run_operation(
+            "overlay_video",
+            [video, overlay],
+            {
+                "position": position,
+                "opacity": opacity,
+                "watermark_scale": watermark_scale,
+            },
+            overlay_video_operation,
+        )
+
+    @server.tool(description="Extract an audio track from video.")
+    async def extract_audio(input: MediaInput, format: str = "mp3") -> dict[str, Any]:
+        """Extract audio through the existing extraction operation."""
+        return await registry.run_operation(
+            "extract_audio", [input], {"format": format}, extract_audio_operation
+        )
+
+    @server.tool(description="Replace a video's audio track with another media input.")
+    async def replace_audio(video: MediaInput, audio: MediaInput) -> dict[str, Any]:
+        """Replace audio through the shared operation."""
+        return await registry.run_operation(
+            "replace_audio", [video, audio], {}, replace_audio_operation
+        )
+
+    @server.tool(description="Remove all audio streams from a video.")
+    async def remove_audio(input: MediaInput) -> dict[str, Any]:
+        """Remove audio through the shared operation."""
+        return await registry.run_operation("remove_audio", [input], {}, remove_audio_operation)
+
+    @server.tool(description="Generate a JPEG thumbnail from a video timestamp.")
+    async def generate_thumbnail(
+        input: MediaInput, timestamp: float = 0, quality: int = 3
+    ) -> dict[str, Any]:
+        """Generate a thumbnail through the shared thumbnail operation."""
+        return await registry.run_operation(
+            "generate_thumbnail",
+            [input],
+            {"timestamp": timestamp, "quality": quality},
+            thumbnail_operation,
+        )
+
+    @server.tool(description="Extract video frames into a ZIP archive.")
+    async def extract_frames(
+        input: MediaInput, fps: float = 1, max_frames: int | None = None
+    ) -> dict[str, Any]:
+        """Extract frames through the shared frame operation."""
+        return await registry.run_operation(
+            "extract_frames",
+            [input],
+            _without_none({"fps": fps, "max_frames": max_frames}),
+            extract_frames_operation,
+        )
+
+    @server.tool(description="Burn SRT, VTT, ASS, or SSA subtitles into video pixels.")
+    async def burn_subtitles(
+        video: MediaInput,
+        subtitles: MediaInput,
+        style: str | None = None,
+    ) -> dict[str, Any]:
+        """Burn subtitles through the existing subtitle operation."""
+        return await registry.run_operation(
+            "burn_subtitles",
+            [video, subtitles],
+            _without_none({"style": style}),
+            burn_subtitles_operation,
+        )
+
+
+def _without_none(values: dict[str, Any]) -> dict[str, Any]:
+    return {key: value for key, value in values.items() if value is not None}

app/mcp/tools/whisper.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+from typing import Any
+
+from mcp.server.fastmcp import FastMCP
+
+from app.mcp.registry import MCPRegistry, MediaInput
+
+
+def register_whisper_tools(server: FastMCP[Any], registry: MCPRegistry) -> None:
+    """Register faster-whisper MCP tools using the shared WhisperService."""
+
+    @server.tool(description="Transcribe speech with faster-whisper on CPU/int8.")
+    async def transcribe(
+        input: MediaInput,
+        model: str | None = None,
+        language: str | None = None,
+        output_format: str = "txt",
+        beam_size: int = 5,
+        vad_filter: bool = True,
+    ) -> dict[str, Any]:
+        """Transcribe media in its source language."""
+        return await registry.run_whisper(
+            "transcribe",
+            input,
+            _options(model, language, "transcribe", output_format, beam_size, vad_filter),
+        )
+
+    @server.tool(description="Translate speech to English with faster-whisper.")
+    async def translate(
+        input: MediaInput,
+        model: str | None = None,
+        language: str | None = None,
+        output_format: str = "txt",
+        beam_size: int = 5,
+        vad_filter: bool = True,
+    ) -> dict[str, Any]:
+        """Translate and transcribe media into English."""
+        return await registry.run_whisper(
+            "translate",
+            input,
+            _options(model, language, "translate", output_format, beam_size, vad_filter),
+        )
+
+    @server.tool(description="Detect spoken language and return transcription metadata.")
+    async def detect_language(input: MediaInput, model: str | None = None) -> dict[str, Any]:
+        """Detect language using Whisper's transcription information."""
+        return await registry.run_whisper(
+            "detect_language",
+            input,
+            _options(model, None, "transcribe", "json", 1, True),
+        )
+
+    @server.tool(description="Generate SubRip (SRT) subtitles from speech.")
+    async def generate_srt(
+        input: MediaInput,
+        model: str | None = None,
+        language: str | None = None,
+        task: str = "transcribe",
+    ) -> dict[str, Any]:
+        """Generate an SRT subtitle file."""
+        return await registry.run_whisper(
+            "generate_srt", input, _options(model, language, task, "srt", 5, True)
+        )
+
+    @server.tool(description="Generate WebVTT subtitles from speech.")
+    async def generate_vtt(
+        input: MediaInput,
+        model: str | None = None,
+        language: str | None = None,
+        task: str = "transcribe",
+    ) -> dict[str, Any]:
+        """Generate a WebVTT subtitle file."""
+        return await registry.run_whisper(
+            "generate_vtt", input, _options(model, language, task, "vtt", 5, True)
+        )
+
+    @server.tool(description="Generate structured JSON transcript data.")
+    async def generate_json(
+        input: MediaInput,
+        model: str | None = None,
+        language: str | None = None,
+        task: str = "transcribe",
+    ) -> dict[str, Any]:
+        """Generate a JSON transcript with timestamped segments."""
+        return await registry.run_whisper(
+            "generate_json", input, _options(model, language, task, "json", 5, True)
+        )
+
+
+def _options(
+    model: str | None,
+    language: str | None,
+    task: str,
+    output_format: str,
+    beam_size: int,
+    vad_filter: bool,
+) -> dict[str, Any]:
+    options: dict[str, Any] = {
+        "task": task,
+        "output_format": output_format,
+        "beam_size": beam_size,
+        "vad_filter": vad_filter,
+    }
+    if model is not None:
+        options["model"] = model
+    if language is not None:
+        options["language"] = language
+    return options

app/mcp/tools/ytdlp.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+from typing import Any
+
+from mcp.server.fastmcp import FastMCP
+
+from app.mcp.registry import MCPRegistry, MediaInput
+
+
+def register_ytdlp_tools(server: FastMCP[Any], registry: MCPRegistry) -> None:
+    """Register yt-dlp tools through the shared YTDLPService and resolver."""
+
+    @server.tool(description="Download the best available video with yt-dlp.")
+    async def download_video(url: str, format_selector: str | None = None) -> dict[str, Any]:
+        """Download one video URL."""
+        params: dict[str, Any] = {"mode": "video"}
+        if format_selector is not None:
+            params["format"] = format_selector
+        return await registry.run_ytdlp("download_video", MediaInput(url=url), params)
+
+    @server.tool(description="Download and convert the best audio with yt-dlp.")
+    async def download_audio(
+        url: str,
+        audio_format: str = "mp3",
+        format_selector: str | None = None,
+    ) -> dict[str, Any]:
+        """Download one URL as audio."""
+        params: dict[str, Any] = {"mode": "audio", "audio_format": audio_format}
+        if format_selector is not None:
+            params["format"] = format_selector
+        return await registry.run_ytdlp("download_audio", MediaInput(url=url), params)
+
+    @server.tool(
+        name="video_metadata", description="Read yt-dlp metadata without downloading media."
+    )
+    async def ytdlp_video_metadata(url: str) -> dict[str, Any]:
+        """Return platform metadata for one video URL."""
+        return await registry.run_ytdlp("video_metadata", MediaInput(url=url), {"mode": "metadata"})
+
+    @server.tool(description="Read bounded playlist metadata without downloading entries.")
+    async def playlist_metadata(url: str, max_entries: int = 100) -> dict[str, Any]:
+        """Return flat metadata for up to max_entries playlist items."""
+        return await registry.run_ytdlp(
+            "playlist_metadata",
+            MediaInput(url=url),
+            {
+                "mode": "metadata",
+                "playlist": True,
+                "max_entries": max_entries,
+            },
+        )
+
+    @server.tool(description="List available yt-dlp video and audio formats for a URL.")
+    async def list_formats(url: str) -> dict[str, Any]:
+        """Return normalized yt-dlp format summaries."""
+        return await registry.run_ytdlp(
+            "list_formats",
+            MediaInput(url=url),
+            {"mode": "metadata", "include_formats": True},
+        )

app/models/__init__.py

@@ -0,0 +1,5 @@
+"""Pydantic domain and transport models."""
+
+from app.models.media import InputMedia, MediaSource, OperationResult, ResolvedRequest
+
+__all__ = ["InputMedia", "MediaSource", "OperationResult", "ResolvedRequest"]

app/models/media.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+from enum import Enum
+from pathlib import Path
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class MediaSource(str, Enum):
+    MULTIPART = "multipart"
+    JSON_URL = "json_url"
+    JSON_BASE64 = "json_base64"
+    N8N_BINARY = "n8n_binary"
+    OCTET_STREAM = "octet_stream"
+    YTDLP = "yt_dlp"
+    LOCAL_PATH = "local_path"
+
+
+class InputMedia(BaseModel):
+    """Source-agnostic media passed to all processing operations."""
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    source: MediaSource
+    filename: str
+    mime_type: str
+    temp_path: Path
+    size: int = Field(ge=0)
+    duration: float | None = None
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+
+class ResolvedRequest(BaseModel):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    request_id: str
+    inputs: list[InputMedia]
+    params: dict[str, Any] = Field(default_factory=dict)
+
+    @property
+    def primary(self) -> InputMedia:
+        return self.inputs[0]
+
+
+class OperationResult(BaseModel):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    path: Path | None = None
+    filename: str | None = None
+    mime_type: str | None = None
+    metadata: dict[str, Any] = Field(default_factory=dict)

app/models/requests.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class URLInput(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    url: str
+    filename: str | None = None
+
+
+class Base64Input(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    base64: str
+    filename: str | None = None
+    mime_type: str | None = None
+
+
+class WhisperOptions(BaseModel):
+    model: Literal["tiny", "base", "small", "medium", "large-v3"] | None = None
+    task: Literal["transcribe", "translate"] = "transcribe"
+    language: str | None = None
+    output_format: Literal["txt", "srt", "vtt", "json", "tsv"] = "json"
+    beam_size: int = Field(default=5, ge=1, le=20)
+    vad_filter: bool = True
+
+
+class YTDLPOptions(BaseModel):
+    mode: Literal["video", "audio", "thumbnail", "metadata"] = "video"
+    format: str | None = None
+    audio_format: Literal["mp3", "m4a", "wav", "opus", "flac"] = "mp3"
+
+
+JsonDict = dict[str, Any]

app/operations/__init__.py

@@ -0,0 +1 @@
+"""Media operation implementations."""

app/operations/common.py

@@ -0,0 +1,114 @@
+from __future__ import annotations
+
+import mimetypes
+import re
+from collections.abc import Sequence
+from pathlib import Path
+from typing import Any
+
+from app.core.exceptions import InputError, ProcessingError
+from app.models.media import InputMedia, OperationResult
+from app.services.ffmpeg_service import FFmpegService
+
+VIDEO_FORMATS = {"mp4", "mkv", "webm", "mov", "avi", "m4v", "mpeg", "ts"}
+AUDIO_FORMATS = {"mp3", "wav", "aac", "m4a", "flac", "ogg", "opus"}
+IMAGE_FORMATS = {"jpg", "jpeg", "png", "webp", "bmp", "tiff", "gif"}
+
+
+def require_inputs(inputs: Sequence[InputMedia], count: int = 1) -> None:
+    if len(inputs) < count:
+        raise InputError(f"This operation requires at least {count} media input(s)")
+
+
+def output_path(output_dir: Path, stem: str, extension: str) -> Path:
+    clean_extension = extension.lower().lstrip(".")
+    if not re.fullmatch(r"[a-z0-9]{2,5}", clean_extension):
+        raise InputError("Invalid output format")
+    output_dir.mkdir(parents=True, exist_ok=True)
+    return output_dir / f"{stem}.{clean_extension}"
+
+
+def format_param(params: dict[str, Any], default: str, allowed: set[str]) -> str:
+    value = str(params.get("format", default)).lower().lstrip(".")
+    if value not in allowed:
+        raise InputError(
+            "Unsupported output format", details={"format": value, "allowed": sorted(allowed)}
+        )
+    return value
+
+
+def video_codecs(extension: str, *, crf: int = 23, preset: str = "medium") -> list[str]:
+    if extension == "webm":
+        return ["-c:v", "libvpx-vp9", "-crf", str(crf), "-b:v", "0", "-c:a", "libopus"]
+    if extension == "avi":
+        return ["-c:v", "mpeg4", "-q:v", "5", "-c:a", "libmp3lame"]
+    if extension in {"mpeg", "ts"}:
+        return ["-c:v", "mpeg2video", "-c:a", "mp2"]
+    return [
+        "-c:v",
+        "libx264",
+        "-preset",
+        preset,
+        "-crf",
+        str(crf),
+        "-c:a",
+        "aac",
+        "-b:a",
+        "128k",
+        "-movflags",
+        "+faststart",
+    ]
+
+
+def audio_codec(extension: str) -> list[str]:
+    return {
+        "mp3": ["-c:a", "libmp3lame", "-b:a", "192k"],
+        "wav": ["-c:a", "pcm_s16le"],
+        "aac": ["-c:a", "aac", "-b:a", "192k"],
+        "m4a": ["-c:a", "aac", "-b:a", "192k"],
+        "flac": ["-c:a", "flac"],
+        "ogg": ["-c:a", "libvorbis", "-q:a", "5"],
+        "opus": ["-c:a", "libopus", "-b:a", "128k"],
+    }[extension]
+
+
+async def execute(
+    ffmpeg: FFmpegService,
+    args: Sequence[str | Path],
+    output: Path,
+    operation: str,
+    metadata: dict[str, Any] | None = None,
+) -> OperationResult:
+    await ffmpeg.run(args, operation=operation)
+    if not output.is_file() or output.stat().st_size == 0:
+        raise ProcessingError("The operation did not produce an output file")
+    return OperationResult(
+        path=output,
+        filename=output.name,
+        mime_type=mimetypes.guess_type(output.name)[0] or "application/octet-stream",
+        metadata={"operation": operation, **(metadata or {})},
+    )
+
+
+def even(value: int) -> int:
+    return max(2, value if value % 2 == 0 else value - 1)
+
+
+def subtitle_filter_path(path: Path) -> str:
+    value = str(path.resolve()).replace("\\", "\\\\")
+    for character in (":", "'", "[", "]", ","):
+        value = value.replace(character, f"\\{character}")
+    return value
+
+
+def atempo_chain(factor: float) -> str:
+    values: list[float] = []
+    remaining = factor
+    while remaining > 2.0:
+        values.append(2.0)
+        remaining /= 2.0
+    while remaining < 0.5:
+        values.append(0.5)
+        remaining /= 0.5
+    values.append(remaining)
+    return ",".join(f"atempo={value:.6g}" for value in values)