给 Agent 配上数据库连接,只解决了“能不能查”的问题。到了业务场景,“收入”究竟按支付时间还是下单时间统计、是否排除退款、当前用户能看哪些租户,才决定答案是否可信。如果把这些判断都交给模型临场生成 SQL,语义错误和越权查询就可能同时出现。
把 MCP 工具收窄为业务能力
MCP 可以负责把工具暴露给 Agent,但工具不一定要叫 execute_sql。更稳妥的边界是:Agent 选择一个已定义的指标并填写参数;语义网关负责解释指标、绑定身份、执行参数化查询,再返回有限的结果。
例如,将“查询已支付订单金额”定义为 get_paid_revenue(start_date, end_date)。其中“已支付”的口径写在服务端,租户身份来自服务端验证的凭据,而不是模型提交的 tenant_id。这样,模型可以决定查哪个时间段,却不能通过改写 SQL 改变指标或权限。
一个可运行的最小网关
下面用 Python 标准库模拟网关。它不是某个 MCP 产品的现成接口,而是可以放在 MCP 工具背后的 HTTP 服务:请求只接受日期,服务端固定查询口径并绑定租户。示例数据和令牌仅用于本机演示。
保存为 gateway.py,运行 python gateway.py:
from datetime import date
from http.server import BaseHTTPRequestHandler, HTTPServer
import json
import sqlite3
TOKENS = {"demo-token": "tenant-a"} # 演示用;生产环境应验证真实身份
DB = sqlite3.connect(":memory:", check_same_thread=False)
DB.execute("CREATE TABLE orders (tenant TEXT, paid_on TEXT, status TEXT, cents INTEGER)")
DB.executemany("INSERT INTO orders VALUES (?, ?, ?, ?)", [
("tenant-a", "2026-01-05", "paid", 1200),
("tenant-a", "2026-01-06", "pending", 300),
("tenant-b", "2026-01-05", "paid", 9900),
])
class Handler(BaseHTTPRequestHandler):
def do_POST(self):
if self.path != "/metrics/paid-revenue":
self.send_error(404)
return
tenant = TOKENS.get(self.headers.get("Authorization", "").removeprefix("Bearer "))
if tenant is None:
self.send_error(401)
return
try:
length = int(self.headers.get("Content-Length", "0"))
if not 0 < length <= 1024:
raise ValueError("invalid body size")
args = json.loads(self.rfile.read(length))
start, end = (date.fromisoformat(args[k]) for k in ("start_date", "end_date"))
if not 0 < (end - start).days <= 31:
raise ValueError("date range must be 1–31 days")
except (ValueError, KeyError, TypeError, json.JSONDecodeError):
self.send_error(400, "invalid date range")
return
cents = DB.execute(
"SELECT COALESCE(SUM(cents), 0) FROM orders "
"WHERE tenant = ? AND status = 'paid' AND paid_on >= ? AND paid_on < ?",
(tenant, start.isoformat(), end.isoformat()),
).fetchone()[0]
body = json.dumps({"metric": "paid_revenue", "amount_cents": cents}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
HTTPServer(("127.0.0.1", 8000), Handler).serve_forever()
另开一个终端验证;end_date 是不包含在内的结束日期:
curl -sS http://127.0.0.1:8000/metrics/paid-revenue \
-H 'Authorization: Bearer demo-token' \
-H 'Content-Type: application/json' \
-d '{"start_date":"2026-01-01","end_date":"2026-02-01"}'
# {"metric": "paid_revenue", "amount_cents": 1200}
接入 MCP 时,可以把同名工具的参数限制为这两个日期,让适配器调用上述端点。适配器应从已认证会话获取令牌,不要让 Agent 自己填写或读取服务端令牌。 这里的 HTTP 接口是演示网关边界,并非声称所有 MCP 客户端都按这种方式配置。
上线前检查边界
固定 SQL 和参数化查询能减少模型拼接 SQL 的风险,但这个小示例还不是生产级权限系统。上线时应使用经过验证的用户身份和服务端权限映射;为指标明确支付、退款、时区等口径;限制查询成本与返回数据量,并记录工具调用和查询审计。对需要自由探索的数据分析任务,也可以提供单独的只读、脱敏环境,不要因此把生产库的任意 SQL 权限交给 Agent。