server.py

import time

from fastmcp import FastMCP
from fastmcp.server.context import Context
from fastmcp.server.middleware import CallNext, Middleware, MiddlewareContext
from fastmcp.tools.tool import ToolResult
from mcp.types import CallToolResult, TextContent
from pydantic import Field

from tools import (
    app,
    audit,
    calendar,
    chrome,
    clipboard,
    config,
    contacts,
    display,
    files,
    messages,
    notes,
    reminders,
    safari,
    screenshot,
    system,
    window,
)
from tools.mcp_result import structured_only


class StructuredFastMCP(FastMCP):
    """FastMCP variant that preserves structured error payloads and ``isError``."""

    async def _call_tool_mcp(
        self, key: str, arguments: dict
    ) -> CallToolResult:
        async with Context(fastmcp=self):
            result = await self._call_tool_middleware(key, arguments)
            payload = result.structured_content
            return CallToolResult(
                content=result.content,
                structuredContent=payload,
                isError=bool(payload and payload.get("ok") is False),
            )


mcp = StructuredFastMCP("Altic-MCP")


@mcp.tool()
def open_app(name: str) -> str:
    """
    Open any mac application by specifying its name. Use this tool
    if you encounter any error or issue mentioning that the app is not
    open

    Args:
        name: Name of the mac app. e.g. "Mail", "Contacts", "Messages" etc.

    Returns:
        A success or failure message
    """
    return app.open_app(name)


@mcp.tool()
async def get_frontmost_app() -> str:
    """
    Get the currently frontmost macOS application.

    Returns:
        JSON string with app name, bundle id, pid, and active state.
    """
    return window.get_frontmost_app()


@mcp.tool()
async def list_windows(
    app_name: str = Field(default=""),
    include_minimized: bool = Field(default=False),
) -> str:
    """
    List manageable macOS windows.

    Args:
        app_name: Optional app name, bundle id, or process name filter
        include_minimized: Include minimized windows when available

    Returns:
        JSON string with window ids, app metadata, titles, frames, and display indexes.
    """
    return window.list_windows(app_name, include_minimized)


@mcp.tool()
async def focus_window(
    app_name: str = Field(default=""),
    window_id: int | None = Field(default=None, ge=1),
    window_index: int | None = Field(default=None, ge=1),
) -> str:
    """
    Focus a macOS window by window id, app name, or frontmost fallback.

    Args:
        app_name: Optional app name, bundle id, or process name
        window_id: Optional CoreGraphics window id
        window_index: Optional 1-based index among the app's windows

    Returns:
        JSON string with focused window metadata, or an error message.
    """
    return window.focus_window(app_name, window_id, window_index)


@mcp.tool()
async def move_window(
    x: int,
    y: int,
    app_name: str = Field(default=""),
    window_id: int | None = Field(default=None, ge=1),
    window_index: int | None = Field(default=None, ge=1),
    display_index: int | None = Field(default=None, ge=1),
) -> str:
    """
    Move a macOS window to a display-aware top-left position.

    Args:
        x: Target top-left x coordinate
        y: Target top-left y coordinate
        app_name: Optional app name, bundle id, or process name
        window_id: Optional CoreGraphics window id
        window_index: Optional 1-based index among the app's windows
        display_index: Optional 1-based display index for clamping placement

    Returns:
        JSON string with final window metadata, or an error message.
    """
    return window.move_window(x, y, app_name, window_id, window_index, display_index)


@mcp.tool()
async def resize_window(
    width: int = Field(ge=1),
    height: int = Field(ge=1),
    app_name: str = Field(default=""),
    window_id: int | None = Field(default=None, ge=1),
    window_index: int | None = Field(default=None, ge=1),
    display_index: int | None = Field(default=None, ge=1),
) -> str:
    """
    Resize a macOS window and clamp it to the selected or current display.

    Args:
        width: Target width in points
        height: Target height in points
        app_name: Optional app name, bundle id, or process name
        window_id: Optional CoreGraphics window id
        window_index: Optional 1-based index among the app's windows
        display_index: Optional 1-based display index for clamping placement

    Returns:
        JSON string with final window metadata, or an error message.
    """
    return window.resize_window(
        width,
        height,
        app_name,
        window_id,
        window_index,
        display_index,
    )


@mcp.tool()
async def center_window(
    app_name: str = Field(default=""),
    window_id: int | None = Field(default=None, ge=1),
    window_index: int | None = Field(default=None, ge=1),
    display_index: int | None = Field(default=None, ge=1),
    width: int | None = Field(default=None, ge=1),
    height: int | None = Field(default=None, ge=1),
) -> str:
    """
    Center a macOS window on its current display or a selected display.

    Args:
        app_name: Optional app name, bundle id, or process name
        window_id: Optional CoreGraphics window id
        window_index: Optional 1-based index among the app's windows
        display_index: Optional 1-based display index
        width: Optional width to apply before centering
        height: Optional height to apply before centering

    Returns:
        JSON string with final window metadata, or an error message.
    """
    return window.center_window(
        app_name,
        window_id,
        window_index,
        display_index,
        width,
        height,
    )


@mcp.tool()
async def tile_windows(
    layout: str = Field(default="columns"),
    app_names: list[str] | None = Field(default=None),
    display_index: int | None = Field(default=None, ge=1),
    padding: int = Field(default=8, ge=0, le=100),
) -> str:
    """
    Tile multiple macOS windows on a display.

    Args:
        layout: Tile layout: columns, rows, or grid
        app_names: Optional app names to tile in order; defaults to visible windows
        display_index: Optional 1-based display index
        padding: Gap around and between windows

    Returns:
        JSON string with final window metadata for tiled windows, or an error message.
    """
    return window.tile_windows(layout, app_names, display_index, padding)


