SQL 解析 API 服务（REST / HTTP）

基于 sqlglot 的 SQL 解析 / 转译 / 格式化 / 提取 / 条件注入服务，使用 FastAPI 暴露为 REST 接口。核心逻辑在 core.py，被本服务（HTTP）复用。

运行

pip install -r requirements.txt
uvicorn main:app --host 0.0.0.0 --port 8000 --reload

启动后访问交互式文档：http://localhost:8000/docs

服务默认开启 CORS（allow_origins=["*"]），便于前端跨域调用。

接口总览

方法	路径	说明
GET	/health	健康检查，返回 sqlglot 版本
GET	/dialects	列出 sqlglot 支持的所有方言
POST	/parse	将 SQL 解析为结构化 AST（JSON）
POST	/transpile	在方言之间互转 SQL（如 MySQL → PostgreSQL）
POST	/format	美化 / 标准化 SQL
POST	/validate	校验 SQL 语法是否合法（不抛异常）
POST	/extract	提取表名、列名、函数、CTE 及别名映射
POST	/add_condition	追加 WHERE 条件（自动别名，支持多表 / 多条件 / 多操作符 / 直拼原始 SQL）

所有接口在 SQL 解析 / 处理失败时会返回 400，错误信息在响应体 detail 字段中。

---

接口详情

1. 健康检查 GET /health

返回服务状态与 sqlglot 版本号。

curl http://localhost:8000/health
# {"status":"ok","sqlglot_version":"30.12.0"}

2. 方言列表 GET /dialects

返回 sqlglot 支持的全部方言名称（按字母序）。

curl http://localhost:8000/dialects
# {"dialects":["bigquery","clickhouse","mysql","postgres", ...]}

3. 解析为 AST POST /parse

请求体：

字段	类型	说明
sql	string	待解析的 SQL
read	string?	输入方言，留空自动识别
pretty	bool	AST 是否美化输出（默认 true）

curl -X POST http://localhost:8000/parse \
  -H "Content-Type: application/json" \
  -d '{"sql": "SELECT a, b FROM t WHERE a > 1"}'
# {"sql":"...","dialect":null,"ast":"SELECT a, b FROM t WHERE a > 1","json":{...}}

返回含 sql / dialect / ast / json 四个字段，其中 json 为 AST 的树形结构。

4. 方言转译 POST /transpile

字段	类型	说明
sql	string	待转译的 SQL
read	string?	输入方言，留空自动识别
write	string	目标方言（必填）

curl -X POST http://localhost:8000/transpile \
  -H "Content-Type: application/json" \
  -d '{"sql": "SELECT IFNULL(a, 0) FROM t", "write": "postgres"}'
# {"read":null,"write":"postgres","sql":"SELECT IFNULL(a, 0) FROM t","transpiled":"SELECT COALESCE(a, 0) FROM t"}

5. 格式化 POST /format

字段	类型	说明
sql	string	待格式化的 SQL
read	string?	输入方言，留空自动识别
write	string?	输出方言，默认与输入一致（可顺带转译）

curl -X POST http://localhost:8000/format \
  -H "Content-Type: application/json" \
  -d '{"sql": "select a,b from t where a>1"}'
# {"sql":"select a,b from t where a>1","formatted":"SELECT a, b\nFROM t\nWHERE a > 1"}

6. 语法校验 POST /validate

字段	类型	说明
sql	string	待校验的 SQL
read	string?	输入方言，留空自动识别

返回不抛异常，valid 为布尔，error 在合法时为 null。

curl -X POST http://localhost:8000/validate \
  -H "Content-Type: application/json" \
  -d '{"sql": "SELECT a FROM t"}'
# {"valid":true,"error":null}

7. 提取元数据 POST /extract

字段	类型	说明
sql	string	待分析的 SQL
read	string?	输入方言，留空自动识别

返回字段：

字段	说明
tables	真实表名（已剔除 CTE 名）
columns	带表限定符的列名，如 t1.id
columns_bare	裸列名（不含来源），如 id
functions	函数调用文本
ctes	CTE 别名
table_aliases	FROM/JOIN 上的显式别名映射 [{alias, table}]
cte_sources	CTE 名 → 内部真实表映射 [{cte, tables}]

curl -X POST http://localhost:8000/extract \
  -H "Content-Type: application/json" \
  -d '{"sql": "SELECT a, b FROM t1 JOIN t2 ON t1.id = t2.id"}'
# {"tables":["t1","t2"],"columns":["a","b","t1.id","t2.id"],
#  "columns_bare":["a","b","id"],"functions":[],"ctes":[],
#  "table_aliases":[],"cte_sources":[]}

8. 追加条件 POST /add_condition

为 SQL 追加 WHERE 条件，支持 多表 / 多条件 / 多操作符 / 直拼原始 SQL。自动定位每个真实表在 SQL 中的引用名（别名优先），把裸列或真实表名限定列改写为别名前缀后，注入到「该表所属查询的作用域」。

