Google 开源了一个14900 Star 项目，让 AI Agent 碰数据库不再是定时炸弹

# 典型的危险实现：给 LLM 一个「execute_sql」工具
tools = [
    {
        "name": "execute_sql",
        "description": "Execute any SQL query on the database",
        "parameters": {
            "query": {"type": "string", "description": "SQL query to execute"}
        }
    }
]

AI Agent / LLM
     ↓ MCP 协议调用工具
MCP Toolbox Server（YAML 配置的工具集合）
     ↓ 参数化查询
数据库（MySQL / PostgreSQL / Redis / ...）

kind: tool
name:get_user_orders
type:postgres-sql
source:prod-postgres
description:"查询指定用户的订单列表，仅返回最近 30 天"
parameters:
-name:user_id
    type:integer
    description:"用户 ID"
-name:limit
    type:integer
    description:"返回条数，最多 50"
statement: |
  SELECT order_id, amount, status, created_at
  FROM orders
  WHERE user_id = $1
    AND created_at > NOW() - INTERVAL '30 days'
  ORDER BY created_at DESC
  LIMIT LEAST($2, 50)

annotations:
  readOnlyHint: true      # 声明这是只读操作
  destructiveHint: false  # 声明不会有破坏性
  idempotentHint: true    # 声明是幂等的

kind: tool
name:update_user_profile
type:postgres-sql
source:prod-postgres
authRequired:
-google-oidc-service    # 必须通过这个认证服务才能调用
parameters:
-name:user_id
    type:string
    authServices:
      -name:google-oidc-service
        field:sub          # 从 JWT 的 sub claim 自动填充，防止越权

docker run --rm -i \
  -v $(pwd)/tools.yaml:/app/tools.yaml \
  -p 5000:5000 \
  us-central1-docker.pkg.dev/database-toolbox/toolbox/toolbox:latest \
  --tools-file /app/tools.yaml

# 数据源定义：一个 PostgreSQL 连接
sources:
my-pg:
    kind:postgres
    host:127.0.0.1
    port:5432
    database:myapp
    user:readonly_user       # 注意：数据库账号本身也只有只读权限
    password:${PG_PASSWORD}# 密码通过环境变量注入，不写死在配置里

# 工具定义：这个工具允许 AI Agent 做什么
tools:
search_products:
    kind:postgres-sql
    source:my-pg
    description:|
      根据关键词搜索商品，支持模糊匹配。
      只返回上架状态的商品，不包含下架和删除的。
    parameters:
      -name:keyword
        kind:string
        description:"搜索关键词"
      -name:max_price
        kind:float
        description:"价格上限（元），不传则不限制"
    statement:|
      SELECT product_id, name, price, category
      FROM products
      WHERE status = 'active'
        AND name ILIKE '%' || $1 || '%'
        AND ($2::float IS NULL OR price <= $2)
      ORDER BY created_at DESC
      LIMIT 20
    annotations:
      readOnlyHint:true

get_order_detail:
    kind:postgres-sql
    source:my-pg
    description:"查询订单详情，包含订单行和物流信息"
    parameters:
      -name:order_id
        kind:string
        description:"订单ID"
    statement:|
      SELECT o.order_id, o.status, o.total_amount,
             oi.product_name, oi.quantity, oi.unit_price,
             s.carrier, s.tracking_number
      FROM orders o
      LEFT JOIN order_items oi ON o.order_id = oi.order_id
      LEFT JOIN shipments s ON o.order_id = s.order_id
      WHERE o.order_id = $1
    annotations:
      readOnlyHint:true

# 工具集：把相关工具打包，按需分配给不同 Agent
toolsets:
customer-service-tools:
    tools:
      -search_products
      -get_order_detail

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/googleapis/mcp-toolbox-sdk-go/core"
)

func main() {
    ctx := context.Background()

    // 连接到本地 Toolbox 服务
    // 生产环境换成 Cloud Run 或 K8s 的地址
    client, err := core.NewToolboxClient("http://localhost:5000")
    if err != nil {
        log.Fatalf("创建 Toolbox 客户端失败: %v", err)
    }

    // 加载指定 toolset 的所有工具
    // 客服 Agent 只能用 customer-service-tools，不是全部工具
    tools, err := client.LoadToolset(ctx, "customer-service-tools")
    if err != nil {
        log.Fatalf("加载工具集失败: %v", err)
    }

    fmt.Printf("加载了 %d 个工具\n", len(tools))

    // 直接调用工具（通常是 AI 框架帮你调用，这里演示手动调用）
    searchTool, err := client.LoadTool(ctx, "search_products")
    if err != nil {
        log.Fatalf("加载工具失败: %v", err)
    }

    // 调用工具：传参数，不传 SQL
    // LLM 永远不会看到 SQL，也不能修改 SQL
    result, err := searchTool.Invoke(ctx, map[string]any{
        "keyword":   "无线耳机",
        "max_price": 500.0,
    })
    if err != nil {
        log.Fatalf("工具调用失败: %v", err)
    }

    fmt.Println("查询结果:", result)
}

go get github.com/googleapis/mcp-toolbox-sdk-go/core

  place_order:
    kind: postgres-sql
    source: my-pg
    description: "下单：原子性地检查库存并创建订单"
    statement: |
      WITH stock_check AS (
        SELECT product_id, stock_qty
        FROM products
        WHERE product_id = $1 AND stock_qty >= $2
        FOR UPDATE  -- 行锁，防止并发超卖
      ),
      deducted AS (
        UPDATE products SET stock_qty = stock_qty - $2
        WHERE product_id = $1
          AND EXISTS (SELECT 1 FROM stock_check)
        RETURNING product_id
      )
      INSERT INTO orders (user_id, product_id, quantity, status)
      SELECT $3, product_id, $2, 'pending'
      FROM deducted
      RETURNING order_id
    annotations:
      readOnlyHint: false
      destructiveHint: true

-e TOOLBOX_URL=https://your-toolbox.example.com

sources:
  my-pg:
    kind:postgres
    host:127.0.0.1
    port:5432
    database:myapp
    user:readonly_user
    password:${PG_PASSWORD}
    pool:
      maxConns:20      # 最大连接数
      minConns:5       # 最小保持连接数
      maxConnLifetime:3600s