moritz/mop-skills
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(skills): add YouTube and Google Maps Platform MCP skills

Browse Source
This commit is contained in:
Moritz Graf
2026-09-18 11:30:35 +02:00
parent ea079c9852
commit 6e98fd83b2
4 changed files with 1239 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/google-maps/SKILL.md
+36
View File
@@ -0,0 +1,36 @@
---
name: google-maps
description: Search places, live opening hours, contact details, routing directions (drive/transit/walk/bike), and static visual maps via Google Maps Platform (Places New, Routes, and Static Maps).
---
# 🗺️ Google Maps Platform (Places, Routes & Visual Maps)
Integration with Google Maps Platform for OpenClaw, enabling local point-of-interest lookups, real-time opening hours, routing directions, and static map images:
1. 📍 **`maps_search_places(query, location="Regensburg, Germany", radius_meters=5000, open_now=False)`**:
- Searches places (bakeries, restaurants, playgrounds, doctors, supermarkets) using Google Places API (New).
- Biased around Stadtamhof / Regensburg by default.
- Returns place name, address, rating, user ratings count, live opening status (`open_now: true/false`, today's opening hours), phone number, category, and direct Google Maps link.
2. 🕒 **`maps_get_place_details(place_id)`**:
- Fetches full weekly opening hours (Monday–Sunday), website, wheelchair accessibility, and top recent reviews.
3. 🚗 **`maps_get_directions(origin, destination, travel_mode="DRIVE"|"TRANSIT"|"WALK"|"BICYCLE")`**:
- Computes route distance, travel duration, and key turn/transit steps using Google Routes API v2.
- Ideal for trip planning, estimating commute time before appointments, or finding walking routes.
4. 🖼️ **`maps_render_static_map(center_or_address, zoom=15, size="600x400")`**:
- Generates a static map PNG image saved to OpenClaw media storage.
- Outputs `MEDIA:<path>` so Telegram displays a visual location photo card directly in the chat!
---
## 🧭 Prompt Routing & Best Practices
- **When the user asks about local shops or opening hours** (*"Ist der Bäcker in Stadtamhof noch offen?"*, *"Wann schließt der Edeka?"*):
- Call `maps_search_places` with the venue name and `location="Stadtamhof, Regensburg"`.
- Report the live status (`Offen` or `Geschlossen`) and today's closing time clearly.
- **When planning outings or dining** (*"Gibt es einen guten Biergarten in der Nähe?"*, *"Finde einen Spielplatz mit Schatten in Regensburg"*):
- Search using `maps_search_places`, highlight ratings and addresses, and offer to calculate walking directions or render a map card.
- **For Google Calendar Appointments**:
- When an appointment has a location or the user asks *"Wann muss ich los zum Arzt?"*, call `maps_get_directions` from `Grabengasse 7, 93059 Regensburg` to the destination to provide accurate departure times.
skills/google-maps/scripts/google_maps_mcp.py
Executable
+598
View File
@@ -0,0 +1,598 @@
#!/usr/bin/env python3
"""
Google Maps Platform MCP Server & CLI Tool for OpenClaw.
Integrates Google Places (New), Routes API, Geocoding, and Static Maps with OpenClaw.
Features:
- maps_search_places: Search restaurants, bakeries, parks, shops with live opening hours & ratings
- maps_get_place_details: Get full weekly schedule (Mo-Su), phone, website, and reviews
- maps_get_directions: Calculate transit, walking, or driving routes with distance & duration
- maps_render_static_map: Generate static map PNG images for Telegram visual previews
"""
import os
import sys
import json
import hashlib
import argparse
import requests
from typing import Dict, Any, List, Optional
# Environment
GOOGLE_MAPS_API_KEY = os.environ.get("GOOGLE_MAPS_API_KEY", "").strip() or os.environ.get("GOOGLE_PLACES_API_KEY", "").strip()
DEFAULT_LOCATION = os.environ.get("MAPS_DEFAULT_LOCATION", "Regensburg, Germany")
MEDIA_OUTPUT_DIR = os.environ.get("MAPS_MEDIA_DIR", "/tmp/openclaw_maps")
def ensure_media_dir() -> str:
"""Ensure output directory for static map images exists."""
os.makedirs(MEDIA_OUTPUT_DIR, exist_ok=True)
return MEDIA_OUTPUT_DIR
# ==============================================================================
# Google Maps Web Services
# ==============================================================================
def geocode_address(address: str, api_key: Optional[str] = None) -> Optional[Dict[str, Any]]:
"""Geocode an address to lat/lng."""
key = api_key or GOOGLE_MAPS_API_KEY
if not key:
return None
url = "https://maps.googleapis.com/maps/api/geocode/json"
params = {"address": address, "key": key, "language": "de"}
try:
resp = requests.get(url, params=params, timeout=10)
data = resp.json()
if data.get("status") == "OK" and data.get("results"):
first = data["results"][0]
loc = first["geometry"]["location"]
return {
"lat": loc["lat"],
"lng": loc["lng"],
"formatted_address": first.get("formatted_address"),
"place_id": first.get("place_id"),
}
except Exception:
pass
return None
def search_places(
query: str,
location: Optional[str] = None,
radius_meters: int = 5000,
open_now: bool = False,
api_key: Optional[str] = None,
) -> Dict[str, Any]:
"""Search places using Google Places API (New)."""
key = api_key or GOOGLE_MAPS_API_KEY
if not key:
return {"success": False, "error": "GOOGLE_MAPS_API_KEY is not set.", "places": []}
url = "https://places.googleapis.com/v1/places:searchText"
headers = {
"Content-Type": "application/json",
"X-Goog-Api-Key": key,
"X-Goog-FieldMask": (
"places.id,places.displayName,places.formattedAddress,places.location,"
"places.rating,places.userRatingCount,places.currentOpeningHours,"
"places.nationalPhoneNumber,places.googleMapsUri,places.priceLevel,places.primaryType"
),
}
body: Dict[str, Any] = {
"textQuery": query,
"languageCode": "de",
"maxResultCount": 8,
}
if open_now:
body["openNow"] = True
loc_str = location or DEFAULT_LOCATION
geo = geocode_address(loc_str, key)
if geo:
body["locationBias"] = {
"circle": {
"center": {"latitude": geo["lat"], "longitude": geo["lng"]},
"radius": float(radius_meters),
}
}
try:
resp = requests.post(url, headers=headers, json=body, timeout=10)
if resp.status_code != 200:
return {"success": False, "error": f"Places API returned {resp.status_code}: {resp.text}", "places": []}
data = resp.json()
raw_places = data.get("places", [])
places = []
for p in raw_places:
opening = p.get("currentOpeningHours", {})
open_status = "Offen" if opening.get("openNow") else ("Geschlossen" if "openNow" in opening else "Keine Angabe")
descriptions = opening.get("weekdayDescriptions", [])
places.append({
"place_id": p.get("id"),
"name": p.get("displayName", {}).get("text", "Unbekannt"),
"address": p.get("formattedAddress", ""),
"rating": p.get("rating"),
"user_ratings_total": p.get("userRatingCount"),
"open_now": opening.get("openNow"),
"status_text": open_status,
"today_hours": descriptions[0] if descriptions else None,
"phone": p.get("nationalPhoneNumber"),
"maps_url": p.get("googleMapsUri"),
"category": p.get("primaryType"),
})
return {
"success": True,
"query": query,
"count": len(places),
"location_bias": loc_str if geo else None,
"places": places,
}
except Exception as e:
return {"success": False, "error": f"Place search failed: {str(e)}", "places": []}
def get_place_details(place_id: str, api_key: Optional[str] = None) -> Dict[str, Any]:
"""Get comprehensive place details including weekly schedule and reviews."""
key = api_key or GOOGLE_MAPS_API_KEY
if not key:
return {"success": False, "error": "GOOGLE_MAPS_API_KEY is not set."}
# Ensure place_id format (v1 API supports raw ID or places/ID)
p_id = place_id.replace("places/", "")
url = f"https://places.googleapis.com/v1/places/{p_id}"
headers = {
"Content-Type": "application/json",
"X-Goog-Api-Key": key,
"X-Goog-FieldMask": (
"id,displayName,formattedAddress,currentOpeningHours,regularOpeningHours,"
"nationalPhoneNumber,internationalPhoneNumber,websiteUri,googleMapsUri,"
"rating,userRatingCount,reviews,accessibilityOptions,priceLevel"
),
}
params = {"languageCode": "de"}
try:
resp = requests.get(url, headers=headers, params=params, timeout=10)
if resp.status_code != 200:
return {"success": False, "error": f"Places API returned {resp.status_code}: {resp.text}"}
p = resp.json()
current_hours = p.get("currentOpeningHours", {})
regular_hours = p.get("regularOpeningHours", {})
schedule = regular_hours.get("weekdayDescriptions") or current_hours.get("weekdayDescriptions") or []
reviews_summary = []
for rev in (p.get("reviews", []) or [])[:3]:
reviews_summary.append({
"author": rev.get("authorAttribution", {}).get("displayName"),
"rating": rev.get("rating"),
"text": rev.get("text", {}).get("text", "")[:300],
"publish_time": rev.get("relativePublishTimeDescription"),
})
return {
"success": True,
"place_id": p_id,
"name": p.get("displayName", {}).get("text", "Unbekannt"),
"address": p.get("formattedAddress", ""),
"rating": p.get("rating"),
"user_ratings_total": p.get("userRatingCount"),
"open_now": current_hours.get("openNow"),
"weekly_hours": schedule,
"phone": p.get("nationalPhoneNumber") or p.get("internationalPhoneNumber"),
"website": p.get("websiteUri"),
"maps_url": p.get("googleMapsUri"),
"accessibility": p.get("accessibilityOptions", {}),
"recent_reviews": reviews_summary,
}
except Exception as e:
return {"success": False, "error": f"Place details fetch failed: {str(e)}"}
def get_directions(
origin: str,
destination: str,
travel_mode: str = "DRIVE",
api_key: Optional[str] = None,
) -> Dict[str, Any]:
"""Calculate route directions, distance, and duration using Google Routes API v2."""
key = api_key or GOOGLE_MAPS_API_KEY
if not key:
return {"success": False, "error": "GOOGLE_MAPS_API_KEY is not set."}
# Normalize travel mode
mode_map = {
"drive": "DRIVE",
"driving": "DRIVE",
"transit": "TRANSIT",
"public_transport": "TRANSIT",
"walk": "WALK",
"walking": "WALK",
"bicycle": "BICYCLE",
"bicycling": "BICYCLE",
"bike": "BICYCLE",
}
norm_mode = mode_map.get(travel_mode.lower(), "DRIVE")
url = "https://routes.googleapis.com/directions/v2:computeRoutes"
headers = {
"Content-Type": "application/json",
"X-Goog-Api-Key": key,
"X-Goog-FieldMask": (
"routes.duration,routes.distanceMeters,routes.description,"
"routes.legs.duration,routes.legs.distanceMeters,routes.legs.steps"
),
}
body = {
"origin": {"address": origin},
"destination": {"address": destination},
"travelMode": norm_mode,
"languageCode": "de",
"units": "METRIC",
}
try:
resp = requests.post(url, headers=headers, json=body, timeout=12)
if resp.status_code != 200:
return {"success": False, "error": f"Routes API returned {resp.status_code}: {resp.text}"}
data = resp.json()
routes = data.get("routes", [])
if not routes:
return {
"success": False,
"origin": origin,
"destination": destination,
"error": "No route found between origin and destination.",
}
primary = routes[0]
dist_m = primary.get("distanceMeters", 0)
dur_str = primary.get("duration", "0s").rstrip("s")
dur_sec = int(dur_str) if dur_str.isdigit() else 0
dist_km = round(dist_m / 1000.0, 1)
dur_min = round(dur_sec / 60)
# Step summary
steps_summary = []
for leg in primary.get("legs", []):
for step in leg.get("steps", [])[:8]:
instr = step.get("navigationInstruction", {}).get("instructions")
dist_step = step.get("distanceMeters", 0)
if instr:
steps_summary.append(f"{instr} ({dist_step}m)")
return {
"success": True,
"origin": origin,
"destination": destination,
"travel_mode": norm_mode,
"distance_meters": dist_m,
"distance_km": dist_km,
"duration_seconds": dur_sec,
"duration_minutes": dur_min,
"duration_formatted": f"{dur_min} Min." if dur_min < 60 else f"{dur_min // 60} Std. {dur_min % 60} Min.",
"description": primary.get("description", ""),
"steps_overview": steps_summary,
}
except Exception as e:
return {"success": False, "error": f"Directions calculation failed: {str(e)}"}
def render_static_map(
center_or_address: str,
zoom: int = 15,
markers: Optional[List[str]] = None,
size: str = "600x400",
maptype: str = "roadmap",
api_key: Optional[str] = None,
) -> Dict[str, Any]:
"""Download a static map PNG image and save to local media directory."""
key = api_key or GOOGLE_MAPS_API_KEY
if not key:
return {"success": False, "error": "GOOGLE_MAPS_API_KEY is not set."}
url = "https://maps.googleapis.com/maps/api/staticmap"
params: Dict[str, Any] = {
"center": center_or_address,
"zoom": zoom,
"size": size,
"maptype": maptype,
"key": key,
"format": "png",
}
# Add markers
marker_list = markers or [center_or_address]
# Format: color:red|label:A|location
params["markers"] = f"color:red|{marker_list[0]}"
try:
resp = requests.get(url, params=params, timeout=12)
if resp.status_code != 200:
return {"success": False, "error": f"Static Maps API returned {resp.status_code}: {resp.text}"}
out_dir = ensure_media_dir()
h = hashlib.md5(f"{center_or_address}_{zoom}_{size}".encode()).hexdigest()[:10]
filename = f"map_{h}.png"
filepath = os.path.join(out_dir, filename)
with open(filepath, "wb") as f:
f.write(resp.content)
return {
"success": True,
"center": center_or_address,
"zoom": zoom,
"file_path": filepath,
"file_name": filename,
"media_uri": f"MEDIA:{filepath}",
"file_size_bytes": len(resp.content),
}
except Exception as e:
return {"success": False, "error": f"Static map generation failed: {str(e)}"}
# ==============================================================================
# MCP Protocol Handlers (JSON-RPC 2.0 stdio)
# ==============================================================================
TOOLS_DEFINITIONS = [
{
"name": "maps_search_places",
"description": "Search local places, venues, shops, restaurants, bakeries, or doctors using Google Places API (New). Returns names, addresses, live opening hours (open now?), ratings, phone numbers, and Google Maps URLs.",
"inputSchema": {
"type": "object",
"required": ["query"],
"properties": {
"query": {
"type": "string",
"description": "What to search for (e.g. 'Bäckerei Stadtamhof', 'Spielplatz mit Schatten', 'Kinderarzt Prüfening').",
},
"location": {
"type": "string",
"description": "City or neighborhood to bias search around (default 'Regensburg, Germany').",
"default": "Regensburg, Germany",
},
"radius_meters": {
"type": "integer",
"description": "Search radius in meters (default 5000).",
"default": 5000,
},
"open_now": {
"type": "boolean",
"description": "Filter strictly to places currently open right now.",
"default": False,
},
},
},
},
{
"name": "maps_get_place_details",
"description": "Get detailed schedule (opening hours Monday through Sunday), contact info, website, wheelchair accessibility, and recent reviews for a specific place ID.",
"inputSchema": {
"type": "object",
"required": ["place_id"],
"properties": {
"place_id": {
"type": "string",
"description": "The Google place ID returned from maps_search_places.",
},
},
},
},
{
"name": "maps_get_directions",
"description": "Compute route directions, travel time, and distance between two locations with traffic estimates.",
"inputSchema": {
"type": "object",
"required": ["origin", "destination"],
"properties": {
"origin": {
"type": "string",
"description": "Starting address or place name (e.g. 'Grabengasse 7, 93059 Regensburg').",
},
"destination": {
"type": "string",
"description": "Target address or place name (e.g. 'Regensburg Hauptbahnhof', 'Westbad Regensburg').",
},
"travel_mode": {
"type": "string",
"description": "Mode of transit: 'DRIVE' (auto), 'TRANSIT' (bus/train), 'WALK' (foot), 'BICYCLE' (bike). Default 'DRIVE'.",
"enum": ["DRIVE", "TRANSIT", "WALK", "BICYCLE"],
"default": "DRIVE",
},
},
},
},
{
"name": "maps_render_static_map",
"description": "Generate a static satellite or roadmap PNG image of a location or route to send directly to Telegram as a visual photo card.",
"inputSchema": {
"type": "object",
"required": ["center_or_address"],
"properties": {
"center_or_address": {
"type": "string",
"description": "Address or location to center the map on.",
},
"zoom": {
"type": "integer",
"description": "Zoom level (1-20). 15 is street/neighborhood level, 12 is city level.",
"default": 15,
},
"size": {
"type": "string",
"description": "Image dimensions in pixels (default '600x400').",
"default": "600x400",
},
},
},
},
]
def run_mcp_server():
"""Run JSON-RPC 2.0 stdio loop."""
while True:
try:
line = sys.stdin.readline()
if not line:
break
line = line.strip()
if not line:
continue
req = json.loads(line)
req_id = req.get("id")
method = req.get("method")
params = req.get("params", {})
if method == "initialize":
res = {
"jsonrpc": "2.0",
"id": req_id,
"result": {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"serverInfo": {
"name": "openclaw-google-maps",
"version": "1.0.0",
},
},
}
print(json.dumps(res), flush=True)
elif method == "notifications/initialized":
pass
elif method == "tools/list":
res = {
"jsonrpc": "2.0",
"id": req_id,
"result": {"tools": TOOLS_DEFINITIONS},
}
print(json.dumps(res), flush=True)
elif method == "tools/call":
tool_name = params.get("name")
args = params.get("arguments", {})
content = []
if tool_name == "maps_search_places":
result = search_places(
query=args.get("query", ""),
location=args.get("location"),
radius_meters=args.get("radius_meters", 5000),
open_now=args.get("open_now", False),
)
content.append({"type": "text", "text": json.dumps(result, indent=2, ensure_ascii=False)})
elif tool_name == "maps_get_place_details":
result = get_place_details(place_id=args.get("place_id", ""))
content.append({"type": "text", "text": json.dumps(result, indent=2, ensure_ascii=False)})
elif tool_name == "maps_get_directions":
result = get_directions(
origin=args.get("origin", ""),
destination=args.get("destination", ""),
travel_mode=args.get("travel_mode", "DRIVE"),
)
content.append({"type": "text", "text": json.dumps(result, indent=2, ensure_ascii=False)})
elif tool_name == "maps_render_static_map":
result = render_static_map(
center_or_address=args.get("center_or_address", ""),
zoom=args.get("zoom", 15),
size=args.get("size", "600x400"),
)
content.append({"type": "text", "text": json.dumps(result, indent=2, ensure_ascii=False)})
else:
content.append({"type": "text", "text": f"Error: Unknown tool '{tool_name}'"})
res = {
"jsonrpc": "2.0",
"id": req_id,
"result": {"content": content},
}
print(json.dumps(res), flush=True)
else:
res = {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32601, "message": f"Method '{method}' not found"},
}
print(json.dumps(res), flush=True)
except Exception as e:
err_res = {
"jsonrpc": "2.0",
"id": None,
"error": {"code": -32603, "message": f"Internal error: {str(e)}"},
}
print(json.dumps(err_res), flush=True)
def run_cli():
"""CLI testing interface."""
parser = argparse.ArgumentParser(description="Google Maps MCP Server & CLI Tool")
subparsers = parser.add_subparsers(dest="command", help="Subcommand")
# Places
p_places = subparsers.add_parser("places", help="Search places")
p_places.add_argument("query", help="Search query")
p_places.add_argument("--location", "-l", default="Regensburg, Germany", help="Location bias")
p_places.add_argument("--radius", "-r", type=int, default=5000, help="Radius in meters")
p_places.add_argument("--open-now", action="store_true", help="Only open now")
# Details
p_det = subparsers.add_parser("details", help="Get place details")
p_det.add_argument("place_id", help="Place ID")
# Directions
p_dir = subparsers.add_parser("directions", help="Get route directions")
p_dir.add_argument("origin", help="Origin address")
p_dir.add_argument("destination", help="Destination address")
p_dir.add_argument("--mode", "-m", choices=["DRIVE", "TRANSIT", "WALK", "BICYCLE"], default="DRIVE")
# Static map
p_static = subparsers.add_parser("static", help="Render static map")
p_static.add_argument("center", help="Center address")
p_static.add_argument("--zoom", "-z", type=int, default=15)
p_static.add_argument("--size", "-s", default="600x400")
args = parser.parse_args()
if args.command == "places":
res = search_places(query=args.query, location=args.location, radius_meters=args.radius, open_now=args.open_now)
print(json.dumps(res, indent=2, ensure_ascii=False))
elif args.command == "details":
res = get_place_details(place_id=args.place_id)
print(json.dumps(res, indent=2, ensure_ascii=False))
elif args.command == "directions":
res = get_directions(origin=args.origin, destination=args.destination, travel_mode=args.mode)
print(json.dumps(res, indent=2, ensure_ascii=False))
elif args.command == "static":
res = render_static_map(center_or_address=args.center, zoom=args.zoom, size=args.size)
print(json.dumps(res, indent=2, ensure_ascii=False))
else:
parser.print_help()
if __name__ == "__main__":
if len(sys.argv) == 1 or (len(sys.argv) == 2 and sys.argv[1] == "--stdio"):
run_mcp_server()
else:
run_cli()
skills/youtube/SKILL.md
+37
View File
@@ -0,0 +1,37 @@
---
name: youtube
description: Search YouTube videos, retrieve video metadata, and extract full transcripts/subtitles with timestamps for instant AI summarization, recipes, or research.
---
# 📺 YouTube Integration (Search & Transcripts)
Full YouTube search and transcript extraction engine for OpenClaw:
1. 🔍 **`youtube_search(query, max_results=5, video_duration="any")`**:
- Searches YouTube via YouTube Data API v3.
- Returns title, channel, video ID, formatted duration, view count, published date, and direct link.
- Durations: `any`, `short` (< 4 min), `medium` (4-20 min), `long` (> 20 min).
2. 📜 **`youtube_get_transcript(video_url_or_id, languages=["de", "en"], include_timestamps=True)`**:
- **Unlimited & 100% Free**: Powered by `youtube-transcript-api` (zero GCP quota consumption).
- Fetches formatted subtitles and speech transcripts with `[MM:SS]` timestamps.
- Automatically prioritizes German (`de`) and English (`en`), including auto-generated subtitles.
- Perfect for:
- Summarizing interviews, podcasts, tech talks, and keynotes.
- Extracting cooking steps and recipes from YouTube food videos for `koch`.
- Extracting command-line instructions, code, or homelab setup guides for `itguy`.
3. ℹ️ **`youtube_video_details(video_url_or_id)`**:
- Fetches full description, tags, view count, like count, and publication details.
---
## 🧭 Prompt Routing & Best Practices
- **When the user pastes a YouTube URL** (e.g. *"Was wird in diesem Video erklärt? https://youtu.be/..."* or *"Fass mir das Video kurz zusammen"*):
- Do NOT waste a turn searching. Directly call `youtube_get_transcript` with the URL!
- Provide a concise executive summary with key takeaways, timestamps for notable moments, and actionable conclusions.
- **For Koch / Recipes**:
- When the user asks for a recipe from a video, pull the transcript via `youtube_get_transcript`, extract the ingredients and steps, and offer to add missing items to the Bring! shopping list.
- **For Research / Recherche**:
- Search relevant videos via `youtube_search` to find relevant lectures, documentaries, or news reports, then extract the transcript of the top hit for deep grounded analysis.
skills/youtube/scripts/youtube_mcp.py
Executable
+568
View File
@@ -0,0 +1,568 @@
#!/usr/bin/env python3
"""
YouTube MCP Server & CLI Tool for OpenClaw.
Integrates YouTube search, video metadata, and free transcript extraction
(subtitles) via YouTube Data API v3 and youtube-transcript-api.
Features:
- youtube_search: Search videos by keyword with title, channel, duration, and views
- youtube_get_transcript: Fetch German and English subtitles with timestamps (100% free, zero quota)
- youtube_video_details: Fetch full video metadata and description
"""
import os
import sys
import re
import json
import argparse
from typing import Dict, Any, List, Optional
from urllib.parse import urlparse, parse_qs
# Environment
YOUTUBE_API_KEY = os.environ.get("YOUTUBE_API_KEY", "").strip()
def extract_video_id(url_or_id: str) -> Optional[str]:
"""Extract 11-character YouTube video ID from various URL formats or raw ID."""
if not url_or_id:
return None
url_or_id = url_or_id.strip()
# If already an 11-char alphanumeric ID
if re.match(r"^[a-zA-Z0-9_-]{11}$", url_or_id):
return url_or_id
# Common URL formats:
# https://www.youtube.com/watch?v=VIDEO_ID
# https://youtu.be/VIDEO_ID
# https://www.youtube.com/shorts/VIDEO_ID
# https://www.youtube.com/embed/VIDEO_ID
parsed = urlparse(url_or_id)
if "youtube.com" in parsed.netloc:
if parsed.path == "/watch":
qs = parse_qs(parsed.query)
if "v" in qs and qs["v"]:
return qs["v"][0]
elif parsed.path.startswith("/shorts/"):
parts = parsed.path.split("/")
if len(parts) >= 3 and len(parts[2]) == 11:
return parts[2]
elif parsed.path.startswith("/embed/"):
parts = parsed.path.split("/")
if len(parts) >= 3 and len(parts[2]) == 11:
return parts[2]
elif "youtu.be" in parsed.netloc:
path = parsed.path.lstrip("/")
# Path might have query params or /
video_id = path.split("?")[0].split("/")[0]
if len(video_id) == 11:
return video_id
# Regex fallback
match = re.search(r"[a-zA-Z0-9_-]{11}", url_or_id)
if match:
return match.group(0)
return None
def parse_iso_duration(duration_str: str) -> str:
"""Convert ISO 8601 duration (PT1H2M30S or PT5M12S) to human readable (01:02:30 or 05:12)."""
if not duration_str or not duration_str.startswith("PT"):
return duration_str or ""
hours = 0
minutes = 0
seconds = 0
pattern = re.compile(r"PT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?")
match = pattern.match(duration_str)
if match:
h, m, s = match.groups()
hours = int(h) if h else 0
minutes = int(m) if m else 0
seconds = int(s) if s else 0
if hours > 0:
return f"{hours}:{minutes:02d}:{seconds:02d}"
return f"{minutes:02d}:{seconds:02d}"
return duration_str
def format_seconds(seconds: float) -> str:
"""Format seconds into HH:MM:SS or MM:SS."""
s = int(seconds)
hours = s // 3600
minutes = (s % 3600) // 60
secs = s % 60
if hours > 0:
return f"{hours:02d}:{minutes:02d}:{secs:02d}"
return f"{minutes:02d}:{secs:02d}"
# ==============================================================================
# YouTube Operations
# ==============================================================================
def search_videos(
query: str,
max_results: int = 5,
video_duration: str = "any",
api_key: Optional[str] = None,
) -> Dict[str, Any]:
"""Search YouTube videos using YouTube Data API v3."""
key = api_key or YOUTUBE_API_KEY
if not key:
return {
"success": False,
"error": "YOUTUBE_API_KEY is not set. Please provide a valid Google Cloud API key.",
"videos": [],
}
try:
from googleapiclient.discovery import build
youtube = build("youtube", "v3", developerKey=key, cache_discovery=False)
search_kwargs = {
"q": query,
"part": "snippet",
"type": "video",
"maxResults": min(max(1, max_results), 20),
}
if video_duration in ["short", "medium", "long"]:
search_kwargs["videoDuration"] = video_duration
search_resp = youtube.search().list(**search_kwargs).execute()
items = search_resp.get("items", [])
if not items:
return {"success": True, "query": query, "count": 0, "videos": []}
video_ids = [it["id"]["videoId"] for it in items if "videoId" in it.get("id", {})]
# Batch fetch details (duration, view count, stats) - only 1 unit!
details_map = {}
if video_ids:
videos_resp = youtube.videos().list(
id=",".join(video_ids),
part="contentDetails,statistics,snippet"
).execute()
for v in videos_resp.get("items", []):
details_map[v["id"]] = v
videos = []
for item in items:
vid = item["id"].get("videoId")
if not vid:
continue
snip = item.get("snippet", {})
v_detail = details_map.get(vid, {})
cd = v_detail.get("contentDetails", {})
stats = v_detail.get("statistics", {})
duration_raw = cd.get("duration", "")
duration_formatted = parse_iso_duration(duration_raw)
videos.append({
"video_id": vid,
"title": snip.get("title", ""),
"channel": snip.get("channelTitle", ""),
"published_at": snip.get("publishedAt", ""),
"duration": duration_formatted,
"view_count": stats.get("viewCount", "N/A"),
"url": f"https://www.youtube.com/watch?v={vid}",
"description_snippet": snip.get("description", "")[:200],
"thumbnail": snip.get("thumbnails", {}).get("high", {}).get("url", ""),
})
return {
"success": True,
"query": query,
"count": len(videos),
"videos": videos,
}
except Exception as e:
return {
"success": False,
"error": f"YouTube search failed: {str(e)}",
"videos": [],
}
def get_transcript(
video_url_or_id: str,
languages: Optional[List[str]] = None,
include_timestamps: bool = True,
) -> Dict[str, Any]:
"""Fetch video transcript/subtitles using youtube-transcript-api."""
video_id = extract_video_id(video_url_or_id)
if not video_id:
return {
"success": False,
"error": f"Could not extract a valid YouTube video ID from '{video_url_or_id}'.",
}
lang_pref = languages or ["de", "en"]
try:
from youtube_transcript_api import YouTubeTranscriptApi
api = YouTubeTranscriptApi()
# Try fetching with preferred languages
transcript_obj = None
detected_lang = None
is_generated = False
# Use .list(video_id) to inspect available tracks and find best match
try:
transcript_list = api.list(video_id)
# Find best match or auto-generated
try:
transcript_obj = transcript_list.find_transcript(lang_pref)
except Exception:
# If no direct match in lang_pref, try any available transcript and translate if possible
for t in transcript_list:
transcript_obj = t
break
if transcript_obj:
detected_lang = transcript_obj.language_code
is_generated = transcript_obj.is_generated
snippets = transcript_obj.fetch()
else:
# Fallback to direct fetch
snippets = api.fetch(video_id, languages=lang_pref)
except Exception:
# Fallback to direct fetch
snippets = api.fetch(video_id, languages=lang_pref)
formatted_lines = []
full_text_chunks = []
total_duration = 0.0
for item in snippets:
# item may be an object (FetchedTranscriptSnippet) or dict
if hasattr(item, "text"):
text = item.text.strip()
start = getattr(item, "start", 0.0)
duration = getattr(item, "duration", 0.0)
elif isinstance(item, dict):
text = item.get("text", "").strip()
start = item.get("start", 0.0)
duration = item.get("duration", 0.0)
else:
text = str(item).strip()
start = 0.0
duration = 0.0
if not text:
continue
timestamp = format_seconds(start)
formatted_lines.append(f"[{timestamp}] {text}")
full_text_chunks.append(text)
total_duration = max(total_duration, start + duration)
full_text = " ".join(full_text_chunks)
return {
"success": True,
"video_id": video_id,
"url": f"https://www.youtube.com/watch?v={video_id}",
"language": detected_lang or "auto",
"is_generated": is_generated,
"duration_seconds": round(total_duration, 1),
"duration_formatted": format_seconds(total_duration),
"snippet_count": len(formatted_lines),
"formatted_transcript": "\n".join(formatted_lines) if include_timestamps else full_text,
"full_text": full_text,
}
except Exception as e:
err_msg = str(e)
if "TranscriptsDisabled" in err_msg or "Subtitles are disabled" in err_msg:
return {
"success": False,
"video_id": video_id,
"error": "Transcripts are disabled for this video by the uploader.",
}
elif "NoTranscriptFound" in err_msg:
return {
"success": False,
"video_id": video_id,
"error": f"No transcript found in requested languages ({lang_pref}).",
}
return {
"success": False,
"video_id": video_id,
"error": f"Failed to retrieve transcript: {err_msg}",
}
def get_video_details(
video_url_or_id: str,
api_key: Optional[str] = None,
) -> Dict[str, Any]:
"""Fetch complete video details and description."""
video_id = extract_video_id(video_url_or_id)
if not video_id:
return {
"success": False,
"error": f"Could not extract a valid YouTube video ID from '{video_url_or_id}'.",
}
key = api_key or YOUTUBE_API_KEY
if not key:
return {
"success": False,
"error": "YOUTUBE_API_KEY is not set.",
}
try:
from googleapiclient.discovery import build
youtube = build("youtube", "v3", developerKey=key, cache_discovery=False)
resp = youtube.videos().list(
id=video_id,
part="snippet,contentDetails,statistics"
).execute()
items = resp.get("items", [])
if not items:
return {
"success": False,
"error": f"Video '{video_id}' not found.",
}
item = items[0]
snip = item.get("snippet", {})
cd = item.get("contentDetails", {})
stats = item.get("statistics", {})
return {
"success": True,
"video_id": video_id,
"title": snip.get("title", ""),
"channel": snip.get("channelTitle", ""),
"published_at": snip.get("publishedAt", ""),
"description": snip.get("description", ""),
"duration": parse_iso_duration(cd.get("duration", "")),
"view_count": stats.get("viewCount", "0"),
"like_count": stats.get("likeCount", "0"),
"comment_count": stats.get("commentCount", "0"),
"tags": snip.get("tags", []),
"url": f"https://www.youtube.com/watch?v={video_id}",
"thumbnail": snip.get("thumbnails", {}).get("maxres", snip.get("thumbnails", {}).get("high", {})).get("url", ""),
}
except Exception as e:
return {
"success": False,
"error": f"Failed to get video details: {str(e)}",
}
# ==============================================================================
# MCP Protocol Handlers (JSON-RPC 2.0 stdio)
# ==============================================================================
TOOLS_DEFINITIONS = [
{
"name": "youtube_search",
"description": "Search YouTube for videos matching a query. Returns video title, channel, video ID, formatted duration, view count, and direct link.",
"inputSchema": {
"type": "object",
"required": ["query"],
"properties": {
"query": {
"type": "string",
"description": "The search term or topic (e.g. 'Ninja Creami Deluxe Rezepte', 'Kubernetes Helm Tutorial').",
},
"max_results": {
"type": "integer",
"description": "Maximum number of videos to return (default 5, max 20).",
"default": 5,
},
"video_duration": {
"type": "string",
"description": "Filter by duration: 'any' (default), 'short' (< 4m), 'medium' (4m-20m), 'long' (> 20m).",
"enum": ["any", "short", "medium", "long"],
"default": "any",
},
},
},
},
{
"name": "youtube_get_transcript",
"description": "Fetch video transcript / subtitles with timestamps. Free and unlimited. Essential for summarizing YouTube videos, extracting recipes, interview takeaways, or step-by-step instructions.",
"inputSchema": {
"type": "object",
"required": ["video_url_or_id"],
"properties": {
"video_url_or_id": {
"type": "string",
"description": "YouTube URL (e.g. 'https://youtu.be/...' or 'https://www.youtube.com/watch?v=...') or 11-char video ID.",
},
"languages": {
"type": "array",
"items": {"type": "string"},
"description": "Preferred language codes in priority order (default ['de', 'en']).",
"default": ["de", "en"],
},
"include_timestamps": {
"type": "boolean",
"description": "Whether to include [HH:MM:SS] timestamps for each segment (default True).",
"default": True,
},
},
},
},
{
"name": "youtube_video_details",
"description": "Fetch complete YouTube video details including full description, channel, publish date, duration, tags, and stats.",
"inputSchema": {
"type": "object",
"required": ["video_url_or_id"],
"properties": {
"video_url_or_id": {
"type": "string",
"description": "YouTube video URL or 11-character video ID.",
},
},
},
},
]
def run_mcp_server():
"""Run JSON-RPC 2.0 stdio loop for OpenClaw."""
while True:
try:
line = sys.stdin.readline()
if not line:
break
line = line.strip()
if not line:
continue
req = json.loads(line)
req_id = req.get("id")
method = req.get("method")
params = req.get("params", {})
if method == "initialize":
res = {
"jsonrpc": "2.0",
"id": req_id,
"result": {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"serverInfo": {
"name": "openclaw-youtube",
"version": "1.0.0",
},
},
}
print(json.dumps(res), flush=True)
elif method == "notifications/initialized":
pass
elif method == "tools/list":
res = {
"jsonrpc": "2.0",
"id": req_id,
"result": {"tools": TOOLS_DEFINITIONS},
}
print(json.dumps(res), flush=True)
elif method == "tools/call":
tool_name = params.get("name")
args = params.get("arguments", {})
content = []
if tool_name == "youtube_search":
result = search_videos(
query=args.get("query", ""),
max_results=args.get("max_results", 5),
video_duration=args.get("video_duration", "any"),
)
content.append({"type": "text", "text": json.dumps(result, indent=2, ensure_ascii=False)})
elif tool_name == "youtube_get_transcript":
result = get_transcript(
video_url_or_id=args.get("video_url_or_id", ""),
languages=args.get("languages", ["de", "en"]),
include_timestamps=args.get("include_timestamps", True),
)
content.append({"type": "text", "text": json.dumps(result, indent=2, ensure_ascii=False)})
elif tool_name == "youtube_video_details":
result = get_video_details(
video_url_or_id=args.get("video_url_or_id", "")
)
content.append({"type": "text", "text": json.dumps(result, indent=2, ensure_ascii=False)})
else:
content.append({"type": "text", "text": f"Error: Unknown tool '{tool_name}'"})
res = {
"jsonrpc": "2.0",
"id": req_id,
"result": {"content": content},
}
print(json.dumps(res), flush=True)
else:
res = {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32601, "message": f"Method '{method}' not found"},
}
print(json.dumps(res), flush=True)
except Exception as e:
err_res = {
"jsonrpc": "2.0",
"id": None,
"error": {"code": -32603, "message": f"Internal error: {str(e)}"},
}
print(json.dumps(err_res), flush=True)
def run_cli():
"""CLI testing interface."""
parser = argparse.ArgumentParser(description="YouTube MCP Server & CLI Tool")
subparsers = parser.add_subparsers(dest="command", help="Subcommand")
# Search
p_search = subparsers.add_parser("search", help="Search videos")
p_search.add_argument("query", help="Search query")
p_search.add_argument("--limit", "-n", type=int, default=5, help="Max results")
p_search.add_argument("--duration", choices=["any", "short", "medium", "long"], default="any")
# Transcript
p_trans = subparsers.add_parser("transcript", help="Get transcript")
p_trans.add_argument("video", help="URL or video ID")
p_trans.add_argument("--languages", "-l", default="de,en", help="Comma separated language preferences")
p_trans.add_argument("--no-timestamps", action="store_true", help="Omit timestamps")
# Details
p_details = subparsers.add_parser("details", help="Get video details")
p_details.add_argument("video", help="URL or video ID")
args = parser.parse_args()
if args.command == "search":
res = search_videos(query=args.query, max_results=args.limit, video_duration=args.duration)
print(json.dumps(res, indent=2, ensure_ascii=False))
elif args.command == "transcript":
langs = [l.strip() for l in args.languages.split(",") if l.strip()]
res = get_transcript(video_url_or_id=args.video, languages=langs, include_timestamps=not args.no_timestamps)
print(json.dumps(res, indent=2, ensure_ascii=False))
elif args.command == "details":
res = get_video_details(video_url_or_id=args.video)
print(json.dumps(res, indent=2, ensure_ascii=False))
else:
parser.print_help()
if __name__ == "__main__":
if len(sys.argv) == 1 or (len(sys.argv) == 2 and sys.argv[1] == "--stdio"):
run_mcp_server()
else:
run_cli()