@mcp.tool()
async def minimize(
    app_name: str = Field(default=""),
    window_id: int | None = Field(default=None, ge=1),
    window_index: int | None = Field(default=None, ge=1),
) -> str:
    """
    Minimize a macOS window.

    Args:
        app_name: Optional app name, bundle id, or process name
        window_id: Optional CoreGraphics window id
        window_index: Optional 1-based index among the app's windows

    Returns:
        JSON string with minimized window metadata, or an error message.
    """
    return window.minimize(app_name, window_id, window_index)


@mcp.tool()
async def hide_app(app_name: str) -> str:
    """
    Hide a running macOS app.

    Args:
        app_name: App name, bundle id, or process name

    Returns:
        JSON string with hidden app metadata, or an error message.
    """
    return window.hide_app(app_name)


@mcp.tool()
async def quit_app(app_name: str) -> str:
    """
    Quit a running macOS app.

    Args:
        app_name: App name, bundle id, or process name

    Returns:
        JSON string with quit app metadata, or an error message.
    """
    return window.quit_app(app_name)


@mcp.tool()
async def find_files(
    query: str,
    root: str = Field(default=""),
    max_results: int | None = Field(default=None, ge=1, le=500),
    include_hidden: bool = Field(default=False),
    kind: str = Field(default="auto"),
) -> str:
    """
    Find files by name using Spotlight first, with Python filesystem fallback.

    Args:
        query: File name or Spotlight query text
        root: Optional directory to search within
        max_results: Maximum number of results to return
        include_hidden: Include hidden files and folders
        kind: Search backend: auto, name, or spotlight

    Returns:
        JSON string with file metadata results or an error message
    """
    return files.find_files(query, root, max_results, include_hidden, kind)


@mcp.tool()
async def list_directory(
    path: str,
    include_hidden: bool = Field(default=False),
    max_results: int = Field(default=200, ge=1, le=1000),
) -> str:
    """
    List immediate children of a directory with metadata.

    Args:
        path: Directory path to list
        include_hidden: Include hidden files and folders
        max_results: Maximum number of children to return

    Returns:
        JSON string with directory entries or an error message
    """
    return files.list_directory(path, include_hidden, max_results)


@mcp.tool()
async def get_file_info(path: str) -> str:
    """
    Get metadata for a file, directory, or missing path.

    Args:
        path: File or directory path

    Returns:
        JSON string with path metadata or an error message
    """
    return files.get_file_info(path)


@mcp.tool()
async def copy_file(
    source: str,
    destination: str,
    overwrite: bool = Field(default=False),
    dry_run: bool = Field(default=False),
) -> str:
    """
    Copy a file with metadata. Directories require copy_directory.

    Args:
        source: Existing source file path
        destination: Destination file path
        overwrite: Allow replacing an existing destination
        dry_run: Return the planned action without copying

    Returns:
        JSON string with operation details or an error message
    """
    return files.copy_file(source, destination, overwrite, dry_run)


@mcp.tool()
async def copy_directory(
    source: str,
    destination: str,
    overwrite: bool = Field(default=False),
    dry_run: bool = Field(default=False),
) -> str:
    """
    Copy a directory tree. Existing destinations fail unless overwrite is true.

    Args:
        source: Existing source directory path
        destination: Destination directory path
        overwrite: Allow merging into an existing destination
        dry_run: Return the planned action without copying

    Returns:
        JSON string with operation details or an error message
    """
    return files.copy_directory(source, destination, overwrite, dry_run)


@mcp.tool()
async def move_file(
    source: str,
    destination: str,
    overwrite: bool = Field(default=False),
    dry_run: bool = Field(default=False),
) -> str:
    """
    Move a file or directory. Existing destinations fail unless overwrite is true.

    Args:
        source: Existing source path
        destination: Destination path
        overwrite: Allow replacing an existing destination
        dry_run: Return the planned action without moving

    Returns:
        JSON string with operation details or an error message
    """
    return files.move_file(source, destination, overwrite, dry_run)


@mcp.tool()
async def rename_file(
    path: str,
    new_name: str,
    overwrite: bool = Field(default=False),
    dry_run: bool = Field(default=False),
) -> str:
    """
    Rename a file or directory within its current parent directory.

    Args:
        path: Existing file or directory path
        new_name: New file name only, not a path
        overwrite: Allow replacing an existing destination
        dry_run: Return the planned action without renaming

    Returns:
        JSON string with operation details or an error message
    """
    return files.rename_file(path, new_name, overwrite, dry_run)


@mcp.tool()
async def trash_file(path: str, dry_run: bool = Field(default=False)) -> str:
    """
    Move a file or directory to the macOS Trash. Permanent delete is not supported.

    Args:
        path: Existing file or directory path
        dry_run: Return the planned action without moving to Trash

    Returns:
        JSON string with operation details or an error message
    """
    return files.trash_file(path, dry_run)


@mcp.tool()
async def reveal_in_finder(path: str) -> str:
    """
    Reveal a file or directory in Finder.

    Args:
        path: Existing file or directory path

    Returns:
        JSON string with operation details or an error message
    """
    return files.reveal_in_finder(path)


@mcp.tool()
async def get_finder_selection() -> str:
    """
    Get the currently selected Finder items as paths and metadata.

    Returns:
        JSON string with selected Finder items or an error message
    """
    return files.get_finder_selection()


