main.py

import json
import os
import shutil
import signal
import subprocess
import sys
from pathlib import Path
from typing import Any

from mcp.server.fastmcp import FastMCP


DEFAULT_TIMEOUT_SECONDS = int(os.environ.get("HEY_MCP_TIMEOUT", "60"))


def resolve_hey_binary() -> str:
    configured = os.environ.get("HEY_CLI_BIN") or os.environ.get("HEY_BIN")
    candidates = [
        configured,
        "/usr/local/bin/hey",
        "/usr/local/sbin/hey",
        "/opt/homebrew/bin/hey",
        str(Path.home() / ".local/bin/hey"),
        shutil.which("hey"),
        str(Path.home() / "dev/hey/cli/bin/hey"),
    ]

    for candidate in candidates:
        if candidate and os.path.isfile(candidate) and os.access(candidate, os.X_OK):
            return candidate

    if configured:
        return configured
    return "hey"


HEY = resolve_hey_binary()
mcp = FastMCP("hey")


def parse_output(stdout: str) -> dict[str, Any] | list[Any] | str:
    try:
        return json.loads(stdout)
    except json.JSONDecodeError:
        return stdout


def run_hey(
    *args: str,
    json_output: bool = True,
    timeout_seconds: int | None = None,
) -> dict[str, Any] | list[Any] | str:
    """Invoke the resolved host HEY binary and return parsed output."""
    command_args = [str(arg) for arg in args]
    cmd = [HEY, *command_args]
    if json_output and "--json" not in command_args and "--agent" not in command_args:
        cmd.append("--json")

    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout_seconds or DEFAULT_TIMEOUT_SECONDS,
        )
    except FileNotFoundError:
        return {
            "ok": False,
            "error": True,
            "code": "hey_not_found",
            "summary": "Could not find the host HEY binary",
            "binary": HEY,
            "hint": "Install hey or set HEY_CLI_BIN to the host binary path.",
        }
    except subprocess.TimeoutExpired as exc:
        return {
            "ok": False,
            "error": True,
            "code": "hey_timeout",
            "summary": f"hey command timed out after {exc.timeout} seconds",
            "stdout": (exc.stdout or "").strip(),
            "stderr": (exc.stderr or "").strip(),
        }

    stdout = result.stdout.strip()
    stderr = result.stderr.strip()

    if result.returncode != 0:
        parsed = parse_output(stdout) if stdout else {}
        if isinstance(parsed, dict):
            parsed.setdefault("ok", False)
            parsed["error"] = True
            parsed["exit_code"] = result.returncode
            if stderr:
                parsed["stderr"] = stderr
            return parsed
        return {
            "ok": False,
            "error": True,
            "code": "hey_command_failed",
            "exit_code": result.returncode,
            "stderr": stderr,
            "stdout": stdout,
        }

    if not stdout:
        return {"ok": True, "data": None}
    return parse_output(stdout)


def add_value(args: list[str], flag: str, value: str | int | None) -> None:
    if value is not None:
        args.extend([flag, str(value)])


def add_bool(args: list[str], flag: str, value: bool | None) -> None:
    if value is not None:
        args.append(f"{flag}={str(value).lower()}")


def add_repeatable(args: list[str], flag: str, values: list[str] | None) -> None:
    if values:
        for value in values:
            args.extend([flag, str(value)])


def add_pagination(args: list[str], limit: int | None, fetch_all: bool) -> None:
    add_value(args, "--limit", limit)
    if fetch_all:
        args.append("--all")


def expand_output_path(path: str) -> str:
    return os.path.abspath(os.path.expanduser(path))


@mcp.tool()
def run_hey_command(
    args: list[str],
    json_output: bool = True,
    timeout_seconds: int | None = None,
) -> Any:
    """Fallback MCP tool for unsupported non-interactive HEY operations.

    Prefer the typed MCP tools when they cover the task. This backstop keeps the
    MCP server usable when the host HEY client gains a new operation before this server adds a
    dedicated wrapper.

    Args:
        args: Argument tokens without the leading `hey`, for example ["boxes"].
        json_output: Append --json unless args already includes --json or --agent.
        timeout_seconds: Optional command timeout override.
    """
    return run_hey(*args, json_output=json_output, timeout_seconds=timeout_seconds)


# Email


@mcp.tool()
def list_boxes(limit: int | None = None, fetch_all: bool = False) -> Any:
    """List HEY mailboxes.

    Args:
        limit: Maximum number of boxes to return.
        fetch_all: Fetch all pages, overriding limit.
    """
    args = ["boxes"]
    add_pagination(args, limit, fetch_all)
    return run_hey(*args)


