improved_server.py•11 kB
#!/usr/bin/env python3
"""
Improved MCP server with better tool descriptions (docstrings) and safety mechanisms.
This version helps prevent unintended tool calls.
"""
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP
from tavily import TavilyClient
import os
from dice_roller import DiceRoller
import requests, json
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
load_dotenv()
mcp = FastMCP("improved-cal-server")
client = TavilyClient(os.getenv("TAVILY_API_KEY"))
# Keep the original tools
@mcp.tool()
def web_search(query: str) -> str:
"""Search the web for information. Only use this when the user explicitly asks for web search or external information."""
search_results = client.get_search_context(query=query)
return search_results
@mcp.tool()
def roll_dice(notation: str, num_rolls: int = 1) -> str:
"""Roll dice using standard notation (e.g., '3d6', '2d10+5'). Only use when user asks about dice or random numbers."""
roller = DiceRoller(notation, num_rolls)
return str(roller)
# Improved Cal.com tools with better descriptions
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
@mcp.tool()
def cal_list_event_types() -> dict:
"""
INFORMATION ONLY: List the user's available Cal.com event types.
Use this when user asks: 'What event types do I have?' or 'Show my meeting types'
This is READ-ONLY and safe to call.
"""
data = _get("/event-types")
return {"event_types": data.get("event_types", [])}
@mcp.tool()
def cal_list_bookings(status: str = "upcoming") -> dict:
"""
INFORMATION ONLY: List the user's bookings with given status.
Use this when user asks: 'Show my bookings' or 'What meetings do I have?'
Status options: 'upcoming' (default), 'cancelled', 'all'
This is READ-ONLY and safe to call.
"""
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:
"""
INFORMATION ONLY: Check availability for a specific event type on a date.
Use this when user asks: 'What times are available on [date]?' or 'Check availability'
Format: date as YYYY-MM-DD, event_type_id as integer
This is READ-ONLY and safe to call.
"""
_require_key()
try:
tz = ZoneInfo(time_zone)
start_local = datetime.fromisoformat(f"{date}T00:00:00").replace(tzinfo=tz)
end_local = start_local + timedelta(days=1)
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 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:
"""
⚠️ CREATES NEW BOOKING: This will create a new meeting appointment.
ONLY use this when user explicitly asks to 'book', 'schedule', or 'create' a meeting.
NEVER call this when user only asks for information about bookings.
Required: start_iso (UTC format like '2025-08-10T16:00:00Z'), event_type_id, user_id,
time_zone, attendee_name, attendee_email
⚠️ This creates actual calendar entries - use with caution!
"""
# Build location object
if location and (location.startswith("integrations:") or location.startswith("http")):
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
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:
result = _post("/bookings", body)
return {
"ok": True,
"message": f"✅ Successfully created booking for {attendee_name}",
"booking": result
}
except Exception as e:
return {"ok": False, "error": str(e), "hint": "Check availability first"}
@mcp.tool()
def cal_cancel_booking(uid: str, reason: str = "Cancelled via MCP") -> dict:
"""
🚨 DESTRUCTIVE ACTION: This will permanently cancel an existing booking.
ONLY use when user explicitly asks to 'cancel' or 'delete' a specific meeting.
NEVER call this unless the user clearly wants to cancel something.
Required: uid (booking unique identifier)
🚨 This is irreversible - the meeting will be cancelled!
"""
try:
data = _get("/bookings", params={"uid": uid})
bookings = data.get("bookings", [])
if not bookings:
return {"ok": False, "error": f"No booking found for uid={uid}"}
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"}
_delete(f"/bookings/{booking_id}", params={"reason": reason})
return {
"ok": True,
"message": f"🚨 Successfully cancelled booking {uid}",
"id": booking_id,
"uid": uid,
"reason": reason
}
except Exception as e:
return {"ok": False, "error": str(e)}
@mcp.tool()
def cal_cancel_booking_by_details(
date: str,
attendee: str,
time_zone: str | None = None,
reason: str = "Cancelled via MCP"
) -> dict:
"""
🚨 DESTRUCTIVE ACTION: Cancel a booking by finding it with date and attendee details.
ONLY use when user explicitly asks to cancel a meeting and provides date/attendee info.
NEVER call this unless the user clearly wants to cancel something.
🚨 This is irreversible - the meeting will be cancelled!
"""
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 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.")
try:
target_date = _parse_date(date)
fallback_tz = ZoneInfo(time_zone) if time_zone else ZoneInfo("America/New_York")
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
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 be more specific",
"matches": [{"id": c.get("id"), "uid": c.get("uid"), "title": c.get("title")} 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,
"message": f"🚨 Successfully cancelled booking for {attendee} on {date}",
"id": booking_id,
"uid": booking.get("uid"),
"reason": reason
}
except Exception as e:
return {"ok": False, "error": str(e)}
if __name__ == "__main__":
mcp.run(transport="stdio")