mcpField
Q
QueryPilot

QueryPilot

本地服务
github.com数据库21 周前更新

关于 QueryPilot

QueryPilot 为使用 SQLAlchemy 数据库的 AI Agent 提供策略管控的 SQL 网关,将只读校验、架构信息与审计、追踪和评测结合起来,衡量查询可靠性。

主要功能

  1. 01仅允许 SELECT 的 SQL 校验
  2. 02数据库架构发现与搜索
  3. 03访问策略、脱敏和行数限制
  4. 04执行 SQL 并提供修正提示
  5. 05审计、追踪与评测工具

适用场景

  1. 01在只读限制下让 Agent 查询数据库
  2. 02依据实时数据库架构生成 SQL
  3. 03检查并诊断 Agent 的 SQL 错误
  4. 04通过评测套件进行 SQL 回归测试

MCP 相关

QueryPilot 向 MCP 客户端提供架构查询、SQL 校验与执行工具,并可选提供自然语言数据库问答。相同工具可通过 stdio 或 Streamable HTTP 提供。

配置

本地服务的配置在各客户端中通用,复制后粘贴到客户端的 MCP 配置文件即可。

mcp.json · 通用配置
{
"mcpServers": {
"nickklos10-querypilot": {
"command": "uvx",
"args": [
"querypilot"
],
"env": {
"QUERYPILOT_DATABASE_URL": "<QUERYPILOT_DATABASE_URL>",
"QUERYPILOT_DIALECT": "<QUERYPILOT_DIALECT>",
"QUERYPILOT_MAX_ROWS": "<QUERYPILOT_MAX_ROWS>",
"QUERYPILOT_TIMEOUT_SECONDS": "<QUERYPILOT_TIMEOUT_SECONDS>",
"QUERYPILOT_ACCESS_POLICY_JSON": "<QUERYPILOT_ACCESS_POLICY_JSON>",
"QUERYPILOT_MCP_TRANSPORT": "<QUERYPILOT_MCP_TRANSPORT>"
}
}
}
}
启动命令uvx querypilot
需要配置的环境变量
QUERYPILOT_DATABASE_URL密钥SQLAlchemy-style database connection URL, e.g. sqlite:///demo.db or postgresql://user:pass@host:5432/db. Equivalent to --database-url.
QUERYPILOT_DIALECTSQL dialect for the connector. Equivalent to --dialect.
QUERYPILOT_MAX_ROWSMaximum rows returned per query. Equivalent to --max-rows.
QUERYPILOT_TIMEOUT_SECONDSQuery execution timeout in seconds. Equivalent to --timeout-seconds.
QUERYPILOT_ACCESS_POLICY_JSONJSON-encoded AccessPolicy (blocked_columns, allowed_columns, row_filters, masking_rules) enforced before execution. Equivalent to --access-policy-json.
QUERYPILOT_MCP_TRANSPORTMCP transport used by `querypilot mcp`. Equivalent to --transport. Most MCP clients should leave this at stdio; streamable-http is for clients that speak Streamable HTTP directly.

本地服务需要在你的电脑上安装运行环境(如 Node.js 18+)。各客户端的配置文件位置不同,请参考客户端文档。

安装包
pypiquerypilot0.1.1

工具

还没有获取到工具列表

本地服务需要在你的电脑上运行后才能读取工具,维护者暂未提供工具列表。

详细介绍

QueryPilot

PyPI Python License: MIT CI

为 AI Agent 提供安全、可衡量的 SQL。

QueryPilot 位于 LLM Agent 与数据库之间,为 Agent 提供适用于任何 SQLAlchemy 数据库、只读且经过策略检查的 SQL 工具。它基于真实架构为模型提供依据,将错误反馈给模型以便修正 SQL,并附带一套评估工具,可验证 SQL 是否正确并确保其持续正确。

querypilot 评估运行终端报告

快速开始

pip install querypilot
querypilot demo

querypilot demo 会创建一个小型 SaaS 数据库(querypilot-demo.db),对其运行经过验证的查询,并展示一个被拦截的 DELETE。然后试试:

querypilot schema --db sqlite:///querypilot-demo.db              # list tables
querypilot schema --db sqlite:///querypilot-demo.db invoices     # columns, keys, sample values
querypilot sql --db sqlite:///querypilot-demo.db "SELECT customer_nme FROM customers"
# Error: SQL validation failed: Unknown column: customer_nme
#   hint: Column 'customer_nme' does not exist; did you mean customers.customer_name, ...?
querypilot ask --db sqlite:///querypilot-demo.db "Which customers have the highest ARR?"

