使用说明
具有列级访问控制的SQL代理文本
一个MCP服务器，提供具有列级访问控制的自然语言到SQL功能。作为一个学习项目，与克劳德一起探索代理模式。
它做什么
- 将自然语言问题转换为BigQuery SQL
- 使用确定性SQL重写强制列级访问控制
- 使用sqlglot解析SQL以进行精确的列检测（无子字符串匹配）
- 通过外键检测自动查找连接路径
建筑
User Query: "What are total sales?"
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Main LLM (generates SQL) │
│ Output: SELECT SUM(amount) FROM sales │
└─────────────────────┬───────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Access Control Check (sqlglot) │
│ • Parse SQL, extract column references │
│ • Check: is sales.amount in restricted columns? │
│ • YES → apply rule │
└─────────────────────┬───────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Deterministic SQL Rewriting │
│ • Find join path: sales → ae_territories → account_execs │
│ • Add JOINs and WHERE clause │
│ Output: │
│ SELECT SUM(amount) FROM sales │
│ INNER JOIN ae_territories ON sales.territory = ... │
│ INNER JOIN account_execs ON ae_territories.ae_id = ... │
│ WHERE account_execs.ae_email = 'bob@acme.com' │
└─────────────────────┬───────────────────────────────────────┘
│
▼
┌───────────┐
│ BigQuery │
└───────────┘
关键改进
- 执行路径中没有LLM -SQL重写是确定性的，不是基于LLM的
- 精确列匹配 -
amount不匹配amount_ytd（使用SQL解析器） - FK自动检测 -通过分析列名查找连接路径
- 多跳加入 -支持
sales → territories → account_execs
设置
-
安装依赖项
python -m venv venv source venv/bin/activate pip install -r requirements.txt -
配置凭据
- 将您的BigQuery服务帐户JSON设置为
mcp_server/service_account.json - 创建
.env使用您的Anthropic API密钥：ANTHROPIC_API_KEY=sk-ant-...
- 将您的BigQuery服务帐户JSON设置为
用法
独立聊天模式
python terminal_chat.py
注： 聊天模式使用 Claude代理SDK 这需要 克劳德代码 待安装。
作为MCP服务器（适用于Claude Desktop或其他MCP客户端）
添加 ~/Library/Application Support/Claude/claude_desktop_config.json:
{
"mcpServers": {
"text2sql": {
"command": "python",
"args": ["/path/to/mcp_server/mcpserver.py"]
}
}
}
设置访问控制
1.配置数据集和用户表
> configure dataset myproject.mydataset
> configure users_table account_execs
2.检测外键（或手动设置）
> detect_foreign_keys
Detected:
sales.territory -> ae_territories.territory
ae_territories.ae_id -> account_execs.ae_id
> set_foreign_key sales territory ae_territories territory # if needed
3.设置访问规则
> set_access_rule sales.amount account_execs.ae_email
Access rule set for 'sales.amount':
Controlled by: account_execs.ae_email
User field: ae_email
Join path: sales.territory = ae_territories.territory -> ae_territories.ae_id = account_execs.ae_id
4.作为用户进行测试
> manage_identity set bob@acme.com
Now simulating: Bob (ae_email: bob@acme.com)
> what are total sales?
[Access control applied: sales.amount filtered by account_execs.ae_email='bob@acme.com']
SQL: SELECT SUM(amount) FROM sales
INNER JOIN ae_territories ON sales.territory = ae_territories.territory
INNER JOIN account_execs ON ae_territories.ae_id = account_execs.ae_id
WHERE account_execs.ae_email = 'bob@acme.com'
可用工具
| 工具 | 目的 |
|---|---|
read_context |
阅读模式/示例草稿栏 |
write_context |
添加到草稿栏 |
run_query |
使用访问控制执行SQL |
preview_query |
预览访问控制而不执行 |
list_schema_columns |
列出所有表和列 |
detect_foreign_keys |
自动检测FK关系 |
set_foreign_key |
手动设置FK关系 |
set_access_rule |
添加访问控制规则 |
list_access_rules |
显示所有规则 |
remove_access_rule |
删除规则 |
manage_identity |
切换模拟用户 |
configure_dataset |
设置BigQuery数据集 |
运作原理
- 管理员设置规则:
sales.amount被控制account_execs.ae_email - FK检测：系统从以下位置查找路径
sales到account_execs - 查询到达:
SELECT SUM(amount) FROM sales - 立柱检查：sqlglot解析SQL，查找
sales.amount（完全匹配） - SQL重写：确定性地添加JOIN+WHERE（无LLM）
- 执行：修改后的SQL对BigQuery运行
文件
| 文件 | 目的 |
|---|---|
mcp_server/mcpserver.py |
基于sqlglot访问控制的MCP服务器 |
mcp_server/config.json |
访问规则和FK定义 |
mcp_server/context.md |
用于模式/示例的Scratchpad |
terminal_chat.py |
独立终端聊天界面 |
test_access_control.py |
门禁单元测试 |
局限性
这是一个 学习原型，未准备好生产：
- FK检测使用命名约定（可能会错过关系）
- 仅限单值筛选器（无IN子句）
- 在受限列检查中不支持子查询
对于生产，请考虑：
- BigQuery行级安全策略
- 每个角色的视图
- 显式FK元数据
许可证
麻省理工学院
