""" title: DuckDB Access Agent description: This tool provides access to a DuckDB database, with configurable database path. version: 0.0.1 required_open_webui_version: 0.5.20 license: MIT """ import duckdb from pydantic import BaseModel, Field class Tools: class Valves(BaseModel): db_path: str = Field( default="/path/to/my.duckdb", description="The path to the DuckDB database file" ) def __init__(self): print("Initializing DuckDB tool class") self.citation = True self.valves = Tools.Valves() def _get_connection(self): # Changed from getconnection to _get_connection to match usage conn = duckdb.connect(self.valves.db_path, read_only=True) return conn def list_all_tables(self) -> str: print(f"Listing all tables in {self.valves.db_path}") try: conn = self._get_connection() # Now matches the method name result = conn.execute("SELECT table_name FROM information_schema.tables WHERE table_type = 'BASE TABLE'") tables = [row[0] for row in result.fetchall()] conn.close() if tables: return "Here is a list of all the tables in the DuckDB database:\n\n" + "\n".join(tables) else: return "No tables found." except Exception as e: return f"Error listing tables: {str(e)}" def table_data_schema(self, table_name: str) -> str: print(f"Describing table: {table_name}") try: conn = self._get_connection() # Now matches the method name result = conn.execute(f""" SELECT column_name, data_type, is_nullable FROM information_schema.columns WHERE table_name = '{table_name}' """) columns = result.fetchall() conn.close() if not columns: return f"No such table: {table_name}" description = f"Table '{table_name}' in the DuckDB database has the following columns:\n" for column in columns: column_name, data_type, is_nullable = column description += f"- {column_name} ({data_type})" if is_nullable == 'YES': description += " [Nullable]" description += "\n" return description except Exception as e: return f"Error describing table: {str(e)}" def execute_read_query(self, query: str) -> str: print(f"Executing query: {query}") try: conn = self._get_connection() # Now matches the method name result = conn.execute(query) rows = result.fetchall() if not rows: return "No data returned from query." column_names = [description[0] for description in result.description] csv_data = f"Query executed successfully. Below is the result of the query {query} in CSV format:\n\n" csv_data += ",".join(column_names) + "\n" for row in rows: csv_data += ",".join(map(str, row)) + "\n" conn.close() return csv_data except Exception as e: return f"Error executing query: {str(e)}"