@mcp.tool()
def list_postings(box: str, limit: int | None = None, fetch_all: bool = False) -> Any:
    """List postings (emails) in a mailbox.

    Args:
        box: Mailbox name (imbox, feedbox, etc.) or numeric ID.
        limit: Maximum number of postings to return.
        fetch_all: Fetch all pages, overriding limit.
    """
    args = ["box", box]
    add_pagination(args, limit, fetch_all)
    return run_hey(*args)


@mcp.tool()
def read_thread(thread_id: str) -> Any:
    """Read a full email thread.

    Args:
        thread_id: The topic/thread ID to read.
    """
    return run_hey("threads", thread_id)


@mcp.tool()
def list_attachments(thread_id: str) -> Any:
    """List attachments in an email thread.

    Args:
        thread_id: The topic/thread ID to inspect.
    """
    return run_hey("attachments", thread_id)


@mcp.tool()
def download_attachments(
    thread_id: str,
    output: str = ".",
    entry_id: str | None = None,
    index: int | None = None,
) -> Any:
    """Download attachments from an email thread onto the host filesystem.

    Args:
        thread_id: The topic/thread ID containing attachments.
        output: Host output directory. Tilde is expanded by the MCP server.
        entry_id: Optional entry ID to narrow the download.
        index: Optional 1-based attachment index within the selected entry.
    """
    args = ["attachments", "download", thread_id, "--output", expand_output_path(output)]
    add_value(args, "--entry", entry_id)
    add_value(args, "--index", index)
    return run_hey(*args, timeout_seconds=max(DEFAULT_TIMEOUT_SECONDS, 120))


@mcp.tool()
def compose_email(
    subject: str,
    message: str,
    to: str | None = None,
    cc: str | None = None,
    bcc: str | None = None,
    thread_id: str | None = None,
    draft: bool = False,
) -> Any:
    """Compose and send a new email, post to an existing thread, or save a draft.

    Args:
        to: Recipient email address(es), comma-separated for multiple. Required
            for new threads and ignored when thread_id is provided.
        subject: Message subject.
        message: Message body.
        cc: CC recipient email address(es), comma-separated for multiple.
        bcc: BCC recipient email address(es), comma-separated for multiple.
        thread_id: Optional thread ID to post message to.
        draft: Save as a draft instead of sending.
    """
    args = ["compose", "--subject", subject, "--message", message]
    if to:
        args.extend(["--to", to])
    if cc:
        args.extend(["--cc", cc])
    if bcc:
        args.extend(["--bcc", bcc])
    if thread_id:
        args.extend(["--thread-id", thread_id])
    if draft:
        args.append("--draft")
    return run_hey(*args)


@mcp.tool()
def reply_to_thread(thread_id: str, message: str, draft: bool = False) -> Any:
    """Reply to an email thread, or save a reply draft.

    Args:
        thread_id: The topic/thread ID to reply to.
        message: Reply message body.
        draft: Save as a draft instead of sending.
    """
    args = ["reply", thread_id, "--message", message]
    if draft:
        args.append("--draft")
    return run_hey(*args)


@mcp.tool()
def list_drafts(limit: int | None = None, fetch_all: bool = False) -> Any:
    """List email drafts.

    Args:
        limit: Maximum number of drafts to return.
        fetch_all: Fetch all pages, overriding limit.
    """
    args = ["drafts"]
    add_pagination(args, limit, fetch_all)
    return run_hey(*args)


@mcp.tool()
def mark_seen(posting_ids: list[str]) -> Any:
    """Mark postings as seen.

    Args:
        posting_ids: Posting IDs to mark as seen.
    """
    return run_hey("seen", *posting_ids)


@mcp.tool()
def mark_unseen(posting_ids: list[str]) -> Any:
    """Mark postings as unseen.

    Args:
        posting_ids: Posting IDs to mark as unseen.
    """
    return run_hey("unseen", *posting_ids)


# Calendars and events


@mcp.tool()
def list_calendars() -> Any:
    """List all HEY calendars."""
    return run_hey("calendars")


