這幾天,我們終於把 Claude 與自己的 MySQL 資料庫直接串接起來。從此,「查一下今年 momo 各月業績」、「比較無痕與棉感的全通路銷量趨勢」這類問題,Claude 直接連資料庫跑 SQL 給你答案。這篇文章記錄整個過程,包含踩到的坑,希望對想做同樣事情的人有幫助。
串接的核心概念是讓 Claude 能透過 HTTP API 送出 SQL 查詢,伺服器收到後去打 MySQL,把結果回傳。架構如下:
中間的 Cloudflare Tunnel 是解決網路問題的關鍵,後面會詳細說明為什麼需要它。
我們的伺服器已經有一個跑在 GCP 上的 Flask 服務,所以直接在現有專案裡加一個新的 route。核心程式碼很簡單:
from flask import Flask, request, jsonify import mysql.connector app = Flask(__name__) @app.route('/webhook/sql-query', methods=['POST']) def sql_query(): # 驗證 auth key auth = request.headers.get('auth') if auth != 'your-secret-key': return jsonify({'error': 'Unauthorized'}), 401 sql = request.json.get('sql', '') # 只允許 SELECT,防止誤操作 if not sql.strip().upper().startswith('SELECT'): return jsonify({'error': 'Only SELECT allowed'}), 400 conn = mysql.connector.connect( host='localhost', user='your_user', password='your_password', database='ragic_database' ) cursor = conn.cursor(dictionary=True) cursor.execute(sql) result = cursor.fetchall() conn.close() return jsonify(result)
服務是用 systemd 管理,部署後 reload 即可:
sudo systemctl restart your-service.service
Flask 跑起來後,我從本地 Mac 打 API 是通的。但當 Claude 嘗試呼叫時,卻一直 timeout。
34.135.250.196,也是 GCP 的 IP。GCP 的網路架構讓同網段的 GCP IP 之間透過外部 IP 的流量會被路由丟棄,所以 Claude 的請求根本到不了我們的 VM。
用 tcpdump 在伺服器上監控,確認 Claude 的封包根本沒有到達 VM,在 GCP 網路層就消失了。這跟防火牆設定無關。
Cloudflare Tunnel 的原理是讓 VM 主動向 Cloudflare 建立連線,外部請求透過 Cloudflare 中繼進來,完全不需要開防火牆,也不受 GCP 網路限制。
# Debian/Ubuntu sudo mkdir -p --mode=0755 /usr/share/keyrings curl -fsSL https://pkg.cloudflare.com/cloudflare-public-v2.gpg | sudo tee /usr/share/keyrings/cloudflare-public-v2.gpg >/dev/null echo 'deb [signed-by=/usr/share/keyrings/cloudflare-public-v2.gpg] https://pkg.cloudflare.com/cloudflared any main' | sudo tee /etc/apt/sources.list.d/cloudflared.list sudo apt-get update && sudo apt-get install cloudflared
選擇「Create a tunnel」→「Select Cloudflared」→「Name your tunnel」,輸入名稱(例如 sql-webhook)。
sudo cloudflared service install eyJhIjoiXXXXX…(你的 token) sudo systemctl start cloudflared sudo systemctl status cloudflared
在 Cloudflare DNS 新增一筆 CNAME 記錄,讓 db.your-domain.com 指向你的 tunnel:
Type: CNAME Name: db Target: {tunnel-id}.cfargotunnel.com Proxy: Proxied(橘色雲朵)
然後用 Cloudflare API 設定 tunnel 的 ingress 規則,告訴 tunnel 要把流量導向本機哪個 port:
curl -X PUT “https://api.cloudflare.com/client/v4/accounts/{account_id}/cfd_tunnel/{tunnel_id}/configurations” -H “X-Auth-Email: ” -H “X-Auth-Key: your-global-api-key” -H “Content-Type: application/json” -d ‘{
"config": { "ingress": [ { "hostname": "db.your-domain.com", "service": "http://localhost:8000" }, { "service": "http_status:404" } ] }
}’
最後,建立 /etc/cloudflared/config.yml,設定 tunnel token 與 ingress:
tunnel: {tunnel-id} token: eyJhIjoiXXXXX…
ingress:
- hostname: db.your-domain.com service: http://localhost:8000
- service: http_status:404
重啟服務:
sudo systemctl restart cloudflared
從任何地方都可以打這個 API 測試:
curl -X POST https://db.your-domain.com/webhook/sql-query -H “Content-Type: application/json” -H ‘auth: your-secret-key’ -d ‘{“sql”: “SELECT 1 AS test”}’
預期回應:
[{“test”: 1}]
串接完成後,還需要讓 Claude 知道這個 API 的用法、資料表結構、以及業務邏輯(例如品名的款式對應)。我們把這些寫成 Skill 檔案,讓 Claude 每次對話都能自動載入:
def query(sql):
import urllib.request, json url = 'https://db.your-domain.com/webhook/sql-query' data = json.dumps({'sql': sql}).encode() req = urllib.request.Request(url, data=data, headers={ 'Content-Type': 'application/json', 'auth': 'your-secret-key', 'User-Agent': 'curl/7.88.1' # 避免被 Cloudflare WAF 擋 }) with urllib.request.urlopen(req, timeout=30) as res: return json.loads(res.read().decode())
curl/7.88.1 或其他瀏覽器 User-Agent。
預設 SSL 模式是 Full,會嘗試用 HTTPS 連 origin server。但 Flask 只跑 HTTP,所以需要改成 Flexible,或讓 Tunnel 的 ingress 指向 http://localhost。
Python 的 urllib 預設 User-Agent 會觸發 Cloudflare 的 bot 偵測,記得手動設定 User-Agent。
SQL 裡用 CAST(field AS UNSIGNED) 遇到空白字串或負數會得到一個巨大的數字。改用 CAST(field AS SIGNED) 或加 NULL check。
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容,请联系我们,一经查实,本站将立刻删除。
如需转载请保留出处:https://51itzy.com/kjqy/269436.html