Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions google/google-calendar-mcp/.python-version
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
3.13
Empty file.
18 changes: 18 additions & 0 deletions google/google-calendar-mcp/client.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
import asyncio
from fastmcp import Client


async def example():
async with Client("http://127.0.0.1:9000/mcp/google-drive") as client:
res = await client.ping()
res = await client.list_tools()
for r in res:
print(r.name)
# rres = await client.call_tool(
# name="get_weather",
# arguments={"city": "New York", "cred_token": "1234567890"},
# )
# print(rres)

if __name__ == "__main__":
asyncio.run(example())
12 changes: 12 additions & 0 deletions google/google-calendar-mcp/pyproject.toml
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
[project]
name = "google-calendar-mcp"
version = "0.1.0"
description = "Obot's google calendar mcp server"
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
"fastmcp>=2.8.0",
"google>=3.0.0",
"google-api-python-client>=2.172.0",
"rfc3339-validator>=0.1.4",
]
710 changes: 710 additions & 0 deletions google/google-calendar-mcp/server.py

Large diffs are not rendered by default.

277 changes: 277 additions & 0 deletions google/google-calendar-mcp/tools/event.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,277 @@
from datetime import datetime, timezone
from tools.helper import (
setup_logger,
get_obot_user_timezone,
)
from googleapiclient.errors import HttpError
from zoneinfo import available_timezones, ZoneInfo

logger = setup_logger(__name__)

DEFAULT_MAX_RESULTS = 250
GOOGLE_EVENT_TYPE_OPTIONS = [
"birthday",
"default",
"focusTime",
"fromGmail",
"outOfOffice",
"workingLocation",
]
MOVABLE_EVENT_TYPES = ["default"]
CALENDAR_EVENT_TYPE_RULES = {
"default": {
"fully_updatable": True,
"updatable_properties": [
"summary",
"description",
"location",
"start",
"end",
"attendees",
"recurrence",
"reminders",
"colorId",
"visibility",
"transparency",
"status",
"extendedProperties",
"attachments",
"guestsCanInviteOthers",
"guestsCanModify",
"guestsCanSeeOtherGuests",
"source",
"sequence",
# All standard properties can be updated
],
"restrictions": [],
"notes": "Most flexible event type with virtually no restrictions on updates.",
},
"fromGmail": {
"fully_updatable": False,
"updatable_properties": [],
"restrictions": [
"Cannot create new fromGmail events via the API",
"Cannot change the organizer",
"Cannot modify core properties like summary, description, location, or times",
],
"notes": "Limited to updating UI and preference properties only.",
},
"birthday": {
"fully_updatable": False,
"updatable_properties": [
"colorId",
"summary",
"reminders",
"start",
"end",
],
"restrictions": [
"Cannot modify birthdayProperties",
"Start/end time updates must remain all-day events spanning exactly one day",
"Timing updates are restricted if linked to a contact",
"Cannot change the organizer",
"Cannot create custom birthday properties via the API",
],
"notes": "Use People API for comprehensive contact birthday management.",
},
"focusTime": {
"fully_updatable": False,
"updatable_properties": [
# Standard properties
"summary",
"description",
"start",
"end",
"reminders",
"colorId",
"visibility",
"transparency",
# Focus time specific properties
"focusTimeProperties",
"focusTimeProperties.autoDeclineMode",
"focusTimeProperties.chatStatus",
"focusTimeProperties.declineMessage",
],
"restrictions": [
"Only available on primary calendars",
"Only for specific Google Workspace users",
"Cannot be created on secondary calendars",
],
"notes": "Used for dedicated focus periods.",
},
"outOfOffice": {
"fully_updatable": False,
"updatable_properties": [
# Standard properties
"summary",
"description",
"start",
"end",
"reminders",
"colorId",
"visibility",
"transparency",
# Out of office specific properties
"outOfOfficeProperties",
"outOfOfficeProperties.autoDeclineMode",
"outOfOfficeProperties.declineMessage",
],
"restrictions": [
"Only available on primary calendars",
"Only for specific Google Workspace users",
"Cannot be created on secondary calendars",
],
"notes": "Represents time away from work.",
},
"workingLocation": {
"fully_updatable": False,
"updatable_properties": [
# Standard properties
"summary",
"description",
"start",
"end",
"reminders",
"colorId",
"visibility",
"transparency",
# Working location specific properties
"workingLocationProperties",
"workingLocationProperties.type",
"workingLocationProperties.homeOffice",
"workingLocationProperties.customLocation",
"workingLocationProperties.customLocation.label",
"workingLocationProperties.officeLocation",
"workingLocationProperties.officeLocation.buildingId",
"workingLocationProperties.officeLocation.floorId",
"workingLocationProperties.officeLocation.floorSectionId",
"workingLocationProperties.officeLocation.deskId",
"workingLocationProperties.officeLocation.label",
],
"restrictions": [
"Only available on primary calendars",
"Only for specific Google Workspace users",
"Cannot be created on secondary calendars",
],
"notes": "Indicates where someone is working.",
},
}