@mcp.tool()
async def get_clipboard_text(
    max_chars: int | None = Field(default=None, ge=1, le=200000),
) -> str:
    """
    Read plain text from the macOS clipboard.

    Args:
        max_chars: Maximum number of characters to return

    Returns:
        JSON string with clipboard text and truncation metadata, or an error message
    """
    return clipboard.get_clipboard_text(max_chars)


@mcp.tool()
async def set_clipboard_text(text: str) -> str:
    """
    Write plain text to the macOS clipboard.

    Args:
        text: Text to place on the clipboard

    Returns:
        JSON string with operation metadata, or an error message
    """
    return clipboard.set_clipboard_text(text)


@mcp.tool()
async def clear_clipboard() -> str:
    """
    Clear clipboard contents.

    Returns:
        JSON string with operation metadata, or an error message
    """
    return clipboard.clear_clipboard()


@mcp.tool()
async def get_clipboard_files() -> str:
    """
    Return file URLs currently available on the macOS clipboard.

    Returns:
        JSON string with copied file paths, or an error message
    """
    return clipboard.get_clipboard_files()


@mcp.tool()
async def set_clipboard_files(paths: list[str]) -> str:
    """
    Put one or more filesystem paths on the macOS clipboard for Finder paste.

    Args:
        paths: Existing file or directory paths to place on the clipboard

    Returns:
        JSON string with operation metadata, or an error message
    """
    return clipboard.set_clipboard_files(paths)


@mcp.tool()
async def save_clipboard_image(output_path: str = Field(default="")) -> str | list[object]:
    """
    Save an image from the macOS clipboard to a PNG file and share it with the model.

    Args:
        output_path: Optional output path for the PNG

    Returns:
        A text status plus image content, or an error message
    """
    return clipboard.save_clipboard_image(output_path)


@mcp.tool()
async def set_clipboard_image(path: str) -> str:
    """
    Put an image file on the macOS clipboard.

    Args:
        path: Existing image file path

    Returns:
        JSON string with operation metadata, or an error message
    """
    return clipboard.set_clipboard_image(path)


@mcp.tool()
def send_imessage(phone_number: str, message: str) -> str:
    """
    Send an imessage to someone in your contacts

    Args:
        phone_number: The phone number of the recipient
        message: The message text to send

    Returns:
        Success or error message
    """
    return messages.send_message(phone_number=phone_number, message=message)


@mcp.tool()
async def list_chats(limit: int = Field(default=50, ge=1, le=500)) -> str:
    """
    List Messages chats available to the macOS Messages app.

    Args:
        limit: Maximum number of chats to return

    Returns:
        Structured JSON with chat ids, names, and account metadata.
    """
    return messages.list_chats(limit)


@mcp.tool()
async def send_file_message(
    recipient: str,
    path: str,
    message: str = Field(default=""),
    recipient_type: str = Field(default="handle"),
) -> str:
    """
    Send a file attachment with Messages to a handle or chat id.

    Args:
        recipient: Phone/email handle, or chat id when recipient_type is "chat"
        path: Existing local file path to send
        message: Optional text message to send before the file
        recipient_type: "handle" or "chat"

    Returns:
        Structured JSON with send metadata or an error message.
    """
    return messages.send_file_message(recipient, path, message, recipient_type)


@mcp.tool()
async def search_contacts(name: str) -> str:
    """
    Search for a phone number from contacts by name. Returns multiple
    options if more than one contact is found or more than one number is found.
    Ask clarifying questions on which number if any following actions are required

    Args:
        name: The name of contact to search for
        ctx: FastMCP context for logging

    Returns:
        A list of matching contacts
    """
    return contacts.search_contacts(name)


@mcp.tool()
async def get_contact(identifier: str) -> str:
    """
    Get a unique Contacts record by id or name match.

    Args:
        identifier: Contact id or name text

    Returns:
        Structured JSON with contact fields, phone numbers, and emails.
    """
    return contacts.get_contact(identifier)


@mcp.tool()
async def create_contact(
    first_name: str,
    last_name: str,
    organization: str = Field(default=""),
    phone: str = Field(default=""),
    email: str = Field(default=""),
    note: str = Field(default=""),
) -> str:
    """
    Create a Contacts record with optional organization, phone, email, and note.

    Args:
        first_name: Contact first name
        last_name: Contact last name
        organization: Optional organization/company
        phone: Optional phone number to add with mobile label
        email: Optional email to add with home label
        note: Optional contact note

    Returns:
        Structured JSON with the created contact.
    """
    return contacts.create_contact(first_name, last_name, organization, phone, email, note)


@mcp.tool()
async def update_contact(
    identifier: str,
    first_name: str = Field(default=""),
    last_name: str = Field(default=""),
    organization: str = Field(default=""),
    phone: str = Field(default=""),
    email: str = Field(default=""),
    note: str = Field(default=""),
) -> str:
    """
    Update a unique Contacts record by id or name match.

    Args:
        identifier: Contact id or name text
        first_name: Optional replacement first name
        last_name: Optional replacement last name
        organization: Optional replacement organization/company
        phone: Optional phone number to add
        email: Optional email address to add
        note: Optional replacement note

    Returns:
        Structured JSON with the updated contact.
    """
    return contacts.update_contact(
        identifier,
        first_name,
        last_name,
        organization,
        phone,
        email,
        note,
    )


@mcp.tool()
async def read_recent_messages(
    phone_number: str, recent_message_count: int = Field(default=25, ge=1, le=200)
) -> str:
    """
    Read the recent X messages from the iMessage app between the user and the
    person with the phone number. X is defined based on the value of recent_message_count

    Args:
        phone_number: The phone number of the person you want to retrieve the chat from
        recent_message_count: The recent messages to retrieve, can be a maximum of 200

    Returns:
        A list of recent messages in the chat
    """
    return messages.read_recent_messages(phone_number, recent_message_count)


