Prepare for Hugging Face Spaces deployment

.gitignore

@@ -0,0 +1,45 @@
+# Python-generated files
+__pycache__/
+*.py[oc]
+*.pyc
+*.pyo
+*.pyd
+build/
+dist/
+wheels/
+*.egg-info/
+*.egg
+
+# Virtual environments
+.venv/
+venv/
+env/
+
+# Environment files
+.env
+.env.local
+.env.development.local
+.env.test.local
+.env.production.local
+
+# IDE files
+.vscode/
+.idea/
+*.swp
+*.swo
+*~
+
+# OS files
+.DS_Store
+Thumbs.db
+
+# Logs
+*.log
+
+# Temporary files
+*.tmp
+*.temp
+
+# Test files
+test_*.py
+debug_*.py

README.md

@@ -0,0 +1,157 @@
+# 🎬 YouTube Creator MetaData Extractor
+
+AI-powered tool for content creators to analyze YouTube videos and generate professional metadata using advanced language models.
+
+## 🚀 Features
+
+- **🔍 Video Search**: Search YouTube videos by keywords with advanced filters
+- **📊 Video Analysis**: Extract comprehensive video metadata (views, likes, duration, etc.)
+- **📝 Transcript Extraction**: Get video transcripts in multiple languages
+- **⏱️ Smart Timecodes**: AI-generated timecodes for better video navigation
+- **🤖 Gemini AI Integration**: Advanced timecode generation using Google's Gemini 2.0
+- **🌐 Multi-language Support**: Works with videos in Ukrainian, Russian, English, and more
+- **📱 URL Flexibility**: Supports all YouTube URL formats (regular, shorts, embed links)
+
+## 🛠️ Setup
+
+### Required API Keys
+
+To use this tool, you need two API keys:
+
+1. **YouTube Data API v3 Key**
+   - Go to [Google Cloud Console](https://console.developers.google.com/)
+   - Create a new project or select existing
+   - Enable "YouTube Data API v3"
+   - Create credentials (API Key)
+
+2. **Gemini API Key** (for AI features)
+   - Visit [Google AI Studio](https://ai.google.dev/)
+   - Get your free API key for Gemini
+
+### Environment Variables
+
+Set these in your Hugging Face Space settings:
+
+```
+YOUTUBE_API_KEY=your_youtube_api_key_here
+GEMINI_API_KEY=your_gemini_api_key_here
+```
+
+## 📖 How to Use
+
+### 1. Video Search
+- Enter keywords to find YouTube videos
+- Filter by upload date, view count, duration
+- Get detailed metadata for any video
+
+### 2. Transcript Analysis
+- Extract transcripts from videos with subtitles
+- Support for auto-generated and manual captions
+- Multiple language detection and support
+
+### 3. Timecode Generation
+
+**Basic Timecodes**: Algorithmic segmentation based on transcript timing
+**AI Timecodes**: Intelligent topic-based segmentation using Gemini AI
+
+**Supported Formats**:
+- **YouTube**: Ready for video descriptions (e.g., `05:30 Topic description`)
+- **Markdown**: Clickable links with timestamps (e.g., `- [05:30](link) Topic`)
+
+**Language Codes**:
+- `uk` - Ukrainian
+- `ru` - Russian  
+- `en` - English
+- And many others (ISO 639-1 standard)
+
+## 🔧 API Reference
+
+This application provides both a web interface and REST API endpoints:
+
+### Search Videos
+```http
+POST /api/search
+{
+  "query": "your search query",
+  "max_results": 10,
+  "order": "relevance"
+}
+```
+
+### Get Video Info
+```http
+POST /api/video_info
+{
+  "video_id": "video_id_or_full_url"
+}
+```
+
+### Extract Transcript
+```http
+POST /api/transcript
+{
+  "video_id": "video_id_or_full_url",
+  "language_code": "uk"
+}
+```
+
+### Generate AI Timecodes
+```http
+POST /api/gemini_timecodes
+{
+  "video_id": "video_id_or_full_url",
+  "language_code": "uk",
+  "format": "youtube",
+  "model": "gemini-2.0-flash-001"
+}
+```
+
+## 🏗️ Architecture
+
+- **Frontend**: Gradio web interface with responsive design
+- **Backend**: FastAPI server with async processing
+- **AI Integration**: Google Gemini 2.0 for intelligent content analysis
+- **APIs**: YouTube Data API v3 for video metadata
+- **Transcript**: YouTube Transcript API for subtitle extraction
+
+## 📁 Project Structure
+
+```
+├── app.py               # Main Gradio application (HF Spaces entry point)
+├── api_server.py        # FastAPI backend server
+├── gemini_helper.py     # Gemini AI integration
+├── utils.py             # Utility functions
+├── models.py            # Data models
+├── mcp_handlers.py      # Model Context Protocol handlers
+├── requirements.txt     # Python dependencies
+└── README.md           # This file
+```
+
+## 🔬 Technology Stack
+
+- **Python 3.13+**
+- **Gradio** - Web interface framework
+- **FastAPI** - High-performance API framework
+- **Google Gemini 2.0** - Advanced language model for content analysis
+- **YouTube APIs** - Official Google APIs for video data
+- **AsyncIO** - Asynchronous processing for better performance
+
+## 🌟 Use Cases
+
+- **Content Creators**: Generate professional timecodes for YouTube videos
+- **Educators**: Extract and analyze educational content structure
+- **Researchers**: Analyze video metadata and transcripts at scale
+- **Marketers**: Research competitor content and trends
+- **Accessibility**: Create better navigation for long-form content
+
+## 📄 License
+
+MIT License - feel free to use in your projects!
+
+## 🤝 Contributing
+
+Contributions welcome! This project is designed to help content creators worldwide.
+
+---
+
+**Made with ❤️ for the YouTube creator community**

api_server.py

@@ -0,0 +1,559 @@
+import os
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+from typing import Dict, List, Optional, Any, Union
+import httpx
+from googleapiclient.discovery import build
+from googleapiclient.errors import HttpError
+import json
+from youtube_transcript_api import YouTubeTranscriptApi
+from youtube_transcript_api.formatters import JSONFormatter
+from dotenv import load_dotenv
+from utils import format_timestamp, extract_video_id
+from models import MCPResponse
+import re
+
+# Загрузка переменных окружения
+load_dotenv()
+
+# Получение API ключа YouTube из переменных окружения
+YOUTUBE_API_KEY = os.getenv("YOUTUBE_API_KEY")
+
+app = FastAPI(
+    title="YouTube MCP API",
+    description="Model Context Protocol (MCP) server for interacting with YouTube API",
+    version="0.1.0",
+)
+
+# Настройка CORS
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Инициализация YouTube API клиента
+def get_youtube_client():
+    if not YOUTUBE_API_KEY:
+        raise HTTPException(status_code=500, detail="YouTube API key is not configured")
+    
+    try:
+        return build("youtube", "v3", developerKey=YOUTUBE_API_KEY)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"YouTube API initialization error: {str(e)}")
+
+# Базовые модели данных для стандартных API запросов
+class SearchRequest(BaseModel):
+    query: str
+    max_results: Optional[int] = 10
+    order: Optional[str] = "relevance"
+    video_duration: Optional[str] = None
+    
+class VideoInfoRequest(BaseModel):
+    video_id: str
+    
+class TranscriptRequest(BaseModel):
+    video_id: str
+    language_code: Optional[str] = None
+
+class MCPRequestData(BaseModel):
+    action: str
+    parameters: Dict[str, Any]
+
+# Добавим новый маршрут для получения доступных языков транскрипта
+class TranscriptLanguagesRequest(BaseModel):
+    video_id: str
+
+# Модель для запроса тайм-кодов
+class TimecodeRequest(BaseModel):
+    video_id: str
+    language_code: Optional[str] = None
+    segment_length: Optional[int] = 60  # Длина сегмента в секундах
+    format: Optional[str] = "youtube"  # youtube, markdown
+
+# Загрузим модуль gemini_helper только после определения базовых моделей
+from gemini_helper import generate_timecodes_with_gemini, DEFAULT_MODEL
+
+# Модель для запроса тайм-кодов с помощью Gemini
+class GeminiTimecodeRequest(BaseModel):
+    video_id: str
+    language_code: Optional[str] = None
+    format: Optional[str] = "youtube"  # youtube, markdown
+    model: Optional[str] = DEFAULT_MODEL  # модель Gemini (если None, используется модель по умолчанию)
+
+# Теперь можно загрузить mcp_handlers
+from mcp_handlers import (
+    MCPQueryRequest, 
+    MCPVideoRequest, 
+    MCPTranscriptRequest,
+    MCPTimecodeRequest,
+    MCPGeminiRequest,
+    process_mcp_search,
+    process_mcp_video_info,
+    process_mcp_transcript,
+    process_mcp_timecodes,
+    process_mcp_gemini_timecodes,
+    create_text_response,
+    create_error_response
+)
+
+def normalize_language_code(language_code: str) -> str:
+    """Normalize language codes, converting common aliases to standard codes."""
+    if not language_code:
+        return language_code
+    
+    language_code = language_code.lower().strip()
+    
+    # Convert 'ua' to 'uk' for Ukrainian
+    if language_code == 'ua':
+        return 'uk'
+    
+    return language_code
+
+# Стандартные API маршруты
+@app.post("/api/search")
+async def search_videos(request: SearchRequest):
+    try:
+        youtube = get_youtube_client()
+        search_response = youtube.search().list(
+            q=request.query,
+            part="snippet",
+            maxResults=request.max_results,
+            type="video",
+            order=request.order,
+            videoDuration=request.video_duration if request.video_duration else None
+        ).execute()
+        
+        results = []
+        for item in search_response.get("items", []):
+            video_id = item["id"]["videoId"]
+            snippet = item["snippet"]
+            
+            results.append({
+                "video_id": video_id,
+                "title": snippet["title"],
+                "description": snippet["description"],
+                "thumbnail": snippet["thumbnails"]["high"]["url"],
+                "channel_title": snippet["channelTitle"],
+                "published_at": snippet["publishedAt"]
+            })
+            
+        return {"content": results}
+    except HttpError as e:
+        return {"error": f"YouTube API error: {str(e)}"}
+    except Exception as e:
+        return {"error": f"Unexpected error: {str(e)}"}
+
+@app.post("/api/video_info")
+async def get_video_info(request: VideoInfoRequest):
+    try:
+        # Извлекаем ID видео из ссылки, если это ссылка
+        video_id = extract_video_id(request.video_id)
+        
+        youtube = get_youtube_client()
+        video_response = youtube.videos().list(
+            part="snippet,contentDetails,statistics",
+            id=video_id
+        ).execute()
+        
+        if not video_response.get("items"):
+            return {"error": "Video not found"}
+        
+        video = video_response["items"][0]
+        snippet = video["snippet"]
+        content_details = video["contentDetails"]
+        statistics = video["statistics"]
+        
+        return {"content": {
+            "video_id": video_id,
+            "title": snippet["title"],
+            "description": snippet["description"],
+            "channel_title": snippet["channelTitle"],
+            "published_at": snippet["publishedAt"],
+            "duration": content_details["duration"],
+            "view_count": statistics.get("viewCount", "0"),
+            "like_count": statistics.get("likeCount", "0"),
+            "comment_count": statistics.get("commentCount", "0"),
+            "tags": snippet.get("tags", [])
+        }}
+    except HttpError as e:
+        return {"error": f"YouTube API error: {str(e)}"}
+    except Exception as e:
+        return {"error": f"Unexpected error: {str(e)}"}
+
+@app.post("/api/transcript")
+async def get_transcript(request: TranscriptRequest):
+    try:
+        # Extract video ID if URL is provided
+        video_id = extract_video_id(request.video_id)
+        
+        # Normalize language code (ua -> uk)
+        normalized_language = normalize_language_code(request.language_code)
+        
+        # Get list of available languages for the video
+        try:
+            available_languages = []
+            transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)
+            for transcript in transcript_list:
+                available_languages.append({
+                    "language": transcript.language,
+                    "language_code": transcript.language_code,
+                    "is_generated": transcript.is_generated,
+                    "is_translatable": transcript.is_translatable
+                })
+        except Exception as e:
+            print(f"Error getting language list: {str(e)}")
+            return {"error": f"Video not found or no transcripts available: {str(e)}"}
+        
+        print(f"Available languages for video {video_id}: {[lang['language_code'] for lang in available_languages]}")
+        
+        # Try to get transcript in requested language
+        final_language = None
+        transcript_list = None
+        
+        if normalized_language:
+            try:
+                print(f"Trying to get transcript in language: {normalized_language}")
+                transcript_list = YouTubeTranscriptApi.get_transcript(video_id, languages=[normalized_language])
+                print(f"Successfully obtained transcript in language: {normalized_language}")
+                final_language = normalized_language
+            except Exception as e:
+                print(f"Failed to get transcript in language {normalized_language}: {str(e)}")
+        
+        # If specific language failed or not requested, try first available
+        if transcript_list is None and available_languages:
+            try:
+                first_language = available_languages[0]['language_code']
+                print(f"Trying to use first available language: {first_language}")
+                transcript_list = YouTubeTranscriptApi.get_transcript(video_id, languages=[first_language])
+                print(f"Successfully obtained transcript in language: {first_language}")
+                final_language = first_language
+            except Exception as e:
+                print(f"Failed to get transcript in language {first_language}: {str(e)}")
+                return {"error": f"Failed to get transcript in any available language: {str(e)}"}
+        
+        if not transcript_list:
+            return {"error": "Transcript for this video is unavailable"}
+        
+        formatted_transcript = []
+        for entry in transcript_list:
+            formatted_transcript.append({
+                "text": entry.get("text", ""),
+                "start": entry.get("start", 0),
+                "duration": entry.get("duration", 0)
+            })
+        
+        response = {"content": formatted_transcript}
+        if final_language:
+            response["used_language"] = final_language
+            
+        return response
+    except Exception as e:
+        return {"error": f"Error getting transcript: {str(e)}"}
+
+@app.post("/api/transcript_languages")
+async def get_transcript_languages(request: TranscriptLanguagesRequest):
+    try:
+        # Извлекаем ID вид��о из ссылки, если это ссылка
+        video_id = extract_video_id(request.video_id)
+        
+        try:
+            print(f"Getting language list for ID: {video_id}")
+            transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)
+            
+            languages = []
+            for transcript in transcript_list:
+                languages.append({
+                    "language_code": transcript.language_code,
+                    "language": transcript.language,
+                    "is_generated": transcript.is_generated
+                })
+            
+            return {"content": languages}
+        except Exception as transcript_error:
+            return {"error": f"Failed to get language list. Details: {str(transcript_error)}"}
+    except Exception as e:
+        return {"error": f"Error getting language list: {str(e)}"}
+
+# MCP эндпоинты
+@app.post("/api/mcp")
+async def mcp_endpoint(request: MCPRequestData):
+    try:
+        youtube = get_youtube_client()
+        
+        if request.action == "search":
+            search_req = MCPQueryRequest(**request.parameters)
+            result = await process_mcp_search(youtube, search_req)
+            return result
+        elif request.action == "video_info":
+            video_req = MCPVideoRequest(**request.parameters)
+            result = await process_mcp_video_info(youtube, video_req)
+            return result
+        elif request.action == "transcript":
+            transcript_req = MCPTranscriptRequest(**request.parameters)
+            result = await process_mcp_transcript(transcript_req)
+            return result
+        elif request.action == "timecodes":
+            timecode_req = MCPTimecodeRequest(**request.parameters)
+            result = await process_mcp_timecodes(youtube, timecode_req)
+            return result
+        elif request.action == "gemini_timecodes":
+            gemini_req = MCPGeminiRequest(**request.parameters)
+            result = await process_mcp_gemini_timecodes(youtube, gemini_req)
+            return result
+        else:
+            return create_error_response(f"Unknown action: {request.action}")
+    except Exception as e:
+        return create_error_response(f"Error processing request: {str(e)}")
+
+# Маршрут для проверки здоровья сервера
+@app.get("/health")
+async def health_check():
+    return {"status": "ok"}
+
+# Информационный маршрут, описывающий возможности API
+@app.get("/")
+async def root():
+    return {
+        "name": "YouTube MCP API",
+        "version": "0.1.0",
+        "description": "Model Context Protocol (MCP) server for interacting with YouTube API",
+        "endpoints": {
+            "standard": [
+                "/api/search - Search videos on YouTube",
+                "/api/video_info - Get video information",
+                "/api/transcript - Get video transcript"
+            ],
+            "mcp": [
+                "/api/mcp - Model Context Protocol endpoint"
+            ]
+        },
+        "actions": {
+            "search": "Search videos on YouTube",
+            "video_info": "Get video information",
+            "transcript": "Get video transcript"
+        }
+    }
+
+@app.post("/api/timecodes")
+async def generate_timecodes(request: TimecodeRequest):
+    try:
+        # Извлекаем ID видео из ссылки, если это ссылка
+        video_id = extract_video_id(request.video_id)
+        print(f"Generating timecodes for ID: {video_id}")
+        
+        # Пытаемся получить список доступных языков
+        available_languages = []
+        try:
+            transcript_list_obj = YouTubeTranscriptApi.list_transcripts(video_id)
+            for transcript in transcript_list_obj:
+                available_languages.append({
+                    "language_code": transcript.language_code,
+                    "language": transcript.language,
+                    "is_generated": transcript.is_generated
+                })
+            print(f"Available languages for video {video_id}: {[lang['language_code'] for lang in available_languages]}")
+        except Exception as e:
+            print(f"Failed to get language list: {str(e)}")
+        
+        # Получаем транскрипт
+        transcript_list = None
+        used_language = None
+        
+        # Если указан язык, пробуем его использовать
+        if request.language_code:
+            try:
+                print(f"Trying to get transcript in language: {request.language_code}")
+                transcript_list = YouTubeTranscriptApi.get_transcript(video_id, languages=[request.language_code])
+                used_language = request.language_code
+                print(f"Successfully obtained transcript in language: {request.language_code}")
+            except Exception as e:
+                print(f"Failed to get transcript in language {request.language_code}: {str(e)}")
+        
+        # Если транскрипт не получен и есть доступные языки, используем первый доступный
+        if not transcript_list and available_languages:
+            try:
+                first_language = available_languages[0]["language_code"]
+                print(f"Trying to use first available language: {first_language}")
+                transcript_list = YouTubeTranscriptApi.get_transcript(video_id, languages=[first_language])
+                used_language = first_language
+                print(f"Successfully obtained transcript in language: {first_language}")
+            except Exception as e:
+                print(f"Failed to get transcript in language {first_language}: {str(e)}")
+        
+        # Если все еще нет транскрипта, пробуем получить на любом языке
+        if not transcript_list:
+            try:
+                print("Trying to get transcript in any available language")
+                transcript_list = YouTubeTranscriptApi.get_transcript(video_id)
+                print("Transcript successfully obtained")
+            except Exception as e:
+                return {"error": f"Transcript not found. Details: {str(e)}"}
+        
+        if not transcript_list:
+            return {"error": "Transcript for this video is unavailable"}
+        
+        # Группируем транскрипт по сегментам
+        segments = []
+        current_segment = {
+            "start": transcript_list[0]["start"],
+            "end": 0,
+            "text": []
+        }
+        
+        segment_length = request.segment_length
+        
+        for entry in transcript_list:
+            start_time = entry["start"]
+            
+            # Если текущий сегмент пустой или запись находится в пределах длины сегмента
+            if not current_segment["text"] or (start_time - current_segment["start"]) <= segment_length:
+                current_segment["text"].append(entry["text"])
+                current_segment["end"] = start_time + entry["duration"]
+            else:
+                # Закрываем текущий сегмент и начинаем новый
+                segments.append(dict(current_segment))
+                current_segment = {
+                    "start": start_time,
+                    "end": start_time + entry["duration"],
+                    "text": [entry["text"]]
+                }
+        
+        # Добавляем последний сегмент
+        if current_segment["text"]:
+            segments.append(current_segment)
+        
+        # Форматируем тайм-коды в соответствии с выбранным форматом
+        format_type = request.format.lower()
+        timecodes = []
+        
+        for segment in segments:
+            start_formatted = format_timestamp(segment["start"])
+            
+            # Суммарный текст сегмента (первые 100 символов)
+            text_summary = " ".join(segment["text"])
+            if len(text_summary) > 100:
+                text_summary = text_summary[:97] + "..."
+            
+            if format_type == "youtube":
+                # Формат для YouTube (для вставки в описание)
+                timecodes.append(f"{start_formatted} {text_summary}")
+            elif format_type == "markdown":
+                # Формат для Markdown
+                youtube_link = f"https://www.youtube.com/watch?v={video_id}&t={int(segment['start'])}"
+                timecodes.append(f"- [{start_formatted}]({youtube_link}) {text_summary}")
+        
+        # Возвращаем тайм-коды и дополнительную информацию
+        response = {
+            "content": {
+                "video_id": video_id,
+                "timecodes": timecodes,
+                "format": format_type,
+                "segment_length": segment_length,
+                "total_segments": len(segments)
+            }
+        }
+        
+        if used_language:
+            response["content"]["used_language"] = used_language
+            
+        return response
+    except Exception as e:
+        return {"error": f"Error generating timecodes: {str(e)}"}
+
+@app.post("/api/gemini_timecodes")
+async def generate_gemini_timecodes(request: GeminiTimecodeRequest):
+    try:
+        # Extract video ID if URL is provided
+        video_id = extract_video_id(request.video_id)
+        print(f"Generating Gemini timecodes for ID: {video_id}")
+        
+        # Normalize language code (ua -> uk)
+        normalized_language = normalize_language_code(request.language_code)
+        
+        # Get list of available languages for the video
+        try:
+            available_languages = []
+            transcript_list_obj = YouTubeTranscriptApi.list_transcripts(video_id)
+            for transcript in transcript_list_obj:
+                available_languages.append({
+                    "language": transcript.language,
+                    "language_code": transcript.language_code,
+                    "is_generated": transcript.is_generated,
+                    "is_translatable": transcript.is_translatable
+                })
+        except Exception as e:
+            print(f"Error getting language list: {str(e)}")
+            return {"error": f"Video not found or no transcripts available: {str(e)}"}
+        
+        print(f"Available languages for video {video_id}: {[lang['language_code'] for lang in available_languages]}")
+        
+        # Try to get transcript in requested language
+        transcript_list = None
+        used_language = None
+        
+        if normalized_language:
+            try:
+                print(f"Trying to get transcript in language: {normalized_language}")
+                transcript_list = YouTubeTranscriptApi.get_transcript(video_id, languages=[normalized_language])
+                used_language = normalized_language
+                print(f"Successfully obtained transcript in language: {normalized_language}")
+            except Exception as e:
+                print(f"Failed to get transcript in language {normalized_language}: {str(e)}")
+        
+        # If specific language failed or not requested, try first available
+        if transcript_list is None and available_languages:
+            try:
+                first_language = available_languages[0]["language_code"]
+                print(f"Trying to use first available language: {first_language}")
+                transcript_list = YouTubeTranscriptApi.get_transcript(video_id, languages=[first_language])
+                used_language = first_language
+                print(f"Successfully obtained transcript in language: {first_language}")
+            except Exception as e:
+                print(f"Failed to get transcript in language {first_language}: {str(e)}")
+                return {"error": f"Failed to get transcript in any available language: {str(e)}"}
+        
+        if not transcript_list:
+            return {"error": "Transcript for this video is unavailable"}
+        
+        # Получаем информацию о видео для заголовка
+        youtube = get_youtube_client()
+        video_title = "YouTube Video"
+        
+        try:
+            video_response = youtube.videos().list(
+                part="snippet",
+                id=video_id
+            ).execute()
+            
+            if video_response.get("items"):
+                video_title = video_response["items"][0]["snippet"]["title"]
+        except Exception as e:
+            print(f"Failed to get video information: {str(e)}")
+        
+        # Отправляем запрос в Gemini с указанием языка
+        result = await generate_timecodes_with_gemini(
+            transcript_entries=transcript_list,
+            video_title=video_title,
+            format_type=request.format,
+            model_name=request.model,
+            language=used_language
+        )
+        
+        if "error" in result:
+            return {"error": result["error"]}
+        
+        # Добавляем информацию о языке транскрипта
+        if used_language:
+            result["used_language"] = used_language
+            
+        return {"content": result}
+    except Exception as e:
+        return {"error": f"Error generating timecodes with Gemini: {str(e)}"}
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="127.0.0.1", port=8080) 

