diff --git a/skills/google-maps/SKILL.md b/skills/google-maps/SKILL.md new file mode 100644 index 0000000..53e7617 --- /dev/null +++ b/skills/google-maps/SKILL.md @@ -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:` 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. diff --git a/skills/google-maps/scripts/google_maps_mcp.py b/skills/google-maps/scripts/google_maps_mcp.py new file mode 100755 index 0000000..5c5c9bd --- /dev/null +++ b/skills/google-maps/scripts/google_maps_mcp.py @@ -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() diff --git a/skills/youtube/SKILL.md b/skills/youtube/SKILL.md new file mode 100644 index 0000000..3cdbd40 --- /dev/null +++ b/skills/youtube/SKILL.md @@ -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. diff --git a/skills/youtube/scripts/youtube_mcp.py b/skills/youtube/scripts/youtube_mcp.py new file mode 100755 index 0000000..76b41d1 --- /dev/null +++ b/skills/youtube/scripts/youtube_mcp.py @@ -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()