@mcp.tool()
async def set_reminder(name: str, datetime: str, list_name: str = "Reminders") -> str:
    """
    Set a reminder

    Args:
        name: The reminder text
        datetime: The time to set the reminder for, must in the following format "YYYY-MM-DD HH:MM"
        list_name: Reminder list, e.g. Work, Personal etc. Defaults to "Reminders"

    Returns:
        A success or error message
    """
    return reminders.set_reminder(name, datetime, list_name)


@mcp.tool()
async def list_reminder_lists() -> str:
    """
    List Apple Reminders lists.

    Returns:
        Structured JSON with list names or an error message.
    """
    return reminders.list_reminder_lists()


@mcp.tool()
async def list_reminders(
    list_name: str = Field(default=""),
    include_completed: bool = Field(default=False),
) -> str:
    """
    List reminders, optionally filtered by list and completion state.

    Args:
        list_name: Optional Reminders list name
        include_completed: Include completed reminders

    Returns:
        Structured JSON with reminder matches or an error message.
    """
    return reminders.list_reminders(list_name, include_completed)


@mcp.tool()
async def search_reminders(
    query: str,
    list_name: str = Field(default=""),
    include_completed: bool = Field(default=False),
) -> str:
    """
    Search reminders by name.

    Args:
        query: Text to match in reminder names
        list_name: Optional Reminders list name
        include_completed: Include completed reminders

    Returns:
        Structured JSON with reminder matches or an error message.
    """
    return reminders.search_reminders(query, list_name, include_completed)


@mcp.tool()
async def complete_reminder(
    identifier: str,
    list_name: str = Field(default=""),
) -> str:
    """
    Mark an exact or uniquely matching reminder as complete.

    Args:
        identifier: Reminder id or name/title text
        list_name: Optional Reminders list name

    Returns:
        Structured JSON with the updated reminder or an error message.
    """
    return reminders.complete_reminder(identifier, list_name)


@mcp.tool()
async def delete_reminder(
    identifier: str,
    list_name: str = Field(default=""),
    dry_run: bool = Field(default=True),
) -> str:
    """
    Delete an exact or uniquely matching reminder. Defaults to dry-run.

    Args:
        identifier: Reminder id or name/title text
        list_name: Optional Reminders list name
        dry_run: Return matching reminder without deleting when true

    Returns:
        Structured JSON with deletion details or an error message.
    """
    return reminders.delete_reminder(identifier, list_name, dry_run)


@mcp.tool()
async def reschedule_reminder(
    identifier: str,
    datetime: str,
    list_name: str = Field(default=""),
) -> str:
    """
    Reschedule an exact or uniquely matching reminder.

    Args:
        identifier: Reminder id or name/title text
        datetime: New due date/time in "YYYY-MM-DD HH:MM" format
        list_name: Optional Reminders list name

    Returns:
        Structured JSON with the updated reminder or an error message.
    """
    return reminders.reschedule_reminder(identifier, datetime, list_name)


@mcp.tool()
async def update_reminder(
    identifier: str,
    list_name: str = Field(default=""),
    name: str = Field(default=""),
    body: str = Field(default=""),
    datetime: str = Field(default=""),
    completed: bool | None = Field(default=None),
    priority: int | None = Field(default=None, ge=0, le=9),
    flagged: bool | None = Field(default=None),
) -> str:
    """
    Update an exact or uniquely matching reminder.

    Args:
        identifier: Reminder id or name/title text
        list_name: Optional Reminders list name
        name: Optional replacement reminder name
        body: Optional replacement reminder notes
        datetime: Optional due date/time in "YYYY-MM-DD HH:MM" format
        completed: Optional completion state
        priority: Optional priority from 0 to 9
        flagged: Optional flagged state

    Returns:
        Structured JSON with the updated reminder or an error message.
    """
    return reminders.update_reminder(
        identifier,
        list_name,
        name,
        body,
        datetime,
        completed,
        priority,
        flagged,
    )


@mcp.tool()
async def show_reminder(
    identifier: str,
    list_name: str = Field(default=""),
) -> str:
    """
    Reveal an exact or uniquely matching reminder in the Reminders app.

    Args:
        identifier: Reminder id or name/title text
        list_name: Optional Reminders list name

    Returns:
        Structured JSON with the shown reminder or an error message.
    """
    return reminders.show_reminder(identifier, list_name)


@mcp.tool()
async def create_note(name: str, body: str, folder: str = Field(default="")) -> str:
    """
    Create a note

    Args:
        name: Title of the note
        body: Content of the note
        folder: Use this if the note has to be created in a specific folder. Uses
        default folder if none is specified

    Returns:
        A success or error message
    """
    return notes.create_note(name, body, folder)


@mcp.tool()
async def list_note_folders() -> str:
    """
    List Apple Notes folders.

    Returns:
        Structured JSON with folder names or an error message.
    """
    return notes.list_note_folders()


@mcp.tool()
async def list_notes(
    folder: str = Field(default=""),
    max_results: int = Field(default=25, ge=1, le=200),
) -> str:
    """
    List Apple Notes, optionally filtered by folder.

    Args:
        folder: Optional Notes folder name
        max_results: Maximum notes to return

    Returns:
        Structured JSON with note metadata or an error message.
    """
    return notes.list_notes(folder, max_results)


@mcp.tool()
async def get_note(identifier: str) -> str:
    """
    Get a note by id or title.

    Args:
        identifier: Note id or title text

    Returns:
        Structured JSON with note content or an error message.
    """
    return notes.get_note(identifier)


