YouTube MCP Server

from typing import Any, Optional, List
import os
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from youtube_transcript_api import YouTubeTranscriptApi, TranscriptsDisabled, NoTranscriptFound
import pickle

# If modifying these scopes, delete the file token.pickle.
SCOPES = ['https://www.googleapis.com/auth/youtube.readonly']

def get_authenticated_service():
    """Get authenticated YouTube API service."""
    creds = None
    # The file token.pickle stores the user's access and refresh tokens
    if os.path.exists('token.pickle'):
        with open('token.pickle', 'rb') as token:
            creds = pickle.load(token)
    
    # If there are no (valid) credentials available, let the user log in.
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            if not os.path.exists('credentials.json'):
                raise FileNotFoundError(
                    "credentials.json not found. Please download it from Google Cloud Console"
                )
            flow = InstalledAppFlow.from_client_secrets_file(
                'credentials.json', SCOPES)
            creds = flow.run_local_server(port=0)
        # Save the credentials for the next run
        with open('token.pickle', 'wb') as token:
            pickle.dump(creds, token)

    return build('youtube', 'v3', credentials=creds)

def get_video_details(video_id: str) -> Optional[dict[str, Any]]:
    """Get detailed information about a video."""
    try:
        youtube = get_authenticated_service()
        request = youtube.videos().list(
            part="snippet,statistics,contentDetails",
            id=video_id
        )
        response = request.execute()
        
        if not response['items']:
            return None
            
        video = response['items'][0]
        snippet = video['snippet']
        stats = video['statistics']
        content = video['contentDetails']
        
        return {
            'id': video_id,
            'title': snippet['title'],
            'description': snippet['description'],
            'channel_id': snippet['channelId'],
            'channel_title': snippet['channelTitle'],
            'published_at': snippet['publishedAt'],
            'duration': content['duration'],
            'view_count': stats.get('viewCount', '0'),
            'like_count': stats.get('likeCount', '0'),
            'comment_count': stats.get('commentCount', '0'),
            'tags': snippet.get('tags', []),
            'thumbnails': snippet['thumbnails']
        }
    except HttpError as e:
        print(f"An HTTP error occurred: {e}")
        return None

def get_channel_info(channel_id: str) -> Optional[dict[str, Any]]:
    """Get information about a channel."""
    try:
        youtube = get_authenticated_service()
        request = youtube.channels().list(
            part="snippet,statistics",
            id=channel_id
        )
        response = request.execute()
        
        if not response['items']:
            return None
            
        channel = response['items'][0]
        snippet = channel['snippet']
        stats = channel['statistics']
        
        return {
            'id': channel_id,
            'title': snippet['title'],
            'description': snippet['description'],
            'subscriber_count': stats.get('subscriberCount', '0'),
            'video_count': stats.get('videoCount', '0'),
            'view_count': stats.get('viewCount', '0'),
            'thumbnails': snippet['thumbnails'],
            'published_at': snippet['publishedAt']
        }
    except HttpError as e:
        print(f"An HTTP error occurred: {e}")
        return None

def get_video_comments(video_id: str, max_results: int = 100) -> list[dict[str, Any]]:
    """Get comments for a video."""
    try:
        youtube = get_authenticated_service()
        comments = []
        
        request = youtube.commentThreads().list(
            part="snippet",
            videoId=video_id,
            maxResults=min(max_results, 100),
            textFormat="plainText"
        )
        
        while request and len(comments) < max_results:
            response = request.execute()
            
            for item in response['items']:
                comment = item['snippet']['topLevelComment']['snippet']
                comments.append({
                    'author': comment['authorDisplayName'],
                    'text': comment['textDisplay'],
                    'like_count': comment['likeCount'],
                    'published_at': comment['publishedAt']
                })
                
                if len(comments) >= max_results:
                    break
            
            # Get the next page of comments
            request = youtube.commentThreads().list_next(request, response)
            
        return comments
    except HttpError as e:
        print(f"An HTTP error occurred: {e}")
        return []

def get_trending_videos(region_code: str = "US", max_results: int = 50) -> list[dict[str, Any]]:
    """Get trending videos for a region."""
    try:
        youtube = get_authenticated_service()
        request = youtube.videos().list(
            part="snippet,statistics",
            chart="mostPopular",
            regionCode=region_code,
            maxResults=min(max_results, 50)
        )
        response = request.execute()
        
        videos = []
        for item in response['items']:
            snippet = item['snippet']
            stats = item['statistics']
            videos.append({
                'id': item['id'],
                'title': snippet['title'],
                'channel_id': snippet['channelId'],
                'channel_title': snippet['channelTitle'],
                'published_at': snippet['publishedAt'],
                'view_count': stats.get('viewCount', '0'),
                'like_count': stats.get('likeCount', '0'),
                'comment_count': stats.get('commentCount', '0'),
                'thumbnails': snippet['thumbnails']
            })
        
        return videos
    except HttpError as e:
        print(f"An HTTP error occurred: {e}")
        return []

def get_related_videos(video_id: str, max_results: int = 25) -> list[dict[str, Any]]:
    """Get videos related to a specific video."""
    try:
        youtube = get_authenticated_service()
        request = youtube.search().list(
            part="snippet",
            relatedToVideoId=video_id,
            type="video",
            maxResults=min(max_results, 25)
        )
        response = request.execute()
        
        videos = []
        for item in response['items']:
            snippet = item['snippet']
            videos.append({
                'id': item['id']['videoId'],
                'title': snippet['title'],
                'channel_id': snippet['channelId'],
                'channel_title': snippet['channelTitle'],
                'published_at': snippet['publishedAt'],
                'description': snippet['description'],
                'thumbnails': snippet['thumbnails']
            })
        
        return videos
    except HttpError as e:
        print(f"An HTTP error occurred: {e}")
        return []

def get_video_transcript(video_id: str) -> Optional[List[dict[str, Any]]]:
    """Get the transcript for a video.
    
    Args:
        video_id: YouTube video ID
        
    Returns:
        List of transcript segments with text and timing information, or None if transcript is not available
    """
    try:
        transcript_list = YouTubeTranscriptApi.get_transcript(video_id)
        return transcript_list
    except (TranscriptsDisabled, NoTranscriptFound):
        return None
    except Exception as e:
        print(f"An error occurred while getting transcript: {e}")
        return None