MCP Tabular Data Analysis Server

Instructions

Execute a SQL query on a SQLite database.

Args:
    db_path: Path to SQLite database file
    query: SQL query to execute (SELECT queries only for safety)
    limit: Maximum number of rows to return (default 100)

Returns:
    Dictionary containing:
    - query: The executed query
    - row_count: Number of rows returned
    - columns: List of column names
    - rows: Query results

Implementation Reference

• src/mcp_tabular/server.py:380-430 (handler)

  The main handler function for the 'query_sqlite' MCP tool. It executes SELECT queries on a SQLite database, adds a LIMIT if not present, returns results as JSON-friendly dict with columns and rows. Includes safety checks for SELECT only and file existence. Uses pandas for query execution and _resolve_path helper.

  @mcp.tool()
  def query_sqlite(
      db_path: str,
      query: str,
      limit: int = 100,
  ) -> dict[str, Any]:
      """
      Execute a SQL query on a SQLite database.
      
      Args:
          db_path: Path to SQLite database file
          query: SQL query to execute (SELECT queries only for safety)
          limit: Maximum number of rows to return (default 100)
      
      Returns:
          Dictionary containing:
          - query: The executed query
          - row_count: Number of rows returned
          - columns: List of column names
          - rows: Query results
      """
      # Basic safety check - only allow SELECT
      query_upper = query.strip().upper()
      if not query_upper.startswith("SELECT"):
          raise ValueError("Only SELECT queries are allowed for safety")
      
      path = _resolve_path(db_path)
      if not path.exists():
          raise FileNotFoundError(
              f"Database not found: {db_path}\n"
              f"Resolved to: {path}\n"
              f"Project root: {_PROJECT_ROOT}"
          )
      
      conn = sqlite3.connect(str(path))
      try:
          # Add LIMIT if not present
          if "LIMIT" not in query_upper:
              query = f"{query.rstrip(';')} LIMIT {limit}"
          
          df = pd.read_sql_query(query, conn)
          
          return {
              "query": query,
              "row_count": len(df),
              "columns": df.columns.tolist(),
              "rows": df.to_dict(orient="records"),
          }
      finally:
          conn.close()