query_sqlite
Execute SQL queries on SQLite databases to retrieve and analyze tabular data, supporting SELECT operations with configurable row limits for data exploration.
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 resultsInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| db_path | Yes | ||
| query | Yes | ||
| limit | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
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()