app.py

@@ -0,0 +1,401 @@
+import gradio as gr
+import json
+import httpx
+import os
+import traceback
+import asyncio
+import threading
+import uvicorn
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from dotenv import load_dotenv
+from utils import format_timestamp, extract_video_id
+
+# Load environment variables
+load_dotenv()
+
+# Import API server components
+from api_server import app as fastapi_app
+
+# Start FastAPI server in background
+def start_fastapi_server():
+    uvicorn.run(fastapi_app, host="0.0.0.0", port=7860)
+
+# Start FastAPI server in a separate thread
+server_thread = threading.Thread(target=start_fastapi_server, daemon=True)
+server_thread.start()
+
+# Wait a moment for server to start
+import time
+time.sleep(2)
+
+# API URL for Hugging Face Spaces
+API_URL = "http://localhost:7860/api"
+
+async def search_youtube(query, max_results, order, video_duration):
+    """Function for searching videos on YouTube."""
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{API_URL}/search",
+                json={
+                    "query": query,
+                    "max_results": max_results,
+                    "order": order,
+                    "video_duration": video_duration if video_duration != "any" else None
+                }
+            )
+            data = response.json()
+            
+            if "error" in data and data["error"]:
+                return f"Error: {data['error']}", None
+                
+            results = data.get("content", [])
+            formatted_results = []
+            
+            for video in results:
+                formatted_results.append(
+                    f"**{video['title']}**\n"
+                    f"ID: {video['video_id']}\n"
+                    f"Channel: {video['channel_title']}\n"
+                    f"Published: {video['published_at']}\n"
+                    f"[Thumbnail]({video['thumbnail']})\n\n"
+                    f"{video['description'][:200]}...\n\n"
+                    f"---\n"
+                )
+            
+            return "\n".join(formatted_results), json.dumps(results, indent=2, ensure_ascii=False)
+    except Exception as e:
+        return f"Error: {str(e)}", None
+
+async def get_video_info(video_id):
+    """Function for getting video information."""
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{API_URL}/video_info",
+                json={"video_id": video_id}
+            )
+            data = response.json()
+            
+            if "error" in data and data["error"]:
+                return f"Error: {data['error']}", None
+                
+            video_info = data.get("content", {})
+            
+            formatted_info = (
+                f"**{video_info.get('title')}**\n\n"
+                f"Channel: {video_info.get('channel_title')}\n"
+                f"Published: {video_info.get('published_at')}\n"
+                f"Views: {video_info.get('view_count')}\n"
+                f"Likes: {video_info.get('like_count')}\n"
+                f"Comments: {video_info.get('comment_count')}\n"
+                f"Duration: {video_info.get('duration')}\n\n"
+                f"**Description:**\n{video_info.get('description')}\n\n"
+                f"**Tags:**\n{', '.join(video_info.get('tags', []))}"
+            )
+            
+            return formatted_info, json.dumps(video_info, indent=2, ensure_ascii=False)
+    except Exception as e:
+        return f"Error: {str(e)}", None
+
+async def get_transcript(video_id, language_code):
+    """Function for getting video transcript."""
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{API_URL}/transcript",
+                json={
+                    "video_id": video_id,
+                    "language_code": language_code if language_code else None
+                }
+            )
+            data = response.json()
+            
+            if "error" in data and data["error"]:
+                return f"Error: {data['error']}", None
+                
+            transcript = data.get("content", [])
+            
+            formatted_transcript = ""
+            for entry in transcript:
+                start_time = entry.get("start", 0)
+                duration = entry.get("duration", 0)
+                end_time = start_time + duration
+                
+                # Format time to hours:minutes:seconds format
+                start_formatted = format_timestamp(start_time)
+                end_formatted = format_timestamp(end_time)
+                
+                formatted_transcript += f"[{start_formatted} - {end_formatted}] {entry.get('text', '')}\n\n"
+            
+            return formatted_transcript, json.dumps(transcript, indent=2, ensure_ascii=False)
+    except Exception as e:
+        return f"Error: {str(e)}", None
+
+async def get_available_languages(video_id):
+    """Function for getting available transcript languages."""
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{API_URL}/transcript_languages",
+                json={"video_id": video_id}
+            )
+            data = response.json()
+            
+            if "error" in data and data["error"]:
+                return f"Error: {data['error']}", None
+                
+            languages = data.get("content", [])
+            
+            formatted_languages = []
+            for lang in languages:
+                status = "Auto-generated" if lang.get("is_generated") else "Official subtitles"
+                translatable = "Translation available" if lang.get("is_translatable") else "Translation not available"
+                formatted_languages.append(
+                    f"{lang.get('language')} ({lang.get('language_code')}): {status}, {translatable}"
+                )
+            
+            return "\n".join(formatted_languages), json.dumps(languages, indent=2, ensure_ascii=False)
+    except Exception as e:
+        return f"Error: {str(e)}", None
+
+async def generate_timecodes(video_id, language_code, segment_length, format_type):
+    """Function for generating timecodes."""
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{API_URL}/timecodes",
+                json={
+                    "video_id": video_id,
+                    "language_code": language_code if language_code else None,
+                    "segment_length": segment_length,
+                    "format": format_type
+                }
+            )
+            data = response.json()
+            
+            if "error" in data and data["error"]:
+                return f"Error: {data['error']}", None
+                
+            timecodes = data.get("content", {}).get("timecodes", [])
+            
+            if format_type == "youtube":
+                formatted_timecodes = "```\n" + "\n".join(timecodes) + "\n```"
+            elif format_type == "markdown":
+                formatted_timecodes = "\n".join(timecodes)
+            
+            return formatted_timecodes, json.dumps(data.get("content", {}), indent=2, ensure_ascii=False)
+    except Exception as e:
+        return f"Error: {str(e)}", None
+
+async def generate_gemini_timecodes(video_id, language_code, format_type, model):
+    """Function for generating timecodes using Gemini."""
+    try:
+        print(f"Sending request to {API_URL}/gemini_timecodes")
+        print(f"Parameters: video_id={video_id}, language_code={language_code}, format={format_type}, model={model}")
+        
+        # Send request to API
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{API_URL}/gemini_timecodes",
+                json={
+                    "video_id": video_id,
+                    "language_code": language_code,
+                    "format": format_type,
+                    "model": model
+                },
+                timeout=120  # Increase timeout for Gemini API
+            )
+            
+            print(f"Response status: {response.status_code}")
+            
+            # Parse response
+            data = response.json()
+            
+            if "error" in data:
+                print(f"Error in API response: {data['error']}")
+                return f"⚠️ Error: {data['error']}", {"error": data['error']}
+            
+            # Extract timecodes from response
+            content = data.get("content", {})
+            timecodes = content.get("timecodes", [])
+            
+            print(f"Received {len(timecodes)} timecodes")
+            
+            # Format timecodes for display
+            if timecodes:
+                timecodes_text = "\n".join(timecodes)
+                
+                # Model and language information
+                model_info = content.get("model", "Unknown")
+                language_info = content.get("detected_language", "Unknown")
+                duration_info = content.get("video_duration_minutes", "Unknown")
+                
+                summary = f"🤖 Model: {model_info}\n🗣️ Language: {language_info}\n⏱️ Duration: {duration_info} min\n📝 Timecodes: {len(timecodes)}"
+                
+                return summary, content  # Return content object instead of timecodes_text
+            else:
+                return "⚠️ No timecodes generated", {"message": "No timecodes generated"}
+            
+    except Exception as e:
+        print(f"Exception during timecode generation: {str(e)}")
+        traceback.print_exc()
+        return f"Error: {str(e)}", {"error": str(e)}
+
+# Create Gradio interface
+with gr.Blocks(title="YouTube MCP", theme=gr.themes.Soft()) as demo:
+    gr.Markdown("# 🎬 YouTube Creator MetaData Extractor")
+    gr.Markdown("This tool helps content creators analyze YouTube videos and generate metadata using AI")
+    gr.Markdown("### Supports all YouTube URL formats: regular links, short links, shorts and embedded videos")
+    gr.Markdown("**💡 Language codes:** uk = Ukrainian, ru = Russian, en = English (ISO 639-1 standard)")
+    gr.Markdown("---")
+    
+    with gr.Tab("🔍 Video Search"):
+        with gr.Row():
+            with gr.Column():
+                search_query = gr.Textbox(label="Search Query", placeholder="Enter your search query...")
+                with gr.Row():
+                    max_results = gr.Slider(minimum=1, maximum=50, value=10, step=1, label="Max Results")
+                    order = gr.Dropdown(
+                        choices=["relevance", "date", "viewCount", "rating", "title"], 
+                        value="relevance", 
+                        label="Sort By"
+                    )
+                    video_duration = gr.Dropdown(
+                        choices=["any", "short", "medium", "long"], 
+                        value="any", 
+                        label="Duration"
+                    )
+                search_button = gr.Button("🔍 Search", variant="primary")
+            
+            with gr.Column():
+                search_results = gr.Markdown(label="Search Results")
+                search_json = gr.JSON(label="JSON Data")
+        
+        search_button.click(
+            search_youtube, 
+            inputs=[search_query, max_results, order, video_duration], 
+            outputs=[search_results, search_json]
+        )
+    
+    with gr.Tab("ℹ️ Video Info"):
+        with gr.Row():
+            with gr.Column():
+                video_id_input = gr.Textbox(
+                    label="Video ID or URL", 
+                    placeholder="Enter video ID or full URL (youtube.com, youtu.be, shorts, embed)..."
+                )
+                get_info_button = gr.Button("📊 Get Info", variant="primary")
+            
+            with gr.Column():
+                video_info_output = gr.Markdown(label="Video Information")
+                video_info_json = gr.JSON(label="JSON Data")
+        
+        get_info_button.click(
+            get_video_info, 
+            inputs=[video_id_input], 
+            outputs=[video_info_output, video_info_json]
+        )
+    
+    with gr.Tab("📝 Transcript"):
+        with gr.Row():
+            with gr.Column():
+                transcript_video_id = gr.Textbox(
+                    label="Video ID or URL", 
+                    placeholder="Enter video ID or full URL..."
+                )
+                language_code = gr.Textbox(label="Language Code (optional)", placeholder="uk (Ukrainian), ru (Russian), en (English), etc...")
+                with gr.Row():
+                    get_transcript_button = gr.Button("📝 Get Transcript", variant="primary")
+                    get_languages_button = gr.Button("🌐 Available Languages")
+            
+            with gr.Column():
+                transcript_output = gr.Markdown(label="Transcript")
+                transcript_json = gr.JSON(label="JSON Data")
+        
+        get_transcript_button.click(
+            get_transcript, 
+            inputs=[transcript_video_id, language_code], 
+            outputs=[transcript_output, transcript_json]
+        )
+        
+        get_languages_button.click(
+            get_available_languages,
+            inputs=[transcript_video_id],
+            outputs=[transcript_output, transcript_json]
+        )
+
+    with gr.Tab("⏱️ Basic Timecodes"):
+        with gr.Row():
+            with gr.Column():
+                timecode_video_id = gr.Textbox(
+                    label="Video ID or URL", 
+                    placeholder="Enter video ID or full URL..."
+                )
+                timecode_language = gr.Textbox(label="Language Code (optional)", placeholder="uk (Ukrainian), ru (Russian), en (English), etc...")
+                segment_length = gr.Slider(minimum=30, maximum=300, value=60, step=30, label="Segment Length (seconds)")
+                format_type = gr.Dropdown(
+                    choices=["youtube", "markdown"], 
+                    value="youtube", 
+                    label="Format"
+                )
+                generate_timecodes_button = gr.Button("⏱️ Generate Timecodes", variant="primary")
+            
+            with gr.Column():
+                timecodes_output = gr.Markdown(label="Timecodes")
+                timecodes_json = gr.JSON(label="JSON Data")
+        
+        generate_timecodes_button.click(
+            generate_timecodes, 
+            inputs=[timecode_video_id, timecode_language, segment_length, format_type], 
+            outputs=[timecodes_output, timecodes_json]
+        )
+
+    with gr.Tab("🤖 AI Timecodes"):
+        with gr.Row():
+            with gr.Column():
+                gemini_video_id = gr.Textbox(
+                    label="Video ID or URL", 
+                    placeholder="Enter video ID or full URL..."
+                )
+                gemini_language = gr.Textbox(label="Language Code (optional)", placeholder="uk (Ukrainian), ru (Russian), en (English), etc...")
+                gemini_format = gr.Dropdown(
+                    choices=["youtube", "markdown"], 
+                    value="youtube", 
+                    label="Format"
+                )
+                gemini_model = gr.Dropdown(
+                    choices=["gemini-2.0-flash-001", "gemini-2.0-pro-001", "gemini-2.0-pro-vision-001"],
+                    value="gemini-2.0-flash-001",
+                    label="AI Model"
+                )
+                generate_gemini_button = gr.Button("🤖 Generate AI Timecodes", variant="primary")
+            
+            with gr.Column():
+                gemini_output = gr.Markdown(label="Generation Info")
+                gemini_timecodes = gr.Textbox(label="AI Timecodes", lines=10, max_lines=20, show_copy_button=True)
+                gemini_json = gr.JSON(label="JSON Data")
+        
+        async def process_gemini_result(video_id, language_code, format_type, model):
+            result = await generate_gemini_timecodes(video_id, language_code, format_type, model)
+            if result is None:
+                return "Error occurred", "", {}
+            
+            summary, json_data = result
+            
+            # Extract timecodes from json_data
+            timecodes = json_data.get("timecodes", [])
+            timecodes_text = "\n".join(timecodes) if timecodes else "No timecodes generated"
+            
+            return summary, timecodes_text, json_data
+        
+        generate_gemini_button.click(
+            process_gemini_result, 
+            inputs=[gemini_video_id, gemini_language, gemini_format, gemini_model], 
+            outputs=[gemini_output, gemini_timecodes, gemini_json]
+        )
+
+# Launch the app
+if __name__ == "__main__":
+    demo.launch() 

gemini_helper.py

@@ -0,0 +1,297 @@
+import os
+from google import genai
+from google.genai import types
+from dotenv import load_dotenv
+from typing import List, Dict, Any, Optional
+import traceback
+
+# Load environment variables
+load_dotenv()
+
+# Get Gemini API key from environment variables
+GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
+print(f"GEMINI_API_KEY is set: {'Yes' if GEMINI_API_KEY else 'No'}")
+
+# Initialize Gemini API
+client = None
+if GEMINI_API_KEY:
+    try:
+        client = genai.Client(api_key=GEMINI_API_KEY)
+        print("Gemini client successfully initialized")
+    except Exception as e:
+        print(f"Error initializing Gemini client: {str(e)}")
+        traceback.print_exc()
+else:
+    print("WARNING: Gemini API key not configured. LLM timecode generation functions will be unavailable.")
+
+# Default Gemini model
+DEFAULT_MODEL = "gemini-2.0-flash-001"
+# Alternative models if main one doesn't work
+ALTERNATIVE_MODELS = ["gemini-1.5-flash-001"]
+
+def format_transcript_for_prompt(transcript_entries: List[Dict[str, Any]], video_duration_seconds: int = None) -> str:
+    """Formats transcript for passing to prompt."""
+    formatted_transcript = ""
+    
+    # Determine maximum time in transcript if video duration is not provided
+    if video_duration_seconds is None:
+        if transcript_entries:
+            last_entry = transcript_entries[-1]
+            max_time = last_entry.get("start", 0) + last_entry.get("duration", 0)
+            video_duration_seconds = int(max_time) + 10  # Add small buffer
+    
+    for entry in transcript_entries:
+        start_time = entry.get("start", 0)
+        text = entry.get("text", "")
+        
+        # Check that time doesn't exceed total video duration
+        if video_duration_seconds and start_time > video_duration_seconds:
+            continue
+            
+        # Format time in hours:minutes:seconds format
+        time_str = format_time_hms(start_time)
+            
+        formatted_transcript += f"[{time_str}] {text}\n"
+    
+    return formatted_transcript
+
+def format_time_hms(seconds: float) -> str:
+    """
+    Formats time in seconds to hours:minutes:seconds format.
+    For videos shorter than an hour, uses minutes:seconds format.
+    """
+    hours = int(seconds // 3600)
+    minutes = int((seconds % 3600) // 60)
+    secs = int(seconds % 60)
+    
+    if hours > 0:
+        return f"{hours:02d}:{minutes:02d}:{secs:02d}"
+    else:
+        return f"{minutes:02d}:{secs:02d}"
+
+def get_timecode_prompt(video_title: str, transcript: str, format_type: str = "youtube", language: str = None, video_duration_minutes: int = None) -> str:
+    """Creates prompt for generating timecodes based on transcript."""
+    
+    # Determine prompt language based on video language
+    if language and (language.lower().startswith('uk') or language.lower().startswith('ua')):
+        target_language = "Ukrainian"
+        example_description = "Discussion of main principles"
+    elif language and language.lower().startswith('ru'):
+        target_language = "Russian"
+        example_description = "Обсуждение основных принципов"
+    else:
+        target_language = "the same language as the video transcript"
+        example_description = "Discussion of main principles"
+    
+    # Determine number of timecodes based on video duration
+    if video_duration_minutes:
+        if video_duration_minutes <= 30:
+            timecode_count = "10-15"
+        elif video_duration_minutes <= 60:
+            timecode_count = "15-20"
+        else:
+            timecode_count = "20-30"
+    else:
+        timecode_count = "15-25"
+    
+    if format_type == "youtube":
+        format_instructions = (
+            f"Format should be: MM:SS Topic description for videos under 1 hour, or HH:MM:SS Topic description for longer videos\n"
+            f"Example: 05:30 {example_description} or 1:05:30 {example_description}\n"
+            f"This format is suitable for YouTube video descriptions."
+        )
+    elif format_type == "markdown":
+        format_instructions = (
+            f"Format should be Markdown: - [MM:SS](link) Topic description for videos under 1 hour, or - [HH:MM:SS](link) Topic description for longer videos\n"
+            f"Example: - [05:30](https://youtu.be/VIDEOID?t=330) {example_description} or - [1:05:30](https://youtu.be/VIDEOID?t=3930) {example_description}\n"
+            f"This format creates clickable links in Markdown."
+        )
+    else:  # txt
+        format_instructions = (
+            f"Format should be: MM:SS - Topic description for videos under 1 hour, or HH:MM:SS - Topic description for longer videos\n"
+            f"Example: 05:30 - {example_description} or 1:05:30 - {example_description}\n"
+            f"This format is suitable for plain text representation."
+        )
+    
+    prompt = f"""
+    You are an expert at creating timestamps for YouTube videos. You have been provided with a transcript of the video "{video_title}".
+    
+    Your task is to create timestamps for the main themes and segments of the video based on the provided transcript.
+    Create timestamp descriptions in {target_language}.
+    
+    {format_instructions}
+    
+    Rules for creating timestamps:
+    1. Select {timecode_count} key video segments
+    2. Use the time markers provided in the transcript to determine the start of each segment
+    3. Create brief (3-7 words) descriptions for each segment that reflect its main theme, using appropriate terminology and style
+    4. Distribute timestamps approximately evenly throughout the video length
+    5. Use MM:SS format for videos under 1 hour (example: 05:30, 45:20), and HH:MM:SS format for videos 1 hour or longer (example: 1:05:30, 1:45:20)
+    6. DO NOT include standard markers like "Video start" or "Video end"
+    7. Ensure a clear structure so viewers can easily navigate through the video
+    8. The first timestamp does NOT have to be 00:00, start with the first meaningful topic
+    
+    Here is the video transcript:
+    
+    {transcript}
+    
+    Create a list of timestamps in the specified format. Reply with ONLY the list of timestamps, without introduction or conclusion.
+    """
+    
+    return prompt
+
+async def generate_timecodes_with_gemini(
+    transcript_entries: List[Dict[str, Any]],
+    video_title: str,
+    format_type: str = "youtube",
+    model_name: Optional[str] = None,
+    language: Optional[str] = None
+) -> Dict[str, Any]:
+    """
+    Generates timecodes using Gemini based on transcript.
+    
+    Args:
+        transcript_entries: List of transcript entries
+        video_title: Video title
+        format_type: Timecode format (youtube, markdown)
+        model_name: Gemini model name (defaults to DEFAULT_MODEL)
+        language: Transcript language (if known)
+        
+    Returns:
+        Dictionary with generation results
+    """
+    if not GEMINI_API_KEY or client is None:
+        return {
+            "error": "Gemini API key is not configured. Please add GEMINI_API_KEY to .env file"
+        }
+    
+    try:
+        print(f"Starting timecode generation with model: {model_name or DEFAULT_MODEL}")
+        
+        # Determine transcript language if not provided
+        detected_language = language
+        if not detected_language:
+            # Simple heuristic for language detection from first 10 segments
+            text_sample = " ".join([entry.get("text", "") for entry in transcript_entries[:10]])
+            
+            # Set of Ukrainian letters that differ from Russian alphabet
+            ukrainian_specific = set("ґєії")
+            # If there's at least one specific Ukrainian letter
+            if any(char in ukrainian_specific for char in text_sample.lower()):
+                detected_language = "uk"
+                print("Detected transcript language: Ukrainian")
+            # Check for Cyrillic in general
+            elif any(ord('а') <= ord(char) <= ord('я') for char in text_sample.lower()):
+                detected_language = "ru"
+                print("Detected transcript language: Russian")
+            else:
+                detected_language = "en"
+                print("Detected transcript language: English (or other)")
+        
+        # Determine video duration (in seconds and minutes)
+        video_duration_seconds = 0
+        if transcript_entries:
+            last_entry = transcript_entries[-1]
+            video_duration_seconds = last_entry.get("start", 0) + last_entry.get("duration", 0)
+            video_duration_minutes = int(video_duration_seconds / 60)
+            print(f"Determined video duration: {video_duration_minutes} minutes ({video_duration_seconds} seconds)")
+        else:
+            video_duration_minutes = None
+        
+        # Format transcript for prompt
+        formatted_transcript = format_transcript_for_prompt(transcript_entries, video_duration_seconds)
+        
+        # Create prompt considering language and duration
+        prompt = get_timecode_prompt(
+            video_title, 
+            formatted_transcript, 
+            format_type, 
+            detected_language, 
+            video_duration_minutes
+        )
+        print(f"Prompt prepared, length: {len(prompt)} characters")
+        
+        # List of models to try
+        models_to_try = [model_name or DEFAULT_MODEL] + [m for m in ALTERNATIVE_MODELS if m != (model_name or DEFAULT_MODEL)]
+        
+        last_error = None
+        for current_model in models_to_try:
+            try:
+                # Use async API client for content generation
+                print(f"Making request to Gemini API with model {current_model}...")
+                response = await client.aio.models.generate_content(
+                    model=current_model,
+                    contents=prompt,
+                    config=types.GenerateContentConfig(
+                        temperature=0.2,  # Low temperature for more deterministic results
+                        max_output_tokens=2048,  # Enough for timecode list
+                    )
+                )
+                print(f"Response received: {type(response)}")
+                
+                # Get response text
+                timecodes_text = response.text
+                print(f"Response text length: {len(timecodes_text)}")
+                
+                # Split into lines and clean
+                timecodes = [line.strip() for line in timecodes_text.split('\n') if line.strip()]
+                
+                # Filter timecodes to remove "video start" and "video end"
+                filtered_timecodes = []
+                for tc in timecodes:
+                    # Extract description (everything after time)
+                    parts = tc.split(" ", 1)
+                    if len(parts) > 1:
+                        time_part, description = parts
+                        # Skip timecodes with "video start" or "video end"
+                        lowercase_desc = description.lower()
+                        if any(phrase in lowercase_desc for phrase in [
+                            "начало видео", "конец видео", "початок відео", "кінець відео", 
+                            "start of video", "end of video", "video start", "video end",
+                            "beginning", "conclusion", "intro", "outro"
+                        ]):
+                            continue
+                    filtered_timecodes.append(tc)
+                
+                # If too many timecodes, select evenly distributed ones
+                max_timecodes = 25  # Maximum recommended number of timecodes
+                if len(filtered_timecodes) > max_timecodes:
+                    print(f"Too many timecodes ({len(filtered_timecodes)}), reducing to {max_timecodes}")
+                    # Calculate step for selecting timecodes evenly
+                    step = len(filtered_timecodes) / max_timecodes
+                    # Select indices for timecodes
+                    indices = [int(i * step) for i in range(max_timecodes)]
+                    # Ensure we have first and last timecode
+                    if indices[-1] != len(filtered_timecodes) - 1:
+                        indices[-1] = len(filtered_timecodes) - 1
+                    # Select timecodes by indices
+                    final_timecodes = [filtered_timecodes[i] for i in indices]
+                else:
+                    final_timecodes = filtered_timecodes
+                
+                print(f"Final timecodes count after processing: {len(final_timecodes)}")
+                
+                return {
+                    "timecodes": final_timecodes,
+                    "format": format_type,
+                    "model": current_model,
+                    "video_title": video_title,
+                    "detected_language": detected_language,
+                    "video_duration_minutes": video_duration_minutes
+                }
+            except Exception as api_error:
+                print(f"Error with model {current_model}: {str(api_error)}")
+                traceback.print_exc()
+                last_error = api_error
+                continue
+        
+        # If all models failed
+        return {
+            "error": f"Failed to execute request with any model. Last error: {str(last_error)}"
+        }
+    except Exception as e:
+        print(f"General error: {str(e)}")
+        traceback.print_exc()
+        return {
+            "error": f"Error generating timecodes with Gemini: {str(e)}"
+        } 

gradio_app.py

@@ -0,0 +1,383 @@
+import gradio as gr
+import json
+import httpx
+import os
+import traceback
+from dotenv import load_dotenv
+from utils import format_timestamp, extract_video_id
+
+# Load environment variables
+load_dotenv()
+
+# API URL for local development
+API_URL = "http://127.0.0.1:8080/api"
+# API URL for Hugging Face Spaces
+# API_URL = "https://your-huggingface-space-url/api"
+
+async def search_youtube(query, max_results, order, video_duration):
+    """Function for searching videos on YouTube."""
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{API_URL}/search",
+                json={
+                    "query": query,
+                    "max_results": max_results,
+                    "order": order,
+                    "video_duration": video_duration if video_duration != "any" else None
+                }
+            )
+            data = response.json()
+            
+            if "error" in data and data["error"]:
+                return f"Error: {data['error']}", None
+                
+            results = data.get("content", [])
+            formatted_results = []
+            
+            for video in results:
+                formatted_results.append(
+                    f"**{video['title']}**\n"
+                    f"ID: {video['video_id']}\n"
+                    f"Channel: {video['channel_title']}\n"
+                    f"Published: {video['published_at']}\n"
+                    f"[Thumbnail]({video['thumbnail']})\n\n"
+                    f"{video['description'][:200]}...\n\n"
+                    f"---\n"
+                )
+            
+            return "\n".join(formatted_results), json.dumps(results, indent=2, ensure_ascii=False)
+    except Exception as e:
+        return f"Error: {str(e)}", None
+
+async def get_video_info(video_id):
+    """Function for getting video information."""
+    try:
+        # No need to extract video ID here, it is done on the server
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{API_URL}/video_info",
+                json={"video_id": video_id}
+            )
+            data = response.json()
+            
+            if "error" in data and data["error"]:
+                return f"Error: {data['error']}", None
+                
+            video_info = data.get("content", {})
+            
+            formatted_info = (
+                f"**{video_info.get('title')}**\n\n"
+                f"Channel: {video_info.get('channel_title')}\n"
+                f"Published: {video_info.get('published_at')}\n"
+                f"Views: {video_info.get('view_count')}\n"
+                f"Likes: {video_info.get('like_count')}\n"
+                f"Comments: {video_info.get('comment_count')}\n"
+                f"Duration: {video_info.get('duration')}\n\n"
+                f"**Description:**\n{video_info.get('description')}\n\n"
+                f"**Tags:**\n{', '.join(video_info.get('tags', []))}"
+            )
+            
+            return formatted_info, json.dumps(video_info, indent=2, ensure_ascii=False)
+    except Exception as e:
+        return f"Error: {str(e)}", None
+
+async def get_transcript(video_id, language_code):
+    """Function for getting video transcript."""
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{API_URL}/transcript",
+                json={
+                    "video_id": video_id,
+                    "language_code": language_code if language_code else None
+                }
+            )
+            data = response.json()
+            
+            if "error" in data and data["error"]:
+                return f"Error: {data['error']}", None
+                
+            transcript = data.get("content", [])
+            
+            formatted_transcript = ""
+            for entry in transcript:
+                start_time = entry.get("start", 0)
+                duration = entry.get("duration", 0)
+                end_time = start_time + duration
+                
+                # Format time to hours:minutes:seconds format
+                start_formatted = format_timestamp(start_time)
+                end_formatted = format_timestamp(end_time)
+                
+                formatted_transcript += f"[{start_formatted} - {end_formatted}] {entry.get('text', '')}\n\n"
+            
+            return formatted_transcript, json.dumps(transcript, indent=2, ensure_ascii=False)
+    except Exception as e:
+        return f"Error: {str(e)}", None
+
+async def get_available_languages(video_id):
+    """Function for getting available transcript languages."""
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{API_URL}/transcript_languages",
+                json={"video_id": video_id}
+            )
+            data = response.json()
+            
+            if "error" in data and data["error"]:
+                return f"Error: {data['error']}", None
+                
+            languages = data.get("content", [])
+            
+            formatted_languages = []
+            for lang in languages:
+                status = "Auto-generated" if lang.get("is_generated") else "Official subtitles"
+                translatable = "Translation available" if lang.get("is_translatable") else "Translation not available"
+                formatted_languages.append(
+                    f"{lang.get('language')} ({lang.get('language_code')}): {status}, {translatable}"
+                )
+            
+            return "\n".join(formatted_languages), json.dumps(languages, indent=2, ensure_ascii=False)
+    except Exception as e:
+        return f"Error: {str(e)}", None
+
+async def generate_timecodes(video_id, language_code, segment_length, format_type):
+    """Function for generating timecodes."""
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{API_URL}/timecodes",
+                json={
+                    "video_id": video_id,
+                    "language_code": language_code if language_code else None,
+                    "segment_length": segment_length,
+                    "format": format_type
+                }
+            )
+            data = response.json()
+            
+            if "error" in data and data["error"]:
+                return f"Error: {data['error']}", None
+                
+            timecodes = data.get("content", {}).get("timecodes", [])
+            
+            if format_type == "youtube":
+                formatted_timecodes = "```\n" + "\n".join(timecodes) + "\n```"
+            elif format_type == "markdown":
+                formatted_timecodes = "\n".join(timecodes)
+            else:
+                formatted_timecodes = "```\n" + "\n".join(timecodes) + "\n```"
+            
+            return formatted_timecodes, json.dumps(data.get("content", {}), indent=2, ensure_ascii=False)
+    except Exception as e:
+        return f"Error: {str(e)}", None
+
+async def generate_gemini_timecodes(video_id, language_code, format_type, model):
+    """Function for generating timecodes using Gemini."""
+    try:
+        print(f"Sending request to {API_URL}/gemini_timecodes")
+        print(f"Parameters: video_id={video_id}, language_code={language_code}, format={format_type}, model={model}")
+        
+        # Send request to API
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{API_URL}/gemini_timecodes",
+                json={
+                    "video_id": video_id,
+                    "language_code": language_code,
+                    "format": format_type,
+                    "model": model
+                },
+                timeout=120  # Increase timeout for Gemini API
+            )
+            
+            print(f"Response status: {response.status_code}")
+            
+            # Parse response
+            data = response.json()
+            
+            if "error" in data:
+                print(f"Error in API response: {data['error']}")
+                return f"⚠️ Error: {data['error']}", {"error": data['error']}
+            
+            # Extract timecodes from response
+            content = data.get("content", {})
+            timecodes = content.get("timecodes", [])
+            
+            print(f"Received {len(timecodes)} timecodes")
+            
+            # Format timecodes for display
+            if timecodes:
+                timecodes_text = "\n".join(timecodes)
+                
+                # Model and language information
+                model_info = content.get("model", "Unknown")
+                language_info = content.get("detected_language", "Unknown")
+                duration_info = content.get("video_duration_minutes", "Unknown")
+                
+                summary = f"🤖 Model: {model_info}\n🗣️ Language: {language_info}\n⏱️ Duration: {duration_info} min\n📝 Timecodes: {len(timecodes)}"
+                
+                return summary, content  # Return content object instead of timecodes_text
+            else:
+                return "⚠️ No timecodes generated", {"message": "No timecodes generated"}
+            
+    except Exception as e:
+        print(f"Exception during timecode generation: {str(e)}")
+        traceback.print_exc()
+        return f"Error: {str(e)}", {"error": str(e)}
+
+# Create Gradio interface
+with gr.Blocks(title="YouTube MCP") as demo:
+    gr.Markdown("# YouTube Model Context Protocol (MCP)")
+    gr.Markdown("This interface allows interaction with YouTube API through MCP protocol")
+    
+    with gr.Tab("Поиск видео"):
+        with gr.Row():
+            with gr.Column():
+                search_query = gr.Textbox(label="Поисковый запрос", placeholder="Введите запрос...")
+                with gr.Row():
+                    max_results = gr.Slider(minimum=1, maximum=50, value=10, step=1, label="Колич��ство результатов")
+                    order = gr.Dropdown(
+                        choices=["relevance", "date", "viewCount", "rating", "title"], 
+                        value="relevance", 
+                        label="Сортировка"
+                    )
+                    video_duration = gr.Dropdown(
+                        choices=["any", "short", "medium", "long"], 
+                        value="any", 
+                        label="Длительность"
+                    )
+                search_button = gr.Button("Поиск")
+            
+            with gr.Column():
+                search_results = gr.Markdown(label="Результаты")
+                search_json = gr.JSON(label="JSON данные")
+        
+        search_button.click(
+            search_youtube, 
+            inputs=[search_query, max_results, order, video_duration], 
+            outputs=[search_results, search_json]
+        )
+    
+    with gr.Tab("Информация о видео"):
+        with gr.Row():
+            with gr.Column():
+                video_id_input = gr.Textbox(
+                    label="ID видео или ссылка на видео", 
+                    placeholder="Введите ID видео или полную ссылку (youtube.com, youtu.be, shorts, embed)..."
+                )
+                get_info_button = gr.Button("Получить информацию")
+            
+            with gr.Column():
+                video_info_output = gr.Markdown(label="Информация о видео")
+                video_info_json = gr.JSON(label="JSON данные")
+        
+        get_info_button.click(
+            get_video_info, 
+            inputs=[video_id_input], 
+            outputs=[video_info_output, video_info_json]
+        )
+    
+    with gr.Tab("Транскрипт видео"):
+        with gr.Row():
+            with gr.Column():
+                transcript_video_id = gr.Textbox(
+                    label="ID видео или ссылка на видео", 
+                    placeholder="Введите ID видео или полную ссылку (youtube.com, youtu.be, shorts, embed)..."
+                )
+                language_code = gr.Textbox(label="Код языка (опционально)", placeholder="ru, en, etc...")
+                with gr.Row():
+                    get_transcript_button = gr.Button("Получить транскрипт")
+                    get_languages_button = gr.Button("Получить доступные языки")
+            
+            with gr.Column():
+                transcript_output = gr.Markdown(label="Транскрипт")
+                transcript_json = gr.JSON(label="JSON данные")
+        
+        get_transcript_button.click(
+            get_transcript, 
+            inputs=[transcript_video_id, language_code], 
+            outputs=[transcript_output, transcript_json]
+        )
+        
+        get_languages_button.click(
+            get_available_languages,
+            inputs=[transcript_video_id],
+            outputs=[transcript_output, transcript_json]
+        )
+
+    with gr.Tab("Тайм-коды"):
+        with gr.Row():
+            with gr.Column():
+                timecode_video_id = gr.Textbox(
+                    label="ID видео или ссылка на видео", 
+                    placeholder="Введите ID видео или полную ссылку (youtube.com, youtu.be, shorts, embed)..."
+                )
+                timecode_language = gr.Textbox(label="Код языка (опционально)", placeholder="ru, en, etc...")
+                segment_length = gr.Slider(minimum=30, maximum=300, value=60, step=30, label="Длина сегмента (секунды)")
+                format_type = gr.Dropdown(
+                    choices=["youtube", "markdown"], 
+                    value="youtube", 
+                    label="Формат тайм-кодов"
+                )
+                generate_timecodes_button = gr.Button("Сгенерировать тайм-коды")
+            
+            with gr.Column():
+                timecodes_output = gr.Markdown(label="Тайм-коды")
+                timecodes_json = gr.JSON(label="JSON данные")
+        
+        generate_timecodes_button.click(
+            generate_timecodes, 
+            inputs=[timecode_video_id, timecode_language, segment_length, format_type], 
+            outputs=[timecodes_output, timecodes_json]
+        )
+
+    with gr.Tab("Gemini Тайм-коды"):
+        with gr.Row():
+            with gr.Column():
+                gemini_video_id = gr.Textbox(
+                    label="ID видео или ссылка на видео", 
+                    placeholder="Введите ID видео или полную ссылку (youtube.com, youtu.be, shorts, embed)..."
+                )
+                gemini_language = gr.Textbox(label="Код языка (опционально)", placeholder="ru, en, etc...")
+                gemini_format = gr.Dropdown(
+                    choices=["youtube", "markdown"], 
+                    value="youtube", 
+                    label="Формат тайм-кодов"
+                )
+                gemini_model = gr.Dropdown(
+                    choices=["gemini-2.0-flash-001", "gemini-2.0-pro-001", "gemini-2.0-pro-vision-001"],
+                    value="gemini-2.0-flash-001",
+                    label="Модель Gemini"
+                )
+                generate_gemini_button = gr.Button("Сгенерировать тайм-коды с Gemini")
+            
+            with gr.Column():
+                gemini_output = gr.Markdown(label="Информация о генерации")
+                gemini_timecodes = gr.Textbox(label="Тайм-коды", lines=10, max_lines=20, show_copy_button=True)
+                gemini_json = gr.JSON(label="JSON данные")
+        
+        async def process_gemini_result(video_id, language_code, format_type, model):
+            result = await generate_gemini_timecodes(video_id, language_code, format_type, model)
+            if result is None:
+                return "Error occurred", "", {}
+            
+            summary, json_data = result
+            
+            # Extract timecodes from json_data
+            timecodes = json_data.get("timecodes", [])
+            timecodes_text = "\n".join(timecodes) if timecodes else "No timecodes generated"
+            
+            return summary, timecodes_text, json_data
+        
+        generate_gemini_button.click(
+            process_gemini_result, 
+            inputs=[gemini_video_id, gemini_language, gemini_format, gemini_model], 
+            outputs=[gemini_output, gemini_timecodes, gemini_json]
+        )
+
+# Запуск приложения
+if __name__ == "__main__":
+    demo.launch() 

main.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+"""
+Unified launcher for YouTube MCP application.
+Provides options to run API server, Gradio UI, or both.
+"""
+
+import argparse
+import asyncio
+import uvicorn
+import threading
+import time
+from dotenv import load_dotenv
+
+# Load environment variables
+load_dotenv()
+
+def start_api_server(host="127.0.0.1", port=8080):
+    """Start FastAPI server."""
+    from api_server import app
+    print(f"Starting API server on http://{host}:{port}")
+    uvicorn.run(app, host=host, port=port)
+
+def start_gradio_ui(host="127.0.0.1", port=8081):
+    """Start Gradio UI."""
+    import gradio_app
+    print(f"Starting Gradio UI on http://{host}:{port}")
+    gradio_app.demo.launch(server_name=host, server_port=port, share=False)
+
+def start_both(api_host="127.0.0.1", api_port=8080, ui_host="127.0.0.1", ui_port=8081):
+    """Start both API server and Gradio UI."""
+    print(f"Starting API server on http://{api_host}:{api_port}")
+    print(f"Starting Gradio UI on http://{ui_host}:{ui_port}")
+    
+    # Start API server in a separate thread
+    api_thread = threading.Thread(
+        target=start_api_server, 
+        args=(api_host, api_port), 
+        daemon=True
+    )
+    api_thread.start()
+    
+    # Wait a moment for API server to start
+    time.sleep(2)
+    
+    # Start Gradio UI in main thread
+    start_gradio_ui(ui_host, ui_port)
+
+def main():
+    parser = argparse.ArgumentParser(description="YouTube MCP Application Launcher")
+    parser.add_argument(
+        "--mode", 
+        choices=["api", "ui", "both"], 
+        default="both",
+        help="Launch mode: 'api' for FastAPI server only, 'ui' for Gradio UI only, 'both' for both services"
+    )
+    parser.add_argument(
+        "--host", 
+        default="127.0.0.1",
+        help="Host address (default: 127.0.0.1)"
+    )
+    parser.add_argument(
+        "--port", 
+        type=int, 
+        default=8080,
+        help="Port number (default: 8080 for API, 8081 for UI in 'both' mode)"
+    )
+    
+    args = parser.parse_args()
+    
+    try:
+        if args.mode == "api":
+            start_api_server(args.host, args.port)
+        elif args.mode == "ui":
+            start_gradio_ui(args.host, args.port)
+        elif args.mode == "both":
+            start_both(args.host, args.port, args.host, args.port + 1)
+    except KeyboardInterrupt:
+        print("\nKeyboard interruption in main thread... closing server.")
+    except Exception as e:
+        print(f"Error starting application: {e}")
+
+if __name__ == "__main__":
+    main()