设置 OPENAI_API_KEY 或 ANTHROPIC_API_KEY 后,ask 会使用 OpenAI 或 Anthropic(--generator openai-compatible 适用于 Ollama、vLLM 和 LM Studio)。如果没有密钥,则会回退到离线演示生成器。

要连接自己的数据库,运行 querypilot init --db postgresql://[email protected]/prod,即可生成一个 querypilot.yaml,其中保存连接信息、业务提示、列描述和访问策略。所有命令、MCP 服务和 QueryPilot.from_config() 都会读取该文件,因此之后可以省略 --db。

在 Python 中:

from querypilot import QueryPilot

qp = QueryPilot.connect("postgresql://[email protected]/prod")  # dialect inferred from the URL

result = qp.execute_sql("SELECT plan, COUNT(*) FROM subscriptions GROUP BY plan")
print(result.sql)   # rewritten with a LIMIT, validated against the live schema
print(result.rows)

连接任意数据库

QueryPilot.connect(url) 会根据 URL 推断 SQL 方言,并检查表、视图、主键和外键以及注释。它也接受 SQLAlchemy Engine。

数据库URL安装
SQLitesqlite:///app.db已包含
PostgreSQLpostgresql://user:pass@host/db已包含
DuckDBduckdb:///analytics.duckdbpip install "querypilot[duckdb]"
MySQL / MariaDBmysql+pymysql://user:pass@host/dbpip install "querypilot[mysql]"
SQL Servermssql+pyodbc://…pip install "querypilot[mssql]"
Snowflakesnowflake://user@account/db/schemapip install "querypilot[snowflake]"
BigQuerybigquery://project/datasetpip install "querypilot[bigquery]"
Redshiftredshift+psycopg2://…pip install "querypilot[redshift]"
Trino / Oracle / Databrickstrino://…、oracle+oracledb://…、databricks://…querypilot[trino]、[oracle]、[databricks]

每条查询都会使用 sqlglot 按目标方言进行验证。验证包括:仅允许一条语句、仅允许 SELECT,以及表和列必须已知。它还会针对各个引擎屏蔽危险函数,例如 pg_read_file、DuckDB 的 read_csv、MySQL 的 LOAD_FILE 和 SQL Server 的 xp_cmdshell。执行时还会增加引擎自身的防护:在 Postgres、MySQL 和 Oracle 上使用只读事务,在 DuckDB 上使用只读连接,并在引擎支持的情况下设置语句超时。每条查询都在始终回滚的事务中运行。对于没有只读会话模式的数据仓库,请使用只读角色连接。

为 Agent 提供工具

每项集成都提供相同的一组精简工具。结果是精简的 JSON,错误会以 {"ok": false, "error": ..., "hints": [...]} 的形式返回,而不是抛出异常,因此 Agent 可以自行纠正。

工具功能
list_tables策略允许访问的表和视图,以及注释
describe_table指定表的列、类型、键、注释和示例值
search_schema查找与问题相关的表和列
validate_sql检查 SQL,并返回重写后的形式、错误和“你是否想要”提示
run_sql验证并运行 SQL,返回行,并在达到 max_rows 时附带 truncated 标志
ask_database自然语言 → 已验证的 SQL → 行(仅当配置了 LLM 生成器时可用)

MCP(Claude Code、Claude Desktop、Cursor、VS Code、Windsurf):

querypilot mcp-config claude-code --db postgresql://analyst@db/prod
# claude mcp add querypilot --env QUERYPILOT_DATABASE_URL=... -- /path/to/uvx --from 'querypilot[mcp]' querypilot mcp
querypilot mcp-config cursor --db sqlite:///app.db   # prints the mcpServers JSON block

生成的配置使用 uvx 的绝对路径,可避免常见的 spawn uvx ENOENT 故障。传入 --runner python 可改为启动当前环境。每个工具都带有 MCP readOnlyHint 注解。querypilot mcp --transport streamable-http --port 8000 可通过 HTTP 提供相同的工具。

OpenAI / Anthropic 工具调用:

tools = qp.as_openai_tools()            # Chat Completions; qp.toolkit().openai_responses_tools() for Responses
tools = qp.as_anthropic_tools()
result = qp.handle_tool_call(tool_call.name, tool_call.arguments)  # dict or JSON string in, dict out

