MCP 服务器

io.github.nickklos10/querypilot

QueryPilot enables AI agents to safely generate, validate, repair, and execute SQL against fixture databases, with eval-driven regression testing to ensure reliability.

README

QueryPilot

Eval-driven SQL reliability for AI agents.

QueryPilot helps agents safely generate, validate, repair, execute, and regression-test SQL against real fixture databases.

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 SELECTs 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

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):

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.

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:

- 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:

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:

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.

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

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:

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/. They all use the bundled demo SQLite fixture, so most need no API key:

Example	Shows	Key?
`01_quickstart.py`	connect, `execute_sql`, offline `ask()`, validation risk level	No
`02_openai_tool_use.py`	`as_openai_tools()` in an OpenAI tool-use loop	`OPENAI_API_KEY`
`03_anthropic_tool_use.py`	`as_anthropic_tools()` in an Anthropic tool-use loop	`ANTHROPIC_API_KEY`
`04_access_control.py`	blocked columns, row filter, and masking	No
`05_custom_eval_suite/`	a custom YAML suite run with `querypilot eval run`/`check`	No
`06_mcp/`	run `querypilot mcp` + a paste-ready Claude MCP config	No

See 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:

.venv/bin/pip install -e ".[openai]"
.venv/bin/pip install -e ".[anthropic]"

OpenAI:

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:

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, 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.

ollama pull llama3.1
.venv/bin/pip install -e ".[openai]"

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:

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:

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:

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:

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:

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:

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:

.venv/bin/pip install -e ".[server]"
querypilot serve --database-url sqlite:///demo.db --dialect sqlite --max-rows 100

Or use environment variables:

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:

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:

.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:

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:

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:

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:

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.

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:

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