Simple MCP

""" YouTube tools for the MCP server. """ import json import logging import os from typing import Optional from config import DOWNLOADS_BASE_PATH, FFMPEG_LOCATION from . import mcp # Set a long timeout (in seconds) for yt-dlp network operations YTDLP_LONG_TIMEOUT = 600 # 10 minutes # All YouTube downloads must be under a "youtube" folder in the downloads directory YOUTUBE_BASE_PATH = os.path.join(DOWNLOADS_BASE_PATH, "youtube") def _safe_import_yt_dlp(): try: import yt_dlp return yt_dlp except ImportError as e: logging.error("yt-dlp import failed: %s", e) return None except Exception as e: logging.error("Unexpected error importing yt-dlp: %s", e) return None def _safe_makedirs(path): try: os.makedirs(path, exist_ok=True) except OSError as e: logging.error("Failed to create directory %s: %s", path, e) def _sanitize_filename(name): # Remove characters not allowed in filenames return "".join(c for c in name if c not in r'\/:*?"<>|').strip() or "untitled" def _split_output_path( output_path: Optional[str], default_base: str, folder_name: str, ext: str ) -> (str, str): """ Helper to determine the target folder and output template for yt-dlp, given an output_path. All downloads are forced under the youtube folder in the downloads directory. If output_path is a directory, use it as a subdirectory under youtube. If output_path is a file (ends with .mp3 or .mp4), use its parent as a subdirectory under youtube and its filename as the output template. Otherwise, use the default base path (which is always YOUTUBE_BASE_PATH). Returns (target_folder, outtmpl) """ try: # Always use YOUTUBE_BASE_PATH as the root base_path = YOUTUBE_BASE_PATH if output_path: output_path = os.path.expanduser(output_path) # If output_path ends with the extension, treat as file if output_path.lower().endswith(ext): # Place file under youtube/ + parent directory (if any) parent = os.path.dirname(output_path) if parent and parent not in (".", ""): target_folder = os.path.join(base_path, parent) else: target_folder = base_path filename = os.path.basename(output_path) outtmpl = os.path.join(target_folder, filename) return target_folder, outtmpl # If output_path is an existing directory, or ends with os.sep, treat as directory elif os.path.isdir(output_path) or output_path.endswith(os.sep): target_folder = os.path.join(base_path, output_path, folder_name) outtmpl = os.path.join(target_folder, "%(title)s.%(ext)s") return target_folder, outtmpl else: # If not a file and not a directory, treat as directory (for compatibility) target_folder = os.path.join(base_path, output_path, folder_name) outtmpl = os.path.join(target_folder, "%(title)s.%(ext)s") return target_folder, outtmpl else: target_folder = os.path.join(base_path, folder_name) outtmpl = os.path.join(target_folder, "%(title)s.%(ext)s") return target_folder, outtmpl except OSError as e: logging.error("Error in _split_output_path (OSError): %s", e) # Fallback to default target_folder = os.path.join(YOUTUBE_BASE_PATH, folder_name) outtmpl = os.path.join(target_folder, "%(title)s.%(ext)s") return target_folder, outtmpl except Exception as e: logging.error("Error in _split_output_path: %s", e) # Fallback to default target_folder = os.path.join(YOUTUBE_BASE_PATH, folder_name) outtmpl = os.path.join(target_folder, "%(title)s.%(ext)s") return target_folder, outtmpl @mcp.tool() async def youtube_video_info(url: str) -> str: """ Get basic information about a YouTube video. Args: url: The URL of the YouTube video. Returns: A formatted string with video title, uploader, duration, and description. """ logging.info("youtube_video_info called with url=%r", url) yt_dlp = _safe_import_yt_dlp() if yt_dlp is None: return "yt-dlp is not installed. Please install it to use this tool." ydl_opts = { "quiet": True, "skip_download": True, "no_warnings": True, "extract_flat": False, "socket_timeout": YTDLP_LONG_TIMEOUT, "retries": 10, "ffmpeg_location": FFMPEG_LOCATION, "logger": logging.getLogger("yt_dlp"), } try: with yt_dlp.YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(url, download=False) title = info.get("title", "N/A") uploader = info.get("uploader", "N/A") duration = info.get("duration", 0) minutes = duration // 60 seconds = duration % 60 description = info.get("description", "") short_desc = ( (description[:200] + "...") if description and len(description) > 200 else description ) # Also return the metadata as JSON metadata_json = json.dumps(info, ensure_ascii=False, indent=2) return ( f"Title: {title}\n" f"Uploader: {uploader}\n" f"Duration: {minutes}m {seconds}s\n" f"Description: {short_desc}\n" f"Metadata:\n{metadata_json}" ) except yt_dlp.utils.DownloadError as e: logging.error("Error fetching YouTube video info (DownloadError): %s", e) return f"Error fetching YouTube video info: {e}" except Exception as e: logging.error("Error fetching YouTube video info: %s", e) return f"Error fetching YouTube video info: {e}" def _extract_folder_name(yt_dlp, url, default_folder, ext): """Extracts folder name and info for a YouTube video.""" try: with yt_dlp.YoutubeDL( { "quiet": True, "skip_download": True, "socket_timeout": YTDLP_LONG_TIMEOUT, "retries": 10, "ffmpeg_location": FFMPEG_LOCATION, "logger": logging.getLogger("yt_dlp"), } ) as ydl: info = ydl.extract_info(url, download=False) video_id = info.get("id", "unknown_id") title = info.get("title", "unknown_title") safe_title = _sanitize_filename(title) folder_name = f"{safe_title}_{video_id}" return folder_name, info except Exception as e: logging.error("Error extracting YouTube info for folder: %s", e) return default_folder, None def _resolve_output_filename(info, ydl, output_path, target_folder, ext, base_path): """Determine the output filename for the downloaded file.""" if output_path and output_path.lower().endswith(ext): parent = os.path.dirname(output_path) if parent and parent not in (".", ""): filename = os.path.abspath( os.path.join(base_path, parent, os.path.basename(output_path)) ) else: filename = os.path.abspath( os.path.join(base_path, os.path.basename(output_path)) ) else: filename = None if info and "requested_downloads" in info and info["requested_downloads"]: filename = info["requested_downloads"][0].get("filepath") or info[ "requested_downloads" ][0].get("_filename") if not filename and info: filename = ydl.prepare_filename(info) filename = filename.rsplit(".", 1)[0] + ext if filename and not os.path.isfile(filename): # Try to find the file in the target folder for f in os.listdir(target_folder): if f.lower().endswith(ext): filename = os.path.abspath(os.path.join(target_folder, f)) break return filename def _handle_download_error(e, media_type): error_str = str(e) if ( "ffprobe and ffmpeg not found" in error_str or "ffmpeg not found" in error_str or "ffprobe not found" in error_str ): msg = ( f"Error downloading YouTube {media_type}: ERROR: Postprocessing: ffprobe and ffmpeg not found. " "Please install or provide the path using --ffmpeg-location" ) logging.error("%s", msg) return msg logging.error(f"Error downloading YouTube {media_type} (DownloadError): %s", e) return f"Error downloading YouTube {media_type}: {e}" @mcp.tool() async def youtube_download_audio(url: str, output_path: Optional[str] = None) -> str: """ Download the audio of a YouTube video as an MP3 file. Args: url: The URL of the YouTube video. output_path: Optional path to save the MP3 file. If not provided, uses default downloads directory. Returns: Success message with file path or error message. """ logging.info( "youtube_download_audio called with url=%r, output_path=%r", url, output_path ) yt_dlp = _safe_import_yt_dlp() if yt_dlp is None: return "yt-dlp is not installed. Please install it to use this tool." folder_name, info = _extract_folder_name(yt_dlp, url, "youtube_audio", ".mp3") target_folder, outtmpl = _split_output_path( output_path, YOUTUBE_BASE_PATH, folder_name, ".mp3" ) _safe_makedirs(target_folder) ydl_opts = { "format": "bestaudio/best", "outtmpl": outtmpl, "quiet": True, "no_warnings": True, "socket_timeout": YTDLP_LONG_TIMEOUT, "retries": 10, "ffmpeg_location": FFMPEG_LOCATION, "postprocessors": [ { "key": "FFmpegExtractAudio", "preferredcodec": "mp3", "preferredquality": "192", } ], "logger": logging.getLogger("yt_dlp"), "progress_hooks": [], "noprogress": True, "dump_single_json": False, } try: with yt_dlp.YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(url, download=True) mp3_filename = _resolve_output_filename( info, ydl, output_path, target_folder, ".mp3", YOUTUBE_BASE_PATH ) if not mp3_filename or not os.path.isfile(mp3_filename): raise FileNotFoundError("MP3 file not found after download.") metadata_json = json.dumps(info, ensure_ascii=False, indent=2) return f"Audio downloaded as: {mp3_filename}\nMetadata:\n{metadata_json}" except yt_dlp.utils.DownloadError as e: return _handle_download_error(e, "audio") except FileNotFoundError as e: logging.error("Error downloading YouTube audio (FileNotFoundError): %s", e) return f"Error downloading YouTube audio: {e}" except Exception as e: logging.error("Error downloading YouTube audio: %s", e) return f"Error downloading YouTube audio: {e}" @mcp.tool() async def youtube_download_video(url: str, output_path: Optional[str] = None) -> str: """ Download a YouTube video in MP4 format. Args: url: The URL of the YouTube video. output_path: Optional path to save the MP4 file. If not provided, uses default downloads directory. Returns: Success message with file path or error message. """ logging.info( "youtube_download_video called with url=%r, output_path=%r", url, output_path ) yt_dlp = _safe_import_yt_dlp() if yt_dlp is None: return "yt-dlp is not installed. Please install it to use this tool." folder_name, info = _extract_folder_name(yt_dlp, url, "youtube_video", ".mp4") target_folder, outtmpl = _split_output_path( output_path, YOUTUBE_BASE_PATH, folder_name, ".mp4" ) _safe_makedirs(target_folder) ydl_opts = { "format": "bestvideo[ext=mp4]+bestaudio[ext=m4a]/mp4", "outtmpl": outtmpl, "quiet": True, "no_warnings": True, "merge_output_format": "mp4", "socket_timeout": YTDLP_LONG_TIMEOUT, "retries": 10, "ffmpeg_location": FFMPEG_LOCATION, "logger": logging.getLogger("yt_dlp"), "progress_hooks": [], "noprogress": True, "dump_single_json": False, } try: with yt_dlp.YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(url, download=True) mp4_filename = _resolve_output_filename( info, ydl, output_path, target_folder, ".mp4", YOUTUBE_BASE_PATH ) if not mp4_filename or not os.path.isfile(mp4_filename): raise FileNotFoundError("MP4 file not found after download.") metadata_json = json.dumps(info, ensure_ascii=False, indent=2) return f"Video downloaded as: {mp4_filename}\nMetadata:\n{metadata_json}" except yt_dlp.utils.DownloadError as e: return _handle_download_error(e, "video") except FileNotFoundError as e: logging.error("Error downloading YouTube video (FileNotFoundError): %s", e) return f"Error downloading YouTube video: {e}" except Exception as e: logging.error("Error downloading YouTube video: %s", e) return f"Error downloading YouTube video: {e}"