myserver.py•10.6 kB
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP
from tavily import TavilyClient
import os
from dice_roller import DiceRoller
# Add imports near the top
import requests, json
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
load_dotenv()
mcp = FastMCP("mcp-server")
client = TavilyClient(os.getenv("TAVILY_API_KEY"))
@mcp.tool()
def web_search(query: str) -> str:
"""Search the web for information about the given query"""
search_results = client.get_search_context(query=query)
return search_results
@mcp.tool()
def roll_dice(notation: str, num_rolls: int = 1) -> str:
"""Roll the dice with the given notation"""
roller = DiceRoller(notation, num_rolls)
return str(roller)
"""
Add your own tool here, and then use it through Cursor!
"""
# Add Cal.com config and small HTTP helpers below your mcp/client setup
CAL_BASE = "https://api.cal.com/v1"
CALCOM_API_KEY = os.getenv("CALCOM_API_KEY")
def _require_key():
if not CALCOM_API_KEY:
raise RuntimeError("Missing CALCOM_API_KEY env var")
def _get(path: str, params: dict | None = None):
_require_key()
p = {"apiKey": CALCOM_API_KEY, **(params or {})}
r = requests.get(f"{CAL_BASE}{path}", params=p, timeout=15)
r.raise_for_status()
return r.json()
def _post(path: str, body: dict):
_require_key()
url = f"{CAL_BASE}{path}"
r = requests.post(f"{url}?apiKey={CALCOM_API_KEY}",
headers={"Content-Type": "application/json"},
data=json.dumps(body), timeout=20)
r.raise_for_status()
return r.json()
def _delete(path: str, params: dict | None = None):
_require_key()
p = {"apiKey": CALCOM_API_KEY, **(params or {})}
r = requests.delete(f"{CAL_BASE}{path}", params=p, timeout=15)
r.raise_for_status()
return True
# Tools
@mcp.tool()
def cal_list_event_types() -> dict:
"""List available Cal.com event types."""
data = _get("/event-types")
return {"event_types": data.get("event_types", [])}
@mcp.tool()
def cal_list_bookings(status: str = "upcoming") -> dict:
"""List bookings. status: upcoming | cancelled | all (uses 'upcoming' by default)."""
params = {"status": status} if status != "all" else {}
data = _get("/bookings", params)
return {"bookings": data.get("bookings", [])}
@mcp.tool()
def cal_get_availability(
event_type_id: int,
date: str,
time_zone: str = "America/New_York"
) -> dict:
"""Get availability for an event type on a given date (YYYY-MM-DD).
Uses /slots which requires startTime and endTime. We compute the full day's
window in the provided time zone and send ISO8601 UTC.
"""
_require_key()
try:
# Compute start/end of the given date in the requested TZ
tz = ZoneInfo(time_zone)
start_local = datetime.fromisoformat(f"{date}T00:00:00").replace(tzinfo=tz)
end_local = start_local + timedelta(days=1)
# Convert to UTC ISO with Z
start_utc = start_local.astimezone(ZoneInfo("UTC")).isoformat().replace("+00:00", "Z")
end_utc = end_local.astimezone(ZoneInfo("UTC")).isoformat().replace("+00:00", "Z")
params = {
"eventTypeId": event_type_id,
"startTime": start_utc,
"endTime": end_utc,
"timeZone": time_zone,
}
data = _get("/slots", params)
return {"ok": True, "endpoint": "/slots", "data": data}
except requests.exceptions.HTTPError as e:
detail = {"ok": False, "status_code": e.response.status_code if getattr(e, "response", None) else None, "error": str(e)}
if getattr(e, "response", None) is not None:
try:
detail["response_json"] = e.response.json()
except Exception:
detail["response_text"] = e.response.text
return detail
except Exception as e:
return {"ok": False, "error": str(e)}
@mcp.tool()
def cal_create_booking(
start_iso: str,
event_type_id: int,
user_id: int,
time_zone: str,
attendee_name: str,
attendee_email: str,
location: str = "",
status: str | None = None
) -> dict:
"""Create a booking. start_iso like 2025-08-10T16:00:00Z (UTC)."""
# Build responses with location formatting that avoids common Cal.com 400s
# Known good defaults: leave location value/optionValue empty strings unless a simple city/address is provided
if location and (location.startswith("integrations:") or location.startswith("http")):
# Force empty to avoid backend rejecting integration/URL here
location_obj = {"value": "", "optionValue": ""}
elif location:
location_obj = {"value": location, "optionValue": ""}
else:
location_obj = {"value": "", "optionValue": ""}
# Try to compute end time from event type length (minutes)
end_iso = None
try:
all_types = _get("/event-types").get("event_types", [])
et = next((t for t in all_types if t.get("id") == event_type_id), None)
if et and et.get("length") and isinstance(et["length"], int):
start_dt = datetime.fromisoformat(start_iso.replace("Z", "+00:00"))
end_dt = start_dt + timedelta(minutes=et["length"])
end_iso = end_dt.isoformat().replace("+00:00", "Z")
except Exception:
pass
body: dict = {
"start": start_iso,
"eventTypeId": event_type_id,
"responses": {
"name": attendee_name,
"email": attendee_email,
"location": location_obj
},
"timeZone": time_zone,
"language": "en",
"userId": user_id,
"metadata": {}
}
if end_iso:
body["end"] = end_iso
if status:
body["status"] = status
try:
return _post("/bookings", body)
except requests.exceptions.HTTPError as e:
detail = {"ok": False, "error": str(e)}
if getattr(e, "response", None) is not None:
detail["status_code"] = e.response.status_code
try:
detail["response_json"] = e.response.json()
except Exception:
detail["response_text"] = e.response.text
detail["hint"] = "Try a different time; ensure slot is available; leave location blank; status optional."
return detail
except requests.exceptions.RequestException as e:
# Surface error details without crashing the tool
return {
"ok": False,
"error": str(e),
"hint": "Try a different time or leave location blank; ensure start is within event type availability."
}
@mcp.tool()
def cal_cancel_booking(uid: str, reason: str = "Cancelled via MCP") -> dict:
"""Cancel a booking by UID."""
data = _get("/bookings", params={"uid": uid})
bookings = data.get("bookings", [])
if not bookings:
return {"ok": False, "error": f"No booking found for uid={uid}"}
# Ensure we cancel the booking matching the exact UID (Cal.com may return multiple)
match = next((b for b in bookings if b.get("uid") == uid), None)
booking_id = match.get("id") if match else None
if not booking_id:
return {"ok": False, "error": "No numeric booking id found for provided UID", "uids": [b.get("uid") for b in bookings]}
_delete(f"/bookings/{booking_id}", params={"reason": reason})
return {"ok": True, "id": booking_id, "uid": uid, "reason": reason}
@mcp.tool()
def cal_cancel_booking_by_details(
date: str,
attendee: str,
time_zone: str | None = None,
reason: str = "Cancelled via MCP"
) -> dict:
"""Cancel a booking by human details (date and attendee name/email).
- date: accepts YYYY-MM-DD or M/D or MM/DD (assumed in the host's time zone if booking has none)
- attendee: substring match on attendee name or email (case-insensitive)
- time_zone: fallback TZ for date matching if booking.user.timeZone is missing
"""
def _parse_date(d: str) -> datetime.date:
d = d.strip()
for fmt in ("%Y-%m-%d", "%m/%d/%Y", "%m/%d", "%m-%d-%Y", "%m-%d"):
try:
dt = datetime.strptime(d, fmt)
# If no year provided, use current year
if fmt in ("%m/%d", "%m-%d"):
dt = dt.replace(year=datetime.now().year)
return dt.date()
except ValueError:
continue
raise ValueError("Unrecognized date format. Use YYYY-MM-DD or MM/DD.")
target_date = _parse_date(date)
fallback_tz = ZoneInfo(time_zone) if time_zone else ZoneInfo("America/New_York")
# Fetch bookings without status filter (Cal.com v1: omit 'status=all')
data = _get("/bookings")
bookings = data.get("bookings", [])
if not bookings:
return {"ok": False, "error": "No bookings found"}
attendee_l = attendee.lower()
candidates = []
for b in bookings:
# Match attendee
attendees = b.get("attendees", [])
match_attendee = False
for a in attendees:
name = (a.get("name") or "").lower()
email = (a.get("email") or "").lower()
if attendee_l in name or attendee_l in email:
match_attendee = True
break
if not match_attendee:
continue
# Match date in booking's user TZ if present, else provided fallback
start_str = b.get("startTime")
if not start_str:
continue
try:
start_utc = datetime.fromisoformat(start_str.replace("Z", "+00:00"))
user_tz = ZoneInfo((b.get("user") or {}).get("timeZone") or fallback_tz.key)
local_date = start_utc.astimezone(user_tz).date()
if local_date == target_date:
candidates.append(b)
except Exception:
continue
if not candidates:
return {"ok": False, "error": "No matching booking found for given date/attendee"}
if len(candidates) > 1:
return {
"ok": False,
"error": "Multiple matches; please refine",
"matches": [{"id": c.get("id"), "uid": c.get("uid"), "title": c.get("title"), "startTime": c.get("startTime")} for c in candidates]
}
booking = candidates[0]
booking_id = booking.get("id")
if not booking_id:
return {"ok": False, "error": "Selected booking missing numeric id"}
_delete(f"/bookings/{booking_id}", params={"reason": reason})
return {"ok": True, "id": booking_id, "uid": booking.get("uid"), "reason": reason}
if __name__ == "__main__":
mcp.run(transport="stdio")