text-to-sql-mcp

一个 MCP 服务器，将自然语言问题转换为针对真实、非平凡多表模式的 SQL——并且从不信任返回的 SQL。每个生成的查询在执行之前都会通过 sqlglot 解析为真实的 AST，并通过验证器运行：非 SELECT 语句、多语句注入、危险函数、幻觉表/列以及大表无限制的全表扫描都会在结构上被拒绝，而不是依赖 LLM 自身的判断。

模型的输出是建议，而不是命令。由验证器决定实际执行什么。

状态

规格中的里程碑 M1–M4 已实现并测试：模式内省、NL→SQL 生成（真实的 Anthropic/OpenAI 后端 + 确定性的离线回退）、AST 验证器、带标签的评估框架和 MCP 服务器包装器。有关故意推迟的内容，请参阅 风险 / 未决问题。

架构

NL question
    │
    ▼
schema introspection (introspection.py)  ──► SQLite catalog (sqlite_master + PRAGMA table_info)
    │  grounds the prompt in the *real* schema, not guessed names
    ▼
LLMClient.generate_sql()  (llm/factory.py picks one)
    │  - AnthropicLLMClient  (real Claude API call, used if ANTHROPIC_API_KEY is set)
    │  - OpenAILLMClient     (real OpenAI API call, used if OPENAI_API_KEY is set)
    │  - RuleBasedLLMClient  (deterministic fixture lookup, the offline default)
    ▼
candidate SQL string  ──────────────────►  never trusted past this point
    │
    ▼
validate_sql()  (validator/ast_validator.py)
    │  1. parseable?                       -- sqlglot.parse()
    │  2. exactly one statement?           -- reject `SELECT ...; DROP ...`
    │  3. root node is SELECT/UNION/       -- allow-list, not a blocklist
    │     INTERSECT/EXCEPT?
    │  4. no SELECT ... INTO?
    │  5. no dangerous function calls?     -- load_extension, readfile, writefile, ...
    │  6. every table exists in schema?
    │  7. every resolvable column exists?  -- best-effort, conservative
    │  8. large table + no WHERE + not     -- the "unbounded full scan" check
    │     a bounded/aggregate result?
    │
    ├── reject ──► {rejected: true, rejection_reason: "..."}
    │
    ▼ ok
execute_query()  (execution.py)  ──► SQLite opened `mode=ro` + `PRAGMA query_only=ON`
    │  (defense in depth: even a validator bug can't write, because the
    │   connection itself refuses)
    ▼
{sql, rows, rejected: false}
    │
    ▼
query_log  (query_log.py)  ──► every call logged, rejection rate reported
                                 separately from accuracy (see below)

上述所有内容都被封装为一个 MCP 服务器（mcp_server.py，基于官方 mcp Python SDK 的 FastMCP 构建），它确切地暴露了规范 API 契约中的两个工具：

• list_schema() -> {tables: [{name, columns: [{name, type}]}]}

• ask(question: str) -> {sql, rows, rejected, rejection_reason}

为什么用 SQLite 而不是 Postgres

规范专门针对 Postgres。本环境没有运行中的 Postgres 服务器，也没有 Docker 守护进程，因此目标数据库改为 SQLite——这是一种深思熟虑的、有文档记录的替代，而不是疏忽。内省层（introspection.py）是唯一真正特定于 SQLite 的部分（用 sqlite_master + PRAGMA table_info 代替 information_schema）；验证器、执行层和 MCP 包装器操作于解析后的 AST，不关心也不知道是哪个数据库产生了该模式。Postgres 升级路径： 将 db/connection.py 中的 sqlite3.connect(..., mode=ro) 替换为针对只读角色打开的 psycopg 连接，将 introspection.py 的两个查询改写为针对 information_schema.tables/columns 的查询，并将 dialect="postgres" 传递给 validate_sql()——sqlglot 原生支持这两种方言，因此 AST 逻辑本身保持不变。

为什么用合成数据集而不是实时开放数据拉取

规范建议使用真实的城市/政府开放数据门户。db/seed.py 改为生成一个合成但逼真的市政数据集——12 张表，模拟真实的许可/检查/违规模式（NYC DOB、Chicago building permits）——从固定种子确定性生成，完全离线。这是一个刻意的权衡，而非懒惰：它使 init-db 在零网络依赖的情况下可重现（没有不稳定的 CI、没有速率限制、没有门户停机），并在发布演示之前就规避了规范本身标记为风险（§13）的许可问题。按照规范自己的标准，该模式确实是真正非平凡的：12 张表，外键深达三跳（payments → violations → properties），以及故意的列名歧义（status 出现在 permits、licenses、violations 和 complaints 上；type 出现在四张不同的表上），这真实地锻炼了验证器的模式接地逻辑。

