mcp_server.py

import sys
import os

sys.path.insert(0, os.path.dirname(__file__))

from mcp.server.fastmcp import FastMCP
import config
from integrations.local_db import init_db

mcp = FastMCP("personal-ops-mcp")

# --- Email tools ---
if config.GMAIL_ENABLED:
    @mcp.tool()
    def list_unread(max_results: int = 10) -> list[dict]:
        """List your unread emails. Returns sender, subject, date, and snippet."""
        from integrations.gmail import list_unread_emails
        return list_unread_emails(max_results)

    @mcp.tool()
    def search_emails(query: str, max_results: int = 10) -> list[dict]:
        """Search your emails. query uses Gmail search syntax e.g. 'from:boss subject:report'."""
        from integrations.gmail import search_email
        return search_email(query, max_results)

    @mcp.tool()
    def read_email(message_id: str) -> str:
        """Read the full body of an email by its message ID."""
        from integrations.gmail import get_email_body
        return get_email_body(message_id)

# --- Calendar tools ---
if config.GOOGLE_CALENDAR_ENABLED:
    @mcp.tool()
    def list_calendar_events(days_ahead: int = 1, max_results: int = 20) -> list[dict]:
        """List upcoming calendar events. days_ahead=1 means today, 7 means next week."""
        from integrations.google_calendar import list_events
        return list_events(days_ahead, max_results)

    @mcp.tool()
    def create_calendar_event(
        title: str,
        start_time: str,
        end_time: str,
        description: str = "",
        location: str = "",
    ) -> dict:
        """Create a calendar event. start_time and end_time must be ISO 8601 e.g. 2025-05-01T10:00:00Z."""
        from integrations.google_calendar import create_event
        return create_event(title, start_time, end_time, description, location)

# --- Task tools ---
if config.TASKS_ENABLED:
    @mcp.tool()
    def list_todos(include_done: bool = False) -> list[dict]:
        """List your tasks. Set include_done=true to also show completed tasks."""
        from integrations.local_db import list_tasks
        return list_tasks(include_done)

    @mcp.tool()
    def create_todo(title: str, due_date: str = "") -> dict:
        """Create a new task. due_date is optional, use YYYY-MM-DD format."""
        from integrations.local_db import create_task
        return create_task(title, due_date)

    @mcp.tool()
    def complete_todo(task_id: int) -> dict:
        """Mark a task as complete by its ID."""
        from integrations.local_db import complete_task
        complete_task(task_id)
        return {"status": "done", "task_id": task_id}

# --- Notes tools ---
if config.NOTES_ENABLED:
    @mcp.tool()
    def search_notes(query: str) -> list[dict]:
        """Search your notes by keyword. Returns matching notes with title, body, and date."""
        from integrations.local_db import search_notes as _search
        return _search(query)

    @mcp.tool()
    def get_note(note_id: int) -> dict:
        """Get the full content of a note by its ID."""
        from integrations.local_db import get_note as _get
        note = _get(note_id)
        return note if note else {"error": f"Note {note_id} not found"}

    @mcp.tool()
    def create_note(title: str, body: str) -> dict:
        """Create a new note with a title and body."""
        from integrations.local_db import create_note as _create
        return _create(title, body)

# --- File tools ---
if config.GOOGLE_DRIVE_ENABLED:
    @mcp.tool()
    def find_file(query: str, max_results: int = 10) -> list[dict]:
        """Search Google Drive for files by name or content."""
        from integrations.drive import search_files
        return search_files(query, max_results)

    @mcp.tool()
    def recent_files(max_results: int = 10) -> list[dict]:
        """List the most recently modified files in Google Drive."""
        from integrations.drive import list_recent_files
        return list_recent_files(max_results)

# --- Daily briefing ---
@mcp.tool()
def daily_briefing() -> dict:
    """Get a morning briefing: unread emails, today's calendar events, and pending tasks."""
    result = {}
    if config.GMAIL_ENABLED:
        try:
            from integrations.gmail import list_unread_emails
            result["unread_emails"] = list_unread_emails(5)
        except Exception as e:
            result["unread_emails"] = {"error": str(e)}
    if config.GOOGLE_CALENDAR_ENABLED:
        try:
            from integrations.google_calendar import list_events
            result["todays_events"] = list_events(days_ahead=1)
        except Exception as e:
            result["todays_events"] = {"error": str(e)}
    if config.TASKS_ENABLED:
        try:
            from integrations.local_db import list_tasks
            result["pending_tasks"] = list_tasks(include_done=False)
        except Exception as e:
            result["pending_tasks"] = {"error": str(e)}
    return result


if __name__ == "__main__":
    init_db()
    mcp.run(transport="stdio")