Agent 框架:

from querypilot.integrations import (
    as_langchain_tools,          # LangChain / LangGraph StructuredTools
    as_llamaindex_tools,         # LlamaIndex FunctionTools
    as_openai_agents_tools,      # OpenAI Agents SDK FunctionTools
    as_claude_agent_mcp_server,  # Claude Agent SDK in-process MCP server
    claude_agent_allowed_tools,
    as_functions,                # plain typed functions: Pydantic AI, smolagents, CrewAI, ADK
)

tools = as_langchain_tools(qp, naming="langchain")  # sql_db_list_tables, sql_db_schema, sql_db_query, ...

naming="langchain" 会复用社区版 SQLDatabaseToolkit(已停止维护)中的工具名称,因此现有提示词仍可正常使用。请安装对应的额外组件:querypilot[langchain]、[llamaindex]、[openai-agents]、[claude-agent] 或 [mcp]。请参阅 examples/07_frameworks.py。

保护任何 Postgres 客户端:querypilot proxy(原型)

代理无需 SDK。它是一个 Postgres 线协议服务,会对每条语句应用相同的验证器和访问策略,无论客户端是什么。这涵盖 psql、psycopg、SQLAlchemy 和 LangChain、现有的 MCP 数据库服务、Cursor 或 Claude Code,以及 BI 工具。将客户端的连接字符串指向代理,而不是数据库:

pip install "querypilot[proxy]"
querypilot proxy --upstream postgresql://readonly@db/prod --listen 127.0.0.1:6543 --policy querypilot.yaml

psql "postgresql://[email protected]:6543/prod" -c "SELECT email FROM customers"
npx -y @modelcontextprotocol/server-postgres postgresql://[email protected]:6543/prod   # unchanged MCP server

随后,每条语句都会经过以下处理:

  • 只读强制执行。 写入、DDL、COPY 以及不在小型允许列表中的 SET 都会被拒绝。允许的语句会在 READ ONLY 事务中运行,且会话默认设为只读,因此 COMMIT; DROP TABLE … 会被拒绝。
  • 行过滤器和行数上限。 数据库收到的是重写后的 SQL,因此策略的 row_filters 在连接、子查询和 CTE 中均有效,且 LIMIT 不会超过 max_rows。
  • 脱敏。 masking_rules 按输出位置应用,因此 upper(email) AS contact 也会被脱敏。
  • Agent 可修复的错误。 拒绝操作时会返回常规 Postgres 错误,其中包含 SQLSTATE、验证器消息和 HINT,例如 Column 'nope' does not exist. Columns of orders: id, customer_id, …。
  • 审计。 每条语句都会在 --audit-log 中生成一条 JSONL 记录,其中包含客户端 SQL、实际运行的 SQL、行数,以及客户端用户和应用程序。

驱动程序握手、格式设置相关的 SET/SHOW,以及 psql \d 和 ORM 反射所需的目录查询,都会通过允许列表。可能读取数据的目录查询(例如 pg_stats 或 query_to_xml(...))会被拒绝。策略文件是常规的 querypilot.yaml,如果省略 --upstream,则会使用其中的 database_url。若不在 loopback 地址上运行,请设置 QUERYPILOT_PROXY_PASSWORD。协议支持范围、已知缺口(尚不支持 TLS 终止和服务端游标),以及故障时默认拒绝的原因,详见 docs/proxy.md。

观察 Agent 已运行的 SQL:querypilot trace 和 hooks

有些查询工具无法置于 QueryPilot 后面,例如 Palantir Foundry 的 MCP SQL 工具。对于这些工具,QueryPilot 会读取 Agent 已执行的操作。trace import 会将 Claude Code 对话记录中的每次查询工具调用与其结果配对。hooks install 会实时记录新的调用。trace doctor 会报告各类别的失败率、重试链、延迟和停滞情况,并针对每个问题指出最适合在工具、提示词还是数据中修复:

querypilot trace doctor                        # reads ~/.claude/projects
querypilot trace doctor --output report.md     # redacted, shareable
querypilot hooks install --guard --max-rows 200

使用 --guard 时,PreToolUse hook 会根据从 Agent 自身的架构调用中缓存的架构,验证每条查询。在查询到达数据仓库前,它会拒绝未知列并给出“你是否想找”的提示,阻止对以 epoch 毫秒存储的 DATE 列进行日期运算,并强制执行行数上限。Foundry 的工具默认已映射;使用 --tool NAME=SQL_FIELD[:DIALECT] 可映射其他工具。恢复的重试链可导出到 eval replay,作为回归测试套件。请参阅 docs/trace.md 和首份 Foundry 报告。