安装

pip install -e .

需要 Python 3.10+。真实 LLM 后端的可选额外依赖（在已有这些依赖的开发环境中已安装；仅在你没有时才需要）：

pip install -e ".[anthropic]"   # anthropic SDK
pip install -e ".[openai]"      # openai SDK

快速入门 — 针对真实 SQLite 数据库的真实演示运行

# 1. Build the demo database (12 tables, ~8,700 rows, deterministic seed 42)
text-to-sql-mcp init-db

# 2. Inspect the schema the model is grounded in
text-to-sql-mcp schema

# 3. Ask a question -- no API key needed, uses the deterministic rule-based backend
text-to-sql-mcp ask "How many permits are there in total?"

backend:  rule-based
sql:      SELECT COUNT(*) AS count FROM permits
rejected: False
rows (1):
[
  {
    "count": 2600
  }
]

一个重连接的问题：

text-to-sql-mcp ask "How many permits does each contractor hold?"

backend:  rule-based
sql:      SELECT c.business_name, COUNT(*) AS permit_count FROM permits p JOIN contractors c ON p.contractor_id = c.contractor_id GROUP BY c.business_name ORDER BY permit_count DESC
rejected: False
rows (50):
[
  { "business_name": "Garcia Builders", "permit_count": 167 },
  { "business_name": "Kim Builders", "permit_count": 144 },
  { "business_name": "Miller Plumbing Co", "permit_count": 119 },
  ...
]

一个歧义问题——故意不默默解析为一种猜测（参见 边界情况）：

text-to-sql-mcp ask "Show me the recent activity."

backend:  rule-based
sql:      AMBIGUOUS: 'Recent activity' could mean permits, inspections, violations, complaints, or payments -- and over what time window. Please specify which type of record and a date range or property.
rejected: True
reason:   Question is ambiguous and was not silently resolved to one interpretation. Clarification needed: ...

证明阻止破坏性 SQL 的是验证器而非模型——这里使用一个 fixture 客户端来代表一个被攻破/提示注入的模型，该模型总是顺从破坏性请求：

python - <<'EOF'
from text_to_sql_mcp.config import get_settings
from text_to_sql_mcp.service import ask

class MaliciousFixtureLLMClient:
    name = "malicious-fixture"
    def generate_sql(self, question, schema):
        return "DROP TABLE permits"

result = ask("Please delete all the permit records.",
             llm_client=MaliciousFixtureLLMClient(), settings=get_settings())
print("sql:     ", result.sql)
print("rejected:", result.rejected)
print("reason:  ", result.rejection_reason)
EOF

sql:      DROP TABLE permits
rejected: True
reason:   Statement type 'Drop' is not a read-only SELECT/UNION/INTERSECT/EXCEPT query. Only SELECT-family statements may be executed.

现在检查操作员看到的内容——拒绝率，与准确率分开报告（见下文）：

text-to-sql-mcp rejection-report

{
  "total_queries": 5,
  "rejected": 3,
  "accepted": 2,
  "rejection_rate": 0.6,
  "rejected_by_reason": {
    "ambiguous_question": 1,
    "generation_failed": 1,
    "not_select": 1
  }
}

（那个 0.6 不是要达到的目标数字——它是本次会话中实际提出的问题组合在这次运行中产生的数字。重新运行 init-db 并重复上述命令可以精确重现它，因为种子数据和基于规则的后端都是确定性的。）

标注评估集上的准确率

text-to-sql-mcp eval

真实输出，基于规则的后端，此种子（25 个问题：8 个简单 / 10 个中等 / 7 个困难，涵盖规范要求的 20–30 个问题）：

{
  "total_questions": 25,
  "correct": 20,
  "accuracy": 0.8,
  "rejected": 6,
  "rejection_rate": 0.24,
  "by_difficulty": {
    "easy":   { "total": 8,  "correct": 8, "accuracy": 1.0 },
    "medium": { "total": 10, "correct": 8, "accuracy": 0.8 },
    "hard":   { "total": 7,  "correct": 4, "accuracy": 0.5714 }
  },
  "by_join_heaviness": {
    "simple":     { "total": 17, "correct": 17, "accuracy": 1.0 },
    "join_heavy": { "total": 8,  "correct": 3,  "accuracy": 0.375 }
  }
}

这个 80% 不是巧合，也不是一个可以表面接受的说法——它是由一个深思熟虑的设计选择产生的：基于规则的后端识别出 25 个问题中的 20 个，并对其余 5 个*报错（raise）*而不是猜测（参见 llm/rule_based.py 的 _UNANSWERED_IDS）。评估框架通过真实的 ask() 管道运行每个问题，并将实际返回的行与针对同一数据库全新执行的金标准查询进行比较——它不是手工维护的、可能悄悄偏离种子数据的预期数字。准确率随难度下降（100% → 80% → 57%），并且在重连接的问题上显著更低（37.5% 对比简单问题的 100%），纯粹是因为基于规则的后端是一个查找表，而不是因为框架或验证器做了任何不同的事情——这正是规范的验收标准所要求的诚实信号（“准确率被测量和报告，而不仅仅是声称”）。

配置了真实的 ANTHROPIC_API_KEY 后，ask()/eval 改为通过 AnthropicLLMClient 路由（参见 哪些需要真实 API 密钥），准确率将反映实际开放式 NL→SQL 质量，而不是 fixture 覆盖范围——本环境未运行该配置，因为这里没有配置 API 密钥，README 也不会为其声称任何数字。

对抗性验证 — 100% 拒绝，经两种方式测试

pytest tests/test_validator_adversarial.py tests/test_service_adversarial.py -v

• tests/test_validator_adversarial.py —— 29 个故意恶意/畸形的 SQL 字符串（DROP、DELETE、UPDATE、INSERT、CREATE TABLE AS SELECT、ALTER、PRAGMA、ATTACH DATABASE、GRANT、VACUUM/REINDEX、通过 ; 的堆叠语句注入、load_extension/readfile/writefile、SELECT ... INTO、空/垃圾输入）直接输入 validate_sql() —— 29/29 被拒绝，另有 2 个专门测试精确固定了如何处理注释夹带的第二条语句（该文件共 33 个测试函数）。

• tests/test_service_adversarial.py —— 在 ask() 级别提供相同的保证，通过一个 总是顺从 对抗性自然语言提示而不是拒绝它的 fixture LLM 客户端——证明阻止执行的是验证器，“而不是指望模型拒绝”（规范对这条验收标准自己的措辞）。即使 fixture 模型从不说“不”，8/8 的对抗性提示仍然最终被拒绝。

验证器的 _ALLOWED_ROOT_TYPES 是一个允许列表（Select/Union/Intersect/Except），而不是危险关键字的阻止列表——sqlglot 识别的每个 DML/DDL/管理语句在构造上都会解析为不同的、不在允许列表中的 AST 节点类型，因此没有需要保持同步的关键字列表，也没有办法通过重命名或伪装让破坏性语句通过。

哪些需要真实 API 密钥，哪些今天可以独立运行

llm/factory.py 自动选择后端：如果设置了 ANTHROPIC_API_KEY 则使用 Anthropic，否则如果设置了 OPENAI_API_KEY 则使用 OpenAI，否则使用基于规则的回退——切换无需更改代码。无论哪个后端生成的 SQL，AST 验证器的行为都是相同的——这正是该架构的实际要点（模型的输出是建议，永不被信任），也是为什么对抗性测试套件和通用验证器测试根本不需要任何 LLM 后端就能证明安全属性的原因。

已处理的边界情况

• 歧义 NL 问题（§9）：提示指示 LLM 回复 AMBIGUOUS: <clarifying question> 而不是 SQL，而不是默默选择一种解释；service.ask() 检测到这种情况并返回 rejected: true，将澄清说明作为原因，绝不执行猜测。参见 test_ask_handles_ambiguous_question_without_silently_guessing。

• 重连接的问题单独跟踪（§9）：EvalQuestion.is_join_heavy + EvalReport.accuracy_by_join_heaviness() —— 参见上面真实的 100% 对比 37.5% 的拆分。

• 伪装成 SELECT 的提示注入（§9）：允许列表根类型检查意味着无论提示如何要求，DROP/DELETE/等都无法通过；参见上面的对抗性测试套件。

• 非常大表的全表扫描（§9）：_find_unfiltered_large_table_scan 会标记在行数超过阈值（默认 500）的表上且没有 WHERE 的 SELECT，并且其结果没有其他方式限定边界（没有 GROUP BY、没有 LIMIT、不是纯聚合）。最后这个子句是超出规范字面措辞的刻意细化：没有它，像 SELECT COUNT(*) FROM permits 这样的普通报表查询会和真正昂贵的 SELECT * FROM permits 一起被拒绝，这会使验证器对真实报表毫无用处。参见 test_pure_aggregate_on_large_table_passes_without_where 对比 test_unfiltered_select_star_on_large_table_is_rejected。