@mcp.tool()
async def append_to_note(identifier: str, body: str) -> str:
    """
    Append plain text to a note by id or title.

    Args:
        identifier: Note id or title text
        body: Plain text to append

    Returns:
        Structured JSON with the updated note or an error message.
    """
    return notes.append_to_note(identifier, body)


@mcp.tool()
async def update_note(identifier: str, body: str) -> str:
    """
    Replace a note body by id or title.

    Args:
        identifier: Note id or title text
        body: Replacement plain text/HTML body

    Returns:
        Structured JSON with the updated note or an error message.
    """
    return notes.update_note(identifier, body)


@mcp.tool()
async def delete_note(identifier: str, dry_run: bool = Field(default=True)) -> str:
    """
    Delete a note by id or title. Defaults to dry-run.

    Args:
        identifier: Note id or title text
        dry_run: Return matching note without deleting when true

    Returns:
        Structured JSON with deletion details or an error message.
    """
    return notes.delete_note(identifier, dry_run)


@mcp.tool()
async def move_note(identifier: str, folder: str) -> str:
    """
    Move a note by id or title to an existing folder.

    Args:
        identifier: Note id or title text
        folder: Destination folder name

    Returns:
        Structured JSON with moved note metadata or an error message.
    """
    return notes.move_note(identifier, folder)


@mcp.tool()
async def search_notes(
    query: str, max_results: int = Field(default=10, ge=1, le=20)
) -> str:
    """
    Search Apple notes based on a query

    Args:
        query: The query string
        max_results: The maximum number of results returned from the tool, defaults to 10

    Returns:
        A list of notes based on search
    """
    return notes.search_notes(query, max_results)


@mcp.tool()
async def decrease_brightness(
    amount: float = Field(default=0.0625, ge=0.0, le=1.0),
) -> str:
    """
    Decrease screen brightness

    Args:
        amount: Amount to decrease brightness by (0.0 to 1.0 scale). Default is 0.0625 (6.25%)

    Returns:
        Success or error message
    """
    return system.decrease_brightness(amount)


@mcp.tool()
async def increase_brightness(
    amount: float = Field(default=0.0625, ge=0.0, le=1.0),
) -> str:
    """
    Increase screen brightness

    Args:
        amount: Amount to increase brightness by (0.0 to 1.0 scale). Default is 0.0625 (6.25%)

    Returns:
        Success or error message
    """
    return system.increase_brightness(amount)


@mcp.tool()
async def turn_up_volume(amount: float = Field(default=6.25, ge=0.0, le=100.0)) -> str:
    """
    Turn up system volume

    Args:
        amount: Amount to increase volume by (0-100 scale). Default is 6.25 (6.25%)

    Returns:
        Success or error message
    """
    return system.turn_up_volume(amount)


@mcp.tool()
async def turn_down_volume(
    amount: float = Field(default=6.25, ge=0.0, le=100.0),
) -> str:
    """
    Turn down system volume

    Args:
        amount: Amount to decrease volume by (0-100 scale). Default is 6.25 (6.25%)

    Returns:
        Success or error message
    """
    return system.turn_down_volume(amount)


@mcp.tool()
async def create_calendar_event(
    title: str,
    start_datetime: str,
    duration_minutes: int,
    calendar_name: str = Field(default=""),
) -> str:
    """
    Create a calendar event in the macOS Calendar app

    Args:
        title: The event title
        start_datetime: Start date and time in format 'YYYY-MM-DD HH:MM' (e.g., '2025-10-30 14:30')
        duration_minutes: Duration of the event in minutes
        calendar_name: Optional calendar name (uses default calendar if not specified)

    Returns:
        Success or error message
    """
    return calendar.create_calendar_event(
        title, start_datetime, duration_minutes, calendar_name
    )


@mcp.tool()
async def list_calendar_events_for_day(date: str) -> str:
    """
    List all calendar events for a specific day

    Args:
        date: Date in format 'YYYY-MM-DD' (e.g., '2025-10-30')

    Returns:
        List of events for the specified day or error message
    """
    return calendar.list_calendar_events_for_day(date)


@mcp.tool()
async def list_calendars() -> str:
    """
    List Apple Calendar calendars.

    Returns:
        Structured JSON with calendar names or an error message.
    """
    return calendar.list_calendars()


@mcp.tool()
async def list_calendar_events(
    start_date: str,
    end_date: str,
    calendar_name: str = Field(default=""),
) -> str:
    """
    List Calendar events across a date range.

    Args:
        start_date: Start date in "YYYY-MM-DD" format
        end_date: End date in "YYYY-MM-DD" format
        calendar_name: Optional calendar name

    Returns:
        Structured JSON with events or an error message.
    """
    return calendar.list_calendar_events(start_date, end_date, calendar_name)


@mcp.tool()
async def search_calendar_events(
    query: str,
    start_date: str = Field(default=""),
    end_date: str = Field(default=""),
    calendar_name: str = Field(default=""),
) -> str:
    """
    Search Calendar events by title, location, or notes.

    Args:
        query: Search text
        start_date: Optional start date in "YYYY-MM-DD" format
        end_date: Optional end date in "YYYY-MM-DD" format
        calendar_name: Optional calendar name

    Returns:
        Structured JSON with matching events or an error message.
    """
    return calendar.search_calendar_events(query, start_date, end_date, calendar_name)


