"""Search YouTube for videos matching a query (no API key required)."""
import jsonutil as json
import logging
from youtube_search import YoutubeSearch
from tools.alter_privileges import has_privilege, PRIVILEGES
logger = logging.getLogger(__name__)
TOOL_NAME = "search_youtube_videos"
TOOL_DESCRIPTION = (
"Search YouTube for videos matching a query. Returns titles, "
"URLs, durations, channels, and thumbnails without requiring "
"an API key."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search term to look for on YouTube.",
},
"limit": {
"type": "integer",
"description": "Maximum number of results (default 5, max 20).",
},
},
"required": ["query"],
}
# ---------------------------------------------------------------------------
# Authorization
# ---------------------------------------------------------------------------
async def _check_web_search_access(ctx) -> str | None:
"""Gate the tool behind the WEB_SEARCH privilege, returning an error or None.
Pulls ``user_id``, ``redis``, and ``config`` off the tool context and
delegates to :func:`tools.alter_privileges.has_privilege` to check whether
the caller holds ``PRIVILEGES["WEB_SEARCH"]``. This keeps anonymous or
unprivileged users from issuing outbound YouTube searches; the actual
privilege lookup reads grant state from Redis.
It is called once at the top of :func:`run` before any query work, so a
missing privilege short-circuits with a ready-to-return JSON error.
Args:
ctx: Tool execution context exposing ``user_id``, ``redis``, and
``config``.
Returns:
str | None: A JSON error string (with ``success: false``) when the user
lacks the privilege, or ``None`` when access is granted.
"""
user_id = getattr(ctx, "user_id", "") or ""
redis = getattr(ctx, "redis", None)
config = getattr(ctx, "config", None)
if not await has_privilege(redis, user_id, PRIVILEGES["WEB_SEARCH"], config):
return json.dumps(
{
"success": False,
"error": "The user does not have the WEB_SEARCH privilege. Ask an admin to grant it.",
}
)
return None
[docs]
async def run(query: str, limit: int = 5, ctx=None) -> str:
"""Execute this tool and return the result.
Args:
query (str): Search query or input string.
limit (int): Maximum number of items.
ctx: Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
auth_err = await _check_web_search_access(ctx)
if auth_err:
return auth_err
if not query or not query.strip():
return json.dumps(
{"error": "Missing required argument: query cannot be empty."}
)
limit = max(1, min(limit, 20))
logger.info(f"search_youtube_videos: query='{query}', limit={limit}")
try:
results = YoutubeSearch(query, max_results=limit).to_dict()
if not results:
return json.dumps(
{
"success": True,
"query": query,
"results": [],
"message": "No videos found for this query.",
}
)
videos = []
for video in results:
video_id = video.get("id", "")
videos.append(
{
"title": video.get("title", "Unknown"),
"id": video_id,
"url": (
f"https://www.youtube.com/watch?v={video_id}"
if video_id
else video.get("url_suffix", "")
),
"duration": video.get("duration", "Unknown"),
"channel": video.get("channel", "Unknown"),
"views": video.get("views", "Unknown"),
"published": video.get("publish_time", "Unknown"),
"thumbnail": (
f"https://i.ytimg.com/vi/{video_id}/hqdefault.jpg"
if video_id
else ""
),
}
)
logger.info(f"Found {len(videos)} videos for query: '{query}'")
return json.dumps(
{
"success": True,
"query": query,
"count": len(videos),
"results": videos,
},
indent=2,
)
except Exception as e:
logger.error(f"Error searching YouTube for '{query}': {e}")
return json.dumps({"error": f"Failed to search YouTube: {str(e)}"})