提供三种入参方式（互斥，优先级：conditions > raw > 兼容旧版）：

方式一 · 结构化（推荐，支持多表 + 多操作符） — 字段 conditions 为条件数组，每个元素：

字段	类型	说明
column	string	列名，可带引用前缀，如 "status" 或 "t1.id"
op	string	操作符，默认 =。支持 = != > >= < <= LIKE IN NOT IN BETWEEN IS NULL IS NOT NULL
value	any?	比较值；IN / NOT IN 用列表，BETWEEN 用 [下限, 上限] 两元素列表，IS NULL 可省略
table	string?	可选，条件所属真实表名（多表场景）；跨表 JOIN 条件留空，注入顶层查询
combine	string	与上一条条件的连接符 AND（默认）或 OR

curl -X POST http://localhost:8000/add_condition \
  -H "Content-Type: application/json" \
  -d '{
    "sql": "SELECT a FROM table1 AS t1 JOIN table2 AS t2 ON t1.id = t2.id WHERE t1.flag = 0",
    "conditions": [
      {"column": "status", "op": "=",   "value": 1, "table": "table1", "combine": "AND"},
      {"column": "amount", "op": ">",   "value": 100, "table": "table1"},
      {"column": "id",     "op": "IN",  "value": [1, 2, 3], "table": "table2", "combine": "OR"},
      {"column": "t1.id",  "op": "=",   "value": "t2.fid"}
    ]
  }'
# SELECT a FROM table1 AS t1 JOIN table2 AS t2 ON t1.id = t2.id
# WHERE t1.flag = 0 AND (((t1.status = 1 AND t1.amount > 100) OR t2.id IN (1, 2, 3)) AND t1.id = 't2.fid')

方式二 · 直拼原始 SQL（自动把真实表名替换为别名） — 字段 raw，直接写 WHERE 片段（用真实表名），系统自动改写为别名后注入：

字段	类型	说明
raw	string | string[]	原始 WHERE 片段（用真实表名写）；多条按 operator 合并
operator	string	多 raw 片段之间的合并方式，AND（默认）或 OR
read	string?	输入方言，留空自动识别

# 片段里用真实表名 table1 / table2 写，自动替换为别名 t1 / t2
curl -X POST http://localhost:8000/add_condition \
  -H "Content-Type: application/json" \
  -d '{"sql": "SELECT a FROM table1 AS t1 JOIN table2 AS t2 ON t1.id = t2.id WHERE t1.flag = 0",
       "raw": "table1.status = 1 AND table2.amount > 100"}'
# SELECT a FROM table1 AS t1 JOIN table2 AS t2 ON t1.id = t2.id
# WHERE t1.flag = 0 AND (t1.status = 1 AND t2.amount > 100)

# 多条 raw 片段按 operator 合并
curl -X POST http://localhost:8000/add_condition \
  -H "Content-Type: application/json" \
  -d '{"sql": "SELECT a FROM table1 AS t1",
       "raw": ["table1.status = 1", "table1.deleted_at IS NULL"], "operator": "OR"}'
# SELECT a FROM table1 AS t1 WHERE table1.status = 1 OR table1.deleted_at IS NULL

方式三 · 兼容旧版（单表 + 原始条件串）：

字段	类型	说明
sql	string	原始 SQL
table	string	真实表名（非别名），用于定位引用名
condition	string | string[]	条件片段；单条字符串或字符串数组（多条件）
operator	string	多条件合并方式，AND（默认）或 OR
read	string?	输入方言，留空自动识别

curl -X POST http://localhost:8000/add_condition \
  -H "Content-Type: application/json" \
  -d '{"sql": "SELECT a FROM table1 AS t1", "table": "table1",
       "condition": ["status = 1", "amount > 100"], "operator": "AND"}'
# SELECT a FROM table1 AS t1 WHERE t1.status = 1 AND t1.amount > 100

选型建议：

• 需要精确控制多表分作用域和每个条件的操作符 → 用 conditions（方式一）。
• 已有现成的 WHERE 片段（用真实表名写）想直接拼上去 → 用 raw（方式二）。
• 老接口 / 单表简单场景 → 用 table + condition（方式三）。

行为说明：

• 自动建立「真实表名 → 别名」映射；raw 直拼片段里用真实表名写的列（如 table1.status）会自动改写为别名（t1.status）。
• 裸列自动加引用名前缀（status = 1 → t1.status = 1）；列已带前缀（如 t1.id）则原样保留。
• 多表：条件按 table 定位到各自所属查询作用域；表位于子查询 / CTE 内部时，条件只注入到那一层。所有表在同一查询（如 JOIN）时，条件合并进同一 WHERE。raw 片段若跨多表，注入到这些表的最近公共查询作用域。
• 多操作符：除 AND / OR 连接符外，每个条件自带 op（IN / BETWEEN / LIKE / IS NULL 等）。
• 同一作用域内多条件按各自 combine 从左到右联结；若目标查询已有 WHERE，则与原有条件 AND 合并。
• 同一真实表在 SQL 中出现多次时取第一个匹配；找不到表 / 操作符不合法 / SQL 解析失败时返回 400 错误。

