Skip to main content
Glama
xieyuhua

sqlglot MCP server

by xieyuhua
OverviewSchemaRelated ServersScoreDiscussions
Python
Remote

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=["*"]），便于前端跨域调用。

Related MCP server: mcp-server-mssql

接口总览

方法

路径

说明

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        # 本文档

This server cannot be installed

F
license - not found
-
quality - not tested
C
maintenance

How are these scores calculated?

Maintenance

–Maintainers
–Response time
–Release cycle
–Releases (12mo)
Commit activity

Resources

Unclaimed servers have limited discoverability.

Looking for Admin?

If you are the server author, to access and configure the admin panel.

Related MCP Servers

View all related MCP servers

Related MCP Connectors

  • Supabase

    MCP server for interacting with the Supabase platform

  • nlqdb — analytical memory for AI agents

    Analytical memory for AI agents: a real Postgres queried in plain English over MCP. One command.

  • Frenchie

    OCR, transcription, file extraction, and image generation for AI agents via MCP.

View all MCP Connectors

Latest Blog Posts

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/xieyuhua/mcp-sqlglot'

If you have feedback or need assistance with the MCP directory API, please join our Discord server