1. 為什么我要自己寫一個(gè) MCP 數(shù)據(jù)庫服務(wù)你可能已經(jīng)習(xí)慣了把表結(jié)構(gòu)復(fù)制粘貼給 AI讓它幫你寫 SQL然后再手動去數(shù)據(jù)庫客戶端里執(zhí)行。這個(gè)流程在表少的時(shí)候還行一旦庫里有幾十張表、字段名還都是縮寫AI 就開始瞎猜寫出來的 SQL 不是字段名對不上就是 JOIN 關(guān)系搞錯。更麻煩的是每次換一個(gè) AI 工具就要重新配一遍 Key、重新貼一遍表結(jié)構(gòu)通道和憑證散落在各個(gè)客戶端里管理起來很亂。Model Context ProtocolMCP解決的正是這件事它給 AI 客戶端和外部數(shù)據(jù)源之間定了一套標(biāo)準(zhǔn)協(xié)議AI 通過 JSON-RPC 調(diào)用你暴露出來的工具而不是靠猜。我這次用 Python 從零搭了一個(gè) MCP 服務(wù)把「列出所有表」「查看字段定義」「執(zhí)行只讀查詢」這幾個(gè)能力暴露出去AI 就能用自然語言直接查庫了。同時(shí)我把模型調(diào)用的 Key 統(tǒng)一收斂到 TaoToken 一個(gè)入口MCP 服務(wù)本身只負(fù)責(zé)數(shù)據(jù)庫這一側(cè)兩邊職責(zé)分開配置一次就能長期用。這篇適合誰會一點(diǎn) Python、手上有 MySQL 或 SQLite、想讓 AI 工具安全查庫的開發(fā)者。下面從環(huán)境準(zhǔn)備講到配置骨架再到自然語言查詢的驗(yàn)證步驟命令都可以直接復(fù)制。2. TaoToken 前置把模型 Key 統(tǒng)一收口MCP 服務(wù)負(fù)責(zé)「查庫」但 AI 客戶端要能理解你的自然語言、決定調(diào)用哪個(gè)工具背后仍然需要模型。如果每個(gè)客戶端各配一套 Key就又回到了分散的老問題。我的做法是所有支持自定義 API 地址的客戶端統(tǒng)一指向 TaoToken 的 API 入口用同一個(gè) Key。TaoToken 在這里的角色是統(tǒng)一的模型接入層兼容常見的 OpenAI 風(fēng)格接口你不需要為每個(gè)工具單獨(dú)申請憑證。先到控制臺創(chuàng)建一個(gè) API Key控制臺https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteAPI Keys 管理https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite創(chuàng)建好之后客戶端里填的 Base URL 用https://taotoken.net/apiKey 填剛生成的那串。如果你只是想先驗(yàn)證模型通不通可以直接在模型對話頁面試一句模型對話https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite注意MCP 服務(wù)本身不負(fù)責(zé)模型調(diào)用它只暴露數(shù)據(jù)庫工具。模型 Key 配在 AI 客戶端那一側(cè)兩邊不要混在一起排障時(shí)才能快速定位是哪一層的問題。如果你后續(xù)要長期跑編碼類 Agent反復(fù)調(diào)用模型可以了解下 Coding Plan額度模型更適合高頻場景Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite3. 可復(fù)制配置MCP 服務(wù)端骨架與 config.toml3.1 目錄結(jié)構(gòu)與依賴先建一個(gè)干凈的項(xiàng)目目錄我習(xí)慣這樣組織mcp-db-python/ ├── server.py ├── db.py ├── config.toml ├── requirements.txt └── .envrequirements.txt里核心就兩個(gè)MCP 的 Python SDK 和數(shù)據(jù)庫驅(qū)動。mcp1.0.0 pymysql1.1.0 python-dotenv1.0.0安裝pip install -r requirements.txt3.2 config.toml 骨架MCP 客戶端讀取服務(wù)的方式通常是在客戶端的配置里聲明一個(gè) stdio 類型的 server。下面這份config.toml是我實(shí)際在用的骨架把命令、參數(shù)、環(huán)境變量都寫清楚客戶端啟動時(shí)會按這個(gè)拉起 Python 進(jìn)程[mcp_servers.db_python] command python args [/absolute/path/to/mcp-db-python/server.py] [mcp_servers.db_python.env] DB_TYPE mysql DB_HOST 127.0.0.1 DB_PORT 3306 DB_USER readonly_user DB_PASS your_password DB_NAME test READ_ONLY true MAX_ROWS 200幾個(gè)參數(shù)值得單獨(dú)說參數(shù)作用建議值DB_TYPE數(shù)據(jù)庫類型mysql / sqliteREAD_ONLY是否強(qiáng)制只讀trueMAX_ROWS單次查詢返回上限100–500DB_USER數(shù)據(jù)庫賬號單獨(dú)建只讀賬號注意args里的路徑一定寫絕對路徑。MCP 客戶端拉起進(jìn)程時(shí)工作目錄不一定是你以為的那個(gè)相對路徑經(jīng)常導(dǎo)致「找不到 server.py」。3.3 只讀校驗(yàn)與工具注冊db.py里最關(guān)鍵的是 SQL 白名單校驗(yàn)只允許 SELECT、SHOW、DESC、EXPLAIN 這類語句其他一律拒絕import re ALLOWED_PREFIX (select, show, desc, describe, explain) def is_read_only(sql: str) - bool: cleaned re.sub(r\s, , sql.strip().lower()) if not cleaned.startswith(ALLOWED_PREFIX): return False forbidden (insert, update, delete, drop, alter, truncate, grant) return not any(word in cleaned for word in forbidden)server.py里用 MCP SDK 注冊工具把「列表」「看結(jié)構(gòu)」「查詢」三件事暴露出去import asyncio from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent from db import list_tables, get_table_schema, run_query app Server(db-python) app.list_tools() async def list_tools(): return [ Tool(namelist_tables, description列出數(shù)據(jù)庫中所有表, inputSchema{type: object, properties: {}}), Tool(nameget_table_schema, description查看指定表的字段定義, inputSchema{type: object, properties: {table: {type: string}}, required: [table]}), Tool(namerun_query, description執(zhí)行只讀 SQL 查詢, inputSchema{type: object, properties: {sql: {type: string}}, required: [sql]}), ] app.call_tool() async def call_tool(name: str, arguments: dict): if name list_tables: return [TextContent(typetext, textstr(list_tables()))] if name get_table_schema: return [TextContent(typetext, textstr(get_table_schema(arguments[table])))] if name run_query: return [TextContent(typetext, textstr(run_query(arguments[sql])))] raise ValueError(funknown tool: {name}) async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options()) if __name__ __main__: asyncio.run(main())run_query內(nèi)部先過is_read_only再執(zhí)行并限制返回行數(shù)def run_query(sql: str): if not is_read_only(sql): return {error: 僅允許只讀查詢} with get_conn() as conn: with conn.cursor() as cur: cur.execute(sql) rows cur.fetchmany(MAX_ROWS) cols [d[0] for d in cur.description] return [dict(zip(cols, row)) for row in rows]4. 驗(yàn)證請求用自然語言查一次庫配置寫完后先單獨(dú)跑一下服務(wù)確認(rèn)能啟動python server.py終端沒有報(bào)錯、進(jìn)程掛起等待輸入就說明 stdio 通道正常。接著在 AI 客戶端里把上面那份config.toml的 server 配置加進(jìn)去重啟客戶端讓它加載 MCP 服務(wù)。然后直接對 AI 說一句自然語言比如幫我看看 test 庫里有哪些表然后告訴我 users 表的字段結(jié)構(gòu)。正常情況下AI 會先調(diào)用list_tables再調(diào)用get_table_schema把結(jié)果整理后回給你。接著再試一句帶條件的查詢查一下 users 表里最近注冊的 10 個(gè)用戶按創(chuàng)建時(shí)間倒序。AI 會生成類似這樣的 SQL 并調(diào)用run_querySELECT * FROM users ORDER BY created_at DESC LIMIT 10;返回結(jié)果會以文本形式回到對話里。如果這一步成功了說明「自然語言 → 工具調(diào)用 → SQL → 結(jié)果」這條鏈路已經(jīng)打通。你可以再故意讓它執(zhí)行一條DELETE觀察服務(wù)是否返回「僅允許只讀查詢」以此確認(rèn)安全校驗(yàn)生效。5. 本篇常見錯排查5.1 客戶端報(bào) server 啟動失敗九成是路徑問題。檢查config.toml里args的server.py是不是絕對路徑以及command用的python在當(dāng)前環(huán)境里能不能找到。如果你用的是虛擬環(huán)境把command換成虛擬環(huán)境里的 python 絕對路徑例如/Users/you/venv/bin/python。5.2 連不上數(shù)據(jù)庫先確認(rèn)數(shù)據(jù)庫賬號密碼和端口再確認(rèn)賬號有沒有對應(yīng)庫的權(quán)限。生產(chǎn)環(huán)境強(qiáng)烈建議單獨(dú)建一個(gè)只讀賬號只授予 SELECT 權(quán)限這樣即使校驗(yàn)邏輯有疏漏也刪不掉數(shù)據(jù)。5.3 查詢返回空或字段名對不上多半是表名大小寫或庫名沒選對。MySQL 在部分系統(tǒng)上表名區(qū)分大小寫get_table_schema返回的字段名以數(shù)據(jù)庫實(shí)際為準(zhǔn)別用 AI 猜的名字去寫 SQL。遇到不確定的表先讓它調(diào)list_tables。5.4 模型側(cè)報(bào)鑒權(quán)失敗這屬于客戶端到模型那一層和 MCP 服務(wù)無關(guān)。檢查客戶端里填的 Base URL 是不是https://taotoken.net/apiKey 有沒有多余空格。想快速確認(rèn) Key 是否可用去模型對話頁面發(fā)一句話即可模型對話https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite5.5 返回行數(shù)太多把上下文撐爆把MAX_ROWS調(diào)小或者在提示詞里要求 AI 先加LIMIT。我一般把上限設(shè)在 200夠日常排查用了。6. 把 Key 和通道固定下來長期用整套跑通之后你會發(fā)現(xiàn)真正省事的地方在于數(shù)據(jù)庫這一側(cè)的能力被固化成了 MCP 工具模型這一側(cè)的憑證被收斂成了一個(gè) Key。以后不管換哪個(gè)支持 MCP 的客戶端只要把config.toml復(fù)制過去、Base URL 和 Key 填同一套就能直接查庫不用再重新貼表結(jié)構(gòu)、重新配通道。如果你要接的是編碼類 Agent需要頻繁調(diào)用模型可以看下 Coding Plan 的額度方式接入細(xì)節(jié)和參數(shù)說明都在文檔里Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite接入文檔https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteAPI Keyshttps://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite最后留一個(gè)我踩過的坑別把生產(chǎn)庫的寫權(quán)限賬號配進(jìn) MCP 服務(wù)哪怕校驗(yàn)寫得再嚴(yán)賬號權(quán)限才是最后一道閘。只讀賬號 只讀校驗(yàn) 行數(shù)上限這三樣配齊再讓 AI 碰數(shù)據(jù)庫才踏實(shí)。