• 模式不匹配（幻觉表/列名）：根据内省到的模式进行结构性检查，而不是与硬编码列表进行字符串匹配——test_unknown_table_is_rejected、test_unknown_column_on_known_table_is_rejected。列存在性检查刻意保守（跳过多个连接表中模棱两可的无限定引用），以避免误拒绝合法查询——参见 _find_unknown_column 的 docstring。

MCP 服务器

text-to-sql-mcp serve

通过 stdio 运行服务器。将任意 MCP 客户端指向它（例如将其添加到 Claude Desktop 的配置中，或使用 mcp Python SDK 的 ClientSession 驱动它）。已在 tests/test_mcp_server.py 中通过 mcp.shared.memory.create_connected_server_and_client_session 进行了端到端测试——一个真实的 ClientSession 通过内存传输与真实的 FastMCP 服务器通信，并像外部 MCP 客户端那样精确调用 list_tools() 和 call_tool(...)，而不仅仅是直接调用底层 Python 函数。

测试

pytest

91 个测试，全部通过。分类如下：

• test_introspection.py — 模式内省准确性（表、列、行数、大表阈值）

• test_rule_based_llm.py — 确定性后端覆盖率，包括其有意的缺口

• test_execution.py — 只读强制（纵深防御）、行数上限截断

• test_validator_general.py — 合法查询通过、模式锚定、有界与无界大表逻辑

• test_validator_adversarial.py — 29 个对抗性用例，100% 拒绝

• test_service_ask.py / test_service_adversarial.py — 端到端 ask()，包括全流水线对抗性证明

• test_eval_runner.py — 评估框架本身（结构、准确率分解、歧义处理）

• test_query_log.py — 日志记录 + 面向操作员的拒绝率报告，包括真实的 ask() 集成测试

• test_mcp_server.py — 通过真实的 MCP ClientSession 进行端到端测试

配置

cp .env.example .env

将 .env.example 复制为 .env 并填入你有的配置——一切都有可用的默认值：

风险 / 未决问题 / 范围缩减

如实说明哪些内容未包含在内，依据规范自身的 §13 以及本作品集的工程判断要求：

• 按照规范的字面表述，本应使用 Postgres 而非 SQLite。 此环境中没有可用的 Postgres 服务器或 Docker 守护进程。上文记录了替换方案 + 升级路径；AST 验证器和执行层设计特意保持方言无关，因此将来无需重写。

• 使用合成数据集，而非从实时开放数据门户拉取。 这是为了离线可复现性并绕开规范本身标记为风险项的许可问题而做出的刻意权衡——参见上文专门部分。

• 基于规则的后端是一个固定查找表，而非通用模型。 这是根据本作品集的环境约束（此处未配置 LLM API 密钥）明确且有意的设计——真实的 Anthropic/OpenAI 后端存在且已完整实现，并共享完全相同的验证器/执行路径；只是在此环境中从未使用真实 API 密钥运行过，因此不声称任何实时生成的准确率数据。

• 列存在性检查是尽力而为，而非穷尽式。 它故意跳过跨多表连接中含义不明的非限定列引用，以免冒险产生误拒——这一点记录在 _find_unknown_column 的文档字符串中。表存在性检查（针对幻觉表更有价值的防护）则没有类似的规避。

• 没有查询结果缓存 / 连接池。 每次 ask() 都会打开一个新的只读 SQLite 连接。在这种规模下（单文件演示数据库）没有问题；但在高 QPS 生产环境使用前需要关注。

• 歧义检测依赖于 LLM 后端遵循 AMBIGUOUS: 约定。 基于规则的后端针对其唯一一个刻意制造歧义的固定问题实现了该约定；真实的 Anthropic/OpenAI 调用通过共享系统提示词（llm/prompt.py）被指示遵循相同的约定，但这只是提示层面的配合，并非由验证器独立强制（开放式的歧义检测不是 AST 检查器能够验证的事情）。

• MISSING_WHERE_LARGE_TABLE 的有界结果启发式是对规范字面表述的细化，严格来说并非限制，但值得作为一项判断性取舍加以说明：它将 GROUP BY、LIMIT 和纯聚合投影视为豁免于缺失 WHERE 检查。关于推理过程以及界定该行为分界两侧的两个测试，请参见边缘情况一节。

许可协议

MIT — 参见 LICENSE。