"""Create an event in Google Calendar."""
import re
from datetime import datetime
from typing import Any, Dict, List, Optional, Type, Union
from uuid import uuid4
from zoneinfo import ZoneInfo
from langchain_core.callbacks import CallbackManagerForToolRun
from pydantic import BaseModel, Field
from langchain_google_community.calendar.base import CalendarBaseTool
from langchain_google_community.calendar.utils import is_all_day_event
[docs]
class CreateEventSchema(BaseModel):
"""Input for CalendarCreateEvent."""
summary: str = Field(..., description="The title of the event.")
start_datetime: str = Field(
...,
description=(
"The start datetime for the event in 'YYYY-MM-DD HH:MM:SS' format."
"If the event is an all-day event, set the time to 'YYYY-MM-DD' format."
"If you do not know the current datetime, use the tool to get it."
),
)
end_datetime: str = Field(
...,
description=(
"The end datetime for the event in 'YYYY-MM-DD HH:MM:SS' format. "
"If the event is an all-day event, set the time to 'YYYY-MM-DD' format."
),
)
timezone: str = Field(..., description="The timezone of the event.")
calendar_id: str = Field(
default="primary", description="The calendar ID to create the event in."
)
recurrence: Optional[Dict[str, Any]] = Field(
default=None,
description=(
"The recurrence of the event. "
"Format: {'FREQ': <'DAILY' or 'WEEKLY'>, 'INTERVAL': <number>, "
"'COUNT': <number or None>, 'UNTIL': <'YYYYMMDD' or None>, "
"'BYDAY': <'MO', 'TU', 'WE', 'TH', 'FR', 'SA', 'SU' or None>}. "
"Use either COUNT or UNTIL, but not both; set the other to None."
),
)
location: Optional[str] = Field(
default=None, description="The location of the event."
)
description: Optional[str] = Field(
default=None, description="The description of the event."
)
attendees: Optional[List[str]] = Field(
default=None, description="A list of attendees' email addresses for the event."
)
reminders: Union[None, bool, List[Dict[str, Any]]] = Field(
default=None,
description=(
"Reminders for the event. "
"Set to True for default reminders, or provide a list like "
"[{'method': 'email', 'minutes': <minutes>}, ...]. "
"Valid methods are 'email' and 'popup'."
),
)
conference_data: Optional[bool] = Field(
default=None, description="Whether to include conference data."
)
color_id: Optional[str] = Field(
default=None,
description=(
"The color ID of the event. None for default. "
"'1': Lavender, '2': Sage, '3': Grape, '4': Flamingo, '5': Banana, "
"'6': Tangerine, '7': Peacock, '8': Graphite, '9': Blueberry, "
"'10': Basil, '11': Tomato."
),
)
transparency: Optional[str] = Field(
default=None,
description=(
"User availability for the event."
"transparent for available and opaque for busy."
),
)
[docs]
class CalendarCreateEvent(CalendarBaseTool): # type: ignore[override, override]
"""Tool that creates an event in Google Calendar."""
name: str = "create_calendar_event"
description: str = (
"Use this tool to create an event. "
"The input must include the summary, start, and end datetime for the event."
)
args_schema: Type[CreateEventSchema] = CreateEventSchema
def _prepare_event(
self,
summary: str,
start_datetime: str,
end_datetime: str,
timezone: str,
recurrence: Optional[Dict[str, Any]] = None,
location: Optional[str] = None,
description: Optional[str] = None,
attendees: Optional[List[str]] = None,
reminders: Union[None, bool, List[Dict[str, Any]]] = None,
conference_data: Optional[bool] = None,
color_id: Optional[str] = None,
transparency: Optional[str] = None,
) -> Dict[str, Any]:
"""Prepare the event body."""
try:
if is_all_day_event(start_datetime, end_datetime):
start = {"date": start_datetime}
end = {"date": end_datetime}
else:
datetime_format = "%Y-%m-%d %H:%M:%S"
start_dt = datetime.strptime(start_datetime, datetime_format)
end_dt = datetime.strptime(end_datetime, datetime_format)
start = {
"dateTime": start_dt.replace(tzinfo=ZoneInfo(timezone)).isoformat(),
"timeZone": timezone,
}
end = {
"dateTime": end_dt.replace(tzinfo=ZoneInfo(timezone)).isoformat(),
"timeZone": timezone,
}
except ValueError as error:
raise ValueError("The datetime format is incorrect.") from error
recurrence_data = None
if recurrence:
if isinstance(recurrence, dict):
recurrence_items = [
f"{k}={v}" for k, v in recurrence.items() if v is not None
]
recurrence_data = "RRULE:" + ";".join(recurrence_items)
attendees_emails: List[Dict[str, str]] = []
if attendees:
email_pattern = r"^[^@]+@[^@]+\.[^@]+$"
for email in attendees:
if not re.match(email_pattern, email):
raise ValueError(f"Invalid email address: {email}")
attendees_emails.append({"email": email})
reminders_info: Dict[str, Union[bool, List[Dict[str, Any]]]] = {}
if reminders is True:
reminders_info.update({"useDefault": True})
elif isinstance(reminders, list):
for reminder in reminders:
if "method" not in reminder or "minutes" not in reminder:
raise ValueError(
"Each reminder must have 'method' and 'minutes' keys."
)
if reminder["method"] not in ["email", "popup"]:
raise ValueError("The reminder method must be 'email' or 'popup")
reminders_info.update({"useDefault": False, "overrides": reminders})
else:
reminders_info.update({"useDefault": False})
conference_data_info = None
if conference_data:
conference_data_info = {
"createRequest": {
"requestId": str(uuid4()),
"conferenceSolutionKey": {"type": "hangoutsMeet"},
}
}
event_body: Dict[str, Any] = {"summary": summary, "start": start, "end": end}
if location:
event_body["location"] = location
if description:
event_body["description"] = description
if recurrence_data:
event_body["recurrence"] = [recurrence_data]
if len(attendees_emails) > 0:
event_body["attendees"] = attendees_emails
if len(reminders_info) > 0:
event_body["reminders"] = reminders_info
if conference_data_info:
event_body["conferenceData"] = conference_data_info
if color_id:
event_body["colorId"] = color_id
if transparency:
event_body["transparency"] = transparency
return event_body
def _run(
self,
summary: str,
start_datetime: str,
end_datetime: str,
timezone: str,
calendar_id: str = "primary",
recurrence: Optional[Dict[str, Any]] = None,
location: Optional[str] = None,
description: Optional[str] = None,
attendees: Optional[List[str]] = None,
reminders: Union[None, bool, List[Dict[str, Any]]] = None,
conference_data: Optional[bool] = None,
color_id: Optional[str] = None,
transparency: Optional[str] = None,
run_manager: Optional[CallbackManagerForToolRun] = None,
) -> str:
"""Run the tool to create an event in Google Calendar."""
try:
body = self._prepare_event(
summary=summary,
start_datetime=start_datetime,
end_datetime=end_datetime,
timezone=timezone,
recurrence=recurrence,
location=location,
description=description,
attendees=attendees,
reminders=reminders,
conference_data=conference_data,
color_id=color_id,
transparency=transparency,
)
conference_version = 1 if conference_data else 0
event = (
self.api_resource.events()
.insert(
calendarId=calendar_id,
body=body,
conferenceDataVersion=conference_version,
)
.execute()
)
return f"Event created: {event.get('htmlLink')}"
except Exception as error:
raise Exception(f"An error occurred: {error}") from error
