YouTube Translate MCP

Instructions

Implementation Reference

• src/youtube_translate_mcp/server.py:386-435 (handler)

  The handler function for the 'get_subtitles' MCP tool. It validates input format, processes the YouTube video to ensure availability using 'process_video' helper, fetches subtitles via 'make_yt_api_request' to the YouTube Translate API, handles errors, and returns the subtitle content as a string.
  @mcp_server.tool(name="get_subtitles", description="Generate subtitle files for a YouTube video") async def get_subtitles(url: str, language: str = "en", format: str = "srt") -> str: """Generate subtitle files for a YouTube video. This tool processes a video and generates subtitle files in the specified format and language. It first checks if the subtitles already exist before processing the video to optimize performance. If the requested language is not available, it automatically requests a translation first. Args: url: The YouTube video URL language: Language code for subtitles (e.g., "en", "fr", "es") format: Subtitle format, either "srt" or "vtt" (default: "srt") Returns: The subtitles content as text """ logger.info(f"Getting subtitles for URL: {url}, language: {language}, format: {format}") # Validate format if format not in ["srt", "vtt"]: error_msg = f"Invalid format: {format}. Must be 'srt' or 'vtt'." logger.error(error_msg) return f"Error: {error_msg}" # Process the video to ensure it's ready success, video_id, error_message = await process_video(url) if not success: logger.error(f"Failed to process video: {error_message}") return f"Error: {error_message}" # Get the subtitles from the API subtitles_response = await make_yt_api_request( f"/api/videos/{video_id}/subtitles", params={"language": language, "format": format} ) if not subtitles_response: error_msg = f"Failed to retrieve subtitles for language: {language}, format: {format}" logger.error(error_msg) return f"Error: {error_msg}" # The response could be the subtitles as text or a JSON object with an error if isinstance(subtitles_response, dict) and "error" in subtitles_response: error_msg = subtitles_response["error"] logger.error(f"API error: {error_msg}") return f"Error: {error_msg}" return subtitles_response
• src/youtube_translate_mcp/server.py:386-386 (registration)

  Registers the 'get_subtitles' tool with the FastMCP server using the @mcp_server.tool decorator, specifying the tool name and description.
  @mcp_server.tool(name="get_subtitles", description="Generate subtitle files for a YouTube video")
• src/youtube_translate_mcp/server.py:108-202 (helper)

  Key helper function called by get_subtitles to extract YouTube ID, submit for processing if needed, and poll until video processing is completed, returning success flag, video_id, and error message.
  async def process_video(url: str) -> tuple[bool, str, str]: """Helper function to submit a video for processing and wait for completion. This function now tries to optimize API calls by: 1. Extracting YouTube ID from URL when possible 2. Checking if video is already processed using YouTube ID directly 3. Only submitting for processing if needed Args: url: The YouTube video URL Returns: A tuple of (success, video_id, error_message) """ try: # Step 1: Try to extract YouTube ID from URL youtube_id = extract_youtube_id(url) video_id = "" if youtube_id: logger.info(f"Extracted YouTube ID: {youtube_id} from URL: {url}") # Step 2: Check if video has already been processed using YouTube ID directly status_response = await make_yt_api_request(f"/api/videos/{youtube_id}") if status_response and "status" in status_response: video_id = youtube_id logger.info(f"Found existing video with YouTube ID: {youtube_id}, status: {status_response.get('status')}") # If video is already processed or processing, we can use this ID if status_response.get("status") == "completed": logger.info(f"Video already processed, using YouTube ID: {youtube_id}") return True, youtube_id, "" elif status_response.get("status") == "processing": # Need to wait for processing to complete logger.info(f"Video already processing, waiting for completion: {youtube_id}") # Continue to polling step below with the YouTube ID video_id = youtube_id elif status_response.get("status") == "error": error_message = status_response.get("message", "Unknown error occurred") logger.error(f"Error with video: {error_message}") return False, youtube_id, f"Error processing video: {error_message}" # Step 3: Submit video for processing if needed (if we don't have a video_id yet) if not video_id: logger.info(f"Submitting video for processing: {url}") submit_response = await make_yt_api_request("/api/videos", method="POST", json_data={"url": url}) if not submit_response or "id" not in submit_response: logger.error("Failed to submit video for processing") return False, "", "Failed to submit video for processing." video_id = submit_response["id"] logger.info(f"Video submitted, received ID: {video_id}") await asyncio.sleep(1) # wait for 1 second before polling # Step 4: Poll for video processing status until it's complete max_attempts = 10 attempts = 0 while attempts < max_attempts: logger.info(f"Checking video status, attempt {attempts+1}/{max_attempts}") status_response = await make_yt_api_request(f"/api/videos/{video_id}") if not status_response: logger.error("Failed to retrieve video status") return False, video_id, "Failed to retrieve video status." status = status_response.get("status") logger.info(f"Video status: {status}") if status == "completed": logger.info(f"Video processing completed for ID: {video_id}") return True, video_id, "" if status == "error": error_message = status_response.get("message", "Unknown error occurred") logger.error(f"Error processing video: {error_message}") return False, video_id, f"Error processing video: {error_message}" # Calculate backoff delay delay = await calculate_backoff_delay(attempts) logger.info(f"Waiting {delay:.1f}s before checking video status again, attempt {attempts+1}/{max_attempts}") await asyncio.sleep(delay) attempts += 1 logger.error("Video processing timeout - too many attempts") return False, video_id, "Video processing timed out. Please try again later." except Exception as e: logger.error(f"Exception during video processing: {str(e)}") return False, "", f"An error occurred: {str(e)}"
• src/youtube_translate_mcp/server.py:69-107 (helper)

  Core helper for making HTTP requests to the YouTube Translate API endpoints, used by get_subtitles to fetch subtitles. Handles GET/POST, logging, timeouts, and returns response text for subtitles endpoints.
  async def make_yt_api_request(endpoint: str, method: str = "GET", params: dict = None, json_data: dict = None) -> dict[str, Any] | str | None: """Make a request to the YouTube Translate API with proper error handling.""" headers = { "X-API-Key": YOUTUBE_TRANSLATE_API_KEY, "Content-Type": "application/json" } url = f"{YT_TRANSLATE_API_BASE}{endpoint}" logger.info(f"Making API request: {method} {url}") if params: logger.info(f"Request params: {params}") if json_data: logger.info(f"Request data: {json_data}") async with httpx.AsyncClient() as client: try: if method.upper() == "GET": response = await client.get(url, headers=headers, params=params, timeout=30.0) elif method.upper() == "POST": response = await client.post(url, headers=headers, params=params, json=json_data, timeout=30.0) else: logger.error(f"ERROR: Invalid HTTP method: {method}") return None response.raise_for_status() logger.info(f"API response status: {response.status_code}") # If the endpoint is for subtitles, directly return the text content if "/subtitles" in endpoint: return response.text # For all other endpoints, return the JSON response return response.json() except Exception as e: logger.error(f"API request error: {str(e)}") return None
• src/youtube_translate_mcp/server.py:387-387 (schema)

  Input schema defined by function parameters: url (str, required), language (str, default 'en'), format (str, default 'srt'). Output: str (subtitle content).
  async def get_subtitles(url: str, language: str = "en", format: str = "srt") -> str: