"""YtMCP Server - Main FastMCP Server Implementation
Production-grade YouTube MCP server with 16 specialized tools
across 4 categories: Search, Video Intelligence, Channel Forensics, and Utilities.
"""
import sys
from mcp.server import FastMCP
from starlette.requests import Request
from starlette.responses import JSONResponse
# Import all tool functions
from .tools.search import (
search_videos,
search_filtered,
get_trending_videos,
find_channels,
find_playlists
)
from .tools.video import (
get_transcript,
get_video_metadata,
get_video_chapters,
get_thumbnail,
get_comments
)
from .tools.channel import (
get_channel_videos,
get_channel_shorts,
get_channel_streams,
get_playlist_items,
get_channel_about
)
from .tools.utils import get_audio_stream_url
# Initialize FastMCP server
mcp = FastMCP(
name="YtMCP",
instructions=(
"YtMCP is a comprehensive YouTube MCP server providing read-only access to YouTube data. "
"It offers 16 specialized tools across 4 categories:\n\n"
"**Category A: Core Discovery (Search)**\n"
"- search_videos: Basic keyword search\n"
"- search_filtered: Advanced search with filters (date, duration, sort)\n"
"- get_trending_videos: Current trending videos\n"
"- find_channels: Search for channels\n"
"- find_playlists: Search for playlists\n\n"
"**Category B: Video Intelligence**\n"
"- get_transcript: Extract time-synced transcripts/subtitles\n"
"- get_video_metadata: Comprehensive video information\n"
"- get_video_chapters: Extract video chapters/key moments\n"
"- get_thumbnail: High-resolution thumbnail URLs\n"
"- get_comments: Fetch top comments (slower operation)\n\n"
"**Category C: Channel & Playlist Forensics**\n"
"- get_channel_videos: List channel videos with sorting\n"
"- get_channel_shorts: List channel Shorts\n"
"- get_channel_streams: List live streams\n"
"- get_playlist_items: Flatten playlist contents\n"
"- get_channel_about: Channel information and description\n\n"
"**Category D: Utilities**\n"
"- get_audio_stream_url: Get direct audio stream URL\n\n"
"All tools implement rate limiting (0.75s delay) to prevent IP bans. "
"Outputs are formatted in Markdown for optimal LLM consumption."
),
website_url="https://github.com/yourusername/ytmcp",
debug=False,
log_level="INFO"
)
# ============================================================================
# HEALTH CHECK ENDPOINT (for production deployments)
# ============================================================================
@mcp.custom_route("/health", methods=["GET"])
async def health_check(request: Request) -> JSONResponse:
"""Health check endpoint for load balancers and monitoring."""
return JSONResponse({
"status": "healthy",
"service": "ytmcp",
"version": "0.1.0",
"tools_count": 16
})
# ============================================================================
# CATEGORY A: CORE DISCOVERY (SEARCH)
# ============================================================================
@mcp.tool()
async def search_videos_tool(query: str, limit: int = 10) -> str:
"""
Search YouTube for videos by keyword.
Args:
query: Search query string
limit: Maximum number of results (1-50, default: 10)
"""
return await search_videos(query, limit)
@mcp.tool()
async def search_filtered_tool(
query: str,
upload_date: str = "any",
duration: str = "any",
sort: str = "relevance",
limit: int = 10
) -> str:
"""
Advanced video search with filters.
Args:
query: Search query string
upload_date: 'hour', 'today', 'week', 'month', 'year', 'any'
duration: 'short' (<4min), 'medium' (4-20min), 'long' (>20min), 'any'
sort: 'relevance', 'upload_date', 'view_count', 'rating'
limit: Maximum results (1-50, default: 10)
"""
return await search_filtered(query, upload_date, duration, sort, limit)
@mcp.tool()
async def get_trending_videos_tool(limit: int = 20) -> str:
"""
Fetch currently trending videos from YouTube.
Args:
limit: Maximum number of trending videos (1-50, default: 20)
"""
return await get_trending_videos(limit)
@mcp.tool()
async def find_channels_tool(query: str, limit: int = 10) -> str:
"""
Search for YouTube channels by name or topic.
Args:
query: Channel search query
limit: Maximum results (1-30, default: 10)
"""
return await find_channels(query, limit)
@mcp.tool()
async def find_playlists_tool(query: str, limit: int = 10) -> str:
"""
Search for YouTube playlists by keyword.
Args:
query: Playlist search query
limit: Maximum results (1-30, default: 10)
"""
return await find_playlists(query, limit)
# ============================================================================
# CATEGORY B: VIDEO INTELLIGENCE
# ============================================================================
@mcp.tool()
async def get_transcript_tool(video_id: str, languages: str = "en") -> str:
"""
Fetch time-synced transcript/subtitles for a YouTube video.
CRITICAL for video content analysis and summarization.
Args:
video_id: YouTube video ID or URL
languages: Comma-separated language codes (e.g., 'en,de,fr')
"""
return await get_transcript(video_id, languages)
@mcp.tool()
async def get_video_metadata_tool(video_id: str) -> str:
"""
Extract comprehensive metadata for a YouTube video.
Includes views, tags, description, likes, publish date, and more.
Args:
video_id: YouTube video ID or URL
"""
return await get_video_metadata(video_id)
@mcp.tool()
async def get_video_chapters_tool(video_id: str) -> str:
"""
Extract video chapters/key moments if available.
Args:
video_id: YouTube video ID or URL
"""
return await get_video_chapters(video_id)
@mcp.tool()
async def get_thumbnail_tool(video_id: str, quality: str = "maxres") -> str:
"""
Get high-resolution thumbnail URL(s) for a video.
Args:
video_id: YouTube video ID or URL
quality: 'maxres', 'sd', 'hq', 'mq', 'default'
"""
return await get_thumbnail(video_id, quality)
@mcp.tool()
async def get_comments_tool(video_id: str, max_comments: int = 20) -> str:
"""
Fetch top comments from a YouTube video.
WARNING: Slower operation. Use strict limits.
Args:
video_id: YouTube video ID or URL
max_comments: Maximum comments (1-50, default: 20)
"""
return await get_comments(video_id, max_comments)
# ============================================================================
# CATEGORY C: CHANNEL & PLAYLIST FORENSICS
# ============================================================================
@mcp.tool()
async def get_channel_videos_tool(channel_id: str, sort_by: str = "newest", limit: int = 30) -> str:
"""
List all videos from a channel with sorting options.
Args:
channel_id: YouTube channel ID, @handle, or URL
sort_by: 'newest', 'oldest', 'popular'
limit: Maximum videos (1-100, default: 30)
"""
return await get_channel_videos(channel_id, sort_by, limit)
@mcp.tool()
async def get_channel_shorts_tool(channel_id: str, limit: int = 30) -> str:
"""
List YouTube Shorts from a specific channel.
Args:
channel_id: YouTube channel ID, @handle, or URL
limit: Maximum shorts (1-100, default: 30)
"""
return await get_channel_shorts(channel_id, limit)
@mcp.tool()
async def get_channel_streams_tool(channel_id: str, limit: int = 20) -> str:
"""
List live streams (past and present) from a channel.
Args:
channel_id: YouTube channel ID, @handle, or URL
limit: Maximum streams (1-50, default: 20)
"""
return await get_channel_streams(channel_id, limit)
@mcp.tool()
async def get_playlist_items_tool(playlist_id: str, limit: int = 50) -> str:
"""
Flatten a playlist into a list of video IDs and titles.
Args:
playlist_id: YouTube playlist ID or URL
limit: Maximum videos (1-500, default: 50)
"""
return await get_playlist_items(playlist_id, limit)
@mcp.tool()
async def get_channel_about_tool(channel_id: str) -> str:
"""
Get detailed channel information including description and statistics.
Args:
channel_id: YouTube channel ID, @handle, or URL
"""
return await get_channel_about(channel_id)
# ============================================================================
# CATEGORY D: UTILITIES
# ============================================================================
@mcp.tool()
async def get_audio_stream_url_tool(video_id: str) -> str:
"""
Get the direct audio stream URL for a YouTube video.
Returns a playable audio URL without downloading.
Args:
video_id: YouTube video ID or URL
"""
return await get_audio_stream_url(video_id)
# ============================================================================
# MAIN ENTRY POINT
# ============================================================================
def main():
"""Main entry point for the YtMCP server."""
import argparse
parser = argparse.ArgumentParser(
description="YtMCP - YouTube Model Context Protocol Server"
)
parser.add_argument(
"--transport",
choices=["stdio", "sse", "streamable-http"],
default="stdio",
help="Transport protocol to use (default: stdio)"
)
parser.add_argument(
"--host",
default="127.0.0.1",
help="Host to bind to for HTTP transports (default: 127.0.0.1)"
)
parser.add_argument(
"--port",
type=int,
default=8000,
help="Port to bind to for HTTP transports (default: 8000)"
)
args = parser.parse_args()
# Update server settings if HTTP transport
if args.transport in ["sse", "streamable-http"]:
mcp.settings.host = args.host
mcp.settings.port = args.port
print(f"🚀 Starting YtMCP Server on {args.transport} transport...", file=sys.stderr)
print(f"📊 Loaded 16 YouTube tools across 4 categories", file=sys.stderr)
print(f"⏱️ Rate limiting: 0.75s delay between requests", file=sys.stderr)
mcp.run(transport=args.transport)
if __name__ == "__main__":
main()
