YouTube Search MCP Server

#!/usr/bin/env python3 """ YouTube Search MCP Server A fast MCP server that provides YouTube search functionality using the YouTube Data API v3. This server implements search tools with comprehensive filtering and pagination support. """ import os import requests from typing import Optional, Dict, List, Any from urllib.parse import urlencode from datetime import datetime import json from fastmcp import FastMCP class YouTubeSearchAPI: """YouTube Search API client with comprehensive search capabilities.""" def __init__(self, api_key: str): self.api_key = api_key self.base_url = 'https://www.googleapis.com/youtube/v3/search' self.session = requests.Session() def search(self, query: str, **kwargs) -> Dict[str, Any]: """ Search YouTube with comprehensive filtering options. Args: query: Search query string **kwargs: Additional search parameters Returns: YouTube API response with search results """ params = { 'part': 'snippet', 'q': query, 'key': self.api_key, 'maxResults': kwargs.get('maxResults', 25), 'type': kwargs.get('type', 'video'), 'order': kwargs.get('order', 'relevance') } # Add optional parameters if provided optional_params = [ 'channelId', 'channelType', 'pageToken', 'publishedAfter', 'publishedBefore', 'regionCode', 'relevanceLanguage', 'videoDuration', 'videoDefinition', 'videoLicense' ] for param in optional_params: if param in kwargs: params[param] = kwargs[param] try: response = self.session.get(self.base_url, params=params) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: raise Exception(f"YouTube API Error: {e}") # Initialize FastMCP server mcp = FastMCP("YouTube Search Server") # Initialize YouTube API client youtube_api = YouTubeSearchAPI(os.getenv('YOUTUBE_API_KEY', '')) @mcp.tool def search_youtube( query: str, max_results: int = 25, search_type: str = "video", order: str = "relevance", channel_id: Optional[str] = None, published_after: Optional[str] = None, published_before: Optional[str] = None, region_code: Optional[str] = None, language: Optional[str] = None, video_duration: Optional[str] = None, video_definition: Optional[str] = None, page_token: Optional[str] = None ) -> Dict[str, Any]: """ Search YouTube videos, channels, or playlists with comprehensive filtering. Args: query: Search query string max_results: Number of results (0-50, default: 25) search_type: Type of content ('video', 'channel', 'playlist', default: 'video') order: Sort order ('relevance', 'date', 'rating', 'viewCount', 'title', default: 'relevance') channel_id: Limit search to specific channel published_after: Results published after this date (RFC 3339 format) published_before: Results published before this date (RFC 3339 format) region_code: Geographic region (ISO 3166-1 alpha-2 country code) language: Language preference (ISO 639-1 two-letter language code) video_duration: Video length filter ('any', 'short', 'medium', 'long') video_definition: Video quality filter ('any', 'high', 'standard') page_token: Pagination token for next/previous page Returns: YouTube search results with metadata """ if not youtube_api.api_key: raise Exception("YouTube API key not found. Please set YOUTUBE_API_KEY environment variable.") # Validate parameters if max_results < 0 or max_results > 50: raise ValueError("max_results must be between 0 and 50") valid_types = ['video', 'channel', 'playlist'] if search_type not in valid_types: raise ValueError(f"search_type must be one of: {valid_types}") valid_orders = ['relevance', 'date', 'rating', 'viewCount', 'title'] if order not in valid_orders: raise ValueError(f"order must be one of: {valid_orders}") # Build search parameters search_params = { 'maxResults': max_results, 'type': search_type, 'order': order } # Add optional parameters if channel_id: search_params['channelId'] = channel_id if published_after: search_params['publishedAfter'] = published_after if published_before: search_params['publishedBefore'] = published_before if region_code: search_params['regionCode'] = region_code if language: search_params['relevanceLanguage'] = language if video_duration: search_params['videoDuration'] = video_duration if video_definition: search_params['videoDefinition'] = video_definition if page_token: search_params['pageToken'] = page_token try: results = youtube_api.search(query, **search_params) # Process results for better readability processed_results = { 'search_info': { 'query': query, 'total_results': results.get('pageInfo', {}).get('totalResults', 0), 'results_per_page': results.get('pageInfo', {}).get('resultsPerPage', 0), 'next_page_token': results.get('nextPageToken'), 'prev_page_token': results.get('prevPageToken') }, 'items': [] } for item in results.get('items', []): processed_item = { 'id': item.get('id', {}), 'title': item.get('snippet', {}).get('title', ''), 'description': item.get('snippet', {}).get('description', ''), 'channel_title': item.get('snippet', {}).get('channelTitle', ''), 'channel_id': item.get('snippet', {}).get('channelId', ''), 'published_at': item.get('snippet', {}).get('publishedAt', ''), 'thumbnails': item.get('snippet', {}).get('thumbnails', {}), 'kind': item.get('kind', ''), 'live_broadcast_content': item.get('snippet', {}).get('liveBroadcastContent', 'none') } # Add URL based on content type if item.get('id', {}).get('kind') == 'youtube#video': processed_item['url'] = f"https://www.youtube.com/watch?v={item['id']['videoId']}" elif item.get('id', {}).get('kind') == 'youtube#channel': processed_item['url'] = f"https://www.youtube.com/channel/{item['id']['channelId']}" elif item.get('id', {}).get('kind') == 'youtube#playlist': processed_item['url'] = f"https://www.youtube.com/playlist?list={item['id']['playlistId']}" processed_results['items'].append(processed_item) return processed_results except Exception as e: return {'error': str(e)} @mcp.tool def search_youtube_videos( query: str, max_results: int = 25, order: str = "relevance", duration: Optional[str] = None, definition: Optional[str] = None, published_after: Optional[str] = None, published_before: Optional[str] = None, region_code: Optional[str] = None ) -> Dict[str, Any]: """ Search specifically for YouTube videos with video-specific filters. Args: query: Search query string max_results: Number of results (0-50, default: 25) order: Sort order ('relevance', 'date', 'rating', 'viewCount', 'title') duration: Video duration filter ('any', 'short', 'medium', 'long') definition: Video quality filter ('any', 'high', 'standard') published_after: Results published after this date (RFC 3339 format) published_before: Results published before this date (RFC 3339 format) region_code: Geographic region (ISO 3166-1 alpha-2 country code) Returns: YouTube video search results """ return search_youtube.fn( query=query, max_results=max_results, search_type="video", order=order, video_duration=duration, video_definition=definition, published_after=published_after, published_before=published_before, region_code=region_code ) @mcp.tool def search_youtube_channels( query: str, max_results: int = 25, order: str = "relevance", region_code: Optional[str] = None ) -> Dict[str, Any]: """ Search specifically for YouTube channels. Args: query: Search query string max_results: Number of results (0-50, default: 25) order: Sort order ('relevance', 'date', 'rating', 'viewCount', 'title') region_code: Geographic region (ISO 3166-1 alpha-2 country code) Returns: YouTube channel search results """ return search_youtube.fn( query=query, max_results=max_results, search_type="channel", order=order, region_code=region_code ) @mcp.tool def search_youtube_playlists( query: str, max_results: int = 25, order: str = "relevance", region_code: Optional[str] = None ) -> Dict[str, Any]: """ Search specifically for YouTube playlists. Args: query: Search query string max_results: Number of results (0-50, default: 25) order: Sort order ('relevance', 'date', 'rating', 'viewCount', 'title') region_code: Geographic region (ISO 3166-1 alpha-2 country code) Returns: YouTube playlist search results """ return search_youtube.fn( query=query, max_results=max_results, search_type="playlist", order=order, region_code=region_code ) @mcp.tool def search_youtube_by_channel( channel_id: str, query: Optional[str] = None, max_results: int = 25, order: str = "date" ) -> Dict[str, Any]: """ Search within a specific YouTube channel. Args: channel_id: YouTube channel ID query: Optional search query within the channel max_results: Number of results (0-50, default: 25) order: Sort order ('relevance', 'date', 'rating', 'viewCount', 'title') Returns: YouTube search results from the specified channel """ return search_youtube.fn( query=query or "", max_results=max_results, search_type="video", order=order, channel_id=channel_id ) @mcp.resource("help://youtube-search-help") def youtube_search_help() -> str: """ Get comprehensive help for YouTube search functionality. Returns: Detailed help documentation for YouTube search tools """ return """ # YouTube Search MCP Server Help This server provides comprehensive YouTube search functionality with the following tools: ## Main Search Tool - **search_youtube**: Full-featured search with all available filters - Supports videos, channels, and playlists - Comprehensive filtering options - Pagination support ## Specialized Search Tools - **search_youtube_videos**: Video-specific search with video filters - **search_youtube_channels**: Channel-specific search - **search_youtube_playlists**: Playlist-specific search - **search_youtube_by_channel**: Search within a specific channel ## Common Parameters - **query**: Search term - **max_results**: Number of results (0-50, default: 25) - **order**: Sort order (relevance, date, rating, viewCount, title) - **region_code**: Geographic region (ISO 3166-1 alpha-2 country code) ## Video-Specific Parameters - **duration**: Video length (any, short, medium, long) - **definition**: Video quality (any, high, standard) - **published_after/before**: Date filtering (RFC 3339 format) ## Setup Requirements 1. Set YOUTUBE_API_KEY environment variable with your Google API key 2. Enable YouTube Data API v3 in Google Cloud Console 3. Monitor quota usage (100 units per search request) ## Response Format All search results include: - Search metadata (total results, pagination tokens) - Processed items with titles, descriptions, URLs - Channel information and publication dates - Thumbnail URLs in multiple sizes ## Rate Limiting - Daily quota: 10,000 units (default) - Each search costs 100 units - Implement caching for frequently searched terms - Use pagination tokens for large result sets ## Error Handling - Invalid API key: Check YOUTUBE_API_KEY environment variable - Quota exceeded: Monitor usage in Google Cloud Console - Invalid parameters: Check parameter values and types - Network errors: Implement retry logic with exponential backoff """ if __name__ == "__main__": # Check for API key if not os.getenv('YOUTUBE_API_KEY'): print("Warning: YOUTUBE_API_KEY environment variable not set") print("Please set your YouTube Data API v3 key before running the server") print("Example: export YOUTUBE_API_KEY='your_api_key_here'") # Run the server # Default: STDIO transport (no port needed) mcp.run() # Alternative transport options: # For HTTP transport on port 8000: # mcp.run(transport="http", host="0.0.0.0", port=8000) # For SSE transport: # mcp.run(transport="sse", host="0.0.0.0", port=8000)