# Directory Structure
```
├── .gitignore
├── LICENSE
├── main.py
├── notebooks
│ ├── __init__.py
│ ├── fetch_transcipts.ipynb
│ └── youtube_serach.ipynb
├── pyproject.toml
├── README.md
├── requirements.txt
├── server.py
├── tools
│ ├── __init__.py
│ ├── fetch_transcripts.py
│ ├── get_channel_metrics.py
│ ├── get_playlist_metrics.py
│ ├── get_video_metrics.py
│ ├── search_channels.py
│ ├── search_playlists.py
│ └── search_videos.py
└── utils
├── __init__.py
├── models.py
└── tool_utils.py
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[codz]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py.cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
.python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# UV
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
uv.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
#poetry.toml
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
# pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python.
# https://pdm-project.org/en/latest/usage/project/#working-with-version-control
#pdm.lock
#pdm.toml
.pdm-python
.pdm-build/
# pixi
# Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control.
#pixi.lock
# Pixi creates a virtual environment in the .pixi directory, just like venv module creates one
# in the .venv directory. It is recommended not to include this directory in version control.
.pixi
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.envrc
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# YouTube Content Management MCP Server
A Model Context Protocol (MCP) server that provides YouTube Data API v3 integration for content discovery and analytics. This server enables AI assistants to search for YouTube videos, channels, playlists, and retrieve detailed metrics for videos, channels, and playlists.
## Features
### Current Tools
- **🎥 search_videos**: Search YouTube for videos with advanced filtering options, including view count, like count, and comment count.
- **📺 search_channels**: Find YouTube channels based on search queries, including subscriber count, video count, and total view count.
- **📋 search_playlists**: Search YouTube for playlists based on search queries.
- **📊 get_video_metrics**: Retrieve statistics (views, likes, comments) for a specific video by ID.
- **📈 get_channel_metrics**: Retrieve statistics (subscribers, total views, video count) for a specific channel by ID.
- **📑 get_playlist_metrics**: Retrieve statistics (item count, total views) for a specific playlist by ID.
### Planned Features
- Playlist creation and management
- Comment retrieval and analysis
- Video upload and management (with proper authentication)
- Trending videos by region
- Video transcription access
## Prerequisites
- Python 3.8 or higher
- YouTube Data API v3 key
- VSCode with MCP extension (for VSCode usage)
- Required Python packages: `google-api-python-client`, `python-dotenv`, `pydantic`
## Getting Your YouTube API Key
1. Go to the [Google Cloud Console](https://console.cloud.google.com/)
2. Create a new project or select an existing one
3. Enable the YouTube Data API v3:
- Navigate to "APIs & Services" > "Library"
- Search for "YouTube Data API v3"
- Click on it and press "Enable"
4. Create credentials:
- Go to "APIs & Services" > "Credentials"
- Click "Create Credentials" > "API Key"
- Copy the generated API key
5. (Recommended) Restrict the API key:
- Click on the API key to edit it
- Under "API restrictions", select "Restrict key"
- Choose "YouTube Data API v3"
- Save the changes
## Installation
1. **Clone or download this repository**
```bash
git clone https://github.com/NastyRunner13/youtube-content-management-mcp
cd youtube-content-management-mcp
```
2. **Install dependencies**
```bash
pip install -r requirements.txt
```
Or if using `uv`:
```bash
uv install
```
3. **Set up your environment** (Optional)
Create a `.env` file in the project root:
```env
YOUTUBE_API_KEY=your_youtube_api_key_here
```
## Usage
### With VSCode (Recommended)
1. **Install the MCP extension** in VSCode
2. **Configure the MCP server** by adding this to your VSCode `settings.json`:
```json
{
"mcp.servers": {
"youtube-content-management": {
"command": "python",
"args": [
"/path/to/youtube-content-management-mcp/main.py"
],
"env": {
"YOUTUBE_API_KEY": "your_youtube_api_key_here"
}
}
}
}
```
**Alternative using uv:**
```json
{
"mcp.servers": {
"youtube-content-management": {
"command": "uv",
"args": [
"--directory",
"/path/to/youtube-content-management-mcp",
"run",
"main.py"
],
"env": {
"YOUTUBE_API_KEY": "your_youtube_api_key_here"
}
}
}
}
```
3. **Restart VSCode** or reload the window
4. **Use the tools** through the MCP panel or by asking your AI assistant
### With Claude Desktop
Add this configuration to your Claude Desktop config file:
**Windows:** `%APPDATA%/Claude/claude_desktop_config.json`
**macOS:** `~/Library/Application Support/Claude/claude_desktop_config.json`
```json
{
"mcpServers": {
"youtube-content-management": {
"command": "python",
"args": ["/path/to/youtube-content-management-mcp/main.py"],
"env": {
"YOUTUBE_API_KEY": "your_youtube_api_key_here"
}
}
}
}
```
### With Other MCP Clients
The server implements the standard MCP protocol and should work with any compatible MCP client. Refer to your client's documentation for configuration instructions.
## Available Tools
### search_videos
Search YouTube for videos with advanced filtering options, including metrics like view count, like count, and comment count.
**Parameters:**
- `query` (string, required): Search query
- `max_results` (integer, optional): Maximum number of results (1-50, default: 25)
- `order` (string, optional): Sort order - "relevance", "date", "rating", "viewCount" (default: "relevance")
- `duration` (string, optional): Video duration - "medium", "long" (default: "medium")
- `published_after` (string, optional): RFC 3339 timestamp (e.g., "2023-01-01T00:00:00Z")
**Example usage:**
```
Search for Python tutorials uploaded in the last year, sorted by view count
```
### search_channels
Find YouTube channels based on search queries, including metrics like subscriber count, video count, and total view count.
**Parameters:**
- `query` (string, required): Search query for channels
- `max_results` (integer, optional): Maximum number of results (1-50, default: 25)
- `published_after` (string, optional): RFC 3339 timestamp (e.g., "2023-01-01T00:00:00Z")
**Example usage:**
```
Find coding tutorial channels
```
### search_playlists
Search YouTube for playlists based on search queries.
**Parameters:**
- `query` (string, required): Search query for playlists
- `max_results` (integer, optional): Maximum number of results (1-50, default: 25)
- `published_after` (string, optional): RFC 3339 timestamp (e.g., "2023-01-01T00:00:00Z")
**Example usage:**
```
Find playlists about machine learning
```
### get_video_metrics
Retrieve statistics for a specific YouTube video, including view count, like count, and comment count.
**Parameters:**
- `video_id` (string, required): The YouTube video ID
**Example usage:**
```
Get metrics for the video with ID dQw4w9WgXcQ
```
### get_channel_metrics
Retrieve statistics for a specific YouTube channel, including subscriber count, total view count, and video count.
**Parameters:**
- `channel_id` (string, required): The YouTube channel ID
**Example usage:**
```
Get metrics for the channel with ID UC_x5XG1OV2P6uZZ5FSM9Ttw
```
### get_playlist_metrics
Retrieve statistics for a specific YouTube playlist, including item count and total view count of all videos.
**Parameters:**
- `playlist_id` (string, required): The YouTube playlist ID
**Example usage:**
```
Get metrics for the playlist with ID PL-osiE80TeTt2d9bfVyTiXJA-UTHn6WwU
```
## Example Interactions
Once the MCP server is configured, you can interact with it through your AI assistant:
**Video Search with Metrics:**
> "Search for machine learning tutorials from the last 6 months, sorted by view count, and show view counts"
**Channel Discovery with Metrics:**
> "Find top cooking channels on YouTube with their subscriber counts"
**Playlist Search:**
> "Show me playlists about Python programming"
**Video Metrics:**
> "Get the view count and like count for the video with ID dQw4w9WgXcQ"
**Channel Metrics:**
> "What are the subscriber count and total views for the channel UC_x5XG1OV2P6uZZ5FSM9Ttw?"
**Playlist Metrics:**
> "How many videos and total views are in the playlist PL-osiE80TeTt2d9bfVyTiXJA-UTHn6WwU?"
## Input Validation
All tools use [Pydantic](https://pydantic-docs.helpmanual.io/) for robust input validation, ensuring:
- Required fields (e.g., `query`, `video_id`) are provided and non-empty.
- Numeric fields (e.g., `max_results`) are within valid ranges (1-50).
- String fields (e.g., `order`, `duration`) match allowed values.
- Timestamps (e.g., `published_after`) follow RFC 3339 format.
Invalid inputs result in clear error messages, improving reliability and user experience.
## Security Notes
- **Never commit your API key** to version control
- Consider using environment variables instead of hardcoding API keys
- Regularly rotate your API keys
- Monitor your API usage in Google Cloud Console
- Set up API key restrictions to limit usage to YouTube Data API v3
## Troubleshooting
### Common Issues
1. **"YouTube API key is not set"**
- Ensure your API key is properly configured in the environment variables
- Check that the key is valid and has YouTube Data API v3 enabled
2. **"quotaExceeded" errors**
- You've hit your daily API quota limit (default: 10,000 units)
- Wait until the quota resets (daily) or increase your quota in Google Cloud Console
- Note: Metrics tools and search tools with metrics may consume more quota due to multiple API calls
3. **"keyInvalid" errors**
- Your API key is invalid or has been revoked
- Generate a new API key and update your configuration
4. **"Invalid input arguments" errors**
- Check the Pydantic error message for details (e.g., missing `query`, invalid `order`)
- Ensure inputs match the tool's parameter requirements
5. **MCP server not starting**
- Check that all dependencies (`google-api-python-client`, `python-dotenv`, `pydantic`) are installed
- Verify the Python path in your configuration is correct
- Check the MCP extension logs for detailed error messages
### Debug Mode
To enable debug logging, add this to your environment:
```json
"env": {
"YOUTUBE_API_KEY": "your_key_here",
"DEBUG": "true"
}
```
## Contributing
We welcome contributions! Areas where you can help:
- Additional YouTube API endpoints (comments, transcriptions)
- Optimizing API quota usage (e.g., batching metrics calls)
- Enhancing Pydantic validation rules
- Performance optimizations
- Documentation improvements
- Testing and bug reports
## API Limits
- **YouTube Data API v3**: 10,000 units per day (default)
- **Search operations**: 100 units per request
- **List operations (videos, channels, playlists)**: 1 unit per request
- **Playlist items**: 5 units per request
- **Rate limiting**: Be mindful of making too many requests in quick succession, especially with metrics tools
## Support
- Create an issue for bugs or feature requests
- Check the [YouTube Data API documentation](https://developers.google.com/youtube/v3) for API-specific questions
- Review MCP protocol documentation for integration issues
- Refer to [Pydantic documentation](https://pydantic-docs.helpmanual.io/) for validation-related questions
```
--------------------------------------------------------------------------------
/notebooks/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/utils/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
mcp[cli]
python-dotenv
google-api-python-client
ipykernel
pydantic
youtube_transcript_api
```
--------------------------------------------------------------------------------
/server.py:
--------------------------------------------------------------------------------
```python
from mcp.server.fastmcp import FastMCP
# This is the shared MCP server instance
mcp = FastMCP("Youtube Content Management MCP")
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "youtube-content-management-mcp"
version = "0.1.0"
description = "This is the MCP server for Youtube Content Management"
readme = "README.md"
requires-python = ">=3.12"
dependencies = []
```
--------------------------------------------------------------------------------
/tools/__init__.py:
--------------------------------------------------------------------------------
```python
# This file marks the 'tools' directory as a Python package.
from tools.search_videos import search_videos
from tools.search_channels import search_channels
from tools.search_playlists import search_playlists
from tools.get_video_metrics import get_video_metrics
from tools.get_channel_metrics import get_channel_metrics
from tools.get_playlist_metrics import get_playlist_metrics
from tools.get_video_metrics import get_video_metrics
from tools.fetch_transcripts import fetch_transcripts
__all__ = [
"search_videos",
"search_channels",
"search_playlists",
"get_video_metrics",
"get_channel_metrics",
"get_playlist_metrics",
"fetch_transcripts"
]
```
--------------------------------------------------------------------------------
/main.py:
--------------------------------------------------------------------------------
```python
from server import mcp
import tools.search_videos # Import the search_videos tool to register it with the MCP server
import tools.search_channels # Import the search_channels tool to register it with the MCP server
import tools.search_playlists # Import the search_playlists tool to register it with the MCP server
import tools.get_video_metrics # Import the get_video_metrics tool to register it with the MCP server
import tools.get_channel_metrics # Import the get_channel_metrics tool to register it with the MCP
import tools.get_playlist_metrics # Import the get_playlist_metrics tool to register it with the MCP server
import tools.fetch_transcripts # Import the fetch_transcripts tool to register it with the MCP server
# Entry point to run the server
if __name__ == "__main__":
mcp.run()
```
--------------------------------------------------------------------------------
/utils/tool_utils.py:
--------------------------------------------------------------------------------
```python
import re
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from dotenv import load_dotenv
import os
load_dotenv()
class YouTubeAPIError(Exception):
"""Custom exception for YouTube API errors"""
pass
def validate_youtube_params(order: str, duration: str, published_after: str | None) -> None:
"""Validate YouTube API parameters"""
valid_orders = {"relevance", "date", "rating", "viewCount"}
valid_durations = {"medium", "long"}
if order not in valid_orders:
raise ValueError(f"Invalid order parameter: {order}. Must be one of {valid_orders}")
if duration not in valid_durations:
raise ValueError(f"Invalid duration parameter: {duration}. Must be one of {valid_durations}")
if published_after and not re.match(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$', published_after):
raise ValueError(f"Invalid published_after format: {published_after}. Must be RFC 3339 (e.g., 2023-01-01T00:00:00Z)")
def get_youtube_client():
"""Get a singleton YouTube API client instance."""
api_key = os.getenv('YOUTUBE_API_KEY')
if not api_key:
raise YouTubeAPIError("YouTube API key is not set in environment variables")
# Cache the client instance (module-level singleton)
if not hasattr(get_youtube_client, 'client'):
get_youtube_client.client = build('youtube', 'v3', developerKey=api_key)
return get_youtube_client.client
```
--------------------------------------------------------------------------------
/tools/get_video_metrics.py:
--------------------------------------------------------------------------------
```python
from server import mcp
from mcp.types import TextContent
from typing import List
from utils.tool_utils import YouTubeAPIError, get_youtube_client
from googleapiclient.errors import HttpError
from utils.models import VideoIdInput
@mcp.tool()
def get_video_metrics(arguments: dict) -> List[TextContent]:
"""Retrieve statistics for a specific YouTube video.
This function queries the YouTube Data API v3 to fetch metrics for a given video,
including view count, like count, and comment count, formatted as a TextContent object.
Args:
arguments: A dictionary containing:
- video_id (str): The YouTube video ID (required).
Returns:
List[TextContent]: A list containing a single TextContent object with a formatted string
including the video's title, view count, like count, and comment count. If the video
is not found, returns a single TextContent with a "No video found" message.
Raises:
YouTubeAPIError: If the API key is missing, the video ID is invalid, the API request fails,
or the input arguments are invalid (via Pydantic).
"""
try:
input_data = VideoIdInput(**arguments)
except ValueError as e:
raise YouTubeAPIError(f"Invalid input arguments: {e}")
youtube = get_youtube_client()
try:
response = youtube.videos().list(
part='snippet,statistics',
id=input_data.video_id
).execute()
items = response.get('items', [])
if not items:
return [TextContent(type="text", text="No video found for the given ID.")]
item = items[0]
video_info = {
'title': item['snippet']['title'],
'view_count': item['statistics'].get('viewCount', '0'),
'like_count': item['statistics'].get('likeCount', '0'),
'comment_count': item['statistics'].get('commentCount', '0')
}
return [TextContent(
type="text",
text=(f"**{video_info['title']}**\n"
f"Views: {video_info['view_count']}\n"
f"Likes: {video_info['like_count']}\n"
f"Comments: {video_info['comment_count']}")
)]
except HttpError as e:
raise YouTubeAPIError(f"YouTube API error for video ID '{input_data.video_id}': {e}")
except Exception as e:
raise YouTubeAPIError(f"Unexpected error for video ID '{input_data.video_id}': {e}")
```
--------------------------------------------------------------------------------
/tools/get_channel_metrics.py:
--------------------------------------------------------------------------------
```python
from server import mcp
from mcp.types import TextContent
from typing import List
from utils.tool_utils import YouTubeAPIError, get_youtube_client
from googleapiclient.errors import HttpError
from utils.models import ChannelIdInput
@mcp.tool()
def get_channel_metrics(arguments: dict) -> List[TextContent]:
"""Retrieve statistics for a specific YouTube channel.
This function queries the YouTube Data API v3 to fetch metrics for a given channel,
including subscriber count, total view count, and video count, formatted as a TextContent object.
Args:
arguments: A dictionary containing:
- channel_id (str): The YouTube channel ID (required).
Returns:
List[TextContent]: A list containing a single TextContent object with a formatted string
including the channel's title, subscriber count, total view count, and video count.
If the channel is not found, returns a single TextContent with a "No channel found" message.
Raises:
YouTubeAPIError: If the API key is missing, the channel ID is invalid, the API request fails,
or the input arguments are invalid (via Pydantic).
"""
try:
input_data = ChannelIdInput(**arguments)
except ValueError as e:
raise YouTubeAPIError(f"Invalid input arguments: {e}")
youtube = get_youtube_client()
try:
response = youtube.channels().list(
part='snippet,statistics',
id=input_data.channel_id
).execute()
items = response.get('items', [])
if not items:
return [TextContent(type="text", text="No channel found for the given ID.")]
item = items[0]
channel_info = {
'title': item['snippet']['title'],
'subscriber_count': item['statistics'].get('subscriberCount', '0'),
'view_count': item['statistics'].get('viewCount', '0'),
'video_count': item['statistics'].get('videoCount', '0')
}
return [TextContent(
type="text",
text=(f"**{channel_info['title']}**\n"
f"Subscribers: {channel_info['subscriber_count']}\n"
f"Total Views: {channel_info['view_count']}\n"
f"Videos: {channel_info['video_count']}")
)]
except HttpError as e:
raise YouTubeAPIError(f"YouTube API error for channel ID '{input_data.channel_id}': {e}")
except Exception as e:
raise YouTubeAPIError(f"Unexpected error for channel ID '{input_data.channel_id}': {e}")
```
--------------------------------------------------------------------------------
/tools/get_playlist_metrics.py:
--------------------------------------------------------------------------------
```python
from server import mcp
from mcp.types import TextContent
from typing import List
from utils.tool_utils import YouTubeAPIError, get_youtube_client
from googleapiclient.errors import HttpError
from utils.models import PlaylistIdInput
@mcp.tool()
def get_playlist_metrics(arguments: dict) -> List[TextContent]:
"""Retrieve statistics for a specific YouTube playlist.
This function queries the YouTube Data API v3 to fetch metrics for a given playlist,
including item count and total view count of all videos, formatted as a TextContent object.
Args:
arguments: A dictionary containing:
- playlist_id (str): The YouTube playlist ID (required).
Returns:
List[TextContent]: A list containing a single TextContent object with a formatted string
including the playlist's title, item count, and total view count. If the playlist
is not found, returns a single TextContent with a "No playlist found" message.
Raises:
YouTubeAPIError: If the API key is missing, the playlist ID is invalid, the API request fails,
or the input arguments are invalid (via Pydantic).
"""
try:
input_data = PlaylistIdInput(**arguments)
except ValueError as e:
raise YouTubeAPIError(f"Invalid input arguments: {e}")
youtube = get_youtube_client()
try:
playlist_response = youtube.playlists().list(
part='snippet',
id=input_data.playlist_id
).execute()
playlist_items = playlist_response.get('items', [])
if not playlist_items:
return [TextContent(type="text", text="No playlist found for the given ID.")]
playlist_title = playlist_items[0]['snippet']['title']
playlist_items_response = youtube.playlistItems().list(
part='contentDetails',
playlistId=input_data.playlist_id,
maxResults=50
).execute()
video_ids = [item['contentDetails']['videoId'] for item in playlist_items_response.get('items', [])]
item_count = len(video_ids)
total_views = 0
if video_ids:
videos_response = youtube.videos().list(
part='statistics',
id=','.join(video_ids)
).execute()
total_views = sum(int(item['statistics'].get('viewCount', 0)) for item in videos_response.get('items', []))
return [TextContent(
type="text",
text=(f"**{playlist_title}**\n"
f"Playlist ID: {input_data.playlist_id}\n"
f"Items: {item_count}\n"
f"Total Views: {total_views}")
)]
except HttpError as e:
raise YouTubeAPIError(f"YouTube API error for playlist ID '{input_data.playlist_id}': {e}")
except Exception as e:
raise YouTubeAPIError(f"Unexpected error for playlist ID '{input_data.playlist_id}': {e}")
```
--------------------------------------------------------------------------------
/tools/search_playlists.py:
--------------------------------------------------------------------------------
```python
from server import mcp
from mcp.types import TextContent
from typing import List
from utils.tool_utils import YouTubeAPIError, get_youtube_client
from googleapiclient.errors import HttpError
from utils.models import SearchPlaylistsInput
@mcp.tool()
def search_playlists(arguments: dict) -> List[TextContent]:
"""Search YouTube for playlists based on a query.
This function queries the YouTube Data API v3 to retrieve playlists matching the provided
search query. Each result includes the playlist title, ID, creation date, and description,
formatted as a single TextContent object.
Args:
arguments: A dictionary containing search parameters:
- query (str): The search query (required).
- max_results (int, optional): Maximum number of results (1 to 50). Defaults to 25.
- published_after (str, optional): RFC 3339 timestamp (e.g., '2023-01-01T00:00:00Z') to filter playlists created after this date.
Returns:
List[TextContent]: A list containing a single TextContent object with a formatted string
listing all found playlists, including their title, playlist ID, creation date, and
truncated description. If no playlists are found, returns a single TextContent with a
"No playlists found" message.
Raises:
YouTubeAPIError: If the API key is missing, the API request fails, or the input arguments are invalid (via Pydantic).
"""
try:
input_data = SearchPlaylistsInput(**arguments)
except ValueError as e:
raise YouTubeAPIError(f"Invalid input arguments: {e}")
youtube = get_youtube_client()
try:
search_params = {
'part': 'snippet',
'q': input_data.query,
'type': 'playlist',
'maxResults': input_data.max_results
}
if input_data.published_after:
search_params['publishedAfter'] = input_data.published_after
search_response = youtube.search().list(**search_params).execute()
playlists = []
for item in search_response.get('items', []):
playlist_info = {
'playlist_id': item['id']['playlistId'],
'title': item['snippet']['title'],
'description': item['snippet']['description'][:200] + ('...' if item['snippet']['description'] else ''),
'published_at': item['snippet']['publishedAt']
}
playlists.append(playlist_info)
if not playlists:
return [TextContent(type="text", text="No playlists found.")]
return [TextContent(
type="text",
text=f"Found {len(playlists)} playlists:\n\n" +
"\n\n".join([f"**{p['title']}**\n"
f"Playlist ID: {p['playlist_id']}\n"
f"Created: {p['published_at']}\n"
f"Description: {p['description']}"
for p in playlists])
)]
except HttpError as e:
raise YouTubeAPIError(f"YouTube API error for query '{input_data.query}': {e}")
except Exception as e:
raise YouTubeAPIError(f"Unexpected error for query '{input_data.query}': {e}")
```
--------------------------------------------------------------------------------
/tools/fetch_transcripts.py:
--------------------------------------------------------------------------------
```python
from server import mcp
from mcp.types import TextContent
from typing import List
from utils.tool_utils import YouTubeAPIError
from youtube_transcript_api import YouTubeTranscriptApi, NoTranscriptFound, TranscriptsDisabled
@mcp.tool()
def fetch_transcripts(arguments: dict) -> List[TextContent]:
"""Retrieve and analyze YouTube video transcripts for detailed information extraction.
This function fetches the complete transcript of a YouTube video, making it ideal for:
- Analyzing video content in detail when users ask about specific topics covered in videos
- Extracting key information, quotes, or explanations from educational content
- Summarizing long-form video content (lectures, tutorials, presentations, interviews)
- Finding specific details or timestamps within video discussions
- Understanding the full context of video content for comprehensive responses
Use this tool when users:
- Ask detailed questions about YouTube video content
- Request summaries or key points from video material
- Need specific information or quotes from video discussions
- Want to understand complex topics explained in video format
- Ask about timestamps or specific moments in videos
Technical Implementation:
This function queries the YouTube Data API v3 to fetch available captions for a video
and retrieves the transcript in the specified language (default: English). The transcript
is returned as a single TextContent object with timestamped text entries. If no transcript
is available, a message indicating the reason is returned.
Args:
arguments: A dictionary containing:
- video_id (str, optional): The YouTube video ID.
- video_url (str, optional): The YouTube video URL (e.g., 'https://www.youtube.com/watch?v=VIDEO_ID').
- language_code (str, optional): Language code for the transcript (e.g., 'en'). Defaults to 'en'.
Either video_id or video_url must be provided.
Returns:
List[TextContent]: A list containing a single TextContent object with the transcript
as a formatted string (timestamp and text). If no transcript is available or the
video is not found, returns a TextContent with an appropriate message.
Raises:
YouTubeAPIError: If the API key is missing, the API request fails, the input arguments
are invalid (via Pydantic), or an unexpected error occurs.
"""
from utils.models import FetchTranscriptsInput # Moved import here to avoid circular import issues
try:
input_data = FetchTranscriptsInput(**arguments)
except ValueError as e:
raise YouTubeAPIError(f"Invalid input arguments: {e}")
ytt_api = YouTubeTranscriptApi()
try:
# Try to fetch transcript directly first
try:
transcript = ytt_api.fetch(
video_id=input_data.video_id,
languages=[input_data.language_code]
)
except NoTranscriptFound:
# If specific language not found, try to get any available transcript
try:
transcript = ytt_api.fetch(video_id=input_data.video_id)
except NoTranscriptFound:
return [TextContent(type="text", text=f"No transcript available for this video in any language.")]
transcript_text = "".join(snippet.text for snippet in transcript.snippets)
if not transcript_text.strip():
return [TextContent(type="text", text="Transcript is empty or unavailable.")]
return [TextContent(
type="text",
text=f"Transcript for video ID {input_data.video_id} (language: {input_data.language_code}):\n\n{transcript_text}"
)]
except TranscriptsDisabled:
return [TextContent(type="text", text="Transcripts are disabled for this video or access is restricted.")]
except NoTranscriptFound:
return [TextContent(type="text", text=f"No transcript available for this video.")]
except Exception as e:
raise YouTubeAPIError(f"Unexpected error for video ID '{input_data.video_id}': {e}")
```
--------------------------------------------------------------------------------
/tools/search_channels.py:
--------------------------------------------------------------------------------
```python
from server import mcp
from mcp.types import TextContent
from typing import List
from utils.tool_utils import YouTubeAPIError, get_youtube_client
from googleapiclient.errors import HttpError
from tools.get_channel_metrics import get_channel_metrics
from utils.models import SearchChannelsInput
@mcp.tool()
def search_channels(arguments: dict) -> List[TextContent]:
"""Search YouTube for channels based on a query, with metrics.
This function queries the YouTube Data API v3 to retrieve channels matching the provided
search query. Each result includes the channel title, ID, creation date, description,
subscriber count, video count, and total view count, formatted as a single TextContent object.
Metrics are fetched using the get_channel_metrics tool.
Args:
arguments: A dictionary containing search parameters:
- query (str): The search query (required).
- max_results (int, optional): Maximum number of results (1 to 50). Defaults to 25.
- published_after (str, optional): RFC 3339 timestamp (e.g., '2023-01-01T00:00:00Z') to filter channels created after this date.
Returns:
List[TextContent]: A list containing a single TextContent object with a formatted string
listing all found channels, including their title, channel ID, creation date,
truncated description, subscriber count, video count, and total view count.
If no channels are found, returns a single TextContent with a "No channels found" message.
Raises:
YouTubeAPIError: If the API key is missing, the API request fails, the input arguments are invalid (via Pydantic), or an unexpected error occurs.
"""
try:
input_data = SearchChannelsInput(**arguments)
except ValueError as e:
raise YouTubeAPIError(f"Invalid input arguments: {e}")
youtube = get_youtube_client()
try:
search_params = {
'part': 'snippet',
'q': input_data.query,
'type': 'channel',
'maxResults': input_data.max_results
}
if input_data.published_after:
search_params['publishedAfter'] = input_data.published_after
search_response = youtube.search().list(**search_params).execute()
channels = []
for item in search_response.get('items', []):
channel_id = item['id']['channelId']
channel_info = {
'channel_id': channel_id,
'title': item['snippet']['title'],
'description': item['snippet']['description'][:200] + ('...' if item['snippet']['description'] else ''),
'published_at': item['snippet']['publishedAt']
}
# Fetch metrics using get_channel_metrics
metrics_response = get_channel_metrics({"channel_id": channel_id})
if metrics_response[0].text.startswith("No channel found"):
continue
metrics_text = metrics_response[0].text.split('\n')
subscriber_count = next((line.split(': ')[1] for line in metrics_text if line.startswith('Subscribers')), '0')
video_count = next((line.split(': ')[1] for line in metrics_text if line.startswith('Videos')), '0')
view_count = next((line.split(': ')[1] for line in metrics_text if line.startswith('Total Views')), '0')
channel_info.update({
'subscriber_count': subscriber_count,
'video_count': video_count,
'view_count': view_count
})
channels.append(channel_info)
if not channels:
return [TextContent(type="text", text="No channels found.")]
return [TextContent(
type="text",
text=f"Found {len(channels)} channels:\n\n" +
"\n\n".join([f"**{c['title']}**\n"
f"Channel ID: {c['channel_id']}\n"
f"Created: {c['published_at']}\n"
f"Subscribers: {c['subscriber_count']}\n"
f"Videos: {c['video_count']}\n"
f"Total Views: {c['view_count']}\n"
f"Description: {c['description']}"
for c in channels])
)]
except HttpError as e:
raise YouTubeAPIError(f"YouTube API error for query '{input_data.query}': {e}")
except Exception as e:
raise YouTubeAPIError(f"Unexpected error for query '{input_data.query}': {e}")
```
--------------------------------------------------------------------------------
/utils/models.py:
--------------------------------------------------------------------------------
```python
from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Optional
import re
class SearchVideosInput(BaseModel):
query: str = Field(..., min_length=1, description="The search query (required)")
max_results: Optional[int] = Field(25, ge=1, le=50, description="Maximum number of results (1 to 50)")
order: Optional[str] = Field("relevance", description="Sort order: relevance, date, rating, viewCount")
duration: Optional[str] = Field("medium", description="Video duration: medium, long")
published_after: Optional[str] = Field(None, description="RFC 3339 timestamp (e.g., 2023-01-01T00:00:00Z)")
@field_validator("order")
@classmethod
def validate_order(cls, v):
valid_orders = {"relevance", "date", "rating", "viewCount"}
if v not in valid_orders:
raise ValueError(f"Invalid order: {v}. Must be one of {valid_orders}")
return v
@field_validator("duration")
@classmethod
def validate_duration(cls, v):
valid_durations = {"medium", "long"}
if v not in valid_durations:
raise ValueError(f"Invalid duration: {v}. Must be one of {valid_durations}")
return v
@field_validator("published_after")
@classmethod
def validate_published_after(cls, v):
if v and not re.match(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$', v):
raise ValueError(f"Invalid published_after format: {v}. Must be RFC 3339 (e.g., 2023-01-01T00:00:00Z)")
return v
class SearchChannelsInput(BaseModel):
query: str = Field(..., min_length=1, description="The search query (required)")
max_results: Optional[int] = Field(25, ge=1, le=50, description="Maximum number of results (1 to 50)")
published_after: Optional[str] = Field(None, description="RFC 3339 timestamp (e.g., 2023-01-01T00:00:00Z)")
@field_validator("published_after")
@classmethod
def validate_published_after(cls, v):
if v and not re.match(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$', v):
raise ValueError(f"Invalid published_after format: {v}. Must be RFC 3339 (e.g., 2023-01-01T00:00:00Z)")
return v
class SearchPlaylistsInput(BaseModel):
query: str = Field(..., min_length=1, description="The search query (required)")
max_results: Optional[int] = Field(25, ge=1, le=50, description="Maximum number of results (1 to 50)")
published_after: Optional[str] = Field(None, description="RFC 3339 timestamp (e.g., 2023-01-01T00:00:00Z)")
@field_validator("published_after")
@classmethod
def validate_published_after(cls, v):
if v and not re.match(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$', v):
raise ValueError(f"Invalid published_after format: {v}. Must be RFC 3339 (e.g., 2023-01-01T00:00:00Z)")
return v
class VideoIdInput(BaseModel):
video_id: str = Field(..., min_length=1, description="The YouTube video ID (required)")
class ChannelIdInput(BaseModel):
channel_id: str = Field(..., min_length=1, description="The YouTube channel ID (required)")
class PlaylistIdInput(BaseModel):
playlist_id: str = Field(..., min_length=1, description="The YouTube playlist ID (required)")
class FetchTranscriptsInput(BaseModel):
video_id: Optional[str] = Field(None, min_length=1, description="The YouTube video ID")
video_url: Optional[str] = Field(None, description="The YouTube video URL")
language_code: Optional[str] = Field("en", description="Language code for the transcript (e.g., 'en')")
@model_validator(mode='before')
@classmethod
def check_id_or_url(cls, values):
# Handle both dict and object inputs
if hasattr(values, '__dict__'):
values = values.__dict__
video_id = values.get("video_id")
video_url = values.get("video_url")
if not video_id and not video_url:
raise ValueError("Either video_id or video_url must be provided")
if video_url:
# Extract video ID from URL
patterns = [
r"(?:v=|v\/|embed\/|youtu.be\/)([A-Za-z0-9_-]{11})",
r"watch\?v=([A-Za-z0-9_-]{11})"
]
for pattern in patterns:
match = re.search(pattern, video_url)
if match:
values["video_id"] = match.group(1)
break
else:
raise ValueError("Invalid YouTube URL: could not extract video ID")
return values
@field_validator("language_code")
@classmethod
def validate_language_code(cls, v):
if not re.match(r'^[a-z]{2}(-[A-Z]{2})?$', v):
raise ValueError(f"Invalid language code: {v}. Must be a valid ISO 639-1 code (e.g., 'en', 'en-US')")
return v
```
--------------------------------------------------------------------------------
/tools/search_videos.py:
--------------------------------------------------------------------------------
```python
from server import mcp
from mcp.types import TextContent
from typing import List
from utils.tool_utils import YouTubeAPIError, get_youtube_client
from googleapiclient.errors import HttpError
from tools.get_video_metrics import get_video_metrics
from utils.models import SearchVideosInput
@mcp.tool()
def search_videos(arguments: dict) -> List[TextContent]:
"""Search YouTube for videos based on a query and optional filters, excluding short videos, with metrics.
This function queries the YouTube Data API v3 to retrieve videos matching the provided
search criteria. Short videos (under 4 minutes) are excluded. Each result includes the
video title, channel, ID, publication date, description, thumbnail URL, view count,
like count, and comment count, formatted as a TextContent object. Metrics are fetched
using the get_video_metrics tool.
Args:
arguments: A dictionary containing search parameters:
- query (str): The search query (required).
- max_results (int, optional): Maximum number of results (1 to 50). Defaults to 25.
- order (str, optional): Sort order ('relevance', 'date', 'rating', 'viewCount'). Defaults to 'relevance'.
- duration (str, optional): Video duration filter ('medium', 'long'). Defaults to 'medium'.
- published_after (str, optional): RFC 3339 timestamp (e.g., '2023-01-01T00:00:00Z') to filter videos uploaded after this date.
Returns:
List[TextContent]: A list of TextContent objects, each containing a formatted string
with video details (title, channel, video ID, publication date, truncated description,
view count, like count, and comment count). If no videos are found, returns a single
TextContent with a "No videos found" message.
Raises:
YouTubeAPIError: If the API key is missing, the API request fails, the input arguments are invalid (via Pydantic), or an unexpected error occurs.
"""
try:
input_data = SearchVideosInput(**arguments)
except ValueError as e:
raise YouTubeAPIError(f"Invalid input arguments: {e}")
youtube = get_youtube_client()
try:
search_params = {
'part': 'snippet',
'q': input_data.query,
'type': 'video',
'maxResults': input_data.max_results,
'order': input_data.order,
'videoDuration': input_data.duration
}
if input_data.published_after:
search_params['publishedAfter'] = input_data.published_after
search_response = youtube.search().list(**search_params).execute()
results = []
for item in search_response.get('items', []):
video_id = item['id']['videoId']
description = item['snippet']['description']
truncated_desc = description[:200] + ('...' if description else '')
video_info = {
'video_id': video_id,
'title': item['snippet']['title'],
'description': truncated_desc,
'channel_title': item['snippet']['channelTitle'],
'published_at': item['snippet']['publishedAt'],
'thumbnail_url': item['snippet']['thumbnails'].get('default', {}).get('url', '')
}
# Fetch metrics using get_video_metrics
metrics_response = get_video_metrics({"video_id": video_id})
if metrics_response[0].text.startswith("No video found"):
continue
metrics_text = metrics_response[0].text.split('\n')
view_count = next((line.split(': ')[1] for line in metrics_text if line.startswith('Views')), '0')
like_count = next((line.split(': ')[1] for line in metrics_text if line.startswith('Likes')), '0')
comment_count = next((line.split(': ')[1] for line in metrics_text if line.startswith('Comments')), '0')
results.append(TextContent(
type="text",
text=(f"**{video_info['title']}**\n"
f"Channel: {video_info['channel_title']}\n"
f"Video ID: {video_info['video_id']}\n"
f"Published: {video_info['published_at']}\n"
f"Views: {view_count}\n"
f"Likes: {like_count}\n"
f"Comments: {comment_count}\n"
f"Description: {video_info['description']}")
))
return results if results else [TextContent(type="text", text="No videos found.")]
except HttpError as e:
raise YouTubeAPIError(f"YouTube API error for query '{input_data.query}': {e}")
except Exception as e:
raise YouTubeAPIError(f"Unexpected error for query '{input_data.query}': {e}")
```