@mcp.tool()
async def update_calendar_event(
    identifier: str,
    title: str = Field(default=""),
    start_datetime: str = Field(default=""),
    duration_minutes: int | None = Field(default=None),
    calendar_name: str = Field(default=""),
    location: str = Field(default=""),
    notes_text: str = Field(default=""),
) -> str:
    """
    Update an exact or uniquely matching Calendar event.

    Args:
        identifier: Event id or title text
        title: Optional replacement title
        start_datetime: Optional replacement start in "YYYY-MM-DD HH:MM" format
        duration_minutes: Optional replacement duration
        calendar_name: Optional calendar name for matching
        location: Optional replacement location
        notes_text: Optional replacement notes

    Returns:
        Structured JSON with updated event metadata or an error message.
    """
    return calendar.update_calendar_event(
        identifier,
        title,
        start_datetime,
        duration_minutes,
        calendar_name,
        location,
        notes_text,
    )


@mcp.tool()
async def delete_calendar_event(
    identifier: str,
    start_date: str = Field(default=""),
    end_date: str = Field(default=""),
    calendar_name: str = Field(default=""),
    dry_run: bool = Field(default=True),
) -> str:
    """
    Delete an exact or uniquely matching Calendar event. Defaults to dry-run.

    Args:
        identifier: Event id or title text
        start_date: Optional date-range start in "YYYY-MM-DD" format
        end_date: Optional date-range end in "YYYY-MM-DD" format
        calendar_name: Optional calendar name
        dry_run: Return matching event without deleting when true

    Returns:
        Structured JSON with deletion details or an error message.
    """
    return calendar.delete_calendar_event(
        identifier,
        start_date,
        end_date,
        calendar_name,
        dry_run,
    )


@mcp.tool()
async def check_calendar_availability(
    start_datetime: str,
    duration_minutes: int,
    calendar_name: str = Field(default=""),
) -> str:
    """
    Check whether a Calendar time slot conflicts with existing events.

    Args:
        start_datetime: Slot start in "YYYY-MM-DD HH:MM" format
        duration_minutes: Slot duration in minutes
        calendar_name: Optional calendar name

    Returns:
        Structured JSON with availability and conflicts or an error message.
    """
    return calendar.check_calendar_availability(
        start_datetime,
        duration_minutes,
        calendar_name,
    )


@mcp.tool()
async def create_recurring_event(
    title: str,
    start_datetime: str,
    duration_minutes: int = Field(ge=1),
    frequency: str = Field(default="weekly"),
    interval: int = Field(default=1, ge=1),
    count: int | None = Field(default=None, ge=1),
    until_date: str = Field(default=""),
    calendar_name: str = Field(default=""),
    location: str = Field(default=""),
    notes_text: str = Field(default=""),
) -> str:
    """
    Create a recurring Calendar event.

    Args:
        title: Event title
        start_datetime: Start date and time in "YYYY-MM-DD HH:MM" format
        duration_minutes: Duration in minutes
        frequency: daily, weekly, monthly, or yearly
        interval: Recurrence interval
        count: Optional occurrence count
        until_date: Optional recurrence end date in "YYYY-MM-DD" format
        calendar_name: Optional target calendar name
        location: Optional event location
        notes_text: Optional event notes

    Returns:
        Structured JSON with the created recurring event or an error message.
    """
    return calendar.create_recurring_event(
        title,
        start_datetime,
        duration_minutes,
        frequency,
        interval,
        count,
        until_date,
        calendar_name,
        location,
        notes_text,
    )


@mcp.tool()
async def open_safari_tab(url: str = Field(default="")) -> str:
    """
    Open a new tab in Safari with optional URL

    Args:
        url: Optional URL to open in the new tab

    Returns:
        Success or error message
    """
    return safari.open_safari_tab(url)


@mcp.tool()
async def close_safari_tab(tab_index: int = Field(default=-1)) -> str:
    """
    Close a Safari tab. Use -1 for current tab or specify tab index (1-based)

    Args:
        tab_index: Tab index to close (-1 for current tab, or 1-based index)

    Returns:
        Success or error message
    """
    return safari.close_safari_tab(tab_index)


@mcp.tool()
async def get_safari_tabs() -> str:
    """
    Get a list of all open Safari tabs with their URLs and titles

    Returns:
        List of tabs with URLs and titles or error message
    """
    return safari.get_safari_tabs()


@mcp.tool()
async def switch_safari_tab(tab_index: int) -> str:
    """
    Switch to a specific Safari tab by index (1-based)

    Args:
        tab_index: Tab index to switch to (must be greater than 0)

    Returns:
        Success or error message
    """
    return safari.switch_safari_tab(tab_index)


@mcp.tool()
async def run_safari_javascript(javascript_code: str) -> str:
    """
    Execute JavaScript code in the current Safari tab and return the result

    Args:
        javascript_code: JavaScript code to execute

    Returns:
        JavaScript execution result or error message
    """
    return safari.run_safari_javascript(javascript_code)


@mcp.tool()
async def navigate_safari(url: str) -> str:
    """
    Navigate to a URL in the current Safari tab

    Args:
        url: URL to navigate to

    Returns:
        Success or error message
    """
    return safari.navigate_safari(url)


@mcp.tool()
async def reload_safari_page() -> str:
    """
    Reload the current Safari page

    Returns:
        Success or error message
    """
    return safari.reload_safari_page()


@mcp.tool()
async def safari_go_back() -> str:
    """
    Navigate back in Safari history

    Returns:
        Success or error message
    """
    return safari.safari_go_back()


@mcp.tool()
async def safari_go_forward() -> str:
    """
    Navigate forward in Safari history

    Returns:
        Success or error message
    """
    return safari.safari_go_forward()


@mcp.tool()
async def open_safari_window(url: str = Field(default="")) -> str:
    """
    Open a new Safari window with optional URL

    Args:
        url: Optional URL to open in the new window

    Returns:
        Success or error message
    """
    return safari.open_safari_window(url)


