别让 Agent 直接查业务库:用 MCP 接入受控的语义网关

2026-09-18 26 预计阅读时间: 1 分钟
来源: oschina.net AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:5 分钟

给 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。


相关推荐