timing_mcp.py

import sqlite3
import os
import logging
from mcp.server.fastmcp import FastMCP

# Silence FastMCP's log output to stderr. Some MCP clients (e.g. Jan)
# terminate the subprocess when they receive unexpected stderr data.
logging.getLogger("mcp").setLevel(logging.WARNING)

# Initialize the MCP Server
mcp = FastMCP("Timing Local Reader")

DB_PATH = os.path.join(
    os.path.expanduser("~"),
    "Library", "Application Support", "info.eurocomp.Timing2", "SQLite.db"
)

def get_connection():
    """Connects to the database in strict read-only mode to prevent locking issues."""
    if not os.path.exists(DB_PATH):
        raise FileNotFoundError(f"Database not found at {DB_PATH}")
    # The ?mode=ro URI parameter enforces read-only access
    uri = f"file:{DB_PATH}?mode=ro"
    return sqlite3.connect(uri, uri=True)

@mcp.tool()
def list_timing_tables() -> str:
    """Lists all tables in the local Timing SQLite database."""
    try:
        with get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
            tables = [row[0] for row in cursor.fetchall()]
            return f"Tables found: {', '.join(tables)}"
    except Exception as e:
        return f"Error: {str(e)}"

@mcp.tool()
def get_timing_schema(table_name: str) -> str:
    """Gets the column schema for a specific table in the Timing database."""
    try:
        with get_connection() as conn:
            cursor = conn.cursor()
            # Validate table_name against actual tables to prevent SQL injection
            cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
            valid_tables = [row[0] for row in cursor.fetchall()]
            if table_name not in valid_tables:
                return f"Error: Table '{table_name}' not found. Valid tables: {', '.join(valid_tables)}"
            cursor.execute(f"PRAGMA table_info({table_name});")
            columns = cursor.fetchall()
            schema_details = "\n".join([f"- {col[1]} ({col[2]})" for col in columns])
            return f"Schema for '{table_name}':\n{schema_details}"
    except Exception as e:
        return f"Error: {str(e)}"

@mcp.tool()
def query_timing_db(sql_query: str, limit: int = 200) -> str:
    """Executes a read-only SQL query against the Timing database and returns the results. limit controls the maximum number of rows returned (default 200)."""
    try:
        with get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(sql_query)

            # Fetch column headers
            columns = [d[0] for d in cursor.description] if cursor.description else []
            results = cursor.fetchmany(limit)

            output = f"Columns: {', '.join(columns)}\nResults:\n"
            for row in results:
                output += f"{row}\n"
            if len(results) == limit:
                output += f"(results truncated at {limit} rows)"

            return output
    except Exception as e:
        return f"SQL Error: {str(e)}"

if __name__ == "__main__":
    mcp.run()