---

错误处理

所有接口在 SQL 解析 / 处理失败时会返回 HTTP 400，错误信息位于响应体的 detail 字段：

{ "detail": "解析失败: ..." }

常见错误与含义：

detail 片段	含义
解析失败: ...	输入 SQL 或条件片段无法被 sqlglot 解析（语法错误）
未在 SQL 中找到表: xxx	table / conditions[].table 指定的真实表在 SQL 中不存在
条件不能为空	传入的 condition / raw 为空
不支持的合并方式: xxx	operator 不是 AND / OR
不支持的操作符: xxx	conditions[].op 不在支持的操作符列表内
IN 操作符的 value 必须为列表	使用 IN / NOT IN 时 value 不是数组
BETWEEN 操作符的 value 必须为包含 [下限, 上限] 的列表	使用 BETWEEN 时 value 不是两元素数组

提示：/validate 接口不会抛异常，可通过其 valid / error 字段预先判断 SQL 是否合法，再调用其它写操作接口。

MCP 服务（供 Agent 调用）

除了 REST 接口，本服务还提供 MCP Server（mcp_server.py），让 Agent 通过 MCP 协议（stdio 传输）直接调用 SQL 解析/校验/权限能力。

特点：零额外依赖，仅依赖 sqlglot（未使用官方 mcp SDK 分叉包，避免版本/兼容风险），通过标准「换行分隔的 JSON-RPC 2.0」与任何 MCP 客户端（Claude Desktop、自定义 Agent 等）对接。

运行（Agent 以子进程方式拉起）：

python mcp_server.py

暴露的工具

工具	说明
validate_sql	校验 SQL 语法是否合法，返回 {valid, error}（不抛异常）
parse_sql	将 SQL 解析为结构化 AST（JSON）
transpile_sql	方言互转（如 MySQL → PostgreSQL）
format_sql	美化 / 标准化 SQL
extract_sql	提取表名、列名、函数、CTE 及别名映射
add_condition	追加 WHERE 条件（多表 / 多条件 / 多操作符 / 直拼原始 SQL）
check_permission	按策略校验 SQL（表/列白黑名单、禁止 DML/DDL 等），仅校验不改写
enforce_permission	校验通过后自动加固 SQL（注入行级过滤、自动补 LIMIT），返回改写结果

权限策略（policy）

check_permission / enforce_permission 共用如下策略（全部可选）：

字段	类型	说明
allow_write	bool	是否允许 DML（INSERT/UPDATE/DELETE），默认 false
allow_ddl	bool	是否允许 DDL（CREATE/DROP/ALTER/TRUNCATE），默认 false
allowed_tables	list	表白名单（非空时，不在其中的表一律拒绝）
denied_tables	list	表黑名单（命中即拒绝，优先级高于白名单）
allowed_columns	dict	列级白名单 {表: [列,...]}（仅校验带表限定符的列）
row_filters	dict	行级过滤 {表: "条件"}，enforce_permission 自动注入 WHERE
max_limit	int	SELECT 自动补 LIMIT 上限（无 LIMIT 时）

enforce_permission 示例（enforce_permission 工具）：

{
  "sql": "SELECT id, amount FROM orders",
  "policy": {
    "allow_write": false,
    "row_filters": { "orders": "tenant_id = 7" },
    "max_limit": 100
  }
}

→ 返回 SELECT id, amount FROM orders WHERE tenant_id = 7 LIMIT 100（行级隔离 + 限流已自动生效）。

Agent 接入配置示例

Claude Desktop / 兼容 MCP 的客户端（claude_desktop_config.json）：

{
  "mcpServers": {
    "sqlglot": {
      "command": "python",
      "args": ["/绝对路径/mcp-sqlglot/http/mcp_server.py"]
    }
  }
}

自定义 Agent（stdio 调用骨架）：向子进程 stdin 逐行写入 JSON-RPC 消息，读取 stdout 每行响应：

→ {"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"agent","version":"1"}}}
← {"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2024-11-05","capabilities":{"tools":{}},"serverInfo":{"name":"sqlglot-mcp","version":"1.0.0"}}}
→ {"jsonrpc":"2.0","id":2,"method":"tools/list"}
→ {"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"enforce_permission","arguments":{"sql":"SELECT * FROM orders","policy":{"row_filters":{"orders":"tenant_id = 7"},"max_limit":100}}}}

---

http/
├── main.py          # FastAPI 应用与全部 REST 接口
├── mcp_server.py    # MCP Server（stdio JSON-RPC，供 Agent 调用，零额外依赖）
├── core.py          # 核心逻辑（解析/转译/格式化/提取/条件注入/权限校验）
├── requirements.txt # 依赖（fastapi, uvicorn, sqlglot, pydantic, ...）
└── README.md        # 本文档