#!/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()