據(jù)庫技術(shù)實戰(zhàn)：讓大模型精準掌控私有數(shù)據(jù)（附全流程代碼）)
1. 當 RAG 答不準“上個月華東退貨率”時問題出在哪RAG 檢索增強生成這套方案我用了快兩年最大的感受是它擅長找“相似的文本”但不擅長算“精確的數(shù)字”。你問“公司差旅報銷標準是多少”它能把制度文檔里那段話撈出來答得挺像樣可你一旦問“上個月華東區(qū)退貨率是多少”它就開始胡編——因為答案根本不在任何一段文本里而是藏在數(shù)據(jù)庫的orders和returns兩張表的聚合結(jié)果中。這就是傳統(tǒng) RAG 的兩大硬傷。第一是檢索精度低向量相似度匹配的是語義不是事實。用戶問“銷量最好的產(chǎn)品”向量庫可能召回一堆標題里帶“銷量”“產(chǎn)品”的營銷文案卻召不回SELECT product_name FROM sales ORDER BY amount DESC LIMIT 1這條真正該執(zhí)行的查詢。第二是上下文割裂文本切塊把表格結(jié)構(gòu)切碎了模型看到的是“華東區(qū) 退貨 1200 件”這種孤立片段既不知道這是哪張表、哪個時間范圍也無法做二次聚合。MCPModel Context Protocol解決的正是這個斷層。它不像 RAG 那樣把數(shù)據(jù)“喂”給模型而是把數(shù)據(jù)庫的操作能力以工具Tool的形式暴露給模型模型先理解你的意圖生成一條 SQL通過 MCP 協(xié)議調(diào)用數(shù)據(jù)庫執(zhí)行再把結(jié)構(gòu)化結(jié)果回填到對話里。整個過程里模型拿到的是真實的、可驗證的行數(shù)據(jù)而不是一段可能被切碎的文本。打個比方RAG 像是給模型一本被撕碎又隨機裝訂的說明書它只能靠猜MCP 像是給模型配了一個會查數(shù)據(jù)庫的助手模型說“幫我查華東區(qū)上月退貨率”助手去查把準確數(shù)字遞回來。適合誰用如果你手上有私有 SQL 數(shù)據(jù)庫MySQL、PostgreSQL 都行又希望大模型能直接回答“有多少”“排第幾”“環(huán)比漲了多少”這類問題那 MCP 數(shù)據(jù)庫就是比純 RAG 更靠譜的路線。下面我會從零搭一套可運行的 MCP Server用 TaoToken 統(tǒng)一 Key 調(diào)用模型完成一次端到端問答驗證。全程代碼可復(fù)制踩過的坑我也會標出來。2. TaoToken 前置準備一個 Key 打通模型調(diào)用與 MCP 聯(lián)調(diào)在動手寫 MCP Server 之前得先把模型調(diào)用這條鏈路理順。MCP 本身只負責“模型 ? 工具”的協(xié)議通信真正做語義解析、生成 SQL 的還是大模型。所以你需要一個穩(wěn)定的模型 API 入口。我用的是 TaoToken原因很簡單它把多家模型的調(diào)用統(tǒng)一成一個 OpenAI 兼容接口Base URL 固定Key 統(tǒng)一切換模型只改一個model字段。對于 MCP 這種需要反復(fù)調(diào)試“模型生成的 SQL 對不對”的場景統(tǒng)一入口能省掉大量換 SDK、換鑒權(quán)方式的麻煩。第一步拿 Key。訪問https://taotoken.net/api-keysdeep link 帶歸因?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite登錄后在控制臺創(chuàng)建一個 API Key。建議單獨建一個用于 MCP 調(diào)試的 Key方便后續(xù)按項目排查用量。第二步確認 Base URL。TaoToken 的 API 入口是https://taotoken.net/api注意這里不加 UTM 參數(shù)直接作為 OpenAI SDK 的base_url使用。模型對話調(diào)試頁在https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite你可以先在那里手動問幾個問題確認 Key 能用、模型響應(yīng)正常再去接 MCP。第三步選模型。MCP 場景對模型的指令遵循能力和結(jié)構(gòu)化輸出能力要求較高因為它要生成合法的 SQL。實測下來Claude 系列和 GPT 系列在“給定 Schema 生成 SQL”這件事上都比較穩(wěn)。你可以在模型對話頁對比幾個模型的輸出挑一個 SQL 生成準確率高的。如果后續(xù)要做長期編碼或 Agent 類任務(wù)可以看下 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite按套餐走比單次調(diào)用更劃算。第四步本地環(huán)境。你需要 Python 3.10以及一個能連的數(shù)據(jù)庫。我用 PostgreSQL 15 做演示MySQL 同理只改連接串和驅(qū)動。裝依賴pip install mcp psycopg2-binary openai python-dotenv這里mcp是官方 Python SDKpsycopg2-binary是 PostgreSQL 驅(qū)動openai用來調(diào) TaoToken 的兼容接口。裝完后建一個.env文件存配置TAOTOKEN_API_KEYsk-你的Key TAOTOKEN_BASE_URLhttps://taotoken.net/api DB_HOSTlocalhost DB_PORT5432 DB_NAMEshop DB_USERreadonly DB_PASSWORDs3cr3t注意數(shù)據(jù)庫賬戶一定用只讀賬戶這是后面安全邊界的基礎(chǔ)。別用超級用戶跑 MCP否則模型一旦生成DROP TABLE就麻煩了。3. 可復(fù)制配置MCP Server 的 Schema 注入與工具注冊這一節(jié)是核心。MCP Server 要做三件事注入數(shù)據(jù)庫 Schema、注冊查詢工具、安全執(zhí)行 SQL。我把它拆成一個server.py逐段講。3.1 Schema 注入讓模型知道表長什么樣模型要生成正確的 SQL必須先知道有哪些表、哪些字段、字段類型是什么。最直接的辦法是啟動時把 Schema 讀出來拼成一段文本作為工具描述的一部分傳給模型。import os import psycopg2 from dotenv import load_dotenv from mcp.server import FastMCP load_dotenv() DB_CONFIG { host: os.getenv(DB_HOST), port: os.getenv(DB_PORT), dbname: os.getenv(DB_NAME), user: os.getenv(DB_USER), password: os.getenv(DB_PASSWORD), } def load_schema() - str: 讀取所有用戶表的字段信息拼成 Schema 文本 sql SELECT table_name, column_name, data_type FROM information_schema.columns WHERE table_schema public ORDER BY table_name, ordinal_position; with psycopg2.connect(**DB_CONFIG) as conn: with conn.cursor() as cur: cur.execute(sql) rows cur.fetchall() schema_map {} for table, column, dtype in rows: schema_map.setdefault(table, []).append(f{column} {dtype}) lines [] for table, cols in schema_map.items(): lines.append(f表 {table}: , .join(cols)) return \n.join(lines) SCHEMA_TEXT load_schema()這段代碼跑完SCHEMA_TEXT大概長這樣表 orders: id integer, product_id integer, region varchar, amount numeric, created_at timestamp 表 returns: id integer, order_id integer, reason varchar, created_at timestamp 表 products: id integer, name varchar, category varchar, price numeric把它塞進工具描述模型每次調(diào)用工具時都能看到當前數(shù)據(jù)庫的真實結(jié)構(gòu)不用你手動維護文檔。3.2 工具注冊把查詢能力暴露給模型MCP 的mcp.tool()裝飾器負責把 Python 函數(shù)注冊成模型可調(diào)用的工具。關(guān)鍵是函數(shù)簽名和 docstring——模型就是靠這些來決定“什么時候調(diào)、傳什么參數(shù)”。mcp FastMCP(nameSmartDBAssistant) mcp.tool() def query_database(sql: str) - dict: 在只讀數(shù)據(jù)庫上執(zhí)行 SELECT 查詢并返回結(jié)果。 當前數(shù)據(jù)庫結(jié)構(gòu)如下 {schema} 只允許 SELECT 語句禁止任何寫操作。 .format(schemaSCHEMA_TEXT) # 安全檢查 normalized sql.strip().lower() if not normalized.startswith(select): return {error: 只允許 SELECT 查詢} forbidden [insert, update, delete, drop, alter, truncate, ;--] if any(word in normalized for word in forbidden): return {error: 檢測到非法關(guān)鍵字查詢被拒絕} try: with psycopg2.connect(**DB_CONFIG) as conn: with conn.cursor() as cur: cur.execute(sql) columns [desc[0] for desc in cur.description] rows cur.fetchall() return { columns: columns, rows: rows[:200], # 限制返回行數(shù)防止上下文爆炸 row_count: len(rows), } except Exception as e: return {error: str(e)}注意幾個細節(jié)。第一docstring里嵌了 Schema模型調(diào)用前就能看到表結(jié)構(gòu)。第二安全檢查做了兩層前綴必須是select且不含寫操作關(guān)鍵字。第三返回結(jié)果限制 200 行避免一次查詢把上下文撐爆。第四異常直接返回error字段模型看到錯誤可以自己調(diào)整 SQL 重試。3.3 啟動配置stdio 與 http 兩種模式MCP Server 支持多種傳輸方式。本地調(diào)試用stdio最簡單客戶端直接拉起進程要跨機器或接遠程客戶端就用http。if __name__ __main__: # 本地調(diào)試用 stdio mcp.run(transportstdio) # 需要遠程訪問時改成 # mcp.run(transporthttp, host0.0.0.0, port8080)如果你用 Claude Code 或 Cline 這類支持 MCP 的客戶端配置通常寫在一個 JSON 里。以 Cline 的 MCP 配置為例路徑一般在~/.cline/mcp_settings.json{ mcpServers: { smart-db: { command: python, args: [/absolute/path/to/server.py], env: { TAOTOKEN_API_KEY: sk-你的Key, TAOTOKEN_BASE_URL: https://taotoken.net/api } } } }這里三件套要寫全Base URL是https://taotoken.net/apiKey是你的 TaoToken KeyModel ID在客戶端側(cè)配置比如claude-3-5-sonnet或gpt-4o。如果你用的是 Codex 的auth.json體系邏輯類似把 Base URL 和 Key 填進對應(yīng)字段即可。CC Switch 用戶則在切換配置里指定這三項。配置完重啟客戶端MCP Server 會被自動拉起工具列表里應(yīng)該能看到query_database。4. 驗證請求從自然語言到 SQL 再到答案的完整鏈路配置好了現(xiàn)在跑一次端到端驗證。我準備了一張orders表和一張returns表塞了點測試數(shù)據(jù)INSERT INTO orders (product_id, region, amount, created_at) VALUES (1, 華東, 299.00, 2024-05-03), (2, 華東, 158.00, 2024-05-11), (1, 華南, 299.00, 2024-05-15), (3, 華東, 89.00, 2024-05-20); INSERT INTO returns (order_id, reason, created_at) VALUES (1, 質(zhì)量問題, 2024-05-08), (3, 尺寸不符, 2024-05-18);現(xiàn)在寫一個客戶端腳本用 TaoToken 調(diào)模型讓模型通過 MCP 工具查數(shù)據(jù)。核心思路是把 MCP 工具的描述轉(zhuǎn)成 OpenAI 的tools格式模型返回tool_calls時執(zhí)行對應(yīng)函數(shù)把結(jié)果回填再讓模型總結(jié)。import json from openai import OpenAI from server import query_database, SCHEMA_TEXT client OpenAI( api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), ) tools [{ type: function, function: { name: query_database, description: f執(zhí)行只讀 SELECT 查詢。數(shù)據(jù)庫結(jié)構(gòu)\n{SCHEMA_TEXT}, parameters: { type: object, properties: { sql: {type: string, description: 要執(zhí)行的 SELECT 語句} }, required: [sql], }, }, }] def ask(question: str) - str: messages [ {role: system, content: 你是數(shù)據(jù)分析助手需要查數(shù)據(jù)時調(diào)用 query_database 工具拿到結(jié)果后用中文簡潔回答。}, {role: user, content: question}, ] resp client.chat.completions.create( modelclaude-3-5-sonnet, messagesmessages, toolstools, tool_choiceauto, ) msg resp.choices[0].message if msg.tool_calls: call msg.tool_calls[0] args json.loads(call.function.arguments) print(f[模型生成的 SQL] {args[sql]}) result query_database(args[sql]) print(f[數(shù)據(jù)庫返回] {result}) messages.append(msg) messages.append({ role: tool, tool_call_id: call.id, content: json.dumps(result, ensure_asciiFalse, defaultstr), }) final client.chat.completions.create( modelclaude-3-5-sonnet, messagesmessages, ) return final.choices[0].message.content return msg.content if __name__ __main__: print(ask(華東區(qū)一共有多少筆訂單總金額是多少))跑起來輸出大概是這樣[模型生成的 SQL] SELECT COUNT(*) AS order_count, SUM(amount) AS total_amount FROM orders WHERE region 華東 [數(shù)據(jù)庫返回] {columns: [order_count, total_amount], rows: [(3, 546.0)], row_count: 1} 華東區(qū)一共有 3 筆訂單總金額為 546.00 元。再試一個需要 JOIN 的問題print(ask(華東區(qū)有退貨的訂單退貨原因分別是什么))模型生成的 SQL 會是SELECT o.id, r.reason FROM orders o JOIN returns r ON o.id r.order_id WHERE o.region 華東返回兩條記錄模型總結(jié)為“華東區(qū)有兩筆退貨原因分別是質(zhì)量問題和尺寸不符”。整個過程里模型沒有猜任何數(shù)字所有事實都來自數(shù)據(jù)庫的真實返回。這就是 MCP 相比 RAG 的核心優(yōu)勢——可驗證。如果你在模型對話頁https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite手動測過同樣的問法會發(fā)現(xiàn)純對話模型要么答不出要么編一個數(shù)字。接上 MCP 后答案有了數(shù)據(jù)支撐。5. 本篇常見錯排查401、local proxy failed 與 reading choices調(diào)試 MCP 數(shù)據(jù)庫這條鏈路報錯基本集中在幾個地方。我把真實遇到過的列出來對照著排查。401 Unauthorized。最常見的原因是 Key 沒傳對或 Base URL 寫錯。檢查.env里TAOTOKEN_API_KEY是不是完整的sk-開頭字符串TAOTOKEN_BASE_URL是不是https://taotoken.net/api注意結(jié)尾沒有多余斜杠也不要加/v1SDK 會自己拼。如果你在 MCP 客戶端的 JSON 配置里寫env確認 Key 沒有多余空格。還有一種情況是 Key 被禁用或額度耗盡去控制臺https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite看下用量。local proxy failed / connection refused。這個報錯通常出現(xiàn)在 MCP 客戶端拉起 Server 時。原因可能是command路徑不對比如你寫了python但系統(tǒng)里是python3或者args里的server.py用了相對路徑客戶端工作目錄不同導致找不到文件。統(tǒng)一用絕對路徑command寫python3的全路徑which python3查一下。如果是 http 模式檢查端口有沒有被占用lsof -i :8080看一眼。Error reading choices / choices 字段為空。這個報錯來自 OpenAI SDK 解析響應(yīng)時。常見原因是模型返回了非標準結(jié)構(gòu)或者model字段填了一個 TaoToken 不支持的模型名。去模型對話頁確認你用的 Model ID 是平臺支持的。另一個可能是請求超時后返回了空 body檢查網(wǎng)絡(luò)和timeout參數(shù)MCP 場景下 SQL 執(zhí)行慢時容易觸發(fā)。OAuth 相關(guān)報錯。如果你用的是 Claude Code 或某些需要 OAuth 授權(quán)的客戶端可能會遇到 token 過期。這類客戶端通常有自己的登錄態(tài)和 TaoToken 的 Key 是兩套體系。確認客戶端的 OAuth 已重新授權(quán)同時 MCP 配置里的envKey 仍然有效。兩者別混。SQL 語法錯誤但模型不重試。有時候模型生成的 SQL 有方言問題比如 PostgreSQL 的LIMIT在別的庫不通用。解決辦法是在工具 docstring 里明確寫“當前數(shù)據(jù)庫為 PostgreSQL”或者在 system prompt 里指定方言。另外把數(shù)據(jù)庫返回的error原樣回填給模型大多數(shù)情況下模型會自己修正重試。返回結(jié)果太大導致上下文超限。如果你查了一張百萬行的表又沒加LIMIT返回的 JSON 會撐爆上下文。我在query_database里做了rows[:200]截斷但更好的做法是在 docstring 里提醒模型“查詢必須帶 LIMIT”。雙保險。6. 把 MCP 接進你的日常工作流跑通一次問答只是起點。真正讓這套方案產(chǎn)生價值是把它接進你每天用的工具里。如果你用 Claude Code 做開發(fā)可以在項目里配一個.mcp.json把數(shù)據(jù)庫 MCP Server 掛上之后寫代碼時直接問“幫我查下 users 表里最近注冊的 10 個用戶”模型會自己調(diào)工具查完再寫進代碼注釋。Claude Code 的接入文檔在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite里面有 MCP 配置的完整示例。如果你更習慣在 IDE 里用 Cline它的 MCP 市場可以直接加載本地 Server配置方式和上面 JSON 一樣。Cline 的好處是工具調(diào)用過程可視化你能看到模型每一步生成了什么 SQL、返回了什么調(diào)試起來很直觀。對于需要長期跑 Agent 任務(wù)的場景比如每天定時拉取銷售數(shù)據(jù)生成報表建議走 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite按套餐計費比單次調(diào)用穩(wěn)定也不用擔心突發(fā)流量把額度打滿。最后說一個我踩過的坑別把生產(chǎn)庫的寫權(quán)限開給 MCP。哪怕模型再聽話也可能因為 Schema 理解偏差生成意外的語句。只讀賬戶 關(guān)鍵字白名單 行數(shù)限制這三層防護一個都別省。數(shù)據(jù)庫連接串里的密碼用環(huán)境變量注入別硬編碼在代碼里提交到倉庫。這套 MCP 數(shù)據(jù)庫的方案本質(zhì)上是用“工具調(diào)用”替代“文本檢索”把大模型從“猜答案”變成“查答案”。RAG 該用還得用比如查制度文檔、產(chǎn)品說明這類非結(jié)構(gòu)化內(nèi)容但一旦涉及精確的數(shù)字、聚合、關(guān)聯(lián)查詢交給 MCP 更靠譜。兩者不是替代關(guān)系是互補。 突破RAG瓶頸！MCP+數(shù)據(jù)庫技術(shù)實戰(zhàn)：讓大模型精準掌控私有數(shù)據(jù)（附全流程代碼）
1. 當 RAG 答不準“上個月華東退貨率”時問題出在哪RAG 檢索增強生成這套方案我用了快兩年最大的感受是它擅長找“相似的文本”但不擅長算“精確的數(shù)字”。你問“公司差旅報銷標準是多少”它能把制度文檔里那段話撈出來答得挺像樣可你一旦問“上個月華東區(qū)退貨率是多少”它就開始胡編——因為答案根本不在任何一段文本里而是藏在數(shù)據(jù)庫的orders和returns兩張表的聚合結(jié)果中。這就是傳統(tǒng) RAG 的兩大硬傷。第一是檢索精度低向量相似度匹配的是語義不是事實。用戶問“銷量最好的產(chǎn)品”向量庫可能召回一堆標題里帶“銷量”“產(chǎn)品”的營銷文案卻召不回SELECT product_name FROM sales ORDER BY amount DESC LIMIT 1這條真正該執(zhí)行的查詢。第二是上下文割裂文本切塊把表格結(jié)構(gòu)切碎了模型看到的是“華東區(qū) 退貨 1200 件”這種孤立片段既不知道這是哪張表、哪個時間范圍也無法做二次聚合。MCPModel Context Protocol解決的正是這個斷層。它不像 RAG 那樣把數(shù)據(jù)“喂”給模型而是把數(shù)據(jù)庫的操作能力以工具Tool的形式暴露給模型模型先理解你的意圖生成一條 SQL通過 MCP 協(xié)議調(diào)用數(shù)據(jù)庫執(zhí)行再把結(jié)構(gòu)化結(jié)果回填到對話里。整個過程里模型拿到的是真實的、可驗證的行數(shù)據(jù)而不是一段可能被切碎的文本。打個比方RAG 像是給模型一本被撕碎又隨機裝訂的說明書它只能靠猜MCP 像是給模型配了一個會查數(shù)據(jù)庫的助手模型說“幫我查華東區(qū)上月退貨率”助手去查把準確數(shù)字遞回來。適合誰用如果你手上有私有 SQL 數(shù)據(jù)庫MySQL、PostgreSQL 都行又希望大模型能直接回答“有多少”“排第幾”“環(huán)比漲了多少”這類問題那 MCP 數(shù)據(jù)庫就是比純 RAG 更靠譜的路線。下面我會從零搭一套可運行的 MCP Server用 TaoToken 統(tǒng)一 Key 調(diào)用模型完成一次端到端問答驗證。全程代碼可復(fù)制踩過的坑我也會標出來。2. TaoToken 前置準備一個 Key 打通模型調(diào)用與 MCP 聯(lián)調(diào)在動手寫 MCP Server 之前得先把模型調(diào)用這條鏈路理順。MCP 本身只負責“模型 ? 工具”的協(xié)議通信真正做語義解析、生成 SQL 的還是大模型。所以你需要一個穩(wěn)定的模型 API 入口。我用的是 TaoToken原因很簡單它把多家模型的調(diào)用統(tǒng)一成一個 OpenAI 兼容接口Base URL 固定Key 統(tǒng)一切換模型只改一個model字段。對于 MCP 這種需要反復(fù)調(diào)試“模型生成的 SQL 對不對”的場景統(tǒng)一入口能省掉大量換 SDK、換鑒權(quán)方式的麻煩。第一步拿 Key。訪問https://taotoken.net/api-keysdeep link 帶歸因?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite登錄后在控制臺創(chuàng)建一個 API Key。建議單獨建一個用于 MCP 調(diào)試的 Key方便后續(xù)按項目排查用量。第二步確認 Base URL。TaoToken 的 API 入口是https://taotoken.net/api注意這里不加 UTM 參數(shù)直接作為 OpenAI SDK 的base_url使用。模型對話調(diào)試頁在https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite你可以先在那里手動問幾個問題確認 Key 能用、模型響應(yīng)正常再去接 MCP。第三步選模型。MCP 場景對模型的指令遵循能力和結(jié)構(gòu)化輸出能力要求較高因為它要生成合法的 SQL。實測下來Claude 系列和 GPT 系列在“給定 Schema 生成 SQL”這件事上都比較穩(wěn)。你可以在模型對話頁對比幾個模型的輸出挑一個 SQL 生成準確率高的。如果后續(xù)要做長期編碼或 Agent 類任務(wù)可以看下 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite按套餐走比單次調(diào)用更劃算。第四步本地環(huán)境。你需要 Python 3.10以及一個能連的數(shù)據(jù)庫。我用 PostgreSQL 15 做演示MySQL 同理只改連接串和驅(qū)動。裝依賴pip install mcp psycopg2-binary openai python-dotenv這里mcp是官方 Python SDKpsycopg2-binary是 PostgreSQL 驅(qū)動openai用來調(diào) TaoToken 的兼容接口。裝完后建一個.env文件存配置TAOTOKEN_API_KEYsk-你的Key TAOTOKEN_BASE_URLhttps://taotoken.net/api DB_HOSTlocalhost DB_PORT5432 DB_NAMEshop DB_USERreadonly DB_PASSWORDs3cr3t注意數(shù)據(jù)庫賬戶一定用只讀賬戶這是后面安全邊界的基礎(chǔ)。別用超級用戶跑 MCP否則模型一旦生成DROP TABLE就麻煩了。3. 可復(fù)制配置MCP Server 的 Schema 注入與工具注冊這一節(jié)是核心。MCP Server 要做三件事注入數(shù)據(jù)庫 Schema、注冊查詢工具、安全執(zhí)行 SQL。我把它拆成一個server.py逐段講。3.1 Schema 注入讓模型知道表長什么樣模型要生成正確的 SQL必須先知道有哪些表、哪些字段、字段類型是什么。最直接的辦法是啟動時把 Schema 讀出來拼成一段文本作為工具描述的一部分傳給模型。import os import psycopg2 from dotenv import load_dotenv from mcp.server import FastMCP load_dotenv() DB_CONFIG { host: os.getenv(DB_HOST), port: os.getenv(DB_PORT), dbname: os.getenv(DB_NAME), user: os.getenv(DB_USER), password: os.getenv(DB_PASSWORD), } def load_schema() - str: 讀取所有用戶表的字段信息拼成 Schema 文本 sql SELECT table_name, column_name, data_type FROM information_schema.columns WHERE table_schema public ORDER BY table_name, ordinal_position; with psycopg2.connect(**DB_CONFIG) as conn: with conn.cursor() as cur: cur.execute(sql) rows cur.fetchall() schema_map {} for table, column, dtype in rows: schema_map.setdefault(table, []).append(f{column} {dtype}) lines [] for table, cols in schema_map.items(): lines.append(f表 {table}: , .join(cols)) return \n.join(lines) SCHEMA_TEXT load_schema()這段代碼跑完SCHEMA_TEXT大概長這樣表 orders: id integer, product_id integer, region varchar, amount numeric, created_at timestamp 表 returns: id integer, order_id integer, reason varchar, created_at timestamp 表 products: id integer, name varchar, category varchar, price numeric把它塞進工具描述模型每次調(diào)用工具時都能看到當前數(shù)據(jù)庫的真實結(jié)構(gòu)不用你手動維護文檔。3.2 工具注冊把查詢能力暴露給模型MCP 的mcp.tool()裝飾器負責把 Python 函數(shù)注冊成模型可調(diào)用的工具。關(guān)鍵是函數(shù)簽名和 docstring——模型就是靠這些來決定“什么時候調(diào)、傳什么參數(shù)”。mcp FastMCP(nameSmartDBAssistant) mcp.tool() def query_database(sql: str) - dict: 在只讀數(shù)據(jù)庫上執(zhí)行 SELECT 查詢并返回結(jié)果。 當前數(shù)據(jù)庫結(jié)構(gòu)如下 {schema} 只允許 SELECT 語句禁止任何寫操作。 .format(schemaSCHEMA_TEXT) # 安全檢查 normalized sql.strip().lower() if not normalized.startswith(select): return {error: 只允許 SELECT 查詢} forbidden [insert, update, delete, drop, alter, truncate, ;--] if any(word in normalized for word in forbidden): return {error: 檢測到非法關(guān)鍵字查詢被拒絕} try: with psycopg2.connect(**DB_CONFIG) as conn: with conn.cursor() as cur: cur.execute(sql) columns [desc[0] for desc in cur.description] rows cur.fetchall() return { columns: columns, rows: rows[:200], # 限制返回行數(shù)防止上下文爆炸 row_count: len(rows), } except Exception as e: return {error: str(e)}注意幾個細節(jié)。第一docstring里嵌了 Schema模型調(diào)用前就能看到表結(jié)構(gòu)。第二安全檢查做了兩層前綴必須是select且不含寫操作關(guān)鍵字。第三返回結(jié)果限制 200 行避免一次查詢把上下文撐爆。第四異常直接返回error字段模型看到錯誤可以自己調(diào)整 SQL 重試。3.3 啟動配置stdio 與 http 兩種模式MCP Server 支持多種傳輸方式。本地調(diào)試用stdio最簡單客戶端直接拉起進程要跨機器或接遠程客戶端就用http。if __name__ __main__: # 本地調(diào)試用 stdio mcp.run(transportstdio) # 需要遠程訪問時改成 # mcp.run(transporthttp, host0.0.0.0, port8080)如果你用 Claude Code 或 Cline 這類支持 MCP 的客戶端配置通常寫在一個 JSON 里。以 Cline 的 MCP 配置為例路徑一般在~/.cline/mcp_settings.json{ mcpServers: { smart-db: { command: python, args: [/absolute/path/to/server.py], env: { TAOTOKEN_API_KEY: sk-你的Key, TAOTOKEN_BASE_URL: https://taotoken.net/api } } } }這里三件套要寫全Base URL是https://taotoken.net/apiKey是你的 TaoToken KeyModel ID在客戶端側(cè)配置比如claude-3-5-sonnet或gpt-4o。如果你用的是 Codex 的auth.json體系邏輯類似把 Base URL 和 Key 填進對應(yīng)字段即可。CC Switch 用戶則在切換配置里指定這三項。配置完重啟客戶端MCP Server 會被自動拉起工具列表里應(yīng)該能看到query_database。4. 驗證請求從自然語言到 SQL 再到答案的完整鏈路配置好了現(xiàn)在跑一次端到端驗證。我準備了一張orders表和一張returns表塞了點測試數(shù)據(jù)INSERT INTO orders (product_id, region, amount, created_at) VALUES (1, 華東, 299.00, 2024-05-03), (2, 華東, 158.00, 2024-05-11), (1, 華南, 299.00, 2024-05-15), (3, 華東, 89.00, 2024-05-20); INSERT INTO returns (order_id, reason, created_at) VALUES (1, 質(zhì)量問題, 2024-05-08), (3, 尺寸不符, 2024-05-18);現(xiàn)在寫一個客戶端腳本用 TaoToken 調(diào)模型讓模型通過 MCP 工具查數(shù)據(jù)。核心思路是把 MCP 工具的描述轉(zhuǎn)成 OpenAI 的tools格式模型返回tool_calls時執(zhí)行對應(yīng)函數(shù)把結(jié)果回填再讓模型總結(jié)。import json from openai import OpenAI from server import query_database, SCHEMA_TEXT client OpenAI( api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), ) tools [{ type: function, function: { name: query_database, description: f執(zhí)行只讀 SELECT 查詢。數(shù)據(jù)庫結(jié)構(gòu)\n{SCHEMA_TEXT}, parameters: { type: object, properties: { sql: {type: string, description: 要執(zhí)行的 SELECT 語句} }, required: [sql], }, }, }] def ask(question: str) - str: messages [ {role: system, content: 你是數(shù)據(jù)分析助手需要查數(shù)據(jù)時調(diào)用 query_database 工具拿到結(jié)果后用中文簡潔回答。}, {role: user, content: question}, ] resp client.chat.completions.create( modelclaude-3-5-sonnet, messagesmessages, toolstools, tool_choiceauto, ) msg resp.choices[0].message if msg.tool_calls: call msg.tool_calls[0] args json.loads(call.function.arguments) print(f[模型生成的 SQL] {args[sql]}) result query_database(args[sql]) print(f[數(shù)據(jù)庫返回] {result}) messages.append(msg) messages.append({ role: tool, tool_call_id: call.id, content: json.dumps(result, ensure_asciiFalse, defaultstr), }) final client.chat.completions.create( modelclaude-3-5-sonnet, messagesmessages, ) return final.choices[0].message.content return msg.content if __name__ __main__: print(ask(華東區(qū)一共有多少筆訂單總金額是多少))跑起來輸出大概是這樣[模型生成的 SQL] SELECT COUNT(*) AS order_count, SUM(amount) AS total_amount FROM orders WHERE region 華東 [數(shù)據(jù)庫返回] {columns: [order_count, total_amount], rows: [(3, 546.0)], row_count: 1} 華東區(qū)一共有 3 筆訂單總金額為 546.00 元。再試一個需要 JOIN 的問題print(ask(華東區(qū)有退貨的訂單退貨原因分別是什么))模型生成的 SQL 會是SELECT o.id, r.reason FROM orders o JOIN returns r ON o.id r.order_id WHERE o.region 華東返回兩條記錄模型總結(jié)為“華東區(qū)有兩筆退貨原因分別是質(zhì)量問題和尺寸不符”。整個過程里模型沒有猜任何數(shù)字所有事實都來自數(shù)據(jù)庫的真實返回。這就是 MCP 相比 RAG 的核心優(yōu)勢——可驗證。如果你在模型對話頁https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite手動測過同樣的問法會發(fā)現(xiàn)純對話模型要么答不出要么編一個數(shù)字。接上 MCP 后答案有了數(shù)據(jù)支撐。5. 本篇常見錯排查401、local proxy failed 與 reading choices調(diào)試 MCP 數(shù)據(jù)庫這條鏈路報錯基本集中在幾個地方。我把真實遇到過的列出來對照著排查。401 Unauthorized。最常見的原因是 Key 沒傳對或 Base URL 寫錯。檢查.env里TAOTOKEN_API_KEY是不是完整的sk-開頭字符串TAOTOKEN_BASE_URL是不是https://taotoken.net/api注意結(jié)尾沒有多余斜杠也不要加/v1SDK 會自己拼。如果你在 MCP 客戶端的 JSON 配置里寫env確認 Key 沒有多余空格。還有一種情況是 Key 被禁用或額度耗盡去控制臺https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite看下用量。local proxy failed / connection refused。這個報錯通常出現(xiàn)在 MCP 客戶端拉起 Server 時。原因可能是command路徑不對比如你寫了python但系統(tǒng)里是python3或者args里的server.py用了相對路徑客戶端工作目錄不同導致找不到文件。統(tǒng)一用絕對路徑command寫python3的全路徑which python3查一下。如果是 http 模式檢查端口有沒有被占用lsof -i :8080看一眼。Error reading choices / choices 字段為空。這個報錯來自 OpenAI SDK 解析響應(yīng)時。常見原因是模型返回了非標準結(jié)構(gòu)或者model字段填了一個 TaoToken 不支持的模型名。去模型對話頁確認你用的 Model ID 是平臺支持的。另一個可能是請求超時后返回了空 body檢查網(wǎng)絡(luò)和timeout參數(shù)MCP 場景下 SQL 執(zhí)行慢時容易觸發(fā)。OAuth 相關(guān)報錯。如果你用的是 Claude Code 或某些需要 OAuth 授權(quán)的客戶端可能會遇到 token 過期。這類客戶端通常有自己的登錄態(tài)和 TaoToken 的 Key 是兩套體系。確認客戶端的 OAuth 已重新授權(quán)同時 MCP 配置里的envKey 仍然有效。兩者別混。SQL 語法錯誤但模型不重試。有時候模型生成的 SQL 有方言問題比如 PostgreSQL 的LIMIT在別的庫不通用。解決辦法是在工具 docstring 里明確寫“當前數(shù)據(jù)庫為 PostgreSQL”或者在 system prompt 里指定方言。另外把數(shù)據(jù)庫返回的error原樣回填給模型大多數(shù)情況下模型會自己修正重試。返回結(jié)果太大導致上下文超限。如果你查了一張百萬行的表又沒加LIMIT返回的 JSON 會撐爆上下文。我在query_database里做了rows[:200]截斷但更好的做法是在 docstring 里提醒模型“查詢必須帶 LIMIT”。雙保險。6. 把 MCP 接進你的日常工作流跑通一次問答只是起點。真正讓這套方案產(chǎn)生價值是把它接進你每天用的工具里。如果你用 Claude Code 做開發(fā)可以在項目里配一個.mcp.json把數(shù)據(jù)庫 MCP Server 掛上之后寫代碼時直接問“幫我查下 users 表里最近注冊的 10 個用戶”模型會自己調(diào)工具查完再寫進代碼注釋。Claude Code 的接入文檔在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite里面有 MCP 配置的完整示例。如果你更習慣在 IDE 里用 Cline它的 MCP 市場可以直接加載本地 Server配置方式和上面 JSON 一樣。Cline 的好處是工具調(diào)用過程可視化你能看到模型每一步生成了什么 SQL、返回了什么調(diào)試起來很直觀。對于需要長期跑 Agent 任務(wù)的場景比如每天定時拉取銷售數(shù)據(jù)生成報表建議走 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite按套餐計費比單次調(diào)用穩(wěn)定也不用擔心突發(fā)流量把額度打滿。最后說一個我踩過的坑別把生產(chǎn)庫的寫權(quán)限開給 MCP。哪怕模型再聽話也可能因為 Schema 理解偏差生成意外的語句。只讀賬戶 關(guān)鍵字白名單 行數(shù)限制這三層防護一個都別省。數(shù)據(jù)庫連接串里的密碼用環(huán)境變量注入別硬編碼在代碼里提交到倉庫。這套 MCP 數(shù)據(jù)庫的方案本質(zhì)上是用“工具調(diào)用”替代“文本檢索”把大模型從“猜答案”變成“查答案”。RAG 該用還得用比如查制度文檔、產(chǎn)品說明這類非結(jié)構(gòu)化內(nèi)容但一旦涉及精確的數(shù)字、聚合、關(guān)聯(lián)查詢交給 MCP 更靠譜。兩者不是替代關(guān)系是互補。