@mcp.tool()
async def close_safari_window() -> str:
    """
    Close the current Safari window

    Returns:
        Success or error message
    """
    return safari.close_safari_window()


@mcp.tool()
async def get_safari_page_info() -> str:
    """
    Get information about the current Safari page including URL, title, text content, and HTML source

    Returns:
        Page information including URL, title, text, and source or error message
    """
    return safari.get_safari_page_info()


@mcp.tool()
async def chrome_open_session(
    start_url: str = Field(default="about:blank"),
    debug_port: int = Field(default=9222, ge=1024, le=65535),
    headless: bool = Field(default=False),
) -> str:
    """
    Open a Chrome automation session for browser control.

    Args:
        start_url: Initial page URL to open
        debug_port: Chrome remote debugging port for CDP sessions
        headless: Launch Chrome headless if debugger needs to be started

    Returns:
        Session identifier and connection details or error message
    """
    return chrome.chrome_open_session(start_url, debug_port, headless)


@mcp.tool()
async def chrome_list_sessions() -> str:
    """
    List active Chrome automation sessions.

    Returns:
        A list of active session IDs
    """
    return chrome.chrome_list_sessions()


@mcp.tool()
async def chrome_navigate(session_id: str, url: str) -> str:
    """
    Navigate an active Chrome CDP session to a URL.

    Args:
        session_id: Session returned by chrome_open_session
        url: URL to load

    Returns:
        Success or error message
    """
    return chrome.chrome_navigate(session_id, url)


@mcp.tool()
async def chrome_wait_for(
    session_id: str,
    selector: str,
    timeout_ms: int = Field(default=10000, ge=100, le=120000),
    poll_ms: int = Field(default=200, ge=50, le=5000),
) -> str:
    """
    Wait for a CSS selector to exist and be visible in the page.

    Args:
        session_id: Session returned by chrome_open_session
        selector: CSS selector to wait for
        timeout_ms: Max wait time in milliseconds
        poll_ms: Poll interval in milliseconds

    Returns:
        Success or timeout/error message
    """
    return chrome.chrome_wait_for(session_id, selector, timeout_ms, poll_ms)


@mcp.tool()
async def chrome_click(session_id: str, selector: str) -> str:
    """
    Click an element in an active Chrome CDP session using a CSS selector.

    Args:
        session_id: Session returned by chrome_open_session
        selector: CSS selector for target element

    Returns:
        Success or error message
    """
    return chrome.chrome_click(session_id, selector)


@mcp.tool()
async def chrome_type(
    session_id: str,
    selector: str,
    text: str,
    clear_first: bool = Field(default=True),
) -> str:
    """
    Type text into an input element in an active Chrome CDP session.

    Args:
        session_id: Session returned by chrome_open_session
        selector: CSS selector for input element
        text: Text to type
        clear_first: Clear existing value before typing

    Returns:
        Success or error message
    """
    return chrome.chrome_type(session_id, selector, text, clear_first)


@mcp.tool()
async def chrome_extract(
    session_id: str,
    selector: str = Field(default=""),
    attribute: str = Field(default=""),
    javascript_expression: str = Field(default=""),
) -> str:
    """
    Extract data from a page by selector or by JavaScript expression.

    Args:
        session_id: Session returned by chrome_open_session
        selector: CSS selector to query
        attribute: Attribute name to read from selected element
        javascript_expression: JavaScript expression to evaluate directly

    Returns:
        Extracted value as JSON or error message
    """
    return chrome.chrome_extract(session_id, selector, attribute, javascript_expression)


@mcp.tool()
async def chrome_screenshot(
    session_id: str,
    output_path: str = Field(default=""),
) -> str:
    """
    Capture a screenshot of the active page in a Chrome CDP session.

    Args:
        session_id: Session returned by chrome_open_session
        output_path: Optional file path for output PNG

    Returns:
        Saved path or error message
    """
    return chrome.chrome_screenshot(session_id, output_path)


@mcp.tool()
async def chrome_close_session(session_id: str) -> str:
    """
    Close a Chrome automation session.

    Args:
        session_id: Session returned by chrome_open_session

    Returns:
        Success or error message
    """
    return chrome.chrome_close_session(session_id)


@mcp.tool()
async def capture_active_screen(
    output_path: str = Field(default=""),
) -> str | list[object]:
    """
    Capture a full screenshot of the display containing the frontmost app and share
    the image directly with the model.

    Args:
        output_path: Optional file path for output PNG

    Returns:
        A text status plus image content, or an error message
    """
    return screenshot.capture_active_screen(output_path)


@mcp.tool()
async def add_screen_glow() -> str:
    """
    Add a visual feedback indicator (orange glow around screen edges) to show that automated actions are in progress.

    IMPORTANT: Call this FIRST before performing any automated actions to provide visual feedback
    to the user that the tool is actively working. This serves as a clear indicator that
    operations are being executed.

    Returns:
        Success or error message
    """
    return display.add_screen_glow()


@mcp.tool()
async def remove_screen_glow() -> str:
    """
    Remove the visual feedback indicator (screen glow) when automated actions are complete.

    IMPORTANT: Call this when all automated actions are complete to stop the visual feedback
    indicator and signal to the user that operations have finished.

    Returns:
        Success or error message
    """
    return display.remove_screen_glow()


@mcp.tool()
async def get_config() -> str:
    """
    Return the current altic-mcp runtime configuration.

    Returns:
        Structured JSON with all config keys and their current values.
    """
    return config.get_config()