@mcp.tool()
def list_recordings(
    calendar_id: str,
    starts_on: str | None = None,
    ends_on: str | None = None,
    limit: int | None = None,
    fetch_all: bool = False,
) -> Any:
    """List recordings (events, todos, habits, etc.) for a calendar.

    Args:
        calendar_id: Calendar ID.
        starts_on: Start date (YYYY-MM-DD, defaults to today).
        ends_on: End date (YYYY-MM-DD, defaults to 30 days from starts_on).
        limit: Maximum number of recordings per type.
        fetch_all: Fetch all pages, overriding limit.
    """
    args = ["recordings", calendar_id]
    add_value(args, "--starts-on", starts_on)
    add_value(args, "--ends-on", ends_on)
    add_pagination(args, limit, fetch_all)
    return run_hey(*args)


@mcp.tool()
def create_event(
    calendar_id: str,
    title: str,
    starts_at: str,
    all_day: bool = False,
    ends_at: str | None = None,
    start_time: str | None = None,
    end_time: str | None = None,
    timezone: str | None = None,
    reminders: list[str] | None = None,
    invitees: list[str] | None = None,
) -> Any:
    """Create a calendar event.

    Args:
        calendar_id: Calendar ID from list_calendars.
        title: Event title.
        starts_at: Start date (YYYY-MM-DD).
        all_day: Whether this is an all-day event.
        ends_at: End date (YYYY-MM-DD, defaults to starts_at).
        start_time: Start time (HH:MM, required for timed events).
        end_time: End time (HH:MM, required for timed events).
        timezone: IANA timezone (for example America/New_York).
        reminders: Reminder durations, for example ["15m", "1h"].
        invitees: Email addresses to invite.
    """
    args = ["event", "create", "--title", title, "--calendar-id", calendar_id, "--starts-at", starts_at]
    if all_day:
        args.append("--all-day")
    add_value(args, "--ends-at", ends_at)
    add_value(args, "--start-time", start_time)
    add_value(args, "--end-time", end_time)
    add_value(args, "--timezone", timezone)
    add_repeatable(args, "--reminder", reminders)
    add_repeatable(args, "--invitee", invitees)
    return run_hey(*args)


@mcp.tool()
def update_event(
    event_id: str,
    title: str | None = None,
    starts_at: str | None = None,
    ends_at: str | None = None,
    all_day: bool | None = None,
    start_time: str | None = None,
    end_time: str | None = None,
    timezone: str | None = None,
    reminders: list[str] | None = None,
    invitees: list[str] | None = None,
) -> Any:
    """Update a calendar event. Only provided fields are changed.

    Args:
        event_id: Event ID to update.
        title: New event title.
        starts_at: New start date (YYYY-MM-DD).
        ends_at: New end date (YYYY-MM-DD).
        all_day: True for all-day, false for timed, omitted to leave unchanged.
        start_time: New start time (HH:MM).
        end_time: New end time (HH:MM).
        timezone: New IANA timezone.
        reminders: Replacement reminder durations.
        invitees: Replacement invitee email addresses.
    """
    args = ["event", "update", event_id]
    add_value(args, "--title", title)
    add_value(args, "--starts-at", starts_at)
    add_value(args, "--ends-at", ends_at)
    add_bool(args, "--all-day", all_day)
    add_value(args, "--start-time", start_time)
    add_value(args, "--end-time", end_time)
    add_value(args, "--timezone", timezone)
    add_repeatable(args, "--reminder", reminders)
    add_repeatable(args, "--invitee", invitees)
    return run_hey(*args)


@mcp.tool()
def delete_event(event_id: str) -> Any:
    """Delete a calendar event.

    Args:
        event_id: Event ID to delete.
    """
    return run_hey("event", "delete", event_id)


# Todos


@mcp.tool()
def list_todos(limit: int | None = None, fetch_all: bool = False) -> Any:
    """List todos.

    Args:
        limit: Maximum number of todos to return.
        fetch_all: Fetch all pages, overriding limit.
    """
    args = ["todo", "list"]
    add_pagination(args, limit, fetch_all)
    return run_hey(*args)


@mcp.tool()
def add_todo(title: str, date: str | None = None) -> Any:
    """Create a new todo.

    Args:
        title: Todo title.
        date: Optional due date (YYYY-MM-DD).
    """
    args = ["todo", "add", "--title", title]
    add_value(args, "--date", date)
    return run_hey(*args)


@mcp.tool()
def complete_todo(todo_id: str) -> Any:
    """Mark a todo as complete.

    Args:
        todo_id: Todo ID to complete.
    """
    return run_hey("todo", "complete", todo_id)


@mcp.tool()
def uncomplete_todo(todo_id: str) -> Any:
    """Mark a todo as incomplete.

    Args:
        todo_id: Todo ID to mark incomplete.
    """
    return run_hey("todo", "uncomplete", todo_id)


