关于 QueryPilot
QueryPilot 为使用 SQLAlchemy 数据库的 AI Agent 提供策略管控的 SQL 网关,将只读校验、架构信息与审计、追踪和评测结合起来,衡量查询可靠性。
主要功能
- 01仅允许 SELECT 的 SQL 校验
- 02数据库架构发现与搜索
- 03访问策略、脱敏和行数限制
- 04执行 SQL 并提供修正提示
- 05审计、追踪与评测工具
适用场景
- 01在只读限制下让 Agent 查询数据库
- 02依据实时数据库架构生成 SQL
- 03检查并诊断 Agent 的 SQL 错误
- 04通过评测套件进行 SQL 回归测试
MCP 相关
QueryPilot 向 MCP 客户端提供架构查询、SQL 校验与执行工具,并可选提供自然语言数据库问答。相同工具可通过 stdio 或 Streamable HTTP 提供。
配置
本地服务的配置在各客户端中通用,复制后粘贴到客户端的 MCP 配置文件即可。
uvx querypilotQUERYPILOT_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+)。各客户端的配置文件位置不同,请参考客户端文档。
工具
本地服务需要在你的电脑上运行后才能读取工具,维护者暂未提供工具列表。
详细介绍
QueryPilot
为 AI Agent 提供安全、可衡量的 SQL。
QueryPilot 位于 LLM Agent 与数据库之间,为 Agent 提供适用于任何 SQLAlchemy 数据库、只读且经过策略检查的 SQL 工具。它基于真实架构为模型提供依据,将错误反馈给模型以便修正 SQL,并附带一套评估工具,可验证 SQL 是否正确并确保其持续正确。
快速开始
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 | 安装 |
|---|---|---|
| SQLite | sqlite:///app.db | 已包含 |
| PostgreSQL | postgresql://user:pass@host/db | 已包含 |
| DuckDB | duckdb:///analytics.duckdb | pip install "querypilot[duckdb]" |
| MySQL / MariaDB | mysql+pymysql://user:pass@host/db | pip install "querypilot[mysql]" |
| SQL Server | mssql+pyodbc://… | pip install "querypilot[mssql]" |
| Snowflake | snowflake://user@account/db/schema | pip install "querypilot[snowflake]" |
| BigQuery | bigquery://project/dataset | pip install "querypilot[bigquery]" |
| Redshift | redshift+psycopg2://… | pip install "querypilot[redshift]" |
| Trino / Oracle / Databricks | trino://…、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 /healthGET /schemaPOST /search-schemaPOST /askPOST /generate-sqlPOST /validate-sqlPOST /execute-sqlPOST /evals/runGET /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 服务。
评论
一个开放协议,规定了 AI 应用如何以统一方式连接外部工具与数据源。客户端负责与模型对话,MCP 服务负责提供工具、资源和提示模板,两者通过标准消息通信,同一个服务可以被不同客户端复用。