@mcp.tool()
async def set_config_value(key: str, value: str | int | float | bool | list[str]) -> str:
    """
    Set a single runtime config value and persist it to disk.

    Changes take effect immediately without restarting the server. Valid keys:
    file_search_max_results, osascript_timeout_seconds, clipboard_max_chars,
    include_hidden_default, allowed_directories, audit_log_enabled,
    search_visit_limit.

    Args:
        key: Config key name
        value: New value (type must match the key)

    Returns:
        Structured JSON confirming the updated key/value.
    """
    return config.set_config_value(key, value)


@mcp.tool()
async def get_recent_tool_calls(
    limit: int = Field(default=50, ge=1, le=500),
) -> str:
    """
    Return recent tool calls from the audit log.

    Args:
        limit: Maximum number of recent calls to return

    Returns:
        Structured JSON with chronological tool call records.
    """
    return audit.get_recent_tool_calls(limit)


@mcp.tool()
async def start_search(
    query: str,
    root: str = Field(default=""),
    max_results: int | None = Field(default=None, ge=1, le=500),
    include_hidden: bool | None = Field(default=None),
    kind: str = Field(default="auto"),
) -> str:
    """
    Start a file search and return the first batch of results.

    Spotlight results are returned in full (search_id is null). Filesystem
    name searches return a search_id for pagination via get_more_search_results.

    Args:
        query: File name or Spotlight query text
        root: Optional directory to search within
        max_results: Maximum results in this batch
        include_hidden: Include hidden files (defaults to config)
        kind: Search backend: auto, name, or spotlight

    Returns:
        JSON with search_id, results, source, and exhausted/truncated flags.
    """
    return files.start_search(query, root, max_results, include_hidden, kind)


@mcp.tool()
async def get_more_search_results(
    search_id: str,
    count: int = Field(default=25, ge=1, le=500),
) -> str:
    """
    Fetch the next batch of results for an active search session.

    Args:
        search_id: Search id returned by start_search
        count: Maximum results to return in this batch

    Returns:
        JSON with the next results batch and exhaustion status.
    """
    return files.get_more_search_results(search_id, count)


@mcp.tool()
async def stop_search(search_id: str) -> str:
    """
    Stop and remove an active search session.

    Args:
        search_id: Search id returned by start_search

    Returns:
        JSON confirming whether the session was stopped.
    """
    return files.stop_search(search_id)


@mcp.tool()
async def list_searches() -> str:
    """
    List all active search sessions.

    Returns:
        JSON with active search ids, queries, and progress.
    """
    return files.list_searches()


@mcp.resource("altic://apps/frontmost")
async def frontmost_app_resource() -> str:
    """Current frontmost macOS application metadata."""
    return window.get_frontmost_app()


@mcp.resource("altic://finder/selection")
async def finder_selection_resource() -> str:
    """Currently selected Finder items with metadata."""
    return files.get_finder_selection()


@mcp.resource("altic://clipboard/text")
async def clipboard_text_resource() -> str:
    """Current plain-text contents of the macOS clipboard."""
    return clipboard.get_clipboard_text()


@mcp.prompt()
def summarize_today_calendar() -> str:
    """Generate a prompt to summarize today's calendar events."""
    return (
        "Use the list_calendar_events tool with today's date range to retrieve "
        "all calendar events for today, then provide a concise summary of the "
        "schedule including event times, titles, and locations."
    )


@mcp.prompt()
def draft_imessage_reply() -> str:
    """Generate a prompt to draft a reply to the latest iMessage conversation."""
    return (
        "Use list_chats to find the most recent Messages conversation, then use "
        "read_recent_messages to read the latest messages in that chat. Draft a "
        "contextually appropriate reply and present it to the user for approval "
        "before sending with send_imessage."
    )


@mcp.prompt()
def find_and_summarize_notes() -> str:
    """Generate a prompt to search and summarize Apple Notes."""
    return (
        "Ask the user what topic they want to search for, then use search_notes "
        "to find matching notes. Retrieve the full content of the top results "
        "with get_note and provide a summary of what was found."
    )


class StructuredResultMiddleware(Middleware):
    """Expose ordinary tool data only through ``structuredContent``."""

    async def on_call_tool(
        self,
        context: MiddlewareContext,
        call_next: CallNext,
    ) -> ToolResult:
        tool_result = await call_next(context)

        # Images, audio, resources, and other rich blocks must remain in content.
        if any(not isinstance(block, TextContent) for block in tool_result.content):
            return tool_result

        if tool_result.structured_content is not None:
            value = tool_result.structured_content
            # FastMCP wraps scalar/string output schemas in {"result": ...}.
            if set(value) == {"result"}:
                value = value["result"]
            return structured_only(value)

        if len(tool_result.content) == 1:
            return structured_only(tool_result.content[0].text)
        if not tool_result.content:
            return structured_only({})
        return structured_only([block.text for block in tool_result.content])


class AuditMiddleware(Middleware):
    """Log every tool call to the audit log with timing and success status."""

    async def on_call_tool(
        self,
        context: MiddlewareContext,
        call_next: CallNext,
    ):
        start = time.monotonic()
        tool_name = context.message.name
        arguments = context.message.arguments
        try:
            result = await call_next(context)
            duration_ms = (time.monotonic() - start) * 1000
            ok = (result.structured_content or {}).get("ok", True)
            audit.log_tool_call(tool_name, arguments, ok, duration_ms)
            return result
        except Exception:
            duration_ms = (time.monotonic() - start) * 1000
            audit.log_tool_call(tool_name, arguments, False, duration_ms)
            raise


mcp.add_middleware(AuditMiddleware())
mcp.add_middleware(StructuredResultMiddleware())


def main():
    mcp.run()


if __name__ == "__main__":
    main()