# 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:
--------------------------------------------------------------------------------
```
1 | # Byte-compiled / optimized / DLL files
2 | __pycache__/
3 | *.py[codz]
4 | *$py.class
5 |
6 | # C extensions
7 | *.so
8 |
9 | # Distribution / packaging
10 | .Python
11 | build/
12 | develop-eggs/
13 | dist/
14 | downloads/
15 | eggs/
16 | .eggs/
17 | lib/
18 | lib64/
19 | parts/
20 | sdist/
21 | var/
22 | wheels/
23 | share/python-wheels/
24 | *.egg-info/
25 | .installed.cfg
26 | *.egg
27 | MANIFEST
28 |
29 | # PyInstaller
30 | # Usually these files are written by a python script from a template
31 | # before PyInstaller builds the exe, so as to inject date/other infos into it.
32 | *.manifest
33 | *.spec
34 |
35 | # Installer logs
36 | pip-log.txt
37 | pip-delete-this-directory.txt
38 |
39 | # Unit test / coverage reports
40 | htmlcov/
41 | .tox/
42 | .nox/
43 | .coverage
44 | .coverage.*
45 | .cache
46 | nosetests.xml
47 | coverage.xml
48 | *.cover
49 | *.py.cover
50 | .hypothesis/
51 | .pytest_cache/
52 | cover/
53 |
54 | # Translations
55 | *.mo
56 | *.pot
57 |
58 | # Django stuff:
59 | *.log
60 | local_settings.py
61 | db.sqlite3
62 | db.sqlite3-journal
63 |
64 | # Flask stuff:
65 | instance/
66 | .webassets-cache
67 |
68 | # Scrapy stuff:
69 | .scrapy
70 |
71 | # Sphinx documentation
72 | docs/_build/
73 |
74 | # PyBuilder
75 | .pybuilder/
76 | target/
77 |
78 | # Jupyter Notebook
79 | .ipynb_checkpoints
80 |
81 | # IPython
82 | profile_default/
83 | ipython_config.py
84 |
85 | # pyenv
86 | # For a library or package, you might want to ignore these files since the code is
87 | # intended to run in multiple environments; otherwise, check them in:
88 | .python-version
89 |
90 | # pipenv
91 | # According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
92 | # However, in case of collaboration, if having platform-specific dependencies or dependencies
93 | # having no cross-platform support, pipenv may install dependencies that don't work, or not
94 | # install all needed dependencies.
95 | #Pipfile.lock
96 |
97 | # UV
98 | # Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
99 | # This is especially recommended for binary packages to ensure reproducibility, and is more
100 | # commonly ignored for libraries.
101 | uv.lock
102 |
103 | # poetry
104 | # Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
105 | # This is especially recommended for binary packages to ensure reproducibility, and is more
106 | # commonly ignored for libraries.
107 | # https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
108 | #poetry.lock
109 | #poetry.toml
110 |
111 | # pdm
112 | # Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
113 | # pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python.
114 | # https://pdm-project.org/en/latest/usage/project/#working-with-version-control
115 | #pdm.lock
116 | #pdm.toml
117 | .pdm-python
118 | .pdm-build/
119 |
120 | # pixi
121 | # Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control.
122 | #pixi.lock
123 | # Pixi creates a virtual environment in the .pixi directory, just like venv module creates one
124 | # in the .venv directory. It is recommended not to include this directory in version control.
125 | .pixi
126 |
127 | # PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
128 | __pypackages__/
129 |
130 | # Celery stuff
131 | celerybeat-schedule
132 | celerybeat.pid
133 |
134 | # SageMath parsed files
135 | *.sage.py
136 |
137 | # Environments
138 | .env
139 | .envrc
140 | .venv
141 | env/
142 | venv/
143 | ENV/
144 | env.bak/
145 | venv.bak/
146 |
147 | # Spyder project settings
148 | .spyderproject
149 | .spyproject
150 |
151 | # Rope project settings
152 | .ropeproject
153 |
154 | # mkdocs documentation
155 | /site
156 |
157 | # mypy
158 | .mypy_cache/
159 | .dmypy.json
160 | dmypy.json
161 |
162 | # Pyre type checker
163 | .pyre/
164 |
165 | # pytype static type analyzer
166 | .pytype/
167 |
168 | # Cython debug symbols
169 | cython_debug/
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # YouTube Content Management MCP Server
2 |
3 | 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.
4 |
5 | ## Features
6 |
7 | ### Current Tools
8 |
9 | - **🎥 search_videos**: Search YouTube for videos with advanced filtering options, including view count, like count, and comment count.
10 | - **📺 search_channels**: Find YouTube channels based on search queries, including subscriber count, video count, and total view count.
11 | - **📋 search_playlists**: Search YouTube for playlists based on search queries.
12 | - **📊 get_video_metrics**: Retrieve statistics (views, likes, comments) for a specific video by ID.
13 | - **📈 get_channel_metrics**: Retrieve statistics (subscribers, total views, video count) for a specific channel by ID.
14 | - **📑 get_playlist_metrics**: Retrieve statistics (item count, total views) for a specific playlist by ID.
15 |
16 | ### Planned Features
17 |
18 | - Playlist creation and management
19 | - Comment retrieval and analysis
20 | - Video upload and management (with proper authentication)
21 | - Trending videos by region
22 | - Video transcription access
23 |
24 | ## Prerequisites
25 |
26 | - Python 3.8 or higher
27 | - YouTube Data API v3 key
28 | - VSCode with MCP extension (for VSCode usage)
29 | - Required Python packages: `google-api-python-client`, `python-dotenv`, `pydantic`
30 |
31 | ## Getting Your YouTube API Key
32 |
33 | 1. Go to the [Google Cloud Console](https://console.cloud.google.com/)
34 | 2. Create a new project or select an existing one
35 | 3. Enable the YouTube Data API v3:
36 | - Navigate to "APIs & Services" > "Library"
37 | - Search for "YouTube Data API v3"
38 | - Click on it and press "Enable"
39 | 4. Create credentials:
40 | - Go to "APIs & Services" > "Credentials"
41 | - Click "Create Credentials" > "API Key"
42 | - Copy the generated API key
43 | 5. (Recommended) Restrict the API key:
44 | - Click on the API key to edit it
45 | - Under "API restrictions", select "Restrict key"
46 | - Choose "YouTube Data API v3"
47 | - Save the changes
48 |
49 | ## Installation
50 |
51 | 1. **Clone or download this repository**
52 | ```bash
53 | git clone https://github.com/NastyRunner13/youtube-content-management-mcp
54 | cd youtube-content-management-mcp
55 | ```
56 |
57 | 2. **Install dependencies**
58 | ```bash
59 | pip install -r requirements.txt
60 | ```
61 |
62 | Or if using `uv`:
63 | ```bash
64 | uv install
65 | ```
66 |
67 | 3. **Set up your environment** (Optional)
68 | Create a `.env` file in the project root:
69 | ```env
70 | YOUTUBE_API_KEY=your_youtube_api_key_here
71 | ```
72 |
73 | ## Usage
74 |
75 | ### With VSCode (Recommended)
76 |
77 | 1. **Install the MCP extension** in VSCode
78 |
79 | 2. **Configure the MCP server** by adding this to your VSCode `settings.json`:
80 |
81 | ```json
82 | {
83 | "mcp.servers": {
84 | "youtube-content-management": {
85 | "command": "python",
86 | "args": [
87 | "/path/to/youtube-content-management-mcp/main.py"
88 | ],
89 | "env": {
90 | "YOUTUBE_API_KEY": "your_youtube_api_key_here"
91 | }
92 | }
93 | }
94 | }
95 | ```
96 |
97 | **Alternative using uv:**
98 | ```json
99 | {
100 | "mcp.servers": {
101 | "youtube-content-management": {
102 | "command": "uv",
103 | "args": [
104 | "--directory",
105 | "/path/to/youtube-content-management-mcp",
106 | "run",
107 | "main.py"
108 | ],
109 | "env": {
110 | "YOUTUBE_API_KEY": "your_youtube_api_key_here"
111 | }
112 | }
113 | }
114 | }
115 | ```
116 |
117 | 3. **Restart VSCode** or reload the window
118 |
119 | 4. **Use the tools** through the MCP panel or by asking your AI assistant
120 |
121 | ### With Claude Desktop
122 |
123 | Add this configuration to your Claude Desktop config file:
124 |
125 | **Windows:** `%APPDATA%/Claude/claude_desktop_config.json`
126 | **macOS:** `~/Library/Application Support/Claude/claude_desktop_config.json`
127 |
128 | ```json
129 | {
130 | "mcpServers": {
131 | "youtube-content-management": {
132 | "command": "python",
133 | "args": ["/path/to/youtube-content-management-mcp/main.py"],
134 | "env": {
135 | "YOUTUBE_API_KEY": "your_youtube_api_key_here"
136 | }
137 | }
138 | }
139 | }
140 | ```
141 |
142 | ### With Other MCP Clients
143 |
144 | The server implements the standard MCP protocol and should work with any compatible MCP client. Refer to your client's documentation for configuration instructions.
145 |
146 | ## Available Tools
147 |
148 | ### search_videos
149 |
150 | Search YouTube for videos with advanced filtering options, including metrics like view count, like count, and comment count.
151 |
152 | **Parameters:**
153 | - `query` (string, required): Search query
154 | - `max_results` (integer, optional): Maximum number of results (1-50, default: 25)
155 | - `order` (string, optional): Sort order - "relevance", "date", "rating", "viewCount" (default: "relevance")
156 | - `duration` (string, optional): Video duration - "medium", "long" (default: "medium")
157 | - `published_after` (string, optional): RFC 3339 timestamp (e.g., "2023-01-01T00:00:00Z")
158 |
159 | **Example usage:**
160 | ```
161 | Search for Python tutorials uploaded in the last year, sorted by view count
162 | ```
163 |
164 | ### search_channels
165 |
166 | Find YouTube channels based on search queries, including metrics like subscriber count, video count, and total view count.
167 |
168 | **Parameters:**
169 | - `query` (string, required): Search query for channels
170 | - `max_results` (integer, optional): Maximum number of results (1-50, default: 25)
171 | - `published_after` (string, optional): RFC 3339 timestamp (e.g., "2023-01-01T00:00:00Z")
172 |
173 | **Example usage:**
174 | ```
175 | Find coding tutorial channels
176 | ```
177 |
178 | ### search_playlists
179 |
180 | Search YouTube for playlists based on search queries.
181 |
182 | **Parameters:**
183 | - `query` (string, required): Search query for playlists
184 | - `max_results` (integer, optional): Maximum number of results (1-50, default: 25)
185 | - `published_after` (string, optional): RFC 3339 timestamp (e.g., "2023-01-01T00:00:00Z")
186 |
187 | **Example usage:**
188 | ```
189 | Find playlists about machine learning
190 | ```
191 |
192 | ### get_video_metrics
193 |
194 | Retrieve statistics for a specific YouTube video, including view count, like count, and comment count.
195 |
196 | **Parameters:**
197 | - `video_id` (string, required): The YouTube video ID
198 |
199 | **Example usage:**
200 | ```
201 | Get metrics for the video with ID dQw4w9WgXcQ
202 | ```
203 |
204 | ### get_channel_metrics
205 |
206 | Retrieve statistics for a specific YouTube channel, including subscriber count, total view count, and video count.
207 |
208 | **Parameters:**
209 | - `channel_id` (string, required): The YouTube channel ID
210 |
211 | **Example usage:**
212 | ```
213 | Get metrics for the channel with ID UC_x5XG1OV2P6uZZ5FSM9Ttw
214 | ```
215 |
216 | ### get_playlist_metrics
217 |
218 | Retrieve statistics for a specific YouTube playlist, including item count and total view count of all videos.
219 |
220 | **Parameters:**
221 | - `playlist_id` (string, required): The YouTube playlist ID
222 |
223 | **Example usage:**
224 | ```
225 | Get metrics for the playlist with ID PL-osiE80TeTt2d9bfVyTiXJA-UTHn6WwU
226 | ```
227 |
228 | ## Example Interactions
229 |
230 | Once the MCP server is configured, you can interact with it through your AI assistant:
231 |
232 | **Video Search with Metrics:**
233 | > "Search for machine learning tutorials from the last 6 months, sorted by view count, and show view counts"
234 |
235 | **Channel Discovery with Metrics:**
236 | > "Find top cooking channels on YouTube with their subscriber counts"
237 |
238 | **Playlist Search:**
239 | > "Show me playlists about Python programming"
240 |
241 | **Video Metrics:**
242 | > "Get the view count and like count for the video with ID dQw4w9WgXcQ"
243 |
244 | **Channel Metrics:**
245 | > "What are the subscriber count and total views for the channel UC_x5XG1OV2P6uZZ5FSM9Ttw?"
246 |
247 | **Playlist Metrics:**
248 | > "How many videos and total views are in the playlist PL-osiE80TeTt2d9bfVyTiXJA-UTHn6WwU?"
249 |
250 | ## Input Validation
251 |
252 | All tools use [Pydantic](https://pydantic-docs.helpmanual.io/) for robust input validation, ensuring:
253 | - Required fields (e.g., `query`, `video_id`) are provided and non-empty.
254 | - Numeric fields (e.g., `max_results`) are within valid ranges (1-50).
255 | - String fields (e.g., `order`, `duration`) match allowed values.
256 | - Timestamps (e.g., `published_after`) follow RFC 3339 format.
257 |
258 | Invalid inputs result in clear error messages, improving reliability and user experience.
259 |
260 | ## Security Notes
261 |
262 | - **Never commit your API key** to version control
263 | - Consider using environment variables instead of hardcoding API keys
264 | - Regularly rotate your API keys
265 | - Monitor your API usage in Google Cloud Console
266 | - Set up API key restrictions to limit usage to YouTube Data API v3
267 |
268 | ## Troubleshooting
269 |
270 | ### Common Issues
271 |
272 | 1. **"YouTube API key is not set"**
273 | - Ensure your API key is properly configured in the environment variables
274 | - Check that the key is valid and has YouTube Data API v3 enabled
275 |
276 | 2. **"quotaExceeded" errors**
277 | - You've hit your daily API quota limit (default: 10,000 units)
278 | - Wait until the quota resets (daily) or increase your quota in Google Cloud Console
279 | - Note: Metrics tools and search tools with metrics may consume more quota due to multiple API calls
280 |
281 | 3. **"keyInvalid" errors**
282 | - Your API key is invalid or has been revoked
283 | - Generate a new API key and update your configuration
284 |
285 | 4. **"Invalid input arguments" errors**
286 | - Check the Pydantic error message for details (e.g., missing `query`, invalid `order`)
287 | - Ensure inputs match the tool's parameter requirements
288 |
289 | 5. **MCP server not starting**
290 | - Check that all dependencies (`google-api-python-client`, `python-dotenv`, `pydantic`) are installed
291 | - Verify the Python path in your configuration is correct
292 | - Check the MCP extension logs for detailed error messages
293 |
294 | ### Debug Mode
295 |
296 | To enable debug logging, add this to your environment:
297 | ```json
298 | "env": {
299 | "YOUTUBE_API_KEY": "your_key_here",
300 | "DEBUG": "true"
301 | }
302 | ```
303 |
304 | ## Contributing
305 |
306 | We welcome contributions! Areas where you can help:
307 | - Additional YouTube API endpoints (comments, transcriptions)
308 | - Optimizing API quota usage (e.g., batching metrics calls)
309 | - Enhancing Pydantic validation rules
310 | - Performance optimizations
311 | - Documentation improvements
312 | - Testing and bug reports
313 |
314 | ## API Limits
315 |
316 | - **YouTube Data API v3**: 10,000 units per day (default)
317 | - **Search operations**: 100 units per request
318 | - **List operations (videos, channels, playlists)**: 1 unit per request
319 | - **Playlist items**: 5 units per request
320 | - **Rate limiting**: Be mindful of making too many requests in quick succession, especially with metrics tools
321 |
322 | ## Support
323 |
324 | - Create an issue for bugs or feature requests
325 | - Check the [YouTube Data API documentation](https://developers.google.com/youtube/v3) for API-specific questions
326 | - Review MCP protocol documentation for integration issues
327 | - Refer to [Pydantic documentation](https://pydantic-docs.helpmanual.io/) for validation-related questions
```
--------------------------------------------------------------------------------
/notebooks/__init__.py:
--------------------------------------------------------------------------------
```python
1 |
```
--------------------------------------------------------------------------------
/utils/__init__.py:
--------------------------------------------------------------------------------
```python
1 |
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
1 | mcp[cli]
2 | python-dotenv
3 | google-api-python-client
4 | ipykernel
5 | pydantic
6 | youtube_transcript_api
```
--------------------------------------------------------------------------------
/server.py:
--------------------------------------------------------------------------------
```python
1 | from mcp.server.fastmcp import FastMCP
2 |
3 | # This is the shared MCP server instance
4 | mcp = FastMCP("Youtube Content Management MCP")
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
1 | [project]
2 | name = "youtube-content-management-mcp"
3 | version = "0.1.0"
4 | description = "This is the MCP server for Youtube Content Management"
5 | readme = "README.md"
6 | requires-python = ">=3.12"
7 | dependencies = []
8 |
```
--------------------------------------------------------------------------------
/tools/__init__.py:
--------------------------------------------------------------------------------
```python
1 | # This file marks the 'tools' directory as a Python package.
2 | from tools.search_videos import search_videos
3 | from tools.search_channels import search_channels
4 | from tools.search_playlists import search_playlists
5 | from tools.get_video_metrics import get_video_metrics
6 | from tools.get_channel_metrics import get_channel_metrics
7 | from tools.get_playlist_metrics import get_playlist_metrics
8 | from tools.get_video_metrics import get_video_metrics
9 | from tools.fetch_transcripts import fetch_transcripts
10 |
11 | __all__ = [
12 | "search_videos",
13 | "search_channels",
14 | "search_playlists",
15 | "get_video_metrics",
16 | "get_channel_metrics",
17 | "get_playlist_metrics",
18 | "fetch_transcripts"
19 | ]
```
--------------------------------------------------------------------------------
/main.py:
--------------------------------------------------------------------------------
```python
1 | from server import mcp
2 | import tools.search_videos # Import the search_videos tool to register it with the MCP server
3 | import tools.search_channels # Import the search_channels tool to register it with the MCP server
4 | import tools.search_playlists # Import the search_playlists tool to register it with the MCP server
5 | import tools.get_video_metrics # Import the get_video_metrics tool to register it with the MCP server
6 | import tools.get_channel_metrics # Import the get_channel_metrics tool to register it with the MCP
7 | import tools.get_playlist_metrics # Import the get_playlist_metrics tool to register it with the MCP server
8 | import tools.fetch_transcripts # Import the fetch_transcripts tool to register it with the MCP server
9 |
10 | # Entry point to run the server
11 | if __name__ == "__main__":
12 | mcp.run()
```
--------------------------------------------------------------------------------
/utils/tool_utils.py:
--------------------------------------------------------------------------------
```python
1 | import re
2 | from googleapiclient.discovery import build
3 | from googleapiclient.errors import HttpError
4 | from dotenv import load_dotenv
5 | import os
6 |
7 | load_dotenv()
8 |
9 | class YouTubeAPIError(Exception):
10 | """Custom exception for YouTube API errors"""
11 | pass
12 |
13 | def validate_youtube_params(order: str, duration: str, published_after: str | None) -> None:
14 | """Validate YouTube API parameters"""
15 | valid_orders = {"relevance", "date", "rating", "viewCount"}
16 | valid_durations = {"medium", "long"}
17 |
18 | if order not in valid_orders:
19 | raise ValueError(f"Invalid order parameter: {order}. Must be one of {valid_orders}")
20 | if duration not in valid_durations:
21 | raise ValueError(f"Invalid duration parameter: {duration}. Must be one of {valid_durations}")
22 | if published_after and not re.match(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$', published_after):
23 | raise ValueError(f"Invalid published_after format: {published_after}. Must be RFC 3339 (e.g., 2023-01-01T00:00:00Z)")
24 |
25 | def get_youtube_client():
26 | """Get a singleton YouTube API client instance."""
27 | api_key = os.getenv('YOUTUBE_API_KEY')
28 | if not api_key:
29 | raise YouTubeAPIError("YouTube API key is not set in environment variables")
30 |
31 | # Cache the client instance (module-level singleton)
32 | if not hasattr(get_youtube_client, 'client'):
33 | get_youtube_client.client = build('youtube', 'v3', developerKey=api_key)
34 |
35 | return get_youtube_client.client
```
--------------------------------------------------------------------------------
/tools/get_video_metrics.py:
--------------------------------------------------------------------------------
```python
1 | from server import mcp
2 | from mcp.types import TextContent
3 | from typing import List
4 | from utils.tool_utils import YouTubeAPIError, get_youtube_client
5 | from googleapiclient.errors import HttpError
6 | from utils.models import VideoIdInput
7 |
8 | @mcp.tool()
9 | def get_video_metrics(arguments: dict) -> List[TextContent]:
10 | """Retrieve statistics for a specific YouTube video.
11 |
12 | This function queries the YouTube Data API v3 to fetch metrics for a given video,
13 | including view count, like count, and comment count, formatted as a TextContent object.
14 |
15 | Args:
16 | arguments: A dictionary containing:
17 | - video_id (str): The YouTube video ID (required).
18 |
19 | Returns:
20 | List[TextContent]: A list containing a single TextContent object with a formatted string
21 | including the video's title, view count, like count, and comment count. If the video
22 | is not found, returns a single TextContent with a "No video found" message.
23 |
24 | Raises:
25 | YouTubeAPIError: If the API key is missing, the video ID is invalid, the API request fails,
26 | or the input arguments are invalid (via Pydantic).
27 | """
28 | try:
29 | input_data = VideoIdInput(**arguments)
30 | except ValueError as e:
31 | raise YouTubeAPIError(f"Invalid input arguments: {e}")
32 |
33 | youtube = get_youtube_client()
34 |
35 | try:
36 | response = youtube.videos().list(
37 | part='snippet,statistics',
38 | id=input_data.video_id
39 | ).execute()
40 |
41 | items = response.get('items', [])
42 | if not items:
43 | return [TextContent(type="text", text="No video found for the given ID.")]
44 |
45 | item = items[0]
46 | video_info = {
47 | 'title': item['snippet']['title'],
48 | 'view_count': item['statistics'].get('viewCount', '0'),
49 | 'like_count': item['statistics'].get('likeCount', '0'),
50 | 'comment_count': item['statistics'].get('commentCount', '0')
51 | }
52 |
53 | return [TextContent(
54 | type="text",
55 | text=(f"**{video_info['title']}**\n"
56 | f"Views: {video_info['view_count']}\n"
57 | f"Likes: {video_info['like_count']}\n"
58 | f"Comments: {video_info['comment_count']}")
59 | )]
60 |
61 | except HttpError as e:
62 | raise YouTubeAPIError(f"YouTube API error for video ID '{input_data.video_id}': {e}")
63 | except Exception as e:
64 | raise YouTubeAPIError(f"Unexpected error for video ID '{input_data.video_id}': {e}")
```
--------------------------------------------------------------------------------
/tools/get_channel_metrics.py:
--------------------------------------------------------------------------------
```python
1 | from server import mcp
2 | from mcp.types import TextContent
3 | from typing import List
4 | from utils.tool_utils import YouTubeAPIError, get_youtube_client
5 | from googleapiclient.errors import HttpError
6 | from utils.models import ChannelIdInput
7 |
8 | @mcp.tool()
9 | def get_channel_metrics(arguments: dict) -> List[TextContent]:
10 | """Retrieve statistics for a specific YouTube channel.
11 |
12 | This function queries the YouTube Data API v3 to fetch metrics for a given channel,
13 | including subscriber count, total view count, and video count, formatted as a TextContent object.
14 |
15 | Args:
16 | arguments: A dictionary containing:
17 | - channel_id (str): The YouTube channel ID (required).
18 |
19 | Returns:
20 | List[TextContent]: A list containing a single TextContent object with a formatted string
21 | including the channel's title, subscriber count, total view count, and video count.
22 | If the channel is not found, returns a single TextContent with a "No channel found" message.
23 |
24 | Raises:
25 | YouTubeAPIError: If the API key is missing, the channel ID is invalid, the API request fails,
26 | or the input arguments are invalid (via Pydantic).
27 | """
28 | try:
29 | input_data = ChannelIdInput(**arguments)
30 | except ValueError as e:
31 | raise YouTubeAPIError(f"Invalid input arguments: {e}")
32 |
33 | youtube = get_youtube_client()
34 |
35 | try:
36 | response = youtube.channels().list(
37 | part='snippet,statistics',
38 | id=input_data.channel_id
39 | ).execute()
40 |
41 | items = response.get('items', [])
42 | if not items:
43 | return [TextContent(type="text", text="No channel found for the given ID.")]
44 |
45 | item = items[0]
46 | channel_info = {
47 | 'title': item['snippet']['title'],
48 | 'subscriber_count': item['statistics'].get('subscriberCount', '0'),
49 | 'view_count': item['statistics'].get('viewCount', '0'),
50 | 'video_count': item['statistics'].get('videoCount', '0')
51 | }
52 |
53 | return [TextContent(
54 | type="text",
55 | text=(f"**{channel_info['title']}**\n"
56 | f"Subscribers: {channel_info['subscriber_count']}\n"
57 | f"Total Views: {channel_info['view_count']}\n"
58 | f"Videos: {channel_info['video_count']}")
59 | )]
60 |
61 | except HttpError as e:
62 | raise YouTubeAPIError(f"YouTube API error for channel ID '{input_data.channel_id}': {e}")
63 | except Exception as e:
64 | raise YouTubeAPIError(f"Unexpected error for channel ID '{input_data.channel_id}': {e}")
```
--------------------------------------------------------------------------------
/tools/get_playlist_metrics.py:
--------------------------------------------------------------------------------
```python
1 | from server import mcp
2 | from mcp.types import TextContent
3 | from typing import List
4 | from utils.tool_utils import YouTubeAPIError, get_youtube_client
5 | from googleapiclient.errors import HttpError
6 | from utils.models import PlaylistIdInput
7 |
8 | @mcp.tool()
9 | def get_playlist_metrics(arguments: dict) -> List[TextContent]:
10 | """Retrieve statistics for a specific YouTube playlist.
11 |
12 | This function queries the YouTube Data API v3 to fetch metrics for a given playlist,
13 | including item count and total view count of all videos, formatted as a TextContent object.
14 |
15 | Args:
16 | arguments: A dictionary containing:
17 | - playlist_id (str): The YouTube playlist ID (required).
18 |
19 | Returns:
20 | List[TextContent]: A list containing a single TextContent object with a formatted string
21 | including the playlist's title, item count, and total view count. If the playlist
22 | is not found, returns a single TextContent with a "No playlist found" message.
23 |
24 | Raises:
25 | YouTubeAPIError: If the API key is missing, the playlist ID is invalid, the API request fails,
26 | or the input arguments are invalid (via Pydantic).
27 | """
28 | try:
29 | input_data = PlaylistIdInput(**arguments)
30 | except ValueError as e:
31 | raise YouTubeAPIError(f"Invalid input arguments: {e}")
32 |
33 | youtube = get_youtube_client()
34 |
35 | try:
36 | playlist_response = youtube.playlists().list(
37 | part='snippet',
38 | id=input_data.playlist_id
39 | ).execute()
40 |
41 | playlist_items = playlist_response.get('items', [])
42 | if not playlist_items:
43 | return [TextContent(type="text", text="No playlist found for the given ID.")]
44 |
45 | playlist_title = playlist_items[0]['snippet']['title']
46 |
47 | playlist_items_response = youtube.playlistItems().list(
48 | part='contentDetails',
49 | playlistId=input_data.playlist_id,
50 | maxResults=50
51 | ).execute()
52 |
53 | video_ids = [item['contentDetails']['videoId'] for item in playlist_items_response.get('items', [])]
54 | item_count = len(video_ids)
55 |
56 | total_views = 0
57 | if video_ids:
58 | videos_response = youtube.videos().list(
59 | part='statistics',
60 | id=','.join(video_ids)
61 | ).execute()
62 | total_views = sum(int(item['statistics'].get('viewCount', 0)) for item in videos_response.get('items', []))
63 |
64 | return [TextContent(
65 | type="text",
66 | text=(f"**{playlist_title}**\n"
67 | f"Playlist ID: {input_data.playlist_id}\n"
68 | f"Items: {item_count}\n"
69 | f"Total Views: {total_views}")
70 | )]
71 |
72 | except HttpError as e:
73 | raise YouTubeAPIError(f"YouTube API error for playlist ID '{input_data.playlist_id}': {e}")
74 | except Exception as e:
75 | raise YouTubeAPIError(f"Unexpected error for playlist ID '{input_data.playlist_id}': {e}")
```
--------------------------------------------------------------------------------
/tools/search_playlists.py:
--------------------------------------------------------------------------------
```python
1 | from server import mcp
2 | from mcp.types import TextContent
3 | from typing import List
4 | from utils.tool_utils import YouTubeAPIError, get_youtube_client
5 | from googleapiclient.errors import HttpError
6 | from utils.models import SearchPlaylistsInput
7 |
8 | @mcp.tool()
9 | def search_playlists(arguments: dict) -> List[TextContent]:
10 | """Search YouTube for playlists based on a query.
11 |
12 | This function queries the YouTube Data API v3 to retrieve playlists matching the provided
13 | search query. Each result includes the playlist title, ID, creation date, and description,
14 | formatted as a single TextContent object.
15 |
16 | Args:
17 | arguments: A dictionary containing search parameters:
18 | - query (str): The search query (required).
19 | - max_results (int, optional): Maximum number of results (1 to 50). Defaults to 25.
20 | - published_after (str, optional): RFC 3339 timestamp (e.g., '2023-01-01T00:00:00Z') to filter playlists created after this date.
21 |
22 | Returns:
23 | List[TextContent]: A list containing a single TextContent object with a formatted string
24 | listing all found playlists, including their title, playlist ID, creation date, and
25 | truncated description. If no playlists are found, returns a single TextContent with a
26 | "No playlists found" message.
27 |
28 | Raises:
29 | YouTubeAPIError: If the API key is missing, the API request fails, or the input arguments are invalid (via Pydantic).
30 | """
31 | try:
32 | input_data = SearchPlaylistsInput(**arguments)
33 | except ValueError as e:
34 | raise YouTubeAPIError(f"Invalid input arguments: {e}")
35 |
36 | youtube = get_youtube_client()
37 |
38 | try:
39 | search_params = {
40 | 'part': 'snippet',
41 | 'q': input_data.query,
42 | 'type': 'playlist',
43 | 'maxResults': input_data.max_results
44 | }
45 |
46 | if input_data.published_after:
47 | search_params['publishedAfter'] = input_data.published_after
48 |
49 | search_response = youtube.search().list(**search_params).execute()
50 |
51 | playlists = []
52 | for item in search_response.get('items', []):
53 | playlist_info = {
54 | 'playlist_id': item['id']['playlistId'],
55 | 'title': item['snippet']['title'],
56 | 'description': item['snippet']['description'][:200] + ('...' if item['snippet']['description'] else ''),
57 | 'published_at': item['snippet']['publishedAt']
58 | }
59 | playlists.append(playlist_info)
60 |
61 | if not playlists:
62 | return [TextContent(type="text", text="No playlists found.")]
63 |
64 | return [TextContent(
65 | type="text",
66 | text=f"Found {len(playlists)} playlists:\n\n" +
67 | "\n\n".join([f"**{p['title']}**\n"
68 | f"Playlist ID: {p['playlist_id']}\n"
69 | f"Created: {p['published_at']}\n"
70 | f"Description: {p['description']}"
71 | for p in playlists])
72 | )]
73 |
74 | except HttpError as e:
75 | raise YouTubeAPIError(f"YouTube API error for query '{input_data.query}': {e}")
76 | except Exception as e:
77 | raise YouTubeAPIError(f"Unexpected error for query '{input_data.query}': {e}")
```
--------------------------------------------------------------------------------
/tools/fetch_transcripts.py:
--------------------------------------------------------------------------------
```python
1 | from server import mcp
2 | from mcp.types import TextContent
3 | from typing import List
4 | from utils.tool_utils import YouTubeAPIError
5 | from youtube_transcript_api import YouTubeTranscriptApi, NoTranscriptFound, TranscriptsDisabled
6 |
7 | @mcp.tool()
8 | def fetch_transcripts(arguments: dict) -> List[TextContent]:
9 | """Retrieve and analyze YouTube video transcripts for detailed information extraction.
10 |
11 | This function fetches the complete transcript of a YouTube video, making it ideal for:
12 | - Analyzing video content in detail when users ask about specific topics covered in videos
13 | - Extracting key information, quotes, or explanations from educational content
14 | - Summarizing long-form video content (lectures, tutorials, presentations, interviews)
15 | - Finding specific details or timestamps within video discussions
16 | - Understanding the full context of video content for comprehensive responses
17 |
18 | Use this tool when users:
19 | - Ask detailed questions about YouTube video content
20 | - Request summaries or key points from video material
21 | - Need specific information or quotes from video discussions
22 | - Want to understand complex topics explained in video format
23 | - Ask about timestamps or specific moments in videos
24 |
25 | Technical Implementation:
26 | This function queries the YouTube Data API v3 to fetch available captions for a video
27 | and retrieves the transcript in the specified language (default: English). The transcript
28 | is returned as a single TextContent object with timestamped text entries. If no transcript
29 | is available, a message indicating the reason is returned.
30 |
31 | Args:
32 | arguments: A dictionary containing:
33 | - video_id (str, optional): The YouTube video ID.
34 | - video_url (str, optional): The YouTube video URL (e.g., 'https://www.youtube.com/watch?v=VIDEO_ID').
35 | - language_code (str, optional): Language code for the transcript (e.g., 'en'). Defaults to 'en'.
36 | Either video_id or video_url must be provided.
37 |
38 | Returns:
39 | List[TextContent]: A list containing a single TextContent object with the transcript
40 | as a formatted string (timestamp and text). If no transcript is available or the
41 | video is not found, returns a TextContent with an appropriate message.
42 |
43 | Raises:
44 | YouTubeAPIError: If the API key is missing, the API request fails, the input arguments
45 | are invalid (via Pydantic), or an unexpected error occurs.
46 | """
47 | from utils.models import FetchTranscriptsInput # Moved import here to avoid circular import issues
48 |
49 | try:
50 | input_data = FetchTranscriptsInput(**arguments)
51 | except ValueError as e:
52 | raise YouTubeAPIError(f"Invalid input arguments: {e}")
53 |
54 | ytt_api = YouTubeTranscriptApi()
55 |
56 | try:
57 | # Try to fetch transcript directly first
58 | try:
59 | transcript = ytt_api.fetch(
60 | video_id=input_data.video_id,
61 | languages=[input_data.language_code]
62 | )
63 | except NoTranscriptFound:
64 | # If specific language not found, try to get any available transcript
65 | try:
66 | transcript = ytt_api.fetch(video_id=input_data.video_id)
67 | except NoTranscriptFound:
68 | return [TextContent(type="text", text=f"No transcript available for this video in any language.")]
69 |
70 | transcript_text = "".join(snippet.text for snippet in transcript.snippets)
71 |
72 | if not transcript_text.strip():
73 | return [TextContent(type="text", text="Transcript is empty or unavailable.")]
74 |
75 | return [TextContent(
76 | type="text",
77 | text=f"Transcript for video ID {input_data.video_id} (language: {input_data.language_code}):\n\n{transcript_text}"
78 | )]
79 |
80 | except TranscriptsDisabled:
81 | return [TextContent(type="text", text="Transcripts are disabled for this video or access is restricted.")]
82 | except NoTranscriptFound:
83 | return [TextContent(type="text", text=f"No transcript available for this video.")]
84 | except Exception as e:
85 | raise YouTubeAPIError(f"Unexpected error for video ID '{input_data.video_id}': {e}")
```
--------------------------------------------------------------------------------
/tools/search_channels.py:
--------------------------------------------------------------------------------
```python
1 | from server import mcp
2 | from mcp.types import TextContent
3 | from typing import List
4 | from utils.tool_utils import YouTubeAPIError, get_youtube_client
5 | from googleapiclient.errors import HttpError
6 | from tools.get_channel_metrics import get_channel_metrics
7 | from utils.models import SearchChannelsInput
8 |
9 | @mcp.tool()
10 | def search_channels(arguments: dict) -> List[TextContent]:
11 | """Search YouTube for channels based on a query, with metrics.
12 |
13 | This function queries the YouTube Data API v3 to retrieve channels matching the provided
14 | search query. Each result includes the channel title, ID, creation date, description,
15 | subscriber count, video count, and total view count, formatted as a single TextContent object.
16 | Metrics are fetched using the get_channel_metrics tool.
17 |
18 | Args:
19 | arguments: A dictionary containing search parameters:
20 | - query (str): The search query (required).
21 | - max_results (int, optional): Maximum number of results (1 to 50). Defaults to 25.
22 | - published_after (str, optional): RFC 3339 timestamp (e.g., '2023-01-01T00:00:00Z') to filter channels created after this date.
23 |
24 | Returns:
25 | List[TextContent]: A list containing a single TextContent object with a formatted string
26 | listing all found channels, including their title, channel ID, creation date,
27 | truncated description, subscriber count, video count, and total view count.
28 | If no channels are found, returns a single TextContent with a "No channels found" message.
29 |
30 | Raises:
31 | YouTubeAPIError: If the API key is missing, the API request fails, the input arguments are invalid (via Pydantic), or an unexpected error occurs.
32 | """
33 | try:
34 | input_data = SearchChannelsInput(**arguments)
35 | except ValueError as e:
36 | raise YouTubeAPIError(f"Invalid input arguments: {e}")
37 |
38 | youtube = get_youtube_client()
39 |
40 | try:
41 | search_params = {
42 | 'part': 'snippet',
43 | 'q': input_data.query,
44 | 'type': 'channel',
45 | 'maxResults': input_data.max_results
46 | }
47 |
48 | if input_data.published_after:
49 | search_params['publishedAfter'] = input_data.published_after
50 |
51 | search_response = youtube.search().list(**search_params).execute()
52 |
53 | channels = []
54 | for item in search_response.get('items', []):
55 | channel_id = item['id']['channelId']
56 | channel_info = {
57 | 'channel_id': channel_id,
58 | 'title': item['snippet']['title'],
59 | 'description': item['snippet']['description'][:200] + ('...' if item['snippet']['description'] else ''),
60 | 'published_at': item['snippet']['publishedAt']
61 | }
62 |
63 | # Fetch metrics using get_channel_metrics
64 | metrics_response = get_channel_metrics({"channel_id": channel_id})
65 | if metrics_response[0].text.startswith("No channel found"):
66 | continue
67 | metrics_text = metrics_response[0].text.split('\n')
68 | subscriber_count = next((line.split(': ')[1] for line in metrics_text if line.startswith('Subscribers')), '0')
69 | video_count = next((line.split(': ')[1] for line in metrics_text if line.startswith('Videos')), '0')
70 | view_count = next((line.split(': ')[1] for line in metrics_text if line.startswith('Total Views')), '0')
71 |
72 | channel_info.update({
73 | 'subscriber_count': subscriber_count,
74 | 'video_count': video_count,
75 | 'view_count': view_count
76 | })
77 | channels.append(channel_info)
78 |
79 | if not channels:
80 | return [TextContent(type="text", text="No channels found.")]
81 |
82 | return [TextContent(
83 | type="text",
84 | text=f"Found {len(channels)} channels:\n\n" +
85 | "\n\n".join([f"**{c['title']}**\n"
86 | f"Channel ID: {c['channel_id']}\n"
87 | f"Created: {c['published_at']}\n"
88 | f"Subscribers: {c['subscriber_count']}\n"
89 | f"Videos: {c['video_count']}\n"
90 | f"Total Views: {c['view_count']}\n"
91 | f"Description: {c['description']}"
92 | for c in channels])
93 | )]
94 |
95 | except HttpError as e:
96 | raise YouTubeAPIError(f"YouTube API error for query '{input_data.query}': {e}")
97 | except Exception as e:
98 | raise YouTubeAPIError(f"Unexpected error for query '{input_data.query}': {e}")
```
--------------------------------------------------------------------------------
/utils/models.py:
--------------------------------------------------------------------------------
```python
1 | from pydantic import BaseModel, Field, field_validator, model_validator
2 | from typing import Optional
3 | import re
4 |
5 | class SearchVideosInput(BaseModel):
6 | query: str = Field(..., min_length=1, description="The search query (required)")
7 | max_results: Optional[int] = Field(25, ge=1, le=50, description="Maximum number of results (1 to 50)")
8 | order: Optional[str] = Field("relevance", description="Sort order: relevance, date, rating, viewCount")
9 | duration: Optional[str] = Field("medium", description="Video duration: medium, long")
10 | published_after: Optional[str] = Field(None, description="RFC 3339 timestamp (e.g., 2023-01-01T00:00:00Z)")
11 |
12 | @field_validator("order")
13 | @classmethod
14 | def validate_order(cls, v):
15 | valid_orders = {"relevance", "date", "rating", "viewCount"}
16 | if v not in valid_orders:
17 | raise ValueError(f"Invalid order: {v}. Must be one of {valid_orders}")
18 | return v
19 |
20 | @field_validator("duration")
21 | @classmethod
22 | def validate_duration(cls, v):
23 | valid_durations = {"medium", "long"}
24 | if v not in valid_durations:
25 | raise ValueError(f"Invalid duration: {v}. Must be one of {valid_durations}")
26 | return v
27 |
28 | @field_validator("published_after")
29 | @classmethod
30 | def validate_published_after(cls, v):
31 | if v and not re.match(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$', v):
32 | raise ValueError(f"Invalid published_after format: {v}. Must be RFC 3339 (e.g., 2023-01-01T00:00:00Z)")
33 | return v
34 |
35 | class SearchChannelsInput(BaseModel):
36 | query: str = Field(..., min_length=1, description="The search query (required)")
37 | max_results: Optional[int] = Field(25, ge=1, le=50, description="Maximum number of results (1 to 50)")
38 | published_after: Optional[str] = Field(None, description="RFC 3339 timestamp (e.g., 2023-01-01T00:00:00Z)")
39 |
40 | @field_validator("published_after")
41 | @classmethod
42 | def validate_published_after(cls, v):
43 | if v and not re.match(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$', v):
44 | raise ValueError(f"Invalid published_after format: {v}. Must be RFC 3339 (e.g., 2023-01-01T00:00:00Z)")
45 | return v
46 |
47 | class SearchPlaylistsInput(BaseModel):
48 | query: str = Field(..., min_length=1, description="The search query (required)")
49 | max_results: Optional[int] = Field(25, ge=1, le=50, description="Maximum number of results (1 to 50)")
50 | published_after: Optional[str] = Field(None, description="RFC 3339 timestamp (e.g., 2023-01-01T00:00:00Z)")
51 |
52 | @field_validator("published_after")
53 | @classmethod
54 | def validate_published_after(cls, v):
55 | if v and not re.match(r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$', v):
56 | raise ValueError(f"Invalid published_after format: {v}. Must be RFC 3339 (e.g., 2023-01-01T00:00:00Z)")
57 | return v
58 |
59 | class VideoIdInput(BaseModel):
60 | video_id: str = Field(..., min_length=1, description="The YouTube video ID (required)")
61 |
62 | class ChannelIdInput(BaseModel):
63 | channel_id: str = Field(..., min_length=1, description="The YouTube channel ID (required)")
64 |
65 | class PlaylistIdInput(BaseModel):
66 | playlist_id: str = Field(..., min_length=1, description="The YouTube playlist ID (required)")
67 |
68 | class FetchTranscriptsInput(BaseModel):
69 | video_id: Optional[str] = Field(None, min_length=1, description="The YouTube video ID")
70 | video_url: Optional[str] = Field(None, description="The YouTube video URL")
71 | language_code: Optional[str] = Field("en", description="Language code for the transcript (e.g., 'en')")
72 |
73 | @model_validator(mode='before')
74 | @classmethod
75 | def check_id_or_url(cls, values):
76 | # Handle both dict and object inputs
77 | if hasattr(values, '__dict__'):
78 | values = values.__dict__
79 |
80 | video_id = values.get("video_id")
81 | video_url = values.get("video_url")
82 |
83 | if not video_id and not video_url:
84 | raise ValueError("Either video_id or video_url must be provided")
85 |
86 | if video_url:
87 | # Extract video ID from URL
88 | patterns = [
89 | r"(?:v=|v\/|embed\/|youtu.be\/)([A-Za-z0-9_-]{11})",
90 | r"watch\?v=([A-Za-z0-9_-]{11})"
91 | ]
92 | for pattern in patterns:
93 | match = re.search(pattern, video_url)
94 | if match:
95 | values["video_id"] = match.group(1)
96 | break
97 | else:
98 | raise ValueError("Invalid YouTube URL: could not extract video ID")
99 |
100 | return values
101 |
102 | @field_validator("language_code")
103 | @classmethod
104 | def validate_language_code(cls, v):
105 | if not re.match(r'^[a-z]{2}(-[A-Z]{2})?$', v):
106 | raise ValueError(f"Invalid language code: {v}. Must be a valid ISO 639-1 code (e.g., 'en', 'en-US')")
107 | return v
```
--------------------------------------------------------------------------------
/tools/search_videos.py:
--------------------------------------------------------------------------------
```python
1 | from server import mcp
2 | from mcp.types import TextContent
3 | from typing import List
4 | from utils.tool_utils import YouTubeAPIError, get_youtube_client
5 | from googleapiclient.errors import HttpError
6 | from tools.get_video_metrics import get_video_metrics
7 | from utils.models import SearchVideosInput
8 |
9 | @mcp.tool()
10 | def search_videos(arguments: dict) -> List[TextContent]:
11 | """Search YouTube for videos based on a query and optional filters, excluding short videos, with metrics.
12 |
13 | This function queries the YouTube Data API v3 to retrieve videos matching the provided
14 | search criteria. Short videos (under 4 minutes) are excluded. Each result includes the
15 | video title, channel, ID, publication date, description, thumbnail URL, view count,
16 | like count, and comment count, formatted as a TextContent object. Metrics are fetched
17 | using the get_video_metrics tool.
18 |
19 | Args:
20 | arguments: A dictionary containing search parameters:
21 | - query (str): The search query (required).
22 | - max_results (int, optional): Maximum number of results (1 to 50). Defaults to 25.
23 | - order (str, optional): Sort order ('relevance', 'date', 'rating', 'viewCount'). Defaults to 'relevance'.
24 | - duration (str, optional): Video duration filter ('medium', 'long'). Defaults to 'medium'.
25 | - published_after (str, optional): RFC 3339 timestamp (e.g., '2023-01-01T00:00:00Z') to filter videos uploaded after this date.
26 |
27 | Returns:
28 | List[TextContent]: A list of TextContent objects, each containing a formatted string
29 | with video details (title, channel, video ID, publication date, truncated description,
30 | view count, like count, and comment count). If no videos are found, returns a single
31 | TextContent with a "No videos found" message.
32 |
33 | Raises:
34 | YouTubeAPIError: If the API key is missing, the API request fails, the input arguments are invalid (via Pydantic), or an unexpected error occurs.
35 | """
36 | try:
37 | input_data = SearchVideosInput(**arguments)
38 | except ValueError as e:
39 | raise YouTubeAPIError(f"Invalid input arguments: {e}")
40 |
41 | youtube = get_youtube_client()
42 |
43 | try:
44 | search_params = {
45 | 'part': 'snippet',
46 | 'q': input_data.query,
47 | 'type': 'video',
48 | 'maxResults': input_data.max_results,
49 | 'order': input_data.order,
50 | 'videoDuration': input_data.duration
51 | }
52 |
53 | if input_data.published_after:
54 | search_params['publishedAfter'] = input_data.published_after
55 |
56 | search_response = youtube.search().list(**search_params).execute()
57 |
58 | results = []
59 | for item in search_response.get('items', []):
60 | video_id = item['id']['videoId']
61 | description = item['snippet']['description']
62 | truncated_desc = description[:200] + ('...' if description else '')
63 | video_info = {
64 | 'video_id': video_id,
65 | 'title': item['snippet']['title'],
66 | 'description': truncated_desc,
67 | 'channel_title': item['snippet']['channelTitle'],
68 | 'published_at': item['snippet']['publishedAt'],
69 | 'thumbnail_url': item['snippet']['thumbnails'].get('default', {}).get('url', '')
70 | }
71 |
72 | # Fetch metrics using get_video_metrics
73 | metrics_response = get_video_metrics({"video_id": video_id})
74 | if metrics_response[0].text.startswith("No video found"):
75 | continue
76 | metrics_text = metrics_response[0].text.split('\n')
77 | view_count = next((line.split(': ')[1] for line in metrics_text if line.startswith('Views')), '0')
78 | like_count = next((line.split(': ')[1] for line in metrics_text if line.startswith('Likes')), '0')
79 | comment_count = next((line.split(': ')[1] for line in metrics_text if line.startswith('Comments')), '0')
80 |
81 | results.append(TextContent(
82 | type="text",
83 | text=(f"**{video_info['title']}**\n"
84 | f"Channel: {video_info['channel_title']}\n"
85 | f"Video ID: {video_info['video_id']}\n"
86 | f"Published: {video_info['published_at']}\n"
87 | f"Views: {view_count}\n"
88 | f"Likes: {like_count}\n"
89 | f"Comments: {comment_count}\n"
90 | f"Description: {video_info['description']}")
91 | ))
92 |
93 | return results if results else [TextContent(type="text", text="No videos found.")]
94 |
95 | except HttpError as e:
96 | raise YouTubeAPIError(f"YouTube API error for query '{input_data.query}': {e}")
97 | except Exception as e:
98 | raise YouTubeAPIError(f"Unexpected error for query '{input_data.query}': {e}")
```