-(4)MCP中間件和MCP鑒權(quán):用TaoToken統(tǒng)一Key打通FastMCP鑒權(quán)鏈路)
1. 自建 FastMCP Server 為什么必須補(bǔ)上鑒權(quán)中間件如果你已經(jīng)用 FastMCP 跑通了一個(gè) HTTP 模式的 MCP Server大概率會(huì)經(jīng)歷這樣一個(gè)階段本地streamablehttp_client連上去list_tools一列工具全出來了調(diào)用也正常于是順手把端口映射到內(nèi)網(wǎng)想著先給同事用著。問題就出在這里——FastMCP 默認(rèn)不校驗(yàn)任何身份任何能訪問到/mcp/這個(gè)路徑的客戶端都能直接發(fā) JSON-RPC 請(qǐng)求把你的工具全調(diào)一遍。我試過在一個(gè)只做了防火墻 IP 白名單的 MCP Server 上做壓力測試只要請(qǐng)求源在允許網(wǎng)段內(nèi)tools/call完全不設(shè)防。這意味著一旦有人把內(nèi)網(wǎng)地址泄露出去或者某臺(tái)被允許的機(jī)器被當(dāng)成跳板你的數(shù)據(jù)庫查詢工具、文件操作工具就全部暴露了。MCP 基于 JSON-RPC 規(guī)范運(yùn)行請(qǐng)求體長這樣{ jsonrpc: 2.0, id: 1, method: tools/call, params: { name: execute_mysql_sql, arguments: { sql: select * from shares_day_info limit 3 } } }注意這里沒有任何身份字段。JSON-RPC 本身是傳輸無關(guān)的協(xié)議鑒權(quán)信息只能掛在傳輸層——HTTP 模式下就是請(qǐng)求頭。所以正確的做法是在 FastMCP 的中間件管道里插一層攔截所有進(jìn)入的 MCP 消息從 HTTP header 里取出憑證做校驗(yàn)不合法就直接拋錯(cuò)讓請(qǐng)求根本到不了工具執(zhí)行階段。FastMCP 中間件采用管道模型請(qǐng)求按添加順序流經(jīng)每個(gè)中間件每個(gè)中間件可以檢查請(qǐng)求、修改請(qǐng)求、調(diào)用call_next()交給下一個(gè)、再檢查響應(yīng)。它提供了從通用到具體的鉤子層級(jí)——on_message管所有消息on_request只管需要響應(yīng)的請(qǐng)求on_call_tool只管工具調(diào)用。鑒權(quán)這種所有請(qǐng)求都要過的邏輯用on_message或直接在__call__里做最穩(wěn)妥因?yàn)楣ぞ甙l(fā)現(xiàn)list_tools本身也是一次請(qǐng)求不攔的話別人照樣能枚舉你有哪些工具。這一篇要解決的就是給自建 FastMCP Server 加一層鑒權(quán)中間件并且用 TaoToken 的統(tǒng)一 Key 體系把憑證管理收斂到一處避免每個(gè) MCP Server 各寫一套 token 表。適合正在把 MCP Server 從局域網(wǎng)自用推向團(tuán)隊(duì)共享的開發(fā)者。下面從環(huán)境準(zhǔn)備、中間件配置、請(qǐng)求驗(yàn)證到報(bào)錯(cuò)排查一步步給可復(fù)制的片段。2. TaoToken 統(tǒng)一 Key 接入 FastMCP 鑒權(quán)鏈路的前置準(zhǔn)備在寫中間件之前先把憑證來源理清楚。最原始的做法是在自己庫里建一張mcp_server_oauth_tokens表存用戶名、token、過期時(shí)間中間件查庫比對(duì)。這個(gè)方案能跑但有幾個(gè)現(xiàn)實(shí)問題每個(gè) MCP Server 都要連一次庫、token 輪換要手動(dòng)改表、多個(gè)服務(wù)之間憑證不互通。當(dāng)你有三四個(gè) MCP Server 時(shí)維護(hù)成本就上來了。更省事的思路是把簽發(fā)和校驗(yàn)憑證這件事交給一個(gè)統(tǒng)一入口MCP Server 只負(fù)責(zé)拿請(qǐng)求頭里的 Key 去問一句這個(gè) Key 有效嗎。TaoToken 在這里扮演的就是這個(gè)統(tǒng)一 Key 層——你可以在它的控制臺(tái)里生成和管理 API KeyMCP Server 側(cè)只需要配置 Base URL、Key、Model ID 三件套里的前兩件用于鑒權(quán)校驗(yàn)工具本身要調(diào)模型時(shí)再補(bǔ)上 Model ID。先做前置準(zhǔn)備。打開 TaoToken 官網(wǎng) https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注冊并進(jìn)入控制臺(tái)在 API Keys 頁面創(chuàng)建一個(gè) Key。這個(gè) Key 就是你后面要寫進(jìn) MCP 客戶端 header 的憑證??刂婆_(tái)地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite API Keys 管理頁在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。拿到 Key 之后本地環(huán)境需要確認(rèn)幾件事。Python 側(cè)裝好 FastMCP 和 HTTP 相關(guān)依賴pip install fastmcp httpx starlette uvicornFastMCP 的中間件基類在fastmcp.server.middleware下HTTP 請(qǐng)求對(duì)象通過get_http_request()獲取。注意這個(gè)函數(shù)只在 HTTP 傳輸下有效標(biāo)準(zhǔn) I/O 傳輸拿不到 header——這也是為什么鑒權(quán)中間件只對(duì) HTTP 模式有意義。如果你同時(shí)支持兩種傳輸中間件里要先判斷傳輸類型否則 stdio 模式下會(huì)直接報(bào)錯(cuò)。配置層面建議把 TaoToken 的校驗(yàn)地址和你的 Key 放進(jìn)環(huán)境變量別硬編碼export TAOTOKEN_API_BASEhttps://taotoken.net/api export TAOTOKEN_API_KEYsk-你的Key export MCP_SERVER_PORT18088這里TAOTOKEN_API_BASE用不帶 UTM 的 API 地址 https://taotoken.net/api 因?yàn)樗浅绦蛘{(diào)用的端點(diǎn)不需要追蹤參數(shù)。Key 從環(huán)境變量讀中間件里用os.environ.get()取這樣換 Key 不用改代碼。還有一點(diǎn)要提前想清楚鑒權(quán)中間件校驗(yàn)的是調(diào)用方有沒有資格訪問這個(gè) MCP Server而 TaoToken 的 Key 校驗(yàn)的是這個(gè) Key 有沒有資格用 TaoToken 的服務(wù)。兩者可以合一——直接把 TaoToken 的 Key 當(dāng)作 MCP Server 的訪問憑證中間件拿它去調(diào)一次 TaoToken 的接口驗(yàn)證有效性。這樣團(tuán)隊(duì)里每個(gè)人用自己的 TaoToken Key你不需要再維護(hù)一張 token 表。下面第三節(jié)就給這個(gè)方案的完整配置。3. FastMCP 鑒權(quán)中間件可復(fù)制配置與 TaoToken Key 校驗(yàn)片段這一節(jié)是核心直接給能跑的代碼。先看中間件本體它攔截所有 MCP 消息從 HTTP header 取Authorization解析出 Bearer token然后校驗(yàn)。import os import logging from fastmcp import FastMCP from fastmcp.server.middleware import Middleware, MiddlewareContext from fastmcp.server.dependencies import get_http_request logger logging.getLogger(mcp.auth) TAOTOKEN_API_BASE os.environ.get(TAOTOKEN_API_BASE, https://taotoken.net/api) TAOTOKEN_API_KEY os.environ.get(TAOTOKEN_API_KEY, ) class AuthMiddleware(Middleware): async def __call__(self, context: MiddlewareContext, call_next): # stdio 傳輸沒有 HTTP 請(qǐng)求對(duì)象直接放行 try: request get_http_request() except Exception: return await call_next(context) authorization request.headers.get(authorization) if not authorization or not authorization.startswith(Bearer ): raise PermissionError(401 Authorization Required) access_token authorization.split( , 1)[1].strip() if not access_token: raise PermissionError(401 Authorization Required) # 校驗(yàn) token 是否有效這里用 TaoToken 的 Key 作為統(tǒng)一憑證 if not await self._verify_token(access_token): raise PermissionError(403 Forbidden) logger.info(auth passed for method%s, context.method) return await call_next(context) async def _verify_token(self, token: str) - bool: # 簡單策略與配置的 Key 比對(duì)生產(chǎn)環(huán)境可換成調(diào)用 TaoToken 校驗(yàn)接口 if TAOTOKEN_API_KEY and token TAOTOKEN_API_KEY: return True # 也可以在這里調(diào)用 TaoToken 的接口做在線校驗(yàn) return False把中間件掛到 FastMCP 實(shí)例上mcp FastMCP(secure-mcp-server) mcp.add_middleware(AuthMiddleware()) mcp.tool() def execute_mysql_sql(sql: str) - str: # 你的工具邏輯 return fexecuted: {sql} if __name__ __main__: mcp.run(transportstreamable-http, host0.0.0.0, port18088)如果你更習(xí)慣用配置文件管理FastMCP 支持從 JSON 讀取服務(wù)定義。下面是一個(gè)mcp_config.json片段把鑒權(quán)相關(guān)的環(huán)境變量和傳輸方式寫進(jìn)去{ mcpServers: { secure-mysql: { transport: streamable-http, url: http://127.0.0.1:18088/mcp/, headers: { Authorization: Bearer ${TAOTOKEN_API_KEY} }, env: { TAOTOKEN_API_BASE: https://taotoken.net/api } } } }注意headers里的Authorization就是客戶端要帶的憑證${TAOTOKEN_API_KEY}從環(huán)境變量注入避免明文寫進(jìn)配置文件。這個(gè) JSON 結(jié)構(gòu)可以直接被支持 MCP 配置的客戶端讀取。如果你用的是 Cline 或 Claude Code 這類工具它們的 MCP 配置通常放在settings.json或claude_desktop_config.json里結(jié)構(gòu)類似{ mcpServers: { secure-mysql: { command: python, args: [-m, your_mcp_server], env: { TAOTOKEN_API_BASE: https://taotoken.net/api, TAOTOKEN_API_KEY: sk-你的Key } } } }這里要強(qiáng)調(diào)三件套的完整性Base URL 用https://taotoken.net/apiKey 用你在控制臺(tái)生成的sk-開頭字符串Model ID 在工具內(nèi)部需要調(diào)模型時(shí)再指定比如claude-sonnet-4-5之類以控制臺(tái)實(shí)際可用的為準(zhǔn)。鑒權(quán)中間件只用到前兩件第三件是工具執(zhí)行階段的事別混在一起。中間件里_verify_token目前是簡單比對(duì)生產(chǎn)環(huán)境建議改成調(diào)用 TaoToken 的校驗(yàn)接口這樣 Key 的吊銷和過期由 TaoToken 側(cè)統(tǒng)一管理你的 MCP Server 不用重啟。調(diào)用方式就是拿access_token去請(qǐng)求 TaoToken 的 API返回 200 即有效。具體接口路徑參考接入文檔 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。配置寫完后啟動(dòng)服務(wù)python your_mcp_server.py看到 uvicorn 監(jiān)聽 18088 端口就說明起來了。下一節(jié)驗(yàn)證請(qǐng)求。4. 帶鑒權(quán)頭的 JSON-RPC 請(qǐng)求驗(yàn)證與成功結(jié)果確認(rèn)服務(wù)起來后先驗(yàn)證未授權(quán)被攔截。用 curl 直接發(fā)一個(gè)不帶 header 的 JSON-RPC 請(qǐng)求curl -X POST http://127.0.0.1:18088/mcp/ \ -H Content-Type: application/json \ -d { jsonrpc: 2.0, id: 1, method: tools/list, params: {} }預(yù)期返回 401 或 403body 里能看到Authorization Required或Forbidden。這一步確認(rèn)中間件確實(shí)攔住了沒有憑證的請(qǐng)求。再驗(yàn)證錯(cuò)誤憑證被拒絕curl -X POST http://127.0.0.1:18088/mcp/ \ -H Content-Type: application/json \ -H Authorization: Bearer wrong-token-12345 \ -d { jsonrpc: 2.0, id: 2, method: tools/list, params: {} }預(yù)期返回 403。如果這里返回了 200 并且列出了工具說明你的_verify_token邏輯有問題檢查是不是把空 token 也放行了。最后驗(yàn)證正確憑證正常返回curl -X POST http://127.0.0.1:18088/mcp/ \ -H Content-Type: application/json \ -H Authorization: Bearer sk-你的Key \ -d { jsonrpc: 2.0, id: 3, method: tools/list, params: {} }成功的話會(huì)返回工具列表類似{ jsonrpc: 2.0, id: 3, result: { tools: [ { name: execute_mysql_sql, description: 執(zhí)行 SQL 查詢, inputSchema: { type: object, properties: { sql: { type: string } } } } ] } }再發(fā)一個(gè)tools/call驗(yàn)證工具能真正執(zhí)行curl -X POST http://127.0.0.1:18088/mcp/ \ -H Content-Type: application/json \ -H Authorization: Bearer sk-你的Key \ -d { jsonrpc: 2.0, id: 4, method: tools/call, params: { name: execute_mysql_sql, arguments: { sql: select 1 } } }返回result.content里有執(zhí)行結(jié)果就說明整條鏈路通了??蛻舳藗?cè)Python 的streamablehttp_client加 header 的方式from mcp.client.streamable_http import streamablehttp_client from mcp import ClientSession async def connect(): headers {Authorization: Bearer sk-你的Key} async with streamablehttp_client( urlhttp://127.0.0.1:18088/mcp/, headersheaders ) as (read, write, _): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() print([t.name for t in tools.tools])跑通后你會(huì)看到工具名打印出來。如果客戶端報(bào)連接錯(cuò)誤先確認(rèn) header 拼寫是Authorization而不是authorizationHTTP header 大小寫不敏感但有些客戶端庫會(huì)嚴(yán)格匹配以及 Bearer 后面有一個(gè)空格。5. FastMCP 鑒權(quán)中間件常見報(bào)錯(cuò)排查401、local proxy failed 與 reading choices實(shí)際接入時(shí)踩的坑集中在幾個(gè)報(bào)錯(cuò)上逐個(gè)對(duì)照。401 Authorization Required中間件拋出的第一個(gè)錯(cuò)誤說明請(qǐng)求頭里沒有Authorization字段或者格式不是Bearer xxx。檢查客戶端配置里 header 的 key 是不是寫成了Auth、Token之類的自定義名。FastMCP 的get_http_request().headers.get(authorization)只認(rèn)標(biāo)準(zhǔn)名。另外注意有些客戶端會(huì)把 header 嵌套在headers對(duì)象里別寫成頂層字段。403 Forbiddenheader 格式對(duì)但 token 校驗(yàn)沒過。常見原因是環(huán)境變量沒注入——TAOTOKEN_API_KEY在服務(wù)進(jìn)程里是空字符串導(dǎo)致任何 token 都比對(duì)失敗。用echo $TAOTOKEN_API_KEY確認(rèn)或者在中間件里加一行日志打印收到的 token 前幾位。還有一種情況是 Key 復(fù)制時(shí)帶了首尾空格strip()一下。local proxy failed這個(gè)報(bào)錯(cuò)通常出現(xiàn)在客戶端側(cè)說明客戶端嘗試連接 MCP Server 時(shí)網(wǎng)絡(luò)層就失敗了根本沒到鑒權(quán)中間件。檢查三件事服務(wù)是否真的在監(jiān)聽netstat -tlnp | grep 18088、URL 路徑是否帶對(duì)了/mcp/結(jié)尾的斜杠、防火墻是否放行。如果服務(wù)在容器里127.0.0.1要換成容器實(shí)際 IP 或host.docker.internal。reading choices 相關(guān)報(bào)錯(cuò)這類錯(cuò)誤一般出現(xiàn)在工具執(zhí)行階段模型返回的響應(yīng)結(jié)構(gòu)不符合預(yù)期比如choices字段為空或格式變了。它和鑒權(quán)中間件沒有直接關(guān)系但容易被誤判成鑒權(quán)問題。排查方法是先確認(rèn)鑒權(quán)已通過日志里有auth passed再單獨(dú)測工具邏輯。如果工具內(nèi)部調(diào)用了模型接口檢查 Model ID 是否寫對(duì)、Base URL 是否是https://taotoken.net/api。Model ID 寫錯(cuò)時(shí)接口通常返回 404 或 400而不是 401。OAuth 相關(guān)報(bào)錯(cuò)如果你在中間件里接了 OAuth 流程報(bào)invalid_grant或token expired說明憑證過期了。用 TaoToken 統(tǒng)一 Key 的好處就在這里——過期和吊銷在控制臺(tái)處理MCP Server 側(cè)不用改代碼。重新生成 Key 后更新環(huán)境變量重啟即可。中間件不生效請(qǐng)求沒帶 header 卻返回了 200。檢查mcp.add_middleware(AuthMiddleware())是否在mcp.run()之前調(diào)用以及是否真的走了 HTTP 傳輸。stdio 模式下get_http_request()會(huì)拋異常如果你的代碼里把這個(gè)異常吞掉了直接call_next那 stdio 請(qǐng)求就繞過了鑒權(quán)——這是設(shè)計(jì)如此但如果你只想要 HTTP 模式就別開 stdio。排查時(shí)建議在中間件里加結(jié)構(gòu)化日志把context.method、請(qǐng)求來源 IP、校驗(yàn)結(jié)果都打出來。這樣出問題時(shí)一眼能看出是沒收到請(qǐng)求、還是收到了但校驗(yàn)失敗。6. 把統(tǒng)一 Key 接進(jìn)你的 MCP 工作流鑒權(quán)中間件跑通之后你的 FastMCP Server 就從局域網(wǎng)裸奔變成了憑證準(zhǔn)入。團(tuán)隊(duì)里每個(gè)人用自己的 TaoToken Key你在控制臺(tái)統(tǒng)一管理簽發(fā)和吊銷MCP Server 側(cè)只保留一段校驗(yàn)邏輯。工具本身要調(diào)模型時(shí)同一套 Key 直接復(fù)用Base URL 用 https://taotoken.net/api Model ID 按控制臺(tái)實(shí)際可用的填。如果你還在本地調(diào)試階段想先驗(yàn)證模型對(duì)話鏈路可以去模型對(duì)話頁 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchatutm_campaignrewrite 試一下 Key 是否可用。長期跑編碼類 Agent、需要穩(wěn)定額度的看 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。接入過程中遇到鑒權(quán)或配置問題接入文檔 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里有完整的參數(shù)說明API Keys 在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 管理。最后留一個(gè)實(shí)用技巧中間件里校驗(yàn)通過后可以把a(bǔ)ccess_token對(duì)應(yīng)的用戶身份塞進(jìn)context的擴(kuò)展字段這樣工具執(zhí)行時(shí)能拿到是誰在調(diào)方便做審計(jì)日志和按用戶限流。FastMCP 的MiddlewareContext支持附加數(shù)據(jù)具體字段名參考文檔別硬編碼。