def can_update_property(event_type, property_name):
"""
Check if a specific property can be updated for a given event type.

Args:
event_type (str): The event type ('default', 'fromGmail', etc.)
property_name (str): The property to check

Returns:
bool: True if the property can be updated, False otherwise
"""
if event_type not in CALENDAR_EVENT_TYPE_RULES:
raise ValueError(f"Unknown event type: {event_type}")

# Default events can update all standard properties
if event_type == "default":
return True

# For other event types, check the specific list
return (
property_name in CALENDAR_EVENT_TYPE_RULES[event_type]["updatable_properties"]
)


def _get_event_type_restrictions(event_type):
"""
Get the list of restrictions for a given event type.

Args:
event_type (str): The event type ('default', 'fromGmail', etc.)

Returns:
list: List of restriction strings
"""
if event_type not in CALENDAR_EVENT_TYPE_RULES:
raise ValueError(f"Unknown event type: {event_type}")

return CALENDAR_EVENT_TYPE_RULES[event_type]["restrictions"]


def _get_updatable_properties(event_type):
if event_type not in CALENDAR_EVENT_TYPE_RULES:
raise ValueError(f"Unknown event type: {event_type}")

return CALENDAR_EVENT_TYPE_RULES[event_type]["updatable_properties"]


# Private helper functions
def is_valid_date(date_string: str) -> bool:
try:
datetime.strptime(date_string, "%Y-%m-%d")
return True
except ValueError:
return False


def is_valid_iana_timezone(timezone: str) -> bool:
return timezone in available_timezones()


def _is_valid_recurrence_line_syntax(line: str) -> bool:
return any(
line.startswith(prefix) for prefix in ("RRULE:", "EXRULE:", "RDATE", "EXDATE")
)


def get_current_time_rfc3339():
try:
timezone = ZoneInfo(get_obot_user_timezone())
except ValueError:
# Invalid timezone, fallback to UTC
timezone = ZoneInfo("UTC")
return datetime.now(timezone).isoformat()


def validate_recurrence_list(recurrence_list: list[str]) -> list[str]:
"""
Parse a string into a list of recurrence rules.

Args:
recurrence (str): The recurrence string to parse

Raises:
ValueError: If the recurrence string is not a valid JSON array of strings, where each string is an RRULE, EXRULE, RDATE, or EXDATE line as defined by the RFC5545.
ValueError: If the recurrence string is not a valid recurrence rule syntax.

Returns:
list: A list of recurrence rules
"""

for r in recurrence_list:
if not _is_valid_recurrence_line_syntax(r):
raise ValueError(
f"Invalid recurrence rule: {r}. It must be a valid RRULE, EXRULE, RDATE, or EXDATE string."
)

return recurrence_list

def get_current_user_email(service) -> str:
"""
Gets the email of the current user, by getting the user_id of the primary calendar.
"""
user_info = service.calendars().get(calendarId="primary").execute()
return user_info["id"]


def has_calendar_write_access(service, calendar_id: str) -> bool:
"Validate if the user has writer access to the calendar"
try:
calendar = service.calendarList().get(calendarId=calendar_id).execute()
return calendar.get("accessRole") in ("owner", "writer")
except HttpError as e:
if e.resp.status == 403:
return False
raise Exception(
f"HttpError retrieving calendar For validating user access to {calendar_id}: {e}"
)
73 changes: 73 additions & 0 deletions google/google-calendar-mcp/tools/helper.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
import sys
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
import os
import logging
from fastmcp.exceptions import ToolError


def setup_logger(name, tool_name: str = "Google Calendar MCP server"):
"""Setup a logger that writes to sys.stderr. This will eventually show up in GPTScript's debugging logs.

Args:
name (str): The name of the logger.

Returns:
logging.Logger: The logger.
"""
# Create a logger
logger = logging.getLogger(name)
logger.setLevel(logging.DEBUG) # Set the logging level

# Create a stream handler that writes to sys.stderr
stderr_handler = logging.StreamHandler(sys.stderr)

# Create a log formatter
formatter = logging.Formatter(
f"[{tool_name} Debugging Log]: %(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
stderr_handler.setFormatter(formatter)

# Add the handler to the logger
logger.addHandler(stderr_handler)

return logger


logger = setup_logger(__name__)


def str_to_bool(value):
"""Convert a string to a boolean."""
return str(value).lower() in ("true", "1", "yes")


def get_client(cred_token: str, service_name: str = "calendar", version: str = "v3"):
creds = Credentials(token=cred_token)
try:
service = build(serviceName=service_name, version=version, credentials=creds)
return service
except HttpError as err:
raise ToolError(f"Failed to build {service_name} client. HttpError: {err}")


def get_obot_user_timezone():
return os.getenv("OBOT_USER_TIMEZONE", "UTC").strip()


def get_user_timezone(service):
"""Fetches the authenticated user's time zone from User's Google Calendar settings."""
try:
settings = service.settings().get(setting="timezone").execute()
return settings.get(
"value", get_obot_user_timezone()
) # Default to Obot's user timezone if not found
except HttpError as err:
if err.status_code == 403:
raise Exception(f"HttpError retrieving user timezone: {err}")
logger.error(f"HttpError retrieving user timezone: {err}")
return "UTC"
except Exception as e:
logger.error(f"Exception retrieving user timezone: {e}")
return "UTC"
Loading
Loading