YouTube Translate MCP

Instructions

Implementation Reference

• src/youtube_translate_mcp/server.py:491-547 (handler)

  The handler function for the 'search_video' tool. It processes the provided YouTube URL to obtain a video ID, makes an API request to search the transcript for the given query, handles errors, and formats the search results showing matches with timestamps and context.
  @mcp_server.tool(name="search_video", description="Search for specific content within a YouTube video's transcript") async def search_video(url: str, query: str) -> str: """Search for specific content within a YouTube video's transcript. This tool processes a video and searches for specific terms or phrases within its transcript. It properly handles "processing" states by polling until completion rather than failing immediately. Args: url: The YouTube video URL query: The search term or phrase to look for Returns: Search results with context from the video transcript """ logger.info(f"Searching video for URL: {url}, query: {query}") # 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}" # Search the video transcript search_response = await make_yt_api_request( f"/api/videos/{video_id}/search", params={"query": query} ) if not search_response: error_msg = f"Failed to search video for query: {query}" logger.error(error_msg) return f"Error: {error_msg}" # Format the search results if isinstance(search_response, dict) and "matches" in search_response: matches = search_response["matches"] if not matches: return f"No matches found for '{query}' in the video." result = f"Found {len(matches)} matches for '{query}':\n\n" for i, match in enumerate(matches, 1): timestamp = match.get("timestamp", "Unknown") text = match.get("text", "") timestamp_str = f"{int(timestamp) // 60}:{int(timestamp) % 60:02d}" if isinstance(timestamp, (int, float)) else timestamp result += f"{i}. [{timestamp_str}] {text}\n\n" return result elif isinstance(search_response, dict) and "error" in search_response: error_msg = search_response["error"] logger.error(f"API error: {error_msg}") return f"Error: {error_msg}" else: error_msg = "Unexpected response format from API." logger.error(error_msg) return f"Error: {error_msg}"
• src/youtube_translate_mcp/server.py:491-491 (registration)

  Registers the 'search_video' tool with the MCP server using the @mcp_server.tool decorator, specifying the name and description.
  @mcp_server.tool(name="search_video", description="Search for specific content within a YouTube video's transcript")
• src/youtube_translate_mcp/server.py:492-492 (schema)

  Function signature defining the input schema (url: str, query: str) and output (str) for the search_video tool.
  async def search_video(url: str, query: str) -> str:
• src/youtube_translate_mcp/server.py:108-203 (helper)

  Helper function that processes the YouTube video URL: extracts ID, checks status, submits if needed, and polls until processing is complete. Crucial for search_video as it ensures the transcript is available.
  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)

  Helper function to make authenticated requests to the YouTube Translate API, used by search_video to call the /search endpoint.
  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