给 Agent 接上数据库并不难：提供表结构、连接串和 SQL 执行接口，一个 Demo 很快就能回答“上周新增多少客户”。真正困难的是进入生产环境之后，如何让模型理解业务含义，同时避免越权查询、凭据泄露和误写数据。
更稳妥的设计不是训练模型“更小心地写 SQL”，而是缩小它能够执行的动作范围：数据库留在服务端，由语义明确、参数受控的工具承接 Agent 请求。
为什么暴露 DDL 和任意 SQL 不够
DDL 描述的是存储结构，并不等于业务语义。面对 cust_st、biz_dt、type=3 这样的字段，模型即使生成了语法正确的 SQL，也可能回答错问题。
常见风险包括：
- 缩写和枚举没有语义：模型不知道
3表示已退款、已取消还是测试数据。 - 多表关联容易放大数据：一对多关联后直接求和，可能重复计算订单金额。
- 权限边界不清晰：用户只能查看自己的租户，但模型生成的 SQL 忘记添加
tenant_id。 - 凭据落到客户端：桌面 Agent 或本地 MCP 一旦保存生产密码，终端失陷就可能暴露数据库。
- 写操作不可预测：缺少
WHERE的UPDATE、错误的日期条件或模型重试，都可能扩大影响。
因此，生产架构最好把“模型推理”和“数据库执行”分开。Agent 负责选择业务工具和填写参数，服务端负责身份验证、权限过滤、SQL 模板、超时、限流和审计。
把数据库能力翻译成业务工具
不要给模型一个名为 execute_sql 的万能工具，可以把高频需求定义为窄接口：
get_customer_summary(tenant_id, days)：查询某租户最近一段时间的客户统计。get_order_detail(order_id)：返回经过字段脱敏的订单详情。search_inventory(sku, warehouse)：查询可售库存，而不是直接读取库存流水表。request_refund(order_id, reason)：创建退款申请，真正执行前进入审批流程。
工具描述应解释业务口径，而不仅是字段类型。例如，“活跃客户”究竟按登录、下单还是付款判断；金额使用元还是分；日期按 UTC 还是业务时区计算。这层语义可以放在工具描述、指标目录、受控视图或服务端查询模板中。
复杂分析也不必完全禁止，但可以分级处理：常用指标走固定工具；探索性查询进入只读副本或隔离的数据仓库；涉及敏感字段和写操作时要求人工确认。
一个可运行的受控查询网关
下面是一个仅使用 Python 标准库的最小示例。它会创建 SQLite 演示数据库，并暴露固定的客户统计接口。Agent 无法提交任意 SQL，租户条件和行数限制由服务端控制。
将代码保存为 safe_db_tool.py，然后运行 python safe_db_tool.py：
import json
import sqlite3
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import parse_qs, urlparse
DB_PATH = 'demo.db'
API_TOKEN = 'replace-me-in-production'
ALLOWED_DAYS = {7, 30, 90}
def init_db():
with sqlite3.connect(DB_PATH) as conn:
conn.executescript('''
CREATE TABLE IF NOT EXISTS customers (
id INTEGER PRIMARY KEY,
tenant_id INTEGER NOT NULL,
name TEXT NOT NULL,
created_at TEXT NOT NULL
);
DELETE FROM customers;
INSERT INTO customers (tenant_id, name, created_at) VALUES
(1, 'Alice', datetime('now', '-2 days')),
(1, 'Bob', datetime('now', '-20 days')),
(2, 'Carol', datetime('now', '-1 day'));
''')
class Handler(BaseHTTPRequestHandler):
def send_json(self, status, payload):
body = json.dumps(payload, ensure_ascii=False).encode('utf-8')
self.send_response(status)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.send_header('Content-Length', str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_GET(self):
if self.headers.get('Authorization') != f'Bearer {API_TOKEN}':
return self.send_json(401, {'error': 'unauthorized'})
parsed = urlparse(self.path)
if parsed.path != '/tools/customer-summary':
return self.send_json(404, {'error': 'not_found'})
params = parse_qs(parsed.query)
try:
tenant_id = int(params['tenant_id'][0])
days = int(params.get('days', ['30'])[0])
except (KeyError, ValueError):
return self.send_json(400, {'error': 'invalid_parameters'})
if tenant_id <= 0 or days not in ALLOWED_DAYS:
return self.send_json(400, {'error': 'unsupported_parameters'})
sql = '''
SELECT id, name, created_at
FROM customers
WHERE tenant_id = ?
AND created_at >= datetime('now', ?)
ORDER BY created_at DESC
LIMIT 100
'''
with sqlite3.connect(DB_PATH) as conn:
conn.row_factory = sqlite3.Row
rows = conn.execute(sql, (tenant_id, f'-{days} days')).fetchall()
self.send_json(200, {
'tenant_id': tenant_id,
'window_days': days,
'customer_count': len(rows),
'customers': [dict(row) for row in rows]
})
if __name__ == '__main__':
init_db()
print('Listening on http://127.0.0.1:8080')
HTTPServer(('127.0.0.1', 8080), Handler).serve_forever()
另开一个终端调用接口：
curl -s \
-H 'Authorization: Bearer replace-me-in-production' \
'http://127.0.0.1:8080/tools/customer-summary?tenant_id=1&days=30'
这个示例刻意体现了几条边界：
- 客户端拿不到数据库凭据。
- Agent 只能选择允许的时间范围。
- SQL 使用参数绑定，不能把输入拼接进语句。
tenant_id被写入服务端查询条件。- 查询自带
LIMIT，不会无限返回数据。
它仍然只是演示。实际部署时，应从已认证用户的令牌或会话中推导租户，而不是相信 Agent 传入的 tenant_id；API Token 也应替换为短期身份令牌，并存放在密钥管理系统中。
生产环境还需要哪些护栏
受控接口解决了任意 SQL 的问题，但不代表系统已经安全。上线前可以逐项检查：
- 最小权限账号：查询服务使用只读账号，只能访问必要的视图或存储过程。
- 服务端注入数据范围：租户、部门和用户权限来自认证上下文，不能由模型自由指定。
- 查询预算：设置数据库超时、最大扫描量、并发限制、返回行数和响应大小。
- 敏感字段治理：身份证号、手机号、密钥和支付信息应脱敏或完全不进入模型上下文。
- 写操作分级：默认禁用写入；必要写入采用幂等键、事务、影响行数上限和人工审批。
- 完整审计：记录用户、Agent、工具名、参数摘要、查询耗时、返回行数和审批结果，但避免在日志中复制敏感数据。
- 失败策略明确：权限不足或参数含糊时直接拒绝，不让模型通过反复改写参数绕过限制。
如果业务确实要求自然语言生成 SQL，可以把它安排在隔离环境中：只连接脱敏后的只读副本，执行前解析 SQL AST，拒绝写语句和未授权表，并用成本估算阻止全表扫描。即便如此，也不要只靠关键词匹配来判断 SQL 是否安全。
采用建议：从窄工具开始，而不是从万能接口开始
第一阶段可以挑选 5 到 10 个高频、口径稳定的查询，把它们做成带明确 Schema 的工具，并用真实问题测试准确率。随后再补充权限、超时、审计和人工审批。
判断设计是否合格，可以问三个问题：
- 模型理解错业务含义时，最坏能造成什么影响？
- Agent 发出的每次请求，能否追溯到具体用户、权限和工具版本？
- 即使客户端完全失陷，攻击者能否拿到生产数据库密码或执行任意 SQL？
给 Agent 接数据库的重点不是让模型拥有更多权限，而是把数据库能力压缩成可理解、可验证、可审计的业务动作。模型可以灵活推理，但数据边界必须由确定性的服务端代码掌握。