"""Interactive compose/confirm cards for email + calendar. Instead of firing an email or booking an event blind, Jarvis opens a *draft card* on the HUD (like the image popup) that Ahmed can eyeball and hand-edit — fix a misheard address, tweak the subject — then Send or Cancel. Either side can drive it: Jarvis (voice) -> email_compose / event_compose (open a card) compose_update (change a field) compose_send / compose_cancel (act on it) Ahmed (HUD) -> edits a field, clicks Send / Cancel / ✕ -> HUD writes control/card_action.json -> ControlWatcher applies it and drops a note in the master's inbox so Jarvis follows up in his own words. The draft lives HERE, in the engine process, shared by the MCP tools (below) and the ControlWatcher. Field edits Ahmed types are stored silently so a later spoken "just send it" sends HIS corrected version. Only Send/Cancel/Close make Jarvis speak — otherwise he'd chatter while Ahmed types. Engine → HUD events (via voice.events.emit): compose {draft_id, to, cc, subject, body, attachments[]} event_card {draft_id, title, start, end, attendees, meet, context[]} card_close {draft_id} Set MCP_COMPOSE=0 to disable (Jarvis falls back to direct email_send). """ from __future__ import annotations import asyncio import os import uuid from datetime import datetime, timedelta, timezone from voice.events import emit # draft_id -> {"kind": "email"|"event", "fields": {...}} _DRAFTS: dict[str, dict] = {} _ORDER: list[str] = [] # creation order, so "latest" resolves a bare command def enabled() -> bool: if os.environ.get("MCP_COMPOSE", "1") == "0": return False from voice import email_control, gcal_control return email_control.enabled() or gcal_control.enabled() def _new_id() -> str: return uuid.uuid4().hex[:8] def latest_id(kind: str | None = None) -> str | None: """Most recent open draft id, optionally filtered by kind.""" for did in reversed(_ORDER): d = _DRAFTS.get(did) if d and (kind is None or d["kind"] == kind): return did return None def get_draft(draft_id: str) -> dict | None: return _DRAFTS.get(draft_id) # --- creation ------------------------------------------------------------- def _parse_attachments(val) -> list[str]: """Normalize an attachments value (list of paths, or a comma-separated string) into a clean list of path strings. Anything else -> [].""" if isinstance(val, (list, tuple)): return [str(p).strip() for p in val if str(p).strip()] if isinstance(val, str): return [p.strip() for p in val.split(",") if p.strip()] return [] def create_email_draft(to: str, cc: str, subject: str, body: str, attachments=None) -> str: did = _new_id() _DRAFTS[did] = {"kind": "email", "fields": { "to": to, "cc": cc, "subject": subject, "body": body, "attachments": _parse_attachments(attachments)}} _ORDER.append(did) _emit_card(did) return did def create_event_draft(title: str, start: str, end: str, attendees: str, meet: bool, context: list[str]) -> str: did = _new_id() _DRAFTS[did] = {"kind": "event", "fields": { "title": title, "start": start, "end": end, "attendees": attendees, "meet": bool(meet)}, "context": context or []} _ORDER.append(did) _emit_card(did) return did def _emit_card(draft_id: str) -> None: d = _DRAFTS.get(draft_id) if not d: return f = d["fields"] if d["kind"] == "email": emit("compose", draft_id=draft_id, to=f.get("to", ""), cc=f.get("cc", ""), subject=f.get("subject", ""), body=f.get("body", ""), attachments=list(f.get("attachments") or [])) else: emit("event_card", draft_id=draft_id, title=f.get("title", ""), start=f.get("start", ""), end=f.get("end", ""), attendees=f.get("attendees", ""), meet=bool(f.get("meet")), context=d.get("context", [])) # --- mutation ------------------------------------------------------------- def update_draft(draft_id: str, fields: dict, reemit: bool) -> dict | None: """Merge `fields` into the draft. reemit=True refreshes the HUD card (use when JARVIS edits); False stores silently (use for Ahmed's own typing, which the HUD already shows).""" d = _DRAFTS.get(draft_id) if not d: return None d["fields"].update({k: v for k, v in fields.items() if v is not None}) if reemit: _emit_card(draft_id) return d def cancel_draft(draft_id: str) -> dict | None: d = _DRAFTS.pop(draft_id, None) if draft_id in _ORDER: _ORDER.remove(draft_id) if d: emit("card_close", draft_id=draft_id) return d # --- dispatch (actually send/create) -------------------------------------- async def send_draft(draft_id: str) -> tuple[bool, str]: """Send the email / create the event for this draft, then close the card. Returns (ok, human_summary).""" d = _DRAFTS.get(draft_id) if not d: return False, "that draft is no longer open." f = d["fields"] try: if d["kind"] == "email": from voice import email_control atts = _parse_attachments(f.get("attachments")) if atts: out = await asyncio.to_thread( email_control._send, f.get("to", ""), f.get("subject", ""), f.get("body", ""), f.get("cc", ""), atts) else: # no attachments -> exact legacy call out = await asyncio.to_thread( email_control._send, f.get("to", ""), f.get("subject", ""), f.get("body", ""), f.get("cc", "")) else: out = await asyncio.to_thread(_create_event, f) cancel_draft(draft_id) # closes the card return True, out except Exception as e: # noqa: BLE001 return False, f"failed: {e}" def _create_event(f: dict) -> str: """Blocking calendar insert — called via asyncio.to_thread.""" from voice import gcal_control svc = gcal_control._service() start = f.get("start", "") end = gcal_control._end_from(start, f.get("end") or None, 30) attendees = [e.strip() for e in str(f.get("attendees", "")).split(",") if e.strip()] body = { "summary": f.get("title", "") or "(untitled)", "start": {"dateTime": start, "timeZone": gcal_control._TZ}, "end": {"dateTime": end, "timeZone": gcal_control._TZ}, } if attendees: body["attendees"] = [{"email": e} for e in attendees] kw = {"calendarId": "primary", "body": body, "sendUpdates": "all" if attendees else "none"} if f.get("meet"): body["conferenceData"] = {"createRequest": { "requestId": uuid.uuid4().hex, "conferenceSolutionKey": {"type": "hangoutsMeet"}}} kw["conferenceDataVersion"] = 1 ev = svc.events().insert(**kw).execute() if f.get("meet"): link = ev.get("hangoutLink", "(no link)") return f"Meeting '{body['summary']}' booked. Meet link: {link}" who = f" — invited {', '.join(attendees)}" if attendees else "" return f"Event '{body['summary']}' booked for {start}{who}." def surrounding_events(start_iso: str) -> list[str]: """Events on the same day as `start_iso`, for the card's context strip.""" try: from voice import gcal_control day = datetime.fromisoformat(start_iso).date() except Exception: # noqa: BLE001 return [] try: svc = gcal_control._service() tmin = datetime(day.year, day.month, day.day, tzinfo=timezone.utc).isoformat() tmax = (datetime(day.year, day.month, day.day, tzinfo=timezone.utc) + timedelta(days=1)).isoformat() res = svc.events().list( calendarId="primary", timeMin=tmin, timeMax=tmax, singleEvents=True, orderBy="startTime", maxResults=12).execute() out = [] for ev in res.get("items", []): when = ev.get("start", {}).get("dateTime") or \ ev.get("start", {}).get("date", "?") hhmm = when[11:16] if "T" in when else "all-day" out.append(f"{hhmm} {ev.get('summary', '(untitled)')}") return out except Exception: # noqa: BLE001 return [] # --- MCP server ----------------------------------------------------------- def _text(msg: str) -> dict: return {"content": [{"type": "text", "text": msg}]} def build_server(): """Build the in-process ``compose`` MCP server.""" from claude_agent_sdk import tool, create_sdk_mcp_server @tool("email_compose", "Open an editable EMAIL card on Ahmed's HUD for him to review before " "sending (preferred over email_send when Ahmed is present). Returns a " "draft_id. Then either he clicks Send, or he tells you to send and " "you call compose_send. `to`/`cc` are comma-separated addresses. " "`attachments` = optional file paths (comma-separated) to stage on " "the card — Ahmed sees them and can add/remove before it sends.", {"to": str, "subject": str, "body": str, "cc": str, "attachments": str}) async def email_compose(args: dict) -> dict: from voice import email_control if not email_control.enabled(): return _text("email is not configured.") did = create_email_draft( str(args.get("to", "")).strip(), str(args.get("cc", "")).strip(), str(args.get("subject", "")), str(args.get("body", "")), args.get("attachments")) return _text(f"Draft {did} is on Ahmed's screen for review. Wait for " f"him to Send/Cancel, or call compose_send('{did}') if he " f"tells you to send it.") @tool("event_compose", "Open an editable CALENDAR event card on Ahmed's HUD to confirm " "before booking. Shows his other events that day for context. Times " "are local ISO like '2026-07-08T15:00:00'. Set meet=true to attach a " "Google Meet link. Returns a draft_id; book it with compose_send.", {"title": str, "start": str, "end": str, "attendees": str, "meet": bool}) async def event_compose(args: dict) -> dict: from voice import gcal_control if not gcal_control.enabled(): return _text("calendar is not configured.") start = str(args.get("start", "")).strip() if not start: return _text("event_compose failed: no start time.") ctx = await asyncio.to_thread(surrounding_events, start) did = create_event_draft( str(args.get("title", "")), start, str(args.get("end", "")).strip(), str(args.get("attendees", "")).strip(), bool(args.get("meet", False)), ctx) note = f" He already has {len(ctx)} event(s) that day." if ctx else "" return _text(f"Event draft {did} is on Ahmed's screen for review.{note} " f"Book it with compose_send('{did}') when he confirms.") @tool("compose_update", "Change a field on an open draft card (email or event) and refresh " "it on Ahmed's screen — e.g. he says 'change the subject' or 'add " "Elie to CC'. Pass draft_id (omit = most recent) and only the fields " "to change. `attachments` (comma-separated file paths) REPLACES the " "draft's attachment list.", {"draft_id": str, "to": str, "cc": str, "subject": str, "body": str, "title": str, "start": str, "end": str, "attendees": str, "meet": bool, "attachments": str}) async def compose_update(args: dict) -> dict: did = str(args.get("draft_id", "")).strip() or latest_id() if not did or not get_draft(did): return _text("no open draft to update.") fields = {k: args[k] for k in ("to", "cc", "subject", "body", "title", "start", "end", "attendees", "meet") if k in args and args[k] != ""} if args.get("attachments") not in (None, ""): fields["attachments"] = _parse_attachments(args["attachments"]) update_draft(did, fields, reemit=True) return _text(f"Updated draft {did}: {', '.join(fields) or '(nothing)'}.") @tool("compose_send", "Send the email / book the event for an open draft card, using its " "CURRENT values (including any edits Ahmed typed). Pass draft_id " "(omit = most recent). Closes the card.", {"draft_id": str}) async def compose_send(args: dict) -> dict: did = str(args.get("draft_id", "")).strip() or latest_id() if not did or not get_draft(did): return _text("no open draft to send.") ok, out = await send_draft(did) return _text(out) @tool("compose_cancel", "Discard an open draft card without sending. Pass draft_id " "(omit = most recent).", {"draft_id": str}) async def compose_cancel(args: dict) -> dict: did = str(args.get("draft_id", "")).strip() or latest_id() if not did or not get_draft(did): return _text("no open draft to cancel.") cancel_draft(did) return _text(f"Discarded draft {did}.") return create_sdk_mcp_server( name="compose", version="1.0.0", tools=[email_compose, event_compose, compose_update, compose_send, compose_cancel])