|
| 1 | +"""Create an event in Google Calendar.""" |
| 2 | + |
| 3 | +import re |
| 4 | +from datetime import datetime |
| 5 | +from typing import Any, Dict, List, Optional, Type, Union |
| 6 | +from uuid import uuid4 |
| 7 | + |
| 8 | +from langchain_core.callbacks import CallbackManagerForToolRun |
| 9 | +from pydantic import BaseModel, Field |
| 10 | + |
| 11 | +from langchain_google_community.calendar.base import CalendarBaseTool |
| 12 | +from langchain_google_community.calendar.utils import is_all_day_event |
| 13 | + |
| 14 | + |
| 15 | +class CreateEventSchema(BaseModel): |
| 16 | + """Input for CalendarCreateEvent.""" |
| 17 | + |
| 18 | + summary: str = Field(..., description="The title of the event.") |
| 19 | + start_datetime: str = Field( |
| 20 | + ..., |
| 21 | + description=( |
| 22 | + "The start datetime for the event in 'YYYY-MM-DD HH:MM:SS' format." |
| 23 | + "If the event is an all-day event, set the time to 'YYYY-MM-DD' format." |
| 24 | + "If you do not know the current datetime, use the tool to get it." |
| 25 | + ), |
| 26 | + ) |
| 27 | + end_datetime: str = Field( |
| 28 | + ..., |
| 29 | + description=( |
| 30 | + "The end datetime for the event in 'YYYY-MM-DD HH:MM:SS' format. " |
| 31 | + "If the event is an all-day event, set the time to 'YYYY-MM-DD' format." |
| 32 | + ), |
| 33 | + ) |
| 34 | + timezone: str = Field(..., description="The timezone of the event.") |
| 35 | + calendar_id: str = Field( |
| 36 | + default="primary", description="The calendar ID to create the event in." |
| 37 | + ) |
| 38 | + recurrence: Optional[Dict[str, Any]] = Field( |
| 39 | + default=None, |
| 40 | + description=( |
| 41 | + "The recurrence of the event. " |
| 42 | + "Format: {'FREQ': <'DAILY' or 'WEEKLY'>, 'INTERVAL': <number>, " |
| 43 | + "'COUNT': <number or None>, 'UNTIL': <'YYYYMMDD' or None>, " |
| 44 | + "'BYDAY': <'MO', 'TU', 'WE', 'TH', 'FR', 'SA', 'SU' or None>}. " |
| 45 | + "Use either COUNT or UNTIL, but not both; set the other to None." |
| 46 | + ), |
| 47 | + ) |
| 48 | + location: Optional[str] = Field( |
| 49 | + default=None, description="The location of the event." |
| 50 | + ) |
| 51 | + description: Optional[str] = Field( |
| 52 | + default=None, description="The description of the event." |
| 53 | + ) |
| 54 | + attendees: Optional[List[str]] = Field( |
| 55 | + default=None, description="A list of attendees' email addresses for the event." |
| 56 | + ) |
| 57 | + reminders: Union[None, bool, List[Dict[str, Any]]] = Field( |
| 58 | + default=None, |
| 59 | + description=( |
| 60 | + "Reminders for the event. " |
| 61 | + "Set to True for default reminders, or provide a list like " |
| 62 | + "[{'method': 'email', 'minutes': <minutes>}, ...]. " |
| 63 | + "Valid methods are 'email' and 'popup'." |
| 64 | + ), |
| 65 | + ) |
| 66 | + conference_data: Optional[bool] = Field( |
| 67 | + default=None, description="Whether to include conference data." |
| 68 | + ) |
| 69 | + color_id: Optional[str] = Field( |
| 70 | + default=None, |
| 71 | + description=( |
| 72 | + "The color ID of the event. None for default. " |
| 73 | + "'1': Lavender, '2': Sage, '3': Grape, '4': Flamingo, '5': Banana, " |
| 74 | + "'6': Tangerine, '7': Peacock, '8': Graphite, '9': Blueberry, " |
| 75 | + "'10': Basil, '11': Tomato." |
| 76 | + ), |
| 77 | + ) |
| 78 | + transparency: Optional[str] = Field( |
| 79 | + default=None, |
| 80 | + description=( |
| 81 | + "User availability for the event." |
| 82 | + "transparent for available and opaque for busy." |
| 83 | + ), |
| 84 | + ) |
| 85 | + |
| 86 | + |
| 87 | +class CalendarCreateEvent(CalendarBaseTool): # type: ignore[override, override] |
| 88 | + """Tool that creates an event in Google Calendar.""" |
| 89 | + |
| 90 | + name: str = "create_calendar_event" |
| 91 | + description: str = ( |
| 92 | + "Use this tool to create an event. " |
| 93 | + "The input must include the summary, start, and end datetime for the event." |
| 94 | + ) |
| 95 | + args_schema: Type[CreateEventSchema] = CreateEventSchema |
| 96 | + |
| 97 | + def _prepare_event( |
| 98 | + self, |
| 99 | + summary: str, |
| 100 | + start_datetime: str, |
| 101 | + end_datetime: str, |
| 102 | + timezone: str, |
| 103 | + recurrence: Optional[Dict[str, Any]] = None, |
| 104 | + location: Optional[str] = None, |
| 105 | + description: Optional[str] = None, |
| 106 | + attendees: Optional[List[str]] = None, |
| 107 | + reminders: Union[None, bool, List[Dict[str, Any]]] = None, |
| 108 | + conference_data: Optional[bool] = None, |
| 109 | + color_id: Optional[str] = None, |
| 110 | + transparency: Optional[str] = None, |
| 111 | + ) -> Dict[str, Any]: |
| 112 | + """Prepare the event body.""" |
| 113 | + try: |
| 114 | + if is_all_day_event(start_datetime, end_datetime): |
| 115 | + start = {"date": start_datetime} |
| 116 | + end = {"date": end_datetime} |
| 117 | + else: |
| 118 | + datetime_format = "%Y-%m-%d %H:%M:%S" |
| 119 | + start_dt = datetime.strptime(start_datetime, datetime_format) |
| 120 | + end_dt = datetime.strptime(end_datetime, datetime_format) |
| 121 | + start = { |
| 122 | + "dateTime": start_dt.astimezone().isoformat(), |
| 123 | + "timeZone": timezone, |
| 124 | + } |
| 125 | + end = { |
| 126 | + "dateTime": end_dt.astimezone().isoformat(), |
| 127 | + "timeZone": timezone, |
| 128 | + } |
| 129 | + except ValueError as error: |
| 130 | + raise ValueError("The datetime format is incorrect.") from error |
| 131 | + recurrence_data = None |
| 132 | + if recurrence: |
| 133 | + if isinstance(recurrence, dict): |
| 134 | + recurrence_items = [ |
| 135 | + f"{k}={v}" for k, v in recurrence.items() if v is not None |
| 136 | + ] |
| 137 | + recurrence_data = "RRULE:" + ";".join(recurrence_items) |
| 138 | + attendees_emails: List[Dict[str, str]] = [] |
| 139 | + if attendees: |
| 140 | + email_pattern = r"^[^@]+@[^@]+\.[^@]+$" |
| 141 | + for email in attendees: |
| 142 | + if not re.match(email_pattern, email): |
| 143 | + raise ValueError(f"Invalid email address: {email}") |
| 144 | + attendees_emails.append({"email": email}) |
| 145 | + reminders_info: Dict[str, Union[bool, List[Dict[str, Any]]]] = {} |
| 146 | + if reminders is True: |
| 147 | + reminders_info.update({"useDefault": True}) |
| 148 | + elif isinstance(reminders, list): |
| 149 | + for reminder in reminders: |
| 150 | + if "method" not in reminder or "minutes" not in reminder: |
| 151 | + raise ValueError( |
| 152 | + "Each reminder must have 'method' and 'minutes' keys." |
| 153 | + ) |
| 154 | + if reminder["method"] not in ["email", "popup"]: |
| 155 | + raise ValueError("The reminder method must be 'email' or 'popup") |
| 156 | + reminders_info.update({"useDefault": False, "overrides": reminders}) |
| 157 | + else: |
| 158 | + reminders_info.update({"useDefault": False}) |
| 159 | + conference_data_info = None |
| 160 | + if conference_data: |
| 161 | + conference_data_info = { |
| 162 | + "createRequest": { |
| 163 | + "requestId": str(uuid4()), |
| 164 | + "conferenceSolutionKey": {"type": "hangoutsMeet"}, |
| 165 | + } |
| 166 | + } |
| 167 | + event_body: Dict[str, Any] = {"summary": summary, "start": start, "end": end} |
| 168 | + if location: |
| 169 | + event_body["location"] = location |
| 170 | + if description: |
| 171 | + event_body["description"] = description |
| 172 | + if recurrence_data: |
| 173 | + event_body["recurrence"] = [recurrence_data] |
| 174 | + if len(attendees_emails) > 0: |
| 175 | + event_body["attendees"] = attendees_emails |
| 176 | + if len(reminders_info) > 0: |
| 177 | + event_body["reminders"] = reminders_info |
| 178 | + if conference_data_info: |
| 179 | + event_body["conferenceData"] = conference_data_info |
| 180 | + if color_id: |
| 181 | + event_body["colorId"] = color_id |
| 182 | + if transparency: |
| 183 | + event_body["transparency"] = transparency |
| 184 | + return event_body |
| 185 | + |
| 186 | + def _run( |
| 187 | + self, |
| 188 | + summary: str, |
| 189 | + start_datetime: str, |
| 190 | + end_datetime: str, |
| 191 | + timezone: str, |
| 192 | + calendar_id: str = "primary", |
| 193 | + recurrence: Optional[Dict[str, Any]] = None, |
| 194 | + location: Optional[str] = None, |
| 195 | + description: Optional[str] = None, |
| 196 | + attendees: Optional[List[str]] = None, |
| 197 | + reminders: Union[None, bool, List[Dict[str, Any]]] = None, |
| 198 | + conference_data: Optional[bool] = None, |
| 199 | + color_id: Optional[str] = None, |
| 200 | + transparency: Optional[str] = None, |
| 201 | + run_manager: Optional[CallbackManagerForToolRun] = None, |
| 202 | + ) -> str: |
| 203 | + """Run the tool to create an event in Google Calendar.""" |
| 204 | + try: |
| 205 | + body = self._prepare_event( |
| 206 | + summary=summary, |
| 207 | + start_datetime=start_datetime, |
| 208 | + end_datetime=end_datetime, |
| 209 | + timezone=timezone, |
| 210 | + recurrence=recurrence, |
| 211 | + location=location, |
| 212 | + description=description, |
| 213 | + attendees=attendees, |
| 214 | + reminders=reminders, |
| 215 | + conference_data=conference_data, |
| 216 | + color_id=color_id, |
| 217 | + transparency=transparency, |
| 218 | + ) |
| 219 | + conference_version = 1 if conference_data else 0 |
| 220 | + event = ( |
| 221 | + self.api_resource.events() |
| 222 | + .insert( |
| 223 | + calendarId=calendar_id, |
| 224 | + body=body, |
| 225 | + conferenceDataVersion=conference_version, |
| 226 | + ) |
| 227 | + .execute() |
| 228 | + ) |
| 229 | + return f"Event created: {event.get('htmlLink')}" |
| 230 | + except Exception as error: |
| 231 | + raise Exception(f"An error occurred: {error}") from error |
0 commit comments