1. 为什么要在 FastMCP 里加一层中间件FastMCP 中间件是什么简单说它是插在 MCP 客户端请求和你的工具函数之间的一道处理链能在请求真正打到业务逻辑之前做鉴权、限流、日志、参数改写在响应返回之前做格式转换、错误包装。适合谁适合已经把 MCP 服务器跑起来、工具数量超过三五个、开始需要统一管控调用入口的开发者。我试过把鉴权和日志硬编码进每个工具函数里工具一多就变成复制粘贴灾难改一处漏三处中间件就是来解决这个问题的。MCPModel Context Protocol本身解决的是模型和外部工具、资源之间的标准化通信但协议规范里并没有规定你必须怎么做认证、怎么记录调用链。FastMCP 从 2.9.0 开始引入的中间件系统就是补上这块工程化能力。它采用管道-过滤器模式请求进来后依次穿过每个中间件每个中间件可以选择继续传递、修改数据或者直接终止并返回错误。这跟传统 Web 框架的中间件思路一致但钩子是按 MCP 操作类型细分的比如工具调用、资源读取、提示获取各有专用钩子。实际场景里一个 MCP 服务器往往要对接多个上游模型通道。如果每个工具自己去读环境变量、自己拼 API Key、自己处理 401代码会非常散。把统一接入逻辑收拢到中间件里工具函数只关心业务通道切换、Key 轮换、失败重试都在中间件层完成这才是可扩展的服务器架构。下面我会从零给出可复制的中间件注册配置、统一 Key 通道接入示例以及请求链路验证动作确认中间件确实按预期拦截和放行。2. TaoToken 统一接入前置准备在写中间件之前先把统一通道准备好。TaoToken 在这里扮演的角色是给你的 MCP 服务器提供一个统一的模型调用入口工具函数不需要各自持有不同的 Key而是通过中间件注入统一的 Base URL 和 Key。官网入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_end API 地址是 https://taotoken.net/api 注意 API 地址不带任何查询参数。你需要准备三样东西我把它叫做三件套Base URL、API Key、Model ID。Base URL 填https://taotoken.net/apiAPI Key 在控制台的 API Keys 页面创建Model ID 按你实际要调用的模型填写。这三件套在后面的中间件配置里会以环境变量形式注入避免硬编码进代码。创建 Key 的路径是控制台的 api-keys 页面登录后新建一个 Key复制出来保存好它只显示一次。如果你还没决定用哪个模型可以先去模型对话页面试一下调用效果确认通道通了再写进中间件。对于长期跑编码类 Agent 的场景Coding Plan 会更划算适合把 MCP 服务器当成常驻服务来用的开发者。这里要强调一点中间件里做统一接入核心是把「通道选择」和「业务逻辑」解耦。工具函数不应该知道当前用的是哪个上游它只调用一个内部函数由中间件负责把请求转发到 TaoToken 的 API 地址并带上正确的 Key 和 Model ID。这样以后换模型、加通道只改中间件配置不动工具代码。环境变量建议这样组织放在.env或者服务器的环境配置里TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_API_KEYsk-你的key TAOTOKEN_MODEL_ID你的模型ID注意 Base URL 结尾不要多加斜杠很多 401 和 404 都是路径拼接问题导致的。Key 不要提交到 Git用.env加.gitignore管理。Model ID 要和你在控制台看到的完全一致大小写敏感。3. 可复制的中间件注册配置这一节给出可以直接抄的配置。FastMCP 中间件通过mcp.add_middleware()注册顺序决定执行链先添加的先执行前置处理后执行后置处理。所以错误处理放最前认证其次限流再次日志放最后。先看一个统一鉴权 通道注入的中间件实现。它的职责是从请求上下文里取出调用方标识校验是否允许然后把 TaoToken 的三件套注入到 FastMCP 上下文供后续工具函数读取。import os from fastmcp.server.middleware import Middleware, MiddlewareContext from fastmcp.exceptions import ToolError class TaoTokenAuthMiddleware(Middleware): 统一鉴权 TaoToken 通道注入中间件 def __init__(self, allowed_tokens: set[str]): self.allowed_tokens allowed_tokens self.base_url os.environ[TAOTOKEN_BASE_URL] self.api_key os.environ[TAOTOKEN_API_KEY] self.model_id os.environ[TAOTOKEN_MODEL_ID] async def on_request(self, context: MiddlewareContext, call_next): headers context.message.get(headers, {}) or {} token headers.get(X-MCP-Token) if not token or token not in self.allowed_tokens: raise ToolError(未授权缺少或无效的 X-MCP-Token) # 把三件套挂到 FastMCP 上下文工具函数可直接读取 ctx context.fastmcp_context ctx.taotoken_base_url self.base_url ctx.taotoken_api_key self.api_key ctx.taotoken_model_id self.model_id return await call_next(context)注册时把中间件按顺序加进去。下面这段是完整的服务器启动配置包含错误处理、鉴权、限流、计时、日志五个中间件from fastmcp import FastMCP from fastmcp.server.middleware.error_handling import ErrorHandlingMiddleware from fastmcp.server.middleware.rate_limiting import RateLimitingMiddleware from fastmcp.server.middleware.timing import TimingMiddleware from fastmcp.server.middleware.logging import StructuredLoggingMiddleware mcp FastMCP(TaoTokenMCP) # 顺序即执行链不要随意调换 mcp.add_middleware(ErrorHandlingMiddleware(include_tracebackTrue, transform_errorsTrue)) mcp.add_middleware(TaoTokenAuthMiddleware(allowed_tokens{dev-token-001})) mcp.add_middleware(RateLimitingMiddleware(max_requests_per_second10.0, burst_capacity20)) mcp.add_middleware(TimingMiddleware()) mcp.add_middleware(StructuredLoggingMiddleware())如果你用配置文件方式管理可以写一个 JSON 片段路径放在项目根目录的mcp_config.json内容如下{ server_name: TaoTokenMCP, middlewares: [ {type: error_handling, include_traceback: true}, {type: taotoken_auth, allowed_tokens: [dev-token-001]}, {type: rate_limiting, max_requests_per_second: 10.0, burst_capacity: 20}, {type: timing}, {type: structured_logging} ], taotoken: { base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, model_id_env: TAOTOKEN_MODEL_ID } }注意 JSON 里不要写死 Key用环境变量名引用。这样配置文件和代码分离部署时只改环境变量。工具函数里读取三件套的方式是ctx.taotoken_base_url、ctx.taotoken_api_key、ctx.taotoken_model_id这三个字段由中间件注入工具本身不碰环境变量。4. 验证请求链路与成功结果配置写完必须验证中间件真的在拦截和放行。验证分三步先确认未带 Token 的请求被拦再确认带正确 Token 的请求放行最后确认工具函数拿到了注入的三件套。第一步构造一个不带X-MCP-Token的调用。用 MCP 客户端或者直接发请求预期结果是收到ToolError错误信息是「未授权缺少或无效的 X-MCP-Token」。如果这一步没拦住说明中间件没注册成功或者钩子写错了。第二步带上正确的 Token 再调一次import asyncio from fastmcp import Client async def main(): async with Client(http://127.0.0.1:8000/mcp) as client: result await client.call_tool( echo_tool, {text: hello}, headers{X-MCP-Token: dev-token-001} ) print(result) asyncio.run(main())预期输出是工具正常返回同时服务端日志里能看到StructuredLoggingMiddleware打出的 JSON 日志包含 method、source、耗时字段。如果日志里出现了tools/call且耗时正常说明请求穿过了整条链。第三步在工具函数里打印注入的三件套确认中间件确实把值传下去了mcp.tool() async def echo_tool(text: str, ctx) - str: print(base_url:, ctx.taotoken_base_url) print(model_id:, ctx.taotoken_model_id) return fecho: {text}调用后服务端应打印出https://taotoken.net/api和你的 Model ID。如果打印出 None说明中间件注入的字段名和工具读取的字段名不一致检查ctx.taotoken_base_url拼写。成功结果的特征是未授权请求被拦、授权请求放行、日志完整、三件套正确注入。四个都满足中间件链路就算通了。这时候你可以在工具函数里用注入的 Key 去调 TaoToken 的 API把模型返回结果包装成 MCP 工具输出。5. 本篇常见错误排查第一个高频错误是 401。表现是工具调用返回未授权但你的 Token 明明是对的。排查顺序先看中间件是否真的注册了mcp.add_middleware有没有被调用再看钩子是不是写成了on_message而请求走的是on_request最后看context.message.get(headers)是否为空有些客户端把 headers 放在别的位置。如果 401 来自上游而不是你的中间件检查TAOTOKEN_API_KEY是否过期去控制台重新生成。第二个错误是local proxy failed。这个通常出现在中间件里做了异步阻塞操作比如在async钩子里调用了同步的requests.get。中间件必须全程async/await同步库会阻塞事件循环导致代理层超时。把同步调用换成httpx.AsyncClient即可。第三个错误是reading choices相关报错。这多半是工具函数里解析上游响应时字段路径不对。TaoToken 返回的是标准 OpenAI 兼容格式choices 在response[choices][0][message][content]。如果你在中间件里改写过响应结构确认没有把 choices 字段吃掉。第四个错误是 OAuth 相关报错。如果你用的是需要 OAuth 的客户端注意 MCP 的鉴权和 OAuth 是两套东西。中间件里的X-MCP-Token是你自己定义的调用方令牌不是 OAuth access token。两者不要混用否则会出现令牌校验互相覆盖。还有一个容易忽略的点中间件顺序错了会导致限流在鉴权之前执行未授权请求也消耗配额。正确顺序是错误处理 → 鉴权 → 限流 → 计时 → 日志。每次改中间件顺序后都要重新跑一遍第 4 节的验证三步。6. 把统一通道接到你的工具里中间件链路通了之后工具函数里就可以直接用注入的三件套去调 TaoToken。下面是一个完整的工具示例它读取上下文里的 Base URL、Key、Model ID发一个 chat 请求把结果返回给 MCP 客户端import httpx from fastmcp import FastMCP mcp FastMCP(TaoTokenMCP) mcp.tool() async def ask_model(prompt: str, ctx) - str: url f{ctx.taotoken_base_url}/v1/chat/completions headers {Authorization: fBearer {ctx.taotoken_api_key}} payload { model: ctx.taotoken_model_id, messages: [{role: user, content: prompt}] } async with httpx.AsyncClient(timeout30) as client: resp await client.post(url, jsonpayload, headersheaders) resp.raise_for_status() data resp.json() return data[choices][0][message][content]注意 URL 拼接Base URL 是https://taotoken.net/apichat 接口路径是/v1/chat/completions拼起来就是https://taotoken.net/api/v1/chat/completions。如果你在 Base URL 结尾加了斜杠会变成双斜杠部分网关会返回 404。工具写完后重新走一遍第 4 节的验证不带 Token 调用应被拦带 Token 调用应返回模型结果。如果返回的是模型内容而不是报错说明从中间件鉴权到通道注入到上游调用整条链路都通了。后续扩展时你只需要在中间件层加新的钩子比如给特定工具加权限标签检查或者在响应返回前做敏感词过滤工具函数本身不用改。这就是中间件带来的可扩展性非功能需求集中在中间件业务逻辑集中在工具两边各自演进互不干扰。需要长期跑 Agent 的话Coding Plan 配合这套中间件架构可以把通道管理和调用管控都收拢到一处。