search_videos
Search YouTube videos by query to find relevant content, retrieve metadata, and support downloading without an API key.
Instructions
Search YouTube for videos matching a query.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | ||
| max_results | No | ||
| output_format | No | json |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- The 'search_videos' function is defined as an MCP tool, handling the search request by validating input, using a provider to perform the search, and formatting the output.
@mcp.tool(name="search_videos") async def search_videos(query: str, max_results: int = 10, output_format: str = "json") -> str: """Search YouTube for videos matching a query.""" try: logger.info(f"Search request: query='{query}', max_results={max_results}") # Validate query validated_query = validate_query(query) # Create search parameters params = SearchParams( query=validated_query, max_results=max_results, ) # Execute search provider = get_search_provider() videos = await provider.search(params) # Format results formatter = get_formatter(output_format) result = formatter.format_videos(videos) logger.info(f"Search completed: found {len(videos)} videos") return result except InvalidQueryError as e: logger.warning(f"Invalid query: {e.message}") return json.dumps({"error": "invalid_query", "message": e.message}) except NetworkError as e: logger.error(f"Network error: {e.message}") return json.dumps( { "error": "network_error", "message": "Failed to connect to YouTube. Please try again.", "details": e.message, } ) except SearchProviderError as e: logger.error(f"Search provider error: {e.message}", exc_info=True) return json.dumps( { "error": "search_failed", "message": "Search operation failed. Please try a different query.", "details": e.message, } ) except Exception: logger.exception("Unexpected error in search_videos") return json.dumps(