让模型生成更好的 SQL

当 QueryPilot 自行编写 SQL(ask、ask_database、评估)时,会采用文本转 SQL 文献中经测量效果提升最大的技术。来源和数据见 docs/research/developer-needs.md。

  • 以实际数据为依据的架构。 提示词会以精简 DDL 的形式展示每张表,包括主键、外键关联路径、注释,以及类似类别的文本列中的真实值:值不超过 10 个时全部展示,否则展示最常见的 3 个。示例值绝不会取自已阻止、屏蔽或经过行过滤的列。
  • 你的业务规则。 将 hints= 和 descriptions= 传给 connect(),或在 querypilot.yaml 中设置,以定义“收入仅指已付款发票”等规则。
  • 仅在需要时进行裁剪。 在 schema_token_budget(默认 6000 个 token)以内时,会发送完整架构。超出时,QueryPilot 会发送相关表,以及连接这些表所需的外键中间表。
  • 逐步收敛的修复循环。 验证错误会附带“你是否想找”的提示。数据库错误在返回给模型前会经过清理。结果为空时会触发一次温和复核。默认最多尝试 3 次。
from querypilot import AnthropicSQLGenerator, QueryPilot

qp = QueryPilot.connect(
    "duckdb:///analytics.duckdb",
    generator=AnthropicSQLGenerator(),
    hints=["Revenue means SUM(amount_cents) / 100 over invoices with status = 'paid'"],
    descriptions={"accounts.segment": "smb, midmarket or enterprise"},
)
answer = qp.ask("Revenue by segment last quarter")

QueryPilot 的存在意义

为 Agent 提供只读 SQL 访问正逐渐成为常见能力。难点在于让这种访问可靠性可测量:证明 Agent 编写的 SQL 正确、安全且快速,并且不会出现回归。

QueryPilot 可以针对执行结果真实可信的评估套件,衡量提示词、模型或配置的每项变更。你可以手动编写套件,也可以通过重放审计日志,将其生成为回归测试集。昨天在生产环境中成功运行的查询,明天也必须继续正常运行。

评估快速入门

从克隆本仓库开始(捆绑的套件会针对 tests/fixtures/demo.db 运行):

pip install -e ".[eval]"
querypilot eval run \
    --suite suites/smoke.yaml \
    --generator demo \
    --report eval-out.json
querypilot eval check \
    --report eval-out.json \
    --baseline .eval/baseline.json \
    --threshold 0.9 \
    --require-safety 1.0

示例输出(节选——完整报告见本 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.

捆绑的 suites/smoke.yaml 会针对一个精简的 SQLite 测试数据库(tests/fixtures/demo.db)运行,因此无需 LLM 密钥即可端到端运行测试框架。在你自己的项目中,querypilot eval init 会创建 suites/ 和 .eval/ 的基础结构。若要对真实生成器进行基准测试,请使用 --generator openai 或 --generator anthropic。

重放审计日志

querypilot eval replay 会将 JSONLAuditSink 写入的 JSONL 审计日志转换为一个 BenchmarkSuite,其黄金 SQL 就是此前执行过的 SQL。重新运行该套件,可针对你自己的生产流量拦截准确性回归——这是 QueryPilot 独有的能力,也是其评估定位的基础。

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

保守的默认设置:仅包含成功的 ask 记录、非空结果且没有启用访问策略的记录。--include-failures、--include-masked、--include-empty 可分别放宽各项筛选条件。

CI 门禁

querypilot eval check 会将 SuiteReport JSON 与阈值和已提交的基线进行比较,并在出现回归时以非零状态退出。项目附带了一个示例 GitHub Actions 工作流,位于 .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

检测到回归时,输出会说明哪些用例出现回归以及具体情况:

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)

有意更改后,在 main 上刷新基线:

querypilot eval run --suite suites/smoke.yaml --generator demo --report .eval/baseline.json
git commit -am "Refresh eval baseline"

编写测试套件

测试套件采用 YAML 或 JSON 格式。每个用例都包含一个问题、一条标准 SQL,以及对候选结果的架构和安全性预期。

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]

通过在同一个测试数据库中执行标准 SQL 和候选 SQL并比较行,来评估结果集的正确性。默认忽略顺序;如果标准 SQL 在顶层包含 ORDER BY,则会自动改为考虑顺序。

示例

可运行的独立示例位于 examples/ 中。它们都使用 随附的演示 SQLite 测试数据库,因此大多数示例无需 API 密钥:

示例展示内容是否需要密钥?
01_quickstart.py连接、execute_sql、离线 ask()、验证风险等级否
02_openai_tool_use.py在 OpenAI 工具使用循环中调用 as_openai_tools()OPENAI_API_KEY
03_anthropic_tool_use.py在 Anthropic 工具使用循环中调用 as_anthropic_tools()ANTHROPIC_API_KEY
04_access_control.py列封锁、行筛选和掩码否
05_custom_eval_suite/使用 querypilot eval run/check 运行自定义 YAML 测试套件否
06_mcp/运行 querypilot mcp + 可直接粘贴的 Claude MCP 配置否
07_frameworks.py在 LangChain、OpenAI Agents SDK 和 Claude Agent SDK 中使用相同工具否

请参阅 examples/README.md 了解设置方法和完整索引。

大模型 SQL 生成

要以适用于生产环境的方式通过自然语言生成 SQL,请接入大模型生成器。QueryPilot 仍将模型输出视为不可信的候选结果:它会进行验证、重写,并可在执行前要求生成器修复。

安装可选的提供商依赖:

pip install "querypilot[openai]"      # or querypilot[anthropic], or querypilot[llm] for both

OpenAI:

from querypilot import QueryPilot
from querypilot.generation import OpenAISQLGenerator

qp = QueryPilot.connect(
    "sqlite:///demo.db",
    generator=OpenAISQLGenerator(model="gpt-5.1"),
)

Anthropic:

from querypilot import QueryPilot
from querypilot.generation import AnthropicSQLGenerator

qp = QueryPilot.connect(
    "sqlite:///demo.db",
    generator=AnthropicSQLGenerator(model="claude-sonnet-4-20250514"),
)
本地/开源模型

任何兼容 OpenAI 的端点——Ollama、vLLM、LM Studio, 或 llama.cpp 的服务端——都可以通过 OpenAICompatibleSQLGenerator 使用。它复用 [openai] extra(无需额外依赖)并使用 Chat Completions API,因此 你可以以 $0 的成本对开源模型进行基准测试。API 密钥为可选项(本地服务会 忽略它);费用报告会显示 $0,而如果服务端返回用量信息,令牌计数仍会记录。

ollama pull llama3.1
pip install "querypilot[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
    ),
)

在评估框架中,使用 --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 也会读取 $QUERYPILOT_BASE_URL;如果未设置,则默认为 Ollama 的 http://localhost:11434/v1。

安全流程始终如下:

question
  -> schema-scoped prompt
  -> model candidate SQL
  -> QueryPilot validation
  -> optional repair
  -> safe execution

评估框架(库)

CLI 是 run_suite 的轻量封装,也可以直接使用 run_suite:

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

返回的 SuiteReport 是一个 Pydantic 模型,包含 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,以及完整的逐用例 case_results 列表。

安全引擎

QueryPilot 会在执行前验证 SQL,包括:

  • 使用 sqlglot 解析
  • 强制单条语句
  • 仅允许只读 SELECT 的策略
  • 检测被禁止的关键字
  • 检查已知表
  • 在可行时检查列
  • 允许/禁止表策略
  • 自动插入 LIMIT 并限制最大行数
  • 对 SELECT * 发出警告或拒绝
  • 检测笛卡尔积连接
  • 结构化策略检查
  • 查询指纹
  • 风险等级:low、medium、high、critical

在 PostgreSQL 生产环境中使用时,请将 QueryPilot 连接到专用的最小权限角色,该角色仅拥有所需架构的 USAGE 权限和表的 SELECT 权限。QueryPilot 会请求只读事务并应用语句超时,但应用层验证不能替代数据库权限。

示例:

validation = qp.validate_sql("SELECT * FROM customers")

print(validation.valid)
print(validation.risk_level)
print(validation.query_fingerprint)
print(validation.policy_checks)

如需更严格的部署:

from querypilot.core.config import SafetyPolicy

qp = QueryPilot.connect(
    "sqlite:///demo.db",
    safety_policy=SafetyPolicy(
        allow_select_star=False,
        reject_cartesian_joins=True,
    ),
)

FastAPI 服务

将 QueryPilot 作为本地安全 SQL 网关运行:

