免费获取学习方案
ARTICLE DETAIL

资讯详情

深耕编程基础知识与建站技术分享的一线实战洞察。

AI系列相关-(4)MCP中间件和MCP鉴权:用TaoToken统一Key打通FastMCP鉴权链路

AI系列相关-(4)MCP中间件和MCP鉴权:用TaoToken统一Key打通FastMCP鉴权链路 1. 自建 FastMCP Server 为什么必须补上鉴权中间件如果你已经用 FastMCP 跑通了一个 HTTP 模式的 MCP Server大概率会经历这样一个阶段本地streamablehttp_client连上去list_tools一列工具全出来了调用也正常于是顺手把端口映射到内网想着先给同事用着。问题就出在这里——FastMCP 默认不校验任何身份任何能访问到/mcp/这个路径的客户端都能直接发 JSON-RPC 请求把你的工具全调一遍。我试过在一个只做了防火墙 IP 白名单的 MCP Server 上做压力测试只要请求源在允许网段内tools/call完全不设防。这意味着一旦有人把内网地址泄露出去或者某台被允许的机器被当成跳板你的数据库查询工具、文件操作工具就全部暴露了。MCP 基于 JSON-RPC 规范运行请求体长这样{ jsonrpc: 2.0, id: 1, method: tools/call, params: { name: execute_mysql_sql, arguments: { sql: select * from shares_day_info limit 3 } } }注意这里没有任何身份字段。JSON-RPC 本身是传输无关的协议鉴权信息只能挂在传输层——HTTP 模式下就是请求头。所以正确的做法是在 FastMCP 的中间件管道里插一层拦截所有进入的 MCP 消息从 HTTP header 里取出凭证做校验不合法就直接抛错让请求根本到不了工具执行阶段。FastMCP 中间件采用管道模型请求按添加顺序流经每个中间件每个中间件可以检查请求、修改请求、调用call_next()交给下一个、再检查响应。它提供了从通用到具体的钩子层级——on_message管所有消息on_request只管需要响应的请求on_call_tool只管工具调用。鉴权这种所有请求都要过的逻辑用on_message或直接在__call__里做最稳妥因为工具发现list_tools本身也是一次请求不拦的话别人照样能枚举你有哪些工具。这一篇要解决的就是给自建 FastMCP Server 加一层鉴权中间件并且用 TaoToken 的统一 Key 体系把凭证管理收敛到一处避免每个 MCP Server 各写一套 token 表。适合正在把 MCP Server 从局域网自用推向团队共享的开发者。下面从环境准备、中间件配置、请求验证到报错排查一步步给可复制的片段。2. TaoToken 统一 Key 接入 FastMCP 鉴权链路的前置准备在写中间件之前先把凭证来源理清楚。最原始的做法是在自己库里建一张mcp_server_oauth_tokens表存用户名、token、过期时间中间件查库比对。这个方案能跑但有几个现实问题每个 MCP Server 都要连一次库、token 轮换要手动改表、多个服务之间凭证不互通。当你有三四个 MCP Server 时维护成本就上来了。更省事的思路是把签发和校验凭证这件事交给一个统一入口MCP Server 只负责拿请求头里的 Key 去问一句这个 Key 有效吗。TaoToken 在这里扮演的就是这个统一 Key 层——你可以在它的控制台里生成和管理 API KeyMCP Server 侧只需要配置 Base URL、Key、Model ID 三件套里的前两件用于鉴权校验工具本身要调模型时再补上 Model ID。先做前置准备。打开 TaoToken 官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册并进入控制台在 API Keys 页面创建一个 Key。这个 Key 就是你后面要写进 MCP 客户端 header 的凭证。控制台地址是 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 之后本地环境需要确认几件事。Python 侧装好 FastMCP 和 HTTP 相关依赖pip install fastmcp httpx starlette uvicornFastMCP 的中间件基类在fastmcp.server.middleware下HTTP 请求对象通过get_http_request()获取。注意这个函数只在 HTTP 传输下有效标准 I/O 传输拿不到 header——这也是为什么鉴权中间件只对 HTTP 模式有意义。如果你同时支持两种传输中间件里要先判断传输类型否则 stdio 模式下会直接报错。配置层面建议把 TaoToken 的校验地址和你的 Key 放进环境变量别硬编码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 因为它是程序调用的端点不需要追踪参数。Key 从环境变量读中间件里用os.environ.get()取这样换 Key 不用改代码。还有一点要提前想清楚鉴权中间件校验的是调用方有没有资格访问这个 MCP Server而 TaoToken 的 Key 校验的是这个 Key 有没有资格用 TaoToken 的服务。两者可以合一——直接把 TaoToken 的 Key 当作 MCP Server 的访问凭证中间件拿它去调一次 TaoToken 的接口验证有效性。这样团队里每个人用自己的 TaoToken Key你不需要再维护一张 token 表。下面第三节就给这个方案的完整配置。3. FastMCP 鉴权中间件可复制配置与 TaoToken Key 校验片段这一节是核心直接给能跑的代码。先看中间件本体它拦截所有 MCP 消息从 HTTP header 取Authorization解析出 Bearer token然后校验。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 请求对象直接放行 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) # 校验 token 是否有效这里用 TaoToken 的 Key 作为统一凭证 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 比对生产环境可换成调用 TaoToken 校验接口 if TAOTOKEN_API_KEY and token TAOTOKEN_API_KEY: return True # 也可以在这里调用 TaoToken 的接口做在线校验 return False把中间件挂到 FastMCP 实例上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)如果你更习惯用配置文件管理FastMCP 支持从 JSON 读取服务定义。下面是一个mcp_config.json片段把鉴权相关的环境变量和传输方式写进去{ 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}从环境变量注入避免明文写进配置文件。这个 JSON 结构可以直接被支持 MCP 配置的客户端读取。如果你用的是 Cline 或 Claude Code 这类工具它们的 MCP 配置通常放在settings.json或claude_desktop_config.json里结构类似{ mcpServers: { secure-mysql: { command: python, args: [-m, your_mcp_server], env: { TAOTOKEN_API_BASE: https://taotoken.net/api, TAOTOKEN_API_KEY: sk-你的Key } } } }这里要强调三件套的完整性Base URL 用https://taotoken.net/apiKey 用你在控制台生成的sk-开头字符串Model ID 在工具内部需要调模型时再指定比如claude-sonnet-4-5之类以控制台实际可用的为准。鉴权中间件只用到前两件第三件是工具执行阶段的事别混在一起。中间件里_verify_token目前是简单比对生产环境建议改成调用 TaoToken 的校验接口这样 Key 的吊销和过期由 TaoToken 侧统一管理你的 MCP Server 不用重启。调用方式就是拿access_token去请求 TaoToken 的 API返回 200 即有效。具体接口路径参考接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。配置写完后启动服务python your_mcp_server.py看到 uvicorn 监听 18088 端口就说明起来了。下一节验证请求。4. 带鉴权头的 JSON-RPC 请求验证与成功结果确认服务起来后先验证未授权被拦截。用 curl 直接发一个不带 header 的 JSON-RPC 请求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: {} }预期返回 401 或 403body 里能看到Authorization Required或Forbidden。这一步确认中间件确实拦住了没有凭证的请求。再验证错误凭证被拒绝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: {} }预期返回 403。如果这里返回了 200 并且列出了工具说明你的_verify_token逻辑有问题检查是不是把空 token 也放行了。最后验证正确凭证正常返回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: {} }成功的话会返回工具列表类似{ jsonrpc: 2.0, id: 3, result: { tools: [ { name: execute_mysql_sql, description: 执行 SQL 查询, inputSchema: { type: object, properties: { sql: { type: string } } } } ] } }再发一个tools/call验证工具能真正执行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里有执行结果就说明整条链路通了。客户端侧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])跑通后你会看到工具名打印出来。如果客户端报连接错误先确认 header 拼写是Authorization而不是authorizationHTTP header 大小写不敏感但有些客户端库会严格匹配以及 Bearer 后面有一个空格。5. FastMCP 鉴权中间件常见报错排查401、local proxy failed 与 reading choices实际接入时踩的坑集中在几个报错上逐个对照。401 Authorization Required中间件抛出的第一个错误说明请求头里没有Authorization字段或者格式不是Bearer xxx。检查客户端配置里 header 的 key 是不是写成了Auth、Token之类的自定义名。FastMCP 的get_http_request().headers.get(authorization)只认标准名。另外注意有些客户端会把 header 嵌套在headers对象里别写成顶层字段。403 Forbiddenheader 格式对但 token 校验没过。常见原因是环境变量没注入——TAOTOKEN_API_KEY在服务进程里是空字符串导致任何 token 都比对失败。用echo $TAOTOKEN_API_KEY确认或者在中间件里加一行日志打印收到的 token 前几位。还有一种情况是 Key 复制时带了首尾空格strip()一下。local proxy failed这个报错通常出现在客户端侧说明客户端尝试连接 MCP Server 时网络层就失败了根本没到鉴权中间件。检查三件事服务是否真的在监听netstat -tlnp | grep 18088、URL 路径是否带对了/mcp/结尾的斜杠、防火墙是否放行。如果服务在容器里127.0.0.1要换成容器实际 IP 或host.docker.internal。reading choices 相关报错这类错误一般出现在工具执行阶段模型返回的响应结构不符合预期比如choices字段为空或格式变了。它和鉴权中间件没有直接关系但容易被误判成鉴权问题。排查方法是先确认鉴权已通过日志里有auth passed再单独测工具逻辑。如果工具内部调用了模型接口检查 Model ID 是否写对、Base URL 是否是https://taotoken.net/api。Model ID 写错时接口通常返回 404 或 400而不是 401。OAuth 相关报错如果你在中间件里接了 OAuth 流程报invalid_grant或token expired说明凭证过期了。用 TaoToken 统一 Key 的好处就在这里——过期和吊销在控制台处理MCP Server 侧不用改代码。重新生成 Key 后更新环境变量重启即可。中间件不生效请求没带 header 却返回了 200。检查mcp.add_middleware(AuthMiddleware())是否在mcp.run()之前调用以及是否真的走了 HTTP 传输。stdio 模式下get_http_request()会抛异常如果你的代码里把这个异常吞掉了直接call_next那 stdio 请求就绕过了鉴权——这是设计如此但如果你只想要 HTTP 模式就别开 stdio。排查时建议在中间件里加结构化日志把context.method、请求来源 IP、校验结果都打出来。这样出问题时一眼能看出是没收到请求、还是收到了但校验失败。6. 把统一 Key 接进你的 MCP 工作流鉴权中间件跑通之后你的 FastMCP Server 就从局域网裸奔变成了凭证准入。团队里每个人用自己的 TaoToken Key你在控制台统一管理签发和吊销MCP Server 侧只保留一段校验逻辑。工具本身要调模型时同一套 Key 直接复用Base URL 用 https://taotoken.net/api Model ID 按控制台实际可用的填。如果你还在本地调试阶段想先验证模型对话链路可以去模型对话页 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchatutm_campaignrewrite 试一下 Key 是否可用。长期跑编码类 Agent、需要稳定额度的看 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。接入过程中遇到鉴权或配置问题接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里有完整的参数说明API Keys 在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 管理。最后留一个实用技巧中间件里校验通过后可以把access_token对应的用户身份塞进context的扩展字段这样工具执行时能拿到是谁在调方便做审计日志和按用户限流。FastMCP 的MiddlewareContext支持附加数据具体字段名参考文档别硬编码。
返回列表