#!/usr/bin/env python3 """ Google Calendar API client for OpenClaw. Supports listing events, responding to invitations, and CRUD operations. """ import os import sys import json import argparse from datetime import datetime, timedelta from pathlib import Path # Google API imports try: from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow from googleapiclient.discovery import build from googleapiclient.errors import HttpError except ImportError: print("Missing dependencies. Run:") print(" pip3 install google-auth google-auth-oauthlib google-api-python-client --user") sys.exit(1) # OAuth scopes - calendar read/write SCOPES = ['https://www.googleapis.com/auth/calendar'] # Config paths CONFIG_DIR = Path.home() / ".config" / "google-calendar" CREDENTIALS_FILE = CONFIG_DIR / "credentials.json" TOKEN_FILE = CONFIG_DIR / "token.json" def ensure_config_dir(): """Create config directory with secure permissions.""" CONFIG_DIR.mkdir(parents=True, exist_ok=True) os.chmod(CONFIG_DIR, 0o700) def get_credentials(): """Get or refresh OAuth credentials.""" creds = None # Load existing token if TOKEN_FILE.exists(): creds = Credentials.from_authorized_user_file(str(TOKEN_FILE), SCOPES) # Refresh or get new credentials if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: creds.refresh(Request()) else: if not CREDENTIALS_FILE.exists(): print(f"Error: {CREDENTIALS_FILE} not found.") print("Download OAuth credentials from Google Cloud Console.") sys.exit(1) flow = InstalledAppFlow.from_client_secrets_file( str(CREDENTIALS_FILE), SCOPES ) creds = flow.run_local_server(port=0) # Save token ensure_config_dir() with open(TOKEN_FILE, 'w') as token: token.write(creds.to_json()) os.chmod(TOKEN_FILE, 0o600) return creds def get_service(): """Get authenticated Calendar API service.""" creds = get_credentials() return build('calendar', 'v3', credentials=creds) def list_calendars(service): """List all calendars.""" result = service.calendarList().list().execute() calendars = result.get('items', []) return calendars def list_events(service, calendar_id='primary', max_results=10, time_min=None): """List upcoming events.""" if time_min is None: time_min = datetime.utcnow().isoformat() + 'Z' events_result = service.events().list( calendarId=calendar_id, timeMin=time_min, maxResults=max_results, singleEvents=True, orderBy='startTime' ).execute() return events_result.get('items', []) def get_pending_invitations(service, calendar_id='primary'): """Get events where user hasn't responded (needsAction).""" # Get events from now to 90 days ahead time_min = datetime.utcnow().isoformat() + 'Z' time_max = (datetime.utcnow() + timedelta(days=90)).isoformat() + 'Z' events_result = service.events().list( calendarId=calendar_id, timeMin=time_min, timeMax=time_max, singleEvents=True, orderBy='startTime' ).execute() events = events_result.get('items', []) pending = [] for event in events: attendees = event.get('attendees', []) for attendee in attendees: if attendee.get('self', False): if attendee.get('responseStatus') == 'needsAction': pending.append(event) break return pending def respond_to_invitation(service, event_id, response_status, calendar_id='primary'): """ Respond to a calendar invitation. response_status: 'accepted', 'declined', 'tentative' """ # Get the event first event = service.events().get(calendarId=calendar_id, eventId=event_id).execute() # Find self in attendees and update response attendees = event.get('attendees', []) for attendee in attendees: if attendee.get('self', False): attendee['responseStatus'] = response_status break # Update the event updated_event = service.events().patch( calendarId=calendar_id, eventId=event_id, body={'attendees': attendees}, sendUpdates='all' # Notify organizer ).execute() return updated_event def create_event(service, title, start, end, description=None, location=None, attendees=None, calendar_id='primary'): """Create a new calendar event.""" event = { 'summary': title, 'start': {'dateTime': start, 'timeZone': 'America/Los_Angeles'}, 'end': {'dateTime': end, 'timeZone': 'America/Los_Angeles'}, } if description: event['description'] = description if location: event['location'] = location if attendees: event['attendees'] = [{'email': email.strip()} for email in attendees.split(',')] created = service.events().insert(calendarId=calendar_id, body=event).execute() return created def update_event(service, event_id, calendar_id='primary', **kwargs): """Update an existing event.""" # Get current event event = service.events().get(calendarId=calendar_id, eventId=event_id).execute() # Update fields if 'title' in kwargs: event['summary'] = kwargs['title'] if 'description' in kwargs: event['description'] = kwargs['description'] if 'location' in kwargs: event['location'] = kwargs['location'] if 'start' in kwargs: event['start'] = {'dateTime': kwargs['start'], 'timeZone': 'America/Los_Angeles'} if 'end' in kwargs: event['end'] = {'dateTime': kwargs['end'], 'timeZone': 'America/Los_Angeles'} updated = service.events().update( calendarId=calendar_id, eventId=event_id, body=event ).execute() return updated def delete_event(service, event_id, calendar_id='primary'): """Delete an event.""" service.events().delete(calendarId=calendar_id, eventId=event_id).execute() return True def format_event(event): """Format event for display.""" start = event['start'].get('dateTime', event['start'].get('date')) end = event['end'].get('dateTime', event['end'].get('date')) output = [] output.append(f"ID: {event['id']}") output.append(f"Title: {event.get('summary', '(No title)')}") output.append(f"Start: {start}") output.append(f"End: {end}") if event.get('location'): output.append(f"Location: {event['location']}") if event.get('organizer'): org = event['organizer'] org_name = org.get('displayName', org.get('email', 'Unknown')) output.append(f"Organizer: {org_name}") # Show your response status for attendee in event.get('attendees', []): if attendee.get('self', False): output.append(f"Your Status: {attendee.get('responseStatus', 'unknown')}") break return '\n'.join(output) def main(): parser = argparse.ArgumentParser(description='Google Calendar CLI') subparsers = parser.add_subparsers(dest='command', help='Commands') # Auth command subparsers.add_parser('auth', help='Authorize with Google') # Calendars command subparsers.add_parser('calendars', help='List calendars') # Events command events_parser = subparsers.add_parser('events', help='List upcoming events') events_parser.add_argument('--count', type=int, default=10, help='Number of events') events_parser.add_argument('--calendar', default='primary', help='Calendar ID or name') # Invitations command inv_parser = subparsers.add_parser('invitations', help='List pending invitations') inv_parser.add_argument('--calendar', default='primary', help='Calendar ID') # Respond command resp_parser = subparsers.add_parser('respond', help='Respond to invitation') resp_parser.add_argument('event_id', help='Event ID') resp_parser.add_argument('--status', required=True, choices=['accepted', 'declined', 'tentative'], help='Response status') resp_parser.add_argument('--calendar', default='primary', help='Calendar ID') # Create command create_parser = subparsers.add_parser('create', help='Create event') create_parser.add_argument('--title', required=True, help='Event title') create_parser.add_argument('--start', required=True, help='Start time (ISO format)') create_parser.add_argument('--end', required=True, help='End time (ISO format)') create_parser.add_argument('--description', help='Description') create_parser.add_argument('--location', help='Location') create_parser.add_argument('--attendees', help='Comma-separated emails') create_parser.add_argument('--calendar', default='primary', help='Calendar ID') # Update command update_parser = subparsers.add_parser('update', help='Update event') update_parser.add_argument('event_id', help='Event ID') update_parser.add_argument('--title', help='New title') update_parser.add_argument('--start', help='New start time') update_parser.add_argument('--end', help='New end time') update_parser.add_argument('--description', help='New description') update_parser.add_argument('--location', help='New location') update_parser.add_argument('--calendar', default='primary', help='Calendar ID') # Delete command delete_parser = subparsers.add_parser('delete', help='Delete event') delete_parser.add_argument('event_id', help='Event ID') delete_parser.add_argument('--calendar', default='primary', help='Calendar ID') args = parser.parse_args() if not args.command: parser.print_help() sys.exit(1) try: if args.command == 'auth': ensure_config_dir() print(f"Config directory: {CONFIG_DIR}") print(f"Looking for credentials at: {CREDENTIALS_FILE}") get_credentials() print("✅ Authorization successful! Token saved.") return service = get_service() if args.command == 'calendars': calendars = list_calendars(service) for cal in calendars: primary = " (primary)" if cal.get('primary') else "" print(f"• {cal['summary']}{primary}") print(f" ID: {cal['id']}") elif args.command == 'events': events = list_events(service, args.calendar, args.count) if not events: print("No upcoming events found.") else: for event in events: print(format_event(event)) print("---") elif args.command == 'invitations': invites = get_pending_invitations(service, args.calendar) if not invites: print("No pending invitations.") else: print(f"Found {len(invites)} pending invitation(s):\n") for event in invites: print(format_event(event)) print("---") elif args.command == 'respond': result = respond_to_invitation(service, args.event_id, args.status, args.calendar) print(f"✅ Responded '{args.status}' to: {result.get('summary', args.event_id)}") elif args.command == 'create': result = create_event( service, args.title, args.start, args.end, args.description, args.location, args.attendees, args.calendar ) print(f"✅ Created event: {result.get('summary')}") print(f" ID: {result['id']}") print(f" Link: {result.get('htmlLink', 'N/A')}") elif args.command == 'update': updates = {} for field in ['title', 'start', 'end', 'description', 'location']: if getattr(args, field, None): updates[field] = getattr(args, field) result = update_event(service, args.event_id, args.calendar, **updates) print(f"✅ Updated event: {result.get('summary')}") elif args.command == 'delete': delete_event(service, args.event_id, args.calendar) print(f"✅ Deleted event: {args.event_id}") except HttpError as error: print(f"API Error: {error}") sys.exit(1) if __name__ == '__main__': main()