QueryPilot
Safe SQL gateway for AI agents: SELECT-only validation, access policies, masking, audit trail, evals
Open source Open in the app JSON README (API)
About
Safe SQL gateway for AI agents: SELECT-only validation, access policies, masking, audit trail, evals
Details
- Kind
- MCP servers
- Topic
- Databases
- Publisher
- nickklos10
- Origin
- official
- Category
- ferramentas
- Transport
- local
- Version
- 0.1.1
- Stars
- 1
- Last push
- 2026-07-11T21:54:02Z
- Repository state
- ativo
- Language
- Python
- License
- MIT
- Added
- 2026-08-29 04:00:54
- Updated
- 2026-08-29 04:00:54
- Origin id
io.github.nickklos10/querypilot
README
# QueryPilot
<!-- mcp-name: io.github.nickklos10/querypilot -->
[](https://pypi.org/project/querypilot/)
[](https://pypi.org/project/querypilot/)
[](LICENSE)
[](https://github.com/nickklos10/QueryPilot/actions/workflows/eval.yml)
Eval-driven SQL reliability for AI agents.
QueryPilot helps agents safely generate, validate, repair, execute, and regression-test SQL against real fixture databases.
<p align="center"><img src="https://raw.githubusercontent.com/nickklos10/QueryPilot/main/docs/assets/eval-report.svg" alt="querypilot eval run terminal report" width="840"></p>
## Why QueryPilot Exists
Read-only SQL access for agents is becoming a commodity. Tools that let an agent list tables, read schemas, and run validated `SELECT`s already exist. What is much harder — and what QueryPilot focuses on — is making that access *measurably reliable*: proving the SQL the agent generates is correct, safe, fast, and not regressing.
Every change to QueryPilot, your prompts, or your model can be measured against an execution-truth eval suite. Suites can be authored by hand or auto-generated by replaying your audit log as a regression set, so the same queries that worked in production yesterday have to keep working tomorrow.
## Quick Demo
```bash
python3 -m venv .venv
.venv/bin/pip install -e ".[dev,eval]"
.venv/bin/querypilot eval init # scaffold suites/ and .eval/
.venv/bin/querypilot eval run \
--suite suites/smoke.yaml \
--generator demo \
--report eval-out.json
.venv/bin/querypilot eval check \
--report eval-out.json \
--baseline .eval/baseline.json \
--threshold 0.9 \
--require-safety 1.0
```
Sample output (abridged — see the full report at the top of this README):
```text
QueryPilot Eval Report
Suite: smoke
Generator: demo
Overall
✅ Pass rate 3 / 3 (100%)
✅ Safety pass rate 0 / 0 (100%)
✅ Correctness 3 / 3 (100%)
✅ P95 latency 18 ms
✅ No threshold violations.
```
The bundled `suites/smoke.yaml` runs against a tiny SQLite fixture (`tests/fixtures/demo.db`) so the harness works end-to-end without an LLM key. To benchmark a real generator, use `--generator openai` or `--generator anthropic`.
## Audit-Log Replay
`querypilot eval replay` turns a JSONL audit log written by `JSONLAuditSink` into a `BenchmarkSuite` whose gold SQL is the SQL that previously executed. Re-running that suite gates accuracy regressions against your own production traffic — the unique-to-QueryPilot capability the eval positioning rests on.
```bash
querypilot eval replay \
--audit-jsonl audit.jsonl \
--fixture-db sqlite:///tests/fixtures/demo.db \
--output suites/replay.yaml
querypilot eval run --suite suites/replay.yaml --generator demo --report replay-out.json
```
Conservative defaults: only successful `ask` records, non-empty results, no active access policy. `--include-failures`, `--include-masked`, `--include-empty` relax each filter.
## CI Gate
`querypilot eval check` compares a `SuiteReport` JSON against thresholds and a committed baseline, exiting non-zero on regression. A sample GitHub Actions workflow ships at `.github/workflows/eval.yml`:
```yaml
- run: querypilot eval run --suite suites/smoke.yaml --generator demo --report eval-out.json
- run: querypilot eval check --report eval-out.json --baseline .eval/baseline.json --threshold 0.9 --require-safety 1.0
```
When a regression is detected the output explains which cases regressed and how:
```text
Regression detected.
Pass rate:
baseline: 96%
current: 89%
Failed cases (regression vs. baseline):
- monthly_revenue_by_segment (was passing -> now result_mismatch)
- top_customers_by_arr (was passing -> now repair_failed)
Latency:
baseline p95: 2100 ms
current p95: 3800 ms (+1700 ms)
```
Refresh the baseline on `main` after a deliberate change:
```bash
querypilot eval run --suite suites/smoke.yaml --generator demo --report .eval/baseline.json
git commit -am "Refresh eval baseline"
```
## Authoring a Suite
Suites are YAML or JSON. Each case carries a question, a gold SQL, and the schema/safety expectations for the candidate.
```yaml
name: saas_revenue_suite
fixture_db: sqlite:///fixtures/demo.db
fixture_dialect: sqlite
thresholds:
pass_rate: 0.95
safety_pass_rate: 1.0
correctness_rate: 0.9
max_p95_latency_ms: 5000
max_avg_cost_usd: 0.01
comparison:
ignore_row_order: true
ignore_column_order: true
float_tolerance: 0.001
normalize_datetimes: true
cases:
- id: top_customers_by_revenue
question: "Top customers by revenue"
gold_sql: |
SELECT customer_name, revenue
FROM customers
ORDER BY revenue DESC
LIMIT 100
expected_tables: [customers]
must_include: ["ORDER BY", "LIMIT"]
must_not_contain: [DELETE, UPDATE, DROP]
tags: [revenue, ranking]
- id: blocks_drop_table
sql: "DROP TABLE customers"
should_pass: false
expected_failure_kind: validation
expected_error_contains: ["Only SELECT queries are allowed"]
tags: [safety, ddl]
```
Result-set correctness is scored by **executing both the gold and candidate SQL** against the same fixture database and comparing rows. Order-insensitive by default; auto-flipped to order-sensitive when the gold SQL has a top-level `ORDER BY`.
## Library Usage
```python
from querypilot import QueryPilot
qp = QueryPilot.connect(
database_url="sqlite:///demo.db",
dialect="sqlite",
readonly=True,
max_rows=100,
)
result = qp.execute_sql("SELECT * FROM customers")
print(result.sql)
print(result.rows)
```
Natural-language `ask()` works offline for simple demo questions through a deterministic generator:
```python
answer = qp.ask("Top customers by revenue")
print(answer.sql)
print(answer.rows)
print(answer.validation.risk_level)
```
## Examples
Runnable, self-contained examples live in [`examples/`](examples/). They all use
the bundled demo SQLite fixture, so most need no API key:
| Example | Shows | Key? |
| --- | --- | --- |
| [`01_quickstart.py`](examples/01_quickstart.py) | connect, `execute_sql`, offline `ask()`, validation risk level | No |
| [`02_openai_tool_use.py`](examples/02_openai_tool_use.py) | `as_openai_tools()` in an OpenAI tool-use loop | `OPENAI_API_KEY` |
| [`03_anthropic_tool_use.py`](examples/03_anthropic_tool_use.py) | `as_anthropic_tools()` in an Anthropic tool-use loop | `ANTHROPIC_API_KEY` |
| [`04_access_control.py`](examples/04_access_control.py) | blocked columns, row filter, and masking | No |
| [`05_custom_eval_suite/`](examples/05_custom_eval_suite/) | a custom YAML suite run with `querypilot eval run`/`check` | No |
| [`06_mcp/`](examples/06_mcp/) | run `querypilot mcp` + a paste-ready Claude MCP config | No |
See [`examples/README.md`](examples/README.md) for setup and the full index.
## LLM SQL Generation
For production-style natural-language SQL generation, plug in an LLM generator. QueryPilot still treats model output as an untrusted candidate: it validates, rewrites, and can ask the generator for a repair before execution.
Install optional provider dependencies:
```bash
.venv/bin/pip install -e ".[openai]"
.venv/bin/pip install -e ".[anthropic]"
```
OpenAI:
```python
from querypilot import QueryPilot
from querypilot.generation import OpenAISQLGenerator
qp = QueryPilot.connect(
"sqlite:///demo.db",
generator=OpenAISQLGenerator(model="gpt-5.1"),
max_generation_attempts=2,
)
```
Anthropic:
```python
from querypilot import QueryPilot
from querypilot.generation import AnthropicSQLGenerator
qp = QueryPilot.connect(
"sqlite:///demo.db",
generator=AnthropicSQLGenerator(model="claude-sonnet-4-20250514"),
max_generation_attempts=2,
)
```
### Local / open models
Any OpenAI-compatible endpoint — [Ollama](https://ollama.com), vLLM, LM Studio,
or llama.cpp's server — works through `OpenAICompatibleSQLGenerator`. It reuses
the `[openai]` extra (no extra dependency) and talks the Chat Completions API, so
you can benchmark open models at **$0**. The API key is optional (local servers
ignore it), and cost reports show `$0` while token counts still flow through when
the server returns usage.
```bash
ollama pull llama3.1
.venv/bin/pip install -e ".[openai]"
```
```python
from querypilot import QueryPilot
from querypilot.generation import OpenAICompatibleSQLGenerator
qp = QueryPilot.connect(
"sqlite:///demo.db",
generator=OpenAICompatibleSQLGenerator(
model="llama3.1",
base_url="http://localhost:11434/v1", # Ollama's default; omit to use it
),
max_generation_attempts=2,
)
```
From the eval harness, add open models to the benchmark matrix with
`--generator openai-compatible`:
```bash
querypilot eval run \
--suite suites/smoke.yaml \
--generator openai-compatible \
--model llama3.1 \
--base-url http://localhost:11434/v1 \
--report eval-out.json
```
`--base-url` also reads `$QUERYPILOT_BASE_URL`, and defaults to Ollama's
`http://localhost:11434/v1` when unset.
The safety loop is always:
```text
question
-> schema-scoped prompt
-> model candidate SQL
-> QueryPilot validation
-> optional repair
-> safe execution
```
## Eval Harness (Library)
The CLI is a thin wrapper around `run_suite`, which is also usable directly:
```python
from querypilot import QueryPilot
from querypilot.evals import (
BenchmarkCase,
BenchmarkSuite,
NullCostTracker,
build_qp_factory,
render_terminal,
run_suite,
)
from querypilot.generation.sql_generator import DemoSQLGenerator
suite = BenchmarkSuite(
name="adhoc",
fixture_db="sqlite:///tests/fixtures/demo.db",
cases=[
BenchmarkCase(
id="count_customers",
question="Count of customers",
gold_sql="SELECT COUNT(*) AS count FROM customers",
expected_tables=["customers"],
),
],
)
qp_factory = build_qp_factory(
database_url="sqlite:///tests/fixtures/demo.db",
generator=DemoSQLGenerator(),
)
report = run_suite(
suite,
qp_factory=qp_factory,
cost_tracker_factory=NullCostTracker,
)
print(render_terminal(report, color=False))
```
The returned `SuiteReport` is a Pydantic model with `pass_rate`, `safety_pass_rate`, `correctness_rate`, `repair_rate`, `p50_latency_ms`, `p95_latency_ms`, `total_prompt_tokens`, `estimated_cost_usd`, `tag_rollups`, `failure_breakdown`, `threshold_violations`, and the full per-case `case_results` list.
## Safety Engine
QueryPilot validates SQL before execution with:
- `sqlglot` parsing
- single-statement enforcement
- SELECT-only read-only policy
- blocked keyword detection
- known table checks
- column checks where feasible
- allowed/blocked table policy
- automatic `LIMIT` insertion and max-row capping
- `SELECT *` warnings or rejection
- Cartesian join detection
- structured policy checks
- query fingerprints
- risk levels: `low`, `medium`, `high`, `critical`
For PostgreSQL production use, connect QueryPilot with a dedicated
least-privilege role that has only the required schema `USAGE` and table
`SELECT` grants. QueryPilot requests a read-only transaction and applies a
statement timeout, but application validation is not a replacement for
database permissions.
Example:
```python
validation = qp.validate_sql("SELECT * FROM customers")
print(validation.valid)
print(validation.risk_level)
print(validation.query_fingerprint)
print(validation.policy_checks)
```
For stricter deployments:
```python
from querypilot.core.config import SafetyPolicy
qp = QueryPilot.connect(
"sqlite:///demo.db",
safety_policy=SafetyPolicy(
allow_select_star=False,
reject_cartesian_joins=True,
),
)
```
## Agent Tool Adapters
QueryPilot exposes tool schemas without requiring SDK dependencies:
```python
openai_tools = qp.as_openai_tools()
anthropic_tools = qp.as_anthropic_tools()
```
Available tools:
- `ask_database`
- `search_schema`
- `validate_sql`
- `execute_sql`
## FastAPI Server
Run QueryPilot as a local safe SQL gateway:
```bash
.venv/bin/pip install -e ".[server]"
querypilot serve --database-url sqlite:///demo.db --dialect sqlite --max-rows 100
```
Or use environment variables:
```bash
export QUERYPILOT_DATABASE_URL=sqlite:///demo.db
export QUERYPILOT_DIALECT=sqlite
querypilot serve
```
Endpoints:
- `GET /health`
- `GET /schema`
- `POST /search-schema`
- `POST /ask`
- `POST /generate-sql`
- `POST /validate-sql`
- `POST /execute-sql`
- `POST /evals/run`
- `GET /audit/recent`
Example:
```bash
curl -X POST http://127.0.0.1:8000/validate-sql \
-H "content-type: application/json" \
-d '{"sql": "SELECT * FROM customers"}'
```
## MCP Server
Run QueryPilot as an MCP-compatible tool server:
```bash
.venv/bin/pip install -e ".[mcp]"
querypilot mcp --database-url sqlite:///demo.db --dialect sqlite
```
If your MCP client launches servers with `uvx`, include the `[mcp]` extra explicitly so the MCP SDK dependency is installed:
```bash
uvx --from 'querypilot[mcp]' querypilot mcp --database-url sqlite:///demo.db --dialect sqlite
```
By default, the MCP command uses stdio transport. For clients that support Streamable HTTP:
```bash
querypilot mcp \
--database-url sqlite:///demo.db \
--dialect sqlite \
--transport streamable-http
```
MCP tools:
- `ask_database`
- `search_schema`
- `validate_sql`
- `execute_sql`
## Audit Trail
QueryPilot records structured audit events for schema search, SQL generation, validation, execution, and full `ask()` flows.
Each audit record can include:
- `audit_id`
- timestamp
- operation
- question
- original SQL
- rewritten SQL
- validation metadata
- execution status
- row count
- execution time
- error
- actor/session/application/trace metadata
Use the default in-memory sink:
```python
from querypilot import QueryPilot
from querypilot.audit import AuditMetadata
qp = QueryPilot.connect(
"sqlite:///demo.db",
audit_metadata=AuditMetadata(
actor="agent-1",
session_id="session-1",
app_name="analytics-agent",
),
)
result = qp.execute_sql("SELECT customer_name FROM customers")
print(result.audit_id)
print(qp.get_audit_records(limit=10))
```
Or persist JSONL audit events:
```python
from querypilot import QueryPilot
from querypilot.audit import JSONLAuditSink
qp = QueryPilot.connect(
"sqlite:///demo.db",
audit_sink=JSONLAuditSink("querypilot-audit.jsonl"),
)
```
## Access Control
Read-only SQL is necessary but not enough. QueryPilot can also enforce column-level and row-level access policies before execution.
```python
from querypilot import QueryPilot
from querypilot.access import AccessPolicy, MaskingRule
qp = QueryPilot.connect(
"sqlite:///demo.db",
access_policy=AccessPolicy(
blocked_columns={
"customers": ["email"],
},
row_filters={
"customers": "tenant_id = 42",
},
masking_rules={
"customers": {
"email": MaskingRule(mode="redact"),
},
},
),
)
```
What this does:
- rejects SQL that selects blocked columns
- rejects SQL outside an allowlist when `allowed_columns` is configured
- injects required row filters such as `tenant_id = 42`
- masks configured result columns after execution
- records the applied access policy in validation, result, answer, and audit metadata
The server and MCP runtimes can also receive access policy JSON:
```bash
querypilot serve \
--database-url sqlite:///demo.db \
--access-policy-json '{
"row_filters": {"customers": "tenant_id = 42"},
"blocked_columns": {"customers": ["ssn"]}
}'
```
## Current Scope
Shipped:
- installable Python package
- SQLite connector
- PostgreSQL connector structure
- schema introspection
- SQL validation and rewriting
- safe read-only execution
- offline demo SQL generation, OpenAI and Anthropic LLM generators with repair loop
- column policies, row filters, and result masking
- in-memory and JSONL audit logging
- FastAPI server runtime
- MCP tool server runtime
- **eval-driven harness**: YAML/JSON suites, execution-truth correctness scoring, safety/repair/latency/cost metrics, per-tag rollups, failure-category breakdown, threshold violations, JSON and screenshot-quality terminal reports
- **audit-log → regression suite** (`querypilot eval replay`)
- **CI regression gate** (`querypilot eval check` against a committed baseline) + sample GitHub Actions workflow
- **`querypilot eval init`** — scaffolds `suites/` and `.eval/` for new projects
## Roadmap
The eval-driven foundation is shipped. Next pillars:
- **Schema-aware grounded generation** — schema embeddings, retrieval, semantic verification of repaired SQL
- **EXPLAIN-plan and cost guards** — per-query row/cost budgets, cardinality-based LIMIT policies, plan analysis
- **Multi-tenant governance** — tenant-scoped row filters, per-actor policy injection, automatic PII detection
- **Cross-dialect transpilation** — write a suite once, run it against SQLite, Postgres, MySQL
- **Multi-database connectors** — Snowflake, BigQuery, Redshift