@mcp.tool()
def delete_todo(todo_id: str) -> Any:
    """Delete a todo.

    Args:
        todo_id: Todo ID to delete.
    """
    return run_hey("todo", "delete", todo_id)


# Habits


@mcp.tool()
def complete_habit(habit_id: str, date: str | None = None) -> Any:
    """Mark a habit as complete for a date.

    Args:
        habit_id: Habit ID.
        date: Date (YYYY-MM-DD, defaults to today).
    """
    args = ["habit", "complete", habit_id]
    add_value(args, "--date", date)
    return run_hey(*args)


@mcp.tool()
def uncomplete_habit(habit_id: str, date: str | None = None) -> Any:
    """Remove a habit completion for a date.

    Args:
        habit_id: Habit ID.
        date: Date (YYYY-MM-DD, defaults to today).
    """
    args = ["habit", "uncomplete", habit_id]
    add_value(args, "--date", date)
    return run_hey(*args)


# Time tracking


@mcp.tool()
def timetrack_start() -> Any:
    """Start time tracking."""
    return run_hey("timetrack", "start")


@mcp.tool()
def timetrack_stop() -> Any:
    """Stop time tracking."""
    return run_hey("timetrack", "stop")


@mcp.tool()
def timetrack_current() -> Any:
    """Show current time tracking status."""
    return run_hey("timetrack", "current")


@mcp.tool()
def timetrack_list(limit: int | None = None, fetch_all: bool = False) -> Any:
    """List time tracks.

    Args:
        limit: Maximum number of tracks to return.
        fetch_all: Fetch all pages, overriding limit.
    """
    args = ["timetrack", "list"]
    add_pagination(args, limit, fetch_all)
    return run_hey(*args)


# Journal


@mcp.tool()
def journal_list(limit: int | None = None, fetch_all: bool = False) -> Any:
    """List journal entries.

    Args:
        limit: Maximum number of journal entries to return.
        fetch_all: Fetch all pages, overriding limit.
    """
    args = ["journal", "list"]
    add_pagination(args, limit, fetch_all)
    return run_hey(*args)


@mcp.tool()
def journal_read(date: str | None = None) -> Any:
    """Read a journal entry.

    Args:
        date: Date to read (YYYY-MM-DD, defaults to today).
    """
    args = ["journal", "read"]
    if date:
        args.append(date)
    return run_hey(*args)


@mcp.tool()
def journal_write(content: str, date: str | None = None) -> Any:
    """Write or edit a journal entry.

    Args:
        content: Journal content.
        date: Date to write (YYYY-MM-DD, defaults to today).
    """
    args = ["journal", "write"]
    if date:
        args.append(date)
    args.extend(["--content", content])
    return run_hey(*args)


# Diagnostics and host HEY client management


@mcp.tool()
def list_commands() -> Any:
    """List commands exposed by the host HEY client."""
    return run_hey("commands")


@mcp.tool()
def auth_status() -> Any:
    """Show host HEY authentication status."""
    return run_hey("auth", "status")


@mcp.tool()
def auth_refresh() -> Any:
    """Refresh the host HEY access token."""
    return run_hey("auth", "refresh")


@mcp.tool()
def auth_token(stored: bool = False) -> Any:
    """Print the host HEY access token.

    Args:
        stored: Only print stored OAuth token, ignoring HEY_TOKEN.
    """
    args = ["auth", "token"]
    if stored:
        args.append("--stored")
    return run_hey(*args)


@mcp.tool()
def auth_logout() -> Any:
    """Clear host HEY credentials."""
    return run_hey("auth", "logout")


@mcp.tool()
def show_config() -> Any:
    """Show host HEY configuration with sources."""
    return run_hey("config", "show")


@mcp.tool()
def set_config(key: str, value: str) -> Any:
    """Set a host HEY configuration value.

    Args:
        key: Configuration key, for example base_url.
        value: Configuration value.
    """
    return run_hey("config", "set", key, value)


@mcp.tool()
def doctor() -> Any:
    """Check host HEY health and configuration."""
    return run_hey("doctor")


@mcp.tool()
def setup() -> Any:
    """Run the host HEY first-run setup operation."""
    return run_hey("setup")


def shutdown_handler(signum: int, frame: Any) -> None:
    sys.exit(0)


if __name__ == "__main__":
    signal.signal(signal.SIGINT, shutdown_handler)
    signal.signal(signal.SIGTERM, shutdown_handler)
    mcp.run(transport="stdio")