Enhanced MCP Server

""" Google Calendar Adapter with WRITE capabilities Enables creating events, setting reminders, blocking time """ from datetime import datetime, timedelta from typing import List, Dict, Any, Optional from googleapiclient.discovery import build from googleapiclient.errors import HttpError import pytz import structlog from ..utils.google_auth import get_credentials logger = structlog.get_logger(__name__) class CalendarAdapter: """Google Calendar adapter with read and write operations""" def __init__(self, calendar_id: str = "primary"): self.calendar_id = calendar_id self.service = None self._initialize_service() def _initialize_service(self): """Initialize Google Calendar service""" try: creds = get_credentials() self.service = build('calendar', 'v3', credentials=creds) logger.info("Calendar service initialized") except Exception as e: logger.error(f"Failed to initialize calendar service: {e}") raise def create_event( self, summary: str, start_time: datetime, end_time: datetime, description: Optional[str] = None, location: Optional[str] = None, attendees: Optional[List[str]] = None, reminders: Optional[List[int]] = None, timezone: str = "America/Los_Angeles" ) -> Dict[str, Any]: """ Create a calendar event Args: summary: Event title start_time: Start datetime end_time: End datetime description: Event description location: Event location attendees: List of attendee emails reminders: List of reminder minutes (e.g., [15, 60] for 15min and 1hr before) timezone: Timezone for the event Returns: Created event details """ try: # Build event body event = { 'summary': summary, 'start': { 'dateTime': start_time.isoformat(), 'timeZone': timezone, }, 'end': { 'dateTime': end_time.isoformat(), 'timeZone': timezone, }, } if description: event['description'] = description if location: event['location'] = location if attendees: event['attendees'] = [{'email': email} for email in attendees] # Set reminders if reminders: event['reminders'] = { 'useDefault': False, 'overrides': [ {'method': 'popup', 'minutes': minutes} for minutes in reminders ] } else: event['reminders'] = {'useDefault': True} # Create event created_event = self.service.events().insert( calendarId=self.calendar_id, body=event ).execute() logger.info( f"Created calendar event: {summary}", event_id=created_event['id'] ) return { "success": True, "event_id": created_event['id'], "event_link": created_event.get('htmlLink'), "summary": created_event['summary'], "start": created_event['start'].get('dateTime'), "end": created_event['end'].get('dateTime') } except HttpError as e: logger.error(f"Calendar API error: {e}") return { "success": False, "error": str(e) } except Exception as e: logger.error(f"Failed to create event: {e}") return { "success": False, "error": str(e) } def block_time( self, start_time: datetime, end_time: datetime, title: str = "Blocked Time", description: Optional[str] = None, reminders: Optional[List[int]] = None ) -> Dict[str, Any]: """ Block time on calendar (convenience method) Args: start_time: Block start time end_time: Block end time title: Block title (default: "Blocked Time") description: Optional description reminders: Reminder minutes before Returns: Created event details """ return self.create_event( summary=title, start_time=start_time, end_time=end_time, description=description, reminders=reminders ) def get_events( self, time_min: Optional[datetime] = None, time_max: Optional[datetime] = None, max_results: int = 10 ) -> Dict[str, Any]: """ Get calendar events Args: time_min: Start time (default: now) time_max: End time (default: 1 week from now) max_results: Maximum events to return Returns: List of events """ try: if time_min is None: time_min = datetime.now(pytz.UTC) if time_max is None: time_max = time_min + timedelta(days=7) events_result = self.service.events().list( calendarId=self.calendar_id, timeMin=time_min.isoformat(), timeMax=time_max.isoformat(), maxResults=max_results, singleEvents=True, orderBy='startTime' ).execute() events = events_result.get('items', []) return { "success": True, "count": len(events), "events": [ { "id": event.get('id'), "summary": event.get('summary', 'No title'), "start": event['start'].get('dateTime', event['start'].get('date')), "end": event['end'].get('dateTime', event['end'].get('date')), "description": event.get('description', ''), "location": event.get('location', ''), "attendees": [ att.get('email') for att in event.get('attendees', []) ] } for event in events ] } except HttpError as e: logger.error(f"Calendar API error: {e}") return { "success": False, "error": str(e) } def update_event( self, event_id: str, summary: Optional[str] = None, start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, description: Optional[str] = None, reminders: Optional[List[int]] = None ) -> Dict[str, Any]: """ Update an existing calendar event Args: event_id: Event ID to update summary: New title (optional) start_time: New start time (optional) end_time: New end time (optional) description: New description (optional) reminders: New reminders (optional) Returns: Updated event details """ try: # Get existing event event = self.service.events().get( calendarId=self.calendar_id, eventId=event_id ).execute() # Update fields if summary: event['summary'] = summary if start_time: event['start']['dateTime'] = start_time.isoformat() if end_time: event['end']['dateTime'] = end_time.isoformat() if description is not None: event['description'] = description if reminders: event['reminders'] = { 'useDefault': False, 'overrides': [ {'method': 'popup', 'minutes': minutes} for minutes in reminders ] } # Update event updated_event = self.service.events().update( calendarId=self.calendar_id, eventId=event_id, body=event ).execute() logger.info(f"Updated calendar event: {event_id}") return { "success": True, "event_id": updated_event['id'], "summary": updated_event['summary'] } except HttpError as e: logger.error(f"Calendar API error: {e}") return { "success": False, "error": str(e) } def delete_event(self, event_id: str) -> Dict[str, Any]: """ Delete a calendar event Args: event_id: Event ID to delete Returns: Success status """ try: self.service.events().delete( calendarId=self.calendar_id, eventId=event_id ).execute() logger.info(f"Deleted calendar event: {event_id}") return { "success": True, "message": f"Event {event_id} deleted successfully" } except HttpError as e: logger.error(f"Calendar API error: {e}") return { "success": False, "error": str(e) }