YouTube Search MCP Server

#!/usr/bin/env python3 """ YouTube MCP Server for Dedalus A Model Context Protocol server that provides YouTube search capabilities via HTTP transport for remote deployment on Dedalus platform. Features: - Remote HTTP server mode (Dedalus deployment) - Local STDIO mode (development) - YouTube search tool - Health check endpoint - Environment-based configuration """ import os import sys from pathlib import Path from dotenv import load_dotenv from mcp.server.fastmcp import FastMCP from starlette.requests import Request from starlette.responses import PlainTextResponse # Import YouTube search functionality from src.search import handle_get_serps # Load environment variables env_path = Path(".") / ".env" if env_path.exists(): load_dotenv(env_path) # Server configuration SERVER_NAME = "YouTube Search MCP" DEFAULT_HOST = "0.0.0.0" DEFAULT_PORT = 8080 # Get configuration from environment HOST = os.getenv("HOST", DEFAULT_HOST) PORT = int(os.getenv("PORT", str(DEFAULT_PORT))) # Create FastMCP server instance mcp = FastMCP( name=SERVER_NAME, host=HOST, port=PORT, instructions="""YouTube MCP Server - Search YouTube videos and channels Available tools: - search_youtube: Search YouTube and return video results with metadata This server supports: - Multiple languages (en, es, fr, de, etc.) - Configurable result limits - Rich video metadata (views, duration, channel info) - Pagination handling Example usage: search_youtube("machine learning tutorial", lang="en", max_results=10) """, ) @mcp.custom_route("/health", methods=["GET"]) async def health_check(request: Request) -> PlainTextResponse: """Health check endpoint for load balancers and monitoring systems. Returns: PlainTextResponse: "OK" with 200 status code when server is healthy """ return PlainTextResponse("OK") @mcp.tool() def search_youtube(query: str, lang: str = "en", max_results: int = 30) -> list[dict]: """Search YouTube and return video results. This tool searches YouTube using the official YouTube API and returns structured video metadata including titles, channels, views, and duration. Args: query: Search query string (e.g., "machine learning tutorial", "Veritasium") lang: Language code for results (e.g., "en", "es", "fr", "de") (default: "en") max_results: Maximum number of results to return, up to 100 (default: 30) Returns: List of search result dictionaries, each containing: - video_id: YouTube video ID (can be used to construct URLs) - video_title: Full video title - channel_id: YouTube channel ID - channel_name: Channel display name - channel_link: Channel URL path (relative) - view_count: Formatted view count (e.g., "1.2M views") - published_date: Human-readable publish time (e.g., "2 days ago") - duration: Video duration string (e.g., "10:23") Example: >>> results = search_youtube("Andrej Karpathy", lang="en", max_results=10) >>> print(results[0]["video_title"]) "Let's build GPT: from scratch, in code, spelled out." >>> results = search_youtube("machine learning tutorial", max_results=5) >>> for video in results: >>> print(f"{video['video_title']} - {video['channel_name']}") Notes: - Results are sorted by YouTube's relevance algorithm - Maximum max_results is capped at 100 per request - Language codes follow ISO 639-1 standard - View counts and publish dates are localized """ # Validate max_results if max_results > 100: max_results = 100 elif max_results < 1: max_results = 1 # Call YouTube search handler results = handle_get_serps( query=query, langauge_code=lang, max_results=max_results ) # Convert Pydantic models to dictionaries return [r.model_dump() for r in results] def main(): """Main entry point for the MCP server. This function determines the transport mode based on environment variables and command-line arguments: - HTTP mode (Dedalus deployment): When PORT is set or --port is provided - STDIO mode (local development): Default when no port is specified Environment Variables: PORT: Port number for HTTP server (default: 8080) HOST: Host address for HTTP server (default: 0.0.0.0) Command-Line Arguments: --port PORT: Override port for HTTP mode --host HOST: Override host for HTTP mode --stdio: Force STDIO transport mode --help: Show help message Returns: Exit code (0 for success, non-zero for errors) """ import argparse # Parse command-line arguments parser = argparse.ArgumentParser( description="YouTube MCP Server - Search YouTube via MCP" ) parser.add_argument( "--port", type=int, help=f"Port for HTTP transport (default: {DEFAULT_PORT})", ) parser.add_argument( "--host", type=str, default=DEFAULT_HOST, help=f"Host for HTTP transport (default: {DEFAULT_HOST})", ) parser.add_argument( "--stdio", action="store_true", help="Force STDIO transport mode (for local development)", ) args = parser.parse_args() # Determine transport mode # Use HTTP if port is specified (from env or CLI) and not forcing stdio use_http = (args.port or os.getenv("PORT")) and not args.stdio if use_http: # HTTP transport mode (for Dedalus deployment) actual_host = args.host if args.host != DEFAULT_HOST else HOST actual_port = args.port if args.port else PORT print(f"[YouTube MCP] Starting HTTP server on {actual_host}:{actual_port}") print(f"[YouTube MCP] Health check: http://{actual_host}:{actual_port}/health") print(f"[YouTube MCP] SSE endpoint: http://{actual_host}:{actual_port}/sse") try: mcp.run(transport="sse") return 0 except Exception as e: print(f"[YouTube MCP] Error starting HTTP server: {e}", file=sys.stderr) return 1 else: # STDIO transport mode (for local development) print("[YouTube MCP] Starting STDIO server", file=sys.stderr) print("[YouTube MCP] Connect via MCP client using stdio", file=sys.stderr) try: mcp.run(transport="stdio") return 0 except Exception as e: print(f"[YouTube MCP] Error starting STDIO server: {e}", file=sys.stderr) return 1 if __name__ == "__main__": sys.exit(main())