pip install "querypilot[server]"
querypilot serve --db sqlite:///demo.db --max-rows 100

或者使用环境变量:

export QUERYPILOT_DATABASE_URL=sqlite:///demo.db
querypilot serve

端点:

  • GET /health
  • GET /schema
  • POST /search-schema
  • POST /ask
  • POST /generate-sql
  • POST /validate-sql
  • POST /execute-sql
  • POST /evals/run
  • GET /audit/recent

示例:

curl -X POST http://127.0.0.1:8000/validate-sql \
  -H "content-type: application/json" \
  -d '{"sql": "SELECT * FROM customers"}'

审计追踪

QueryPilot 会记录架构搜索、SQL 生成、验证、执行和完整 ask() 流程的结构化审计事件。

每条审计记录可以包含:

  • audit_id
  • 时间戳
  • 操作
  • 问题
  • 原始 SQL
  • 重写后的 SQL
  • 验证元数据
  • 执行状态
  • 行数
  • 执行时间
  • 错误
  • 操作者/会话/应用/追踪元数据

使用默认的内存接收器:

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

或者持久化 JSONL 审计事件:

from querypilot import QueryPilot
from querypilot.audit import JSONLAuditSink

qp = QueryPilot.connect(
    "sqlite:///demo.db",
    audit_sink=JSONLAuditSink("querypilot-audit.jsonl"),
)

访问控制

只读 SQL 必不可少,但还不够。QueryPilot 还可以在执行前强制实施列级和行级访问策略。

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"),
            },
        },
    ),
)

其作用如下:

  • 拒绝查询被屏蔽列的 SQL
  • 配置 allowed_columns 时,拒绝不在允许列表中的 SQL
  • 注入必需的行筛选条件,例如 tenant_id = 42
  • 执行后遮蔽配置的结果列
  • 在验证、结果、回答和审计元数据中记录已应用的访问策略

服务和 MCP 运行时也可以接收访问策略 JSON:

querypilot serve \
  --database-url sqlite:///demo.db \
  --access-policy-json '{
    "row_filters": {"customers": "tenant_id = 42"},
    "blocked_columns": {"customers": ["ssn"]}
  }'

当前范围

已发布:

  • pip install querypilot、querypilot demo/schema/sql/ask/init/mcp-config,以及 querypilot.yaml 项目配置
  • 内置 SQLite 和 PostgreSQL 连接器,并可通过 SQLAlchemy extras 支持 DuckDB、MySQL/MariaDB、SQL Server、Snowflake、BigQuery、Redshift、Trino、Oracle 和 Databricks
  • 感知方言的 SQL 验证和重写(sqlglot)、针对各引擎的危险函数阻止列表,以及分层只读执行
  • 一个可供 MCP、OpenAI、Anthropic、LangChain、LlamaIndex、OpenAI Agents SDK、Claude Agent SDK 和普通函数使用的 Agent 工具包
  • 基于上下文的生成:键、外键、注释、符合策略要求的样例值、业务提示,以及带阈值门控的架构剪枝
  • OpenAI、Anthropic 和兼容 OpenAI 的生成器(Ollama、vLLM、LM Studio),并针对验证错误、数据库错误和空结果提供修复循环
  • 列策略、行筛选和结果遮蔽
  • 内存和 JSONL 审计日志记录
  • FastAPI 服务和 MCP 服务运行时
  • 评估驱动的测试框架:YAML/JSON 测试套件、基于执行事实的正确性评分、安全性/修复/延迟/成本指标、按标签汇总、失败类别细分、阈值违规记录、JSON 和终端报告
  • 审计日志 → 回归测试套件(querypilot eval replay)、CI 回归门禁(querypilot eval check)以及 querypilot eval init
  • SaaSPulse 防污染基准(benchmarks/)

路线图

排名后的路线图及其依据见 docs/roadmap.md 和 docs/research/developer-needs.md。接下来将推出:通过结果投票进行多候选生成、从已验证查询中检索少样本示例、数据仓库成本防护(试运行、计费字节数上限)以及多数据库 MCP 服务。

评论

支持 Markdown

第一次接触 MCP?

从协议是什么,到在客户端里装上第一个服务,这几条回答覆盖了大多数入门问题。

一个开放协议,规定了 AI 应用如何以统一方式连接外部工具与数据源。客户端负责与模型对话,MCP 服务负责提供工具、资源和提示模板,两者通过标准消息通信,同一个服务可以被不同客户端复用。