# Copyright (c) 2025 Dedalus Labs, Inc. and its contributors
# SPDX-License-Identifier: MIT
"""Google Calendar API operations for MCP server.
Read-only access to the Google Calendar API v3 using OAuth Bearer tokens.
This repo supports two ways to provide auth:
- Preferred: run `uv run python -m src.gcal_auth` once to store tokens locally,
then gcal-mcp will auto-refresh tokens as needed.
- Advanced/manual: set `GCAL_ACCESS_TOKEN` directly.
"""
from typing import Any
from urllib.parse import urlencode
from dotenv import load_dotenv
from pydantic import BaseModel
from dedalus_mcp import HttpMethod, HttpRequest, get_context, tool
from dedalus_mcp.auth import Connection, SecretKeys
load_dotenv()
# --- Connection --------------------------------------------------------------
def get_gcal_connection(*, interactive_auth: bool = False) -> Connection:
"""Create the Dedalus connection for Google Calendar.
Args:
interactive_auth: If True, may open a browser to authenticate when
tokens are missing. Use this for local CLI flows, not headless
deployments.
"""
# Ensure token is available (and refresh it if possible).
from .gcal_oauth import ensure_gcal_access_token
ensure_gcal_access_token(interactive=interactive_auth)
return Connection(
name="gcal",
secrets=SecretKeys(token="GCAL_ACCESS_TOKEN"),
base_url="https://www.googleapis.com/calendar/v3",
auth_header_format="Bearer {api_key}",
)
# --- Response Models ---------------------------------------------------------
class GCalResult(BaseModel):
"""Generic Google Calendar API result."""
success: bool
data: Any = None
error: str | None = None
class Calendar(BaseModel):
"""Google Calendar."""
id: str
summary: str
description: str | None = None
time_zone: str | None = None
access_role: str | None = None
class Event(BaseModel):
"""Google Calendar event."""
id: str
summary: str | None = None
description: str | None = None
location: str | None = None
start: dict[str, str] | None = None
end: dict[str, str] | None = None
status: str | None = None
html_link: str | None = None
attendees: list[dict[str, Any]] | None = None
organizer: dict[str, str] | None = None
recurrence: list[str] | None = None
class FreeBusyInfo(BaseModel):
"""Free/busy information for a calendar."""
calendar_id: str
busy: list[dict[str, str]]
# --- Helper ------------------------------------------------------------------
async def _request(
method: HttpMethod,
path: str,
params: dict[str, Any] | None = None,
body: dict[str, Any] | None = None,
) -> GCalResult:
"""Make a Google Calendar API request via the enclave dispatch."""
ctx = get_context()
# Build path with query params
if params:
query_string = urlencode({k: v for k, v in params.items() if v is not None})
if query_string:
path = f"{path}?{query_string}"
request = HttpRequest(method=method, path=path, body=body)
response = await ctx.dispatch("gcal", request)
if response.success:
return GCalResult(success=True, data=response.response.body)
msg = response.error.message if response.error else "Request failed"
return GCalResult(success=False, error=msg)
# --- Calendar Tools ----------------------------------------------------------
@tool(description="List all calendars accessible by the user")
async def gcal_list_calendars() -> GCalResult:
"""List all calendars in the user's calendar list.
Returns calendars with their IDs, summaries, descriptions, and access roles.
"""
return await _request(
HttpMethod.GET,
"/users/me/calendarList",
params={
"minAccessRole": "reader",
},
)
@tool(description="Get details of a specific calendar")
async def gcal_get_calendar(calendar_id: str) -> GCalResult:
"""Get metadata for a specific calendar.
Args:
calendar_id: The calendar ID (use "primary" for the main calendar)
"""
return await _request(
HttpMethod.GET,
f"/calendars/{calendar_id}",
)
# --- Event Tools -------------------------------------------------------------
@tool(description="List events from a calendar")
async def gcal_list_events(
calendar_id: str = "primary",
time_min: str | None = None,
time_max: str | None = None,
max_results: int = 25,
single_events: bool = True,
order_by: str = "startTime",
) -> GCalResult:
"""List events from a calendar within a time range.
Args:
calendar_id: The calendar ID (use "primary" for the main calendar)
time_min: Start of time range in RFC3339 format (e.g., "2024-01-01T00:00:00Z")
time_max: End of time range in RFC3339 format (e.g., "2024-12-31T23:59:59Z")
max_results: Maximum number of events to return (1-2500, default 25)
single_events: Whether to expand recurring events into instances (default True)
order_by: Order of events ("startTime" or "updated")
"""
max_results = max(1, min(2500, max_results))
params: dict[str, Any] = {
"maxResults": str(max_results),
"singleEvents": str(single_events).lower(),
}
if time_min:
params["timeMin"] = time_min
if time_max:
params["timeMax"] = time_max
if single_events:
params["orderBy"] = order_by
return await _request(
HttpMethod.GET,
f"/calendars/{calendar_id}/events",
params=params,
)
@tool(description="Get a specific event by ID")
async def gcal_get_event(calendar_id: str, event_id: str) -> GCalResult:
"""Get details of a specific event.
Args:
calendar_id: The calendar ID (use "primary" for the main calendar)
event_id: The event ID to retrieve
"""
return await _request(
HttpMethod.GET,
f"/calendars/{calendar_id}/events/{event_id}",
)
@tool(description="Search for events by text query")
async def gcal_search_events(
query: str,
calendar_id: str = "primary",
time_min: str | None = None,
time_max: str | None = None,
max_results: int = 25,
) -> GCalResult:
"""Search for events matching a text query.
Args:
query: Free text search query (searches summary, description, location, etc.)
calendar_id: The calendar ID (use "primary" for the main calendar)
time_min: Start of time range in RFC3339 format
time_max: End of time range in RFC3339 format
max_results: Maximum number of events to return (1-2500, default 25)
"""
max_results = max(1, min(2500, max_results))
params: dict[str, Any] = {
"q": query,
"maxResults": str(max_results),
"singleEvents": "true",
"orderBy": "startTime",
}
if time_min:
params["timeMin"] = time_min
if time_max:
params["timeMax"] = time_max
return await _request(
HttpMethod.GET,
f"/calendars/{calendar_id}/events",
params=params,
)
@tool(description="Get instances of a recurring event")
async def gcal_get_event_instances(
calendar_id: str,
event_id: str,
time_min: str | None = None,
time_max: str | None = None,
max_results: int = 25,
) -> GCalResult:
"""Get all instances of a recurring event.
Args:
calendar_id: The calendar ID (use "primary" for the main calendar)
event_id: The ID of the recurring event
time_min: Start of time range in RFC3339 format
time_max: End of time range in RFC3339 format
max_results: Maximum number of instances to return (1-2500, default 25)
"""
max_results = max(1, min(2500, max_results))
params: dict[str, Any] = {
"maxResults": str(max_results),
}
if time_min:
params["timeMin"] = time_min
if time_max:
params["timeMax"] = time_max
return await _request(
HttpMethod.GET,
f"/calendars/{calendar_id}/events/{event_id}/instances",
params=params,
)
# --- Free/Busy Tools ---------------------------------------------------------
@tool(description="Query free/busy information for calendars")
async def gcal_get_freebusy(
time_min: str,
time_max: str,
calendar_ids: list[str] | None = None,
time_zone: str | None = None,
) -> GCalResult:
"""Query free/busy information for one or more calendars.
Args:
time_min: Start of time range in RFC3339 format (e.g., "2024-01-01T00:00:00Z")
time_max: End of time range in RFC3339 format (e.g., "2024-01-31T23:59:59Z")
calendar_ids: List of calendar IDs to query (defaults to ["primary"])
time_zone: IANA timezone (e.g., "America/Los_Angeles")
"""
if calendar_ids is None:
calendar_ids = ["primary"]
body: dict[str, Any] = {
"timeMin": time_min,
"timeMax": time_max,
"items": [{"id": cal_id} for cal_id in calendar_ids],
}
if time_zone:
body["timeZone"] = time_zone
return await _request(
HttpMethod.POST,
"/freeBusy",
body=body,
)
# --- Settings Tools ----------------------------------------------------------
@tool(description="Get user's calendar settings")
async def gcal_get_settings() -> GCalResult:
"""Get all calendar settings for the current user.
Returns settings like timezone, date format, time format, etc.
"""
return await _request(
HttpMethod.GET,
"/users/me/settings",
)
@tool(description="Get a specific calendar setting")
async def gcal_get_setting(setting_id: str) -> GCalResult:
"""Get a specific calendar setting.
Args:
setting_id: The setting ID (e.g., "timezone", "dateFieldOrder", "timeZone")
"""
return await _request(
HttpMethod.GET,
f"/users/me/settings/{setting_id}",
)
# --- Color Tools -------------------------------------------------------------
@tool(description="Get available calendar and event colors")
async def gcal_get_colors() -> GCalResult:
"""Get the available color definitions for calendars and events.
Returns color IDs and their corresponding background/foreground colors.
"""
return await _request(
HttpMethod.GET,
"/colors",
)
# --- Export ------------------------------------------------------------------
gcal_tools = [
# Calendars
gcal_list_calendars,
gcal_get_calendar,
# Events
gcal_list_events,
gcal_get_event,
gcal_search_events,
gcal_get_event_instances,
# Free/Busy
gcal_get_freebusy,
# Settings
gcal_get_settings,
gcal_get_setting,
# Colors
gcal_get_colors,
]