mcp_handlers.py

@@ -0,0 +1,478 @@
+from fastapi import Request, HTTPException
+from typing import Dict, List, Any, Optional, Union
+from pydantic import BaseModel
+import json
+import httpx
+from googleapiclient.discovery import build
+from googleapiclient.errors import HttpError
+from youtube_transcript_api import YouTubeTranscriptApi
+from youtube_transcript_api.formatters import JSONFormatter
+from gemini_helper import generate_timecodes_with_gemini, DEFAULT_MODEL
+from models import MCPResponse
+from utils import format_timestamp, extract_video_id
+
+# Data models for MCP
+class MCPQueryRequest(BaseModel):
+    query: str
+    max_results: Optional[int] = 10
+    
+class MCPVideoRequest(BaseModel):
+    video_id: str
+    
+class MCPTranscriptRequest(BaseModel):
+    video_id: str
+    language_code: Optional[str] = None
+
+# Model for timecode requests through MCP
+class MCPTimecodeRequest(BaseModel):
+    video_id: str
+    language_code: Optional[str] = None
+    segment_length: Optional[int] = 60  # Segment length in seconds
+    format: Optional[str] = "youtube"  # youtube, markdown
+
+# Model for Gemini timecode requests through MCP
+class MCPGeminiRequest(BaseModel):
+    video_id: str
+    language_code: Optional[str] = None
+    format: Optional[str] = "youtube"  # youtube, markdown
+    model: Optional[str] = DEFAULT_MODEL  # Gemini model
+
+# Functions for processing MCP requests
+async def process_mcp_search(youtube_client, request: MCPQueryRequest) -> List[MCPResponse]:
+    """Process MCP request for video search."""
+    try:
+        search_response = youtube_client.search().list(
+            q=request.query,
+            part="snippet",
+            maxResults=request.max_results,
+            type="video"
+        ).execute()
+        
+        results = []
+        for item in search_response.get("items", []):
+            video_id = item["id"]["videoId"]
+            snippet = item["snippet"]
+            
+            # Create MCP format response
+            video_data = {
+                "video_id": video_id,
+                "title": snippet["title"],
+                "description": snippet["description"],
+                "thumbnail": snippet["thumbnails"]["high"]["url"],
+                "channel_title": snippet["channelTitle"],
+                "published_at": snippet["publishedAt"]
+            }
+            
+            # Format markdown for video display
+            markdown_text = (
+                f"## {snippet['title']}\n"
+                f"**Channel:** {snippet['channelTitle']}\n"
+                f"**Published:** {snippet['publishedAt']}\n\n"
+                f"[![Thumbnail]({snippet['thumbnails']['high']['url']})](https://www.youtube.com/watch?v={video_id})\n\n"
+                f"{snippet['description'][:300]}...\n\n"
+                f"[Watch on YouTube](https://www.youtube.com/watch?v={video_id})"
+            )
+            
+            results.append(MCPResponse(
+                type="youtube_video",
+                markdown=markdown_text,
+                data=video_data
+            ))
+        
+        return results
+    except HttpError as e:
+        raise HTTPException(status_code=500, detail=f"YouTube API error: {str(e)}")
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Unexpected error: {str(e)}")
+
+async def process_mcp_video_info(youtube_client, request: MCPVideoRequest) -> MCPResponse:
+    """Process MCP request for video information."""
+    try:
+        # Extract video ID from URL if it's a URL
+        video_id = extract_video_id(request.video_id)
+        
+        video_response = youtube_client.videos().list(
+            part="snippet,contentDetails,statistics",
+            id=video_id
+        ).execute()
+        
+        if not video_response.get("items"):
+            return MCPResponse(
+                type="error",
+                error="Video not found"
+            )
+        
+        video_item = video_response["items"][0]["snippet"]
+        content_details = video_response["items"][0].get("contentDetails", {})
+        statistics = video_response["items"][0].get("statistics", {})
+        
+        # Get detailed video information
+        video_data = {
+            "video_id": video_id,
+            "title": video_item.get("title"),
+            "channel_title": video_item.get("channelTitle"),
+            "published_at": video_item.get("publishedAt"),
+            "view_count": statistics.get("viewCount"),
+            "like_count": statistics.get("likeCount"),
+            "comment_count": statistics.get("commentCount"),
+            "duration": content_details.get("duration"),
+            "thumbnail": video_item.get("thumbnails", {}).get("high", {}).get("url")
+        }
+        
+        return MCPResponse(
+            type="text",
+            content=f"Video information:\n{json.dumps(video_data, indent=2, ensure_ascii=False)}"
+        )
+    except HttpError as e:
+        return MCPResponse(
+            type="error",
+            error=f"YouTube API error: {str(e)}"
+        )
+    except Exception as e:
+        return MCPResponse(
+            type="error",
+            error=f"Unexpected error: {str(e)}"
+        )
+
+async def process_mcp_transcript(request: MCPTranscriptRequest) -> MCPResponse:
+    """Process MCP request for video transcript."""
+    try:
+        # Extract video ID from URL if it's a URL
+        video_id = extract_video_id(request.video_id)
+        
+        try:
+            languages = [request.language_code] if request.language_code else None
+            transcript_list = YouTubeTranscriptApi.get_transcript(video_id, languages=languages)
+        except Exception as transcript_error:
+            if request.language_code:
+                try:
+                    print(f"Failed to get transcript in language {request.language_code}, trying to get available transcripts")
+                    transcript_list = YouTubeTranscriptApi.get_transcript(video_id)
+                except Exception as fallback_error:
+                    return MCPResponse(
+                        type="error",
+                        error=f"Transcript not found. Details: {str(fallback_error)}"
+                    )
+            else:
+                return MCPResponse(
+                    type="error",
+                    error=f"Failed to get transcript. Details: {str(transcript_error)}"
+                )
+        
+        if not transcript_list:
+            return MCPResponse(
+                type="error",
+                error="Transcript for this video is unavailable"
+            )
+        
+        formatted_transcript = []
+        for entry in transcript_list:
+            formatted_transcript.append({
+                "text": entry.get("text", ""),
+                "start": entry.get("start", 0),
+                "duration": entry.get("duration", 0)
+            })
+        
+        # Format markdown for transcript display
+        markdown_text = "# Transcript\n\n"
+        for entry in formatted_transcript:
+            start_time = entry.get("start")
+            duration = entry.get("duration")
+            end_time = start_time + duration
+            text = entry.get("text")
+            
+            # Convert time to hours:minutes:seconds format
+            start_formatted = format_timestamp(start_time)
+            end_formatted = format_timestamp(end_time)
+            
+            markdown_text += f"[{start_formatted} - {end_formatted}] {text}\n\n"
+        
+        return MCPResponse(
+            type="youtube_transcript",
+            markdown=markdown_text,
+            data={
+                "video_id": video_id,
+                "transcript": formatted_transcript
+            }
+        )
+    except Exception as e:
+        return MCPResponse(
+            type="error",
+            error=f"Error getting transcript: {str(e)}"
+        )
+
+# Function for creating text response in MCP format
+def create_text_response(text: str) -> MCPResponse:
+    """Creates text response in MCP format."""
+    return MCPResponse(
+        type="text",
+        text=text
+    )
+
+# Function for creating error response in MCP format
+def create_error_response(error_message: str) -> MCPResponse:
+    """Creates error response in MCP format."""
+    return MCPResponse(
+        type="error",
+        error=error_message
+    )
+
+# Function for formatting time to hours:minutes:seconds format
+def format_timestamp(seconds):
+    """Formats time in seconds to hours:minutes:seconds format."""
+    hours = int(seconds // 3600)
+    minutes = int((seconds % 3600) // 60)
+    secs = int(seconds % 60)
+    
+    if hours > 0:
+        return f"{hours:02d}:{minutes:02d}:{secs:02d}"
+    else:
+        return f"{minutes:02d}:{secs:02d}"
+
+async def process_mcp_timecodes(youtube_client, request: MCPTimecodeRequest) -> MCPResponse:
+    """Process MCP request for timecode generation."""
+    try:
+        # Extract video ID from URL if it's a URL
+        video_id = extract_video_id(request.video_id)
+        
+        # Get transcript
+        try:
+            languages = [request.language_code] if request.language_code else None
+            transcript_list = YouTubeTranscriptApi.get_transcript(video_id, languages=languages)
+        except Exception as transcript_error:
+            if request.language_code:
+                try:
+                    print(f"Failed to get transcript in language {request.language_code}, trying to get available transcripts")
+                    transcript_list = YouTubeTranscriptApi.get_transcript(video_id)
+                except Exception as fallback_error:
+                    return MCPResponse(
+                        type="error",
+                        error=f"Transcript not found. Details: {str(fallback_error)}"
+                    )
+            else:
+                return MCPResponse(
+                    type="error",
+                    error=f"Failed to get transcript. Details: {str(transcript_error)}"
+                )
+        
+        if not transcript_list:
+            return MCPResponse(
+                type="error",
+                error="Transcript for this video is unavailable"
+            )
+        
+        # Group transcript into segments
+        segments = []
+        current_segment = {
+            "start": transcript_list[0]["start"],
+            "end": 0,
+            "text": []
+        }
+        
+        segment_length = request.segment_length
+        
+        for entry in transcript_list:
+            start_time = entry["start"]
+            
+            # If current segment is empty or entry is within segment length
+            if not current_segment["text"] or (start_time - current_segment["start"]) <= segment_length:
+                current_segment["text"].append(entry["text"])
+                current_segment["end"] = start_time + entry["duration"]
+            else:
+                # Close current segment and start new
+                segments.append(dict(current_segment))
+                current_segment = {
+                    "start": start_time,
+                    "end": start_time + entry["duration"],
+                    "text": [entry["text"]]
+                }
+        
+        # Add last segment
+        if current_segment["text"]:
+            segments.append(current_segment)
+        
+        # Format timecodes according to selected format
+        format_type = request.format.lower()
+        timecodes = []
+        
+        for segment in segments:
+            start_formatted = format_timestamp(segment["start"])
+            
+            # Summary text of segment (first 100 characters)
+            text_summary = " ".join(segment["text"])
+            if len(text_summary) > 100:
+                text_summary = text_summary[:97] + "..."
+            
+            if format_type == "youtube":
+                # Format for YouTube (for embedding in description)
+                timecodes.append(f"{start_formatted} {text_summary}")
+            elif format_type == "markdown":
+                # Format for Markdown
+                youtube_link = f"https://www.youtube.com/watch?v={video_id}&t={int(segment['start'])}"
+                timecodes.append(f"- [{start_formatted}]({youtube_link}) {text_summary}")
+                
+        # Create markdown with timecodes
+        markdown_text = f"# Timecodes for Video\n\n"
+        
+        if format_type == "youtube":
+            markdown_text += "```\n"
+            markdown_text += "\n".join(timecodes)
+            markdown_text += "\n```"
+        elif format_type == "markdown":
+            markdown_text += "\n".join(timecodes)
+            
+        # Get video information for title
+        try:
+            video_response = youtube_client.videos().list(
+                part="snippet",
+                id=video_id
+            ).execute()
+            
+            if video_response.get("items"):
+                video_title = video_response["items"][0]["snippet"]["title"]
+                markdown_text = f"# Timecodes for Video: {video_title}\n\n" + markdown_text[markdown_text.find('\n\n') + 2:]
+        except Exception as e:
+            print(f"Failed to get video information: {str(e)}")
+            video_title = "YouTube Video"
+        
+        return MCPResponse(
+            type="youtube_timecodes",
+            markdown=markdown_text,
+            data={
+                "video_id": video_id,
+                "timecodes": timecodes,
+                "format": format_type,
+                "segment_length": segment_length,
+                "total_segments": len(segments)
+            }
+        )
+    except Exception as e:
+        return MCPResponse(
+            type="error",
+            error=f"Error generating timecodes: {str(e)}"
+        )
+
+async def process_mcp_gemini_timecodes(youtube_client, request: MCPGeminiRequest) -> MCPResponse:
+    """Process MCP request for Gemini timecode generation."""
+    try:
+        # Get transcript
+        try:
+            languages = [request.language_code] if request.language_code else None
+            transcript_list = YouTubeTranscriptApi.get_transcript(request.video_id, languages=languages)
+        except Exception as transcript_error:
+            if request.language_code:
+                try:
+                    print(f"Failed to get transcript in language {request.language_code}, trying to get available transcripts")
+                    transcript_list = YouTubeTranscriptApi.get_transcript(request.video_id)
+                except Exception as fallback_error:
+                    return MCPResponse(
+                        type="error",
+                        error=f"Transcript not found. Details: {str(fallback_error)}"
+                    )
+            else:
+                return MCPResponse(
+                    type="error",
+                    error=f"Failed to get transcript. Details: {str(transcript_error)}"
+                )
+        
+        if not transcript_list:
+            return MCPResponse(
+                type="error",
+                error="Transcript for this video is unavailable"
+            )
+        
+        # Get video information for title
+        try:
+            video_response = youtube_client.videos().list(
+                part="snippet",
+                id=request.video_id
+            ).execute()
+            
+            if video_response.get("items"):
+                video_title = video_response["items"][0]["snippet"]["title"]
+        except Exception as e:
+            print(f"Failed to get video information: {str(e)}")
+            video_title = "YouTube Video"
+        
+        # Send request to Gemini
+        result = await generate_timecodes_with_gemini(
+            transcript_entries=transcript_list,
+            video_title=video_title,
+            format_type=request.format,
+            model_name=request.model
+        )
+        
+        if "error" in result:
+            return MCPResponse(
+                type="error",
+                error=result["error"]
+            )
+        
+        # Create markdown with timecodes
+        timecodes = result.get("timecodes", [])
+        format_type = result.get("format", "youtube")
+        
+        markdown_text = f"# Timecodes for Video: {video_title}\n\n"
+        
+        if format_type == "youtube":
+            markdown_text += "```\n"
+            markdown_text += "\n".join(timecodes)
+            markdown_text += "\n```"
+        elif format_type == "markdown":
+            markdown_text += "\n".join(timecodes)
+            
+        return MCPResponse(
+            type="youtube_gemini_timecodes",
+            markdown=markdown_text,
+            data=result
+        )
+    except Exception as e:
+        return MCPResponse(
+            type="error",
+            error=f"Error generating timecodes with Gemini: {str(e)}"
+        )
+
+async def process_mcp_transcript_languages(request: MCPVideoRequest) -> MCPResponse:
+    """Process MCP request for getting list of available transcript languages."""
+    try:
+        # Extract video ID from URL if it's a URL
+        video_id = extract_video_id(request.video_id)
+        
+        try:
+            transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)
+            
+            languages = []
+            for transcript in transcript_list:
+                languages.append({
+                    "language_code": transcript.language_code,
+                    "language": transcript.language,
+                    "is_generated": transcript.is_generated,
+                    "is_translatable": transcript.is_translatable
+                })
+            
+            # Format markdown for displaying language list
+            markdown_text = "# Available Transcript Languages\n\n"
+            for language in languages:
+                lang_type = "Auto-generated" if language["is_generated"] else "Manually added"
+                translatable = "Available for translation" if language.get("is_translatable", False) else "Not available for translation"
+                
+                markdown_text += f"- **{language['language']}** ({language['language_code']}): {lang_type}, {translatable}\n"
+            
+            return MCPResponse(
+                type="youtube_transcript_languages",
+                markdown=markdown_text,
+                data={
+                    "video_id": video_id,
+                    "languages": languages
+                }
+            )
+        except Exception as transcript_error:
+            return MCPResponse(
+                type="error",
+                error=f"Failed to get language list. Details: {str(transcript_error)}"
+            )
+    except Exception as e:
+        return MCPResponse(
+            type="error",
+            error=f"Error getting language list: {str(e)}"
+        ) 

models.py

@@ -0,0 +1,10 @@
+from typing import Dict, Any, Optional
+from pydantic import BaseModel
+
+class MCPResponse(BaseModel):
+    """Response model for MCP API."""
+    type: str
+    text: Optional[str] = None
+    markdown: Optional[str] = None
+    data: Optional[Dict[str, Any]] = None
+    error: Optional[str] = None 

pyproject.toml

@@ -0,0 +1,17 @@
+[project]
+name = "youtube"
+version = "0.1.0"
+description = "YouTube API integration for Model Context Protocol (MCP)"
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = [
+    "fastapi>=0.104.0",
+    "uvicorn>=0.23.2",
+    "pydantic>=2.4.2",
+    "httpx>=0.25.0",
+    "python-dotenv>=1.0.0",
+    "google-api-python-client>=2.122.0",
+    "gradio>=4.4.0",
+    "youtube-transcript-api>=0.6.1",
+    "google-genai>=0.3.0"
+]

requirements.txt

@@ -0,0 +1,9 @@
+fastapi>=0.104.0
+uvicorn>=0.23.2
+pydantic>=2.4.2
+httpx>=0.25.0
+python-dotenv>=1.0.0
+google-api-python-client>=2.122.0
+gradio>=4.4.0
+youtube-transcript-api>=0.6.1
+google-genai>=0.3.0 

utils.py

@@ -0,0 +1,57 @@
+import re
+
+def format_timestamp(seconds):
+    """Formats time in seconds to hours:minutes:seconds format."""
+    hours = int(seconds // 3600)
+    minutes = int((seconds % 3600) // 60)
+    secs = int(seconds % 60)
+    
+    if hours > 0:
+        return f"{hours:02d}:{minutes:02d}:{secs:02d}"
+    else:
+        return f"{minutes:02d}:{secs:02d}"
+
+def extract_video_id(video_id_or_url):
+    """
+    Extracts video ID from a string that can be either an ID or full YouTube URL.
+    
+    Supported formats:
+    - Simple ID (e.g., dQw4w9WgXcQ)
+    - https://www.youtube.com/watch?v=dQw4w9WgXcQ
+    - https://youtu.be/dQw4w9WgXcQ
+    - https://youtube.com/shorts/dQw4w9WgXcQ
+    - https://www.youtube.com/embed/dQw4w9WgXcQ
+    - https://youtube.com/live/dQw4w9WgXcQ
+    
+    Returns:
+    - Video ID or original string if ID not found
+    """
+    print(f"Processing input value: {video_id_or_url}")
+    
+    # If input string is empty or None, return empty string
+    if not video_id_or_url:
+        print("Empty video ID")
+        return ""
+    
+    # Check for simple ID (without special characters)
+    if re.match(r'^[a-zA-Z0-9_-]{11}$', video_id_or_url):
+        print(f"Found simple ID: {video_id_or_url}")
+        return video_id_or_url
+    
+    # Check for nested URLs (when URL is part of another URL)
+    inner_url_match = re.search(r'https?://(?:www\.)?(?:youtube\.com|youtu\.be).*?(?=&|$|\s)', video_id_or_url)
+    if inner_url_match:
+        inner_url = inner_url_match.group(0)
+        print(f"Found nested URL: {inner_url}")
+        video_id_or_url = inner_url
+    
+    # Check for standard youtube.com/watch?v= link
+    match = re.search(r'(?:youtube\.com/watch\?v=|youtu\.be/|youtube\.com/shorts/|youtube\.com/embed/|youtube\.com/live/)([a-zA-Z0-9_-]{11})', video_id_or_url)
+    if match:
+        video_id = match.group(1)
+        print(f"Extracted ID from URL: {video_id}")
+        return video_id
+    
+    # If failed to extract ID, return original string
+    print(f"Failed to extract ID, returning original value: {video_id_or_url}")
+    return video_id_or_url