ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

FastMCP 服务器框架之自定义 HTTP 路由:把 custom_route 装饰器改到 TaoToken

FastMCP 服务器框架之自定义 HTTP 路由:把 custom_route 装饰器改到 TaoToken 1. 为什么要在 FastMCP 里加自定义 HTTP 路由FastMCP 默认只暴露 MCP 协议相关的端点客户端通过 stdio 或 SSE 跟它对话。但真实项目里你往往还需要一些「协议之外」的 HTTP 接口OAuth 回调要接收授权服务器重定向、健康检查要给负载均衡探活、管理后台要拉运行指标。这些需求用 MCP 的 tool 或 resource 都不合适因为它们本质上是普通 REST 请求不是模型调用的工具。custom_route装饰器就是为这个场景准备的。它让你在同一个 FastMCP 服务器实例上挂载任意 HTTP 路径处理函数拿到的是 Starlette 的Request对象返回Response对象跟写普通 Web 接口几乎一样。我试过把健康检查、OAuth 回调、内部管理 API 全塞进一个 MCP 服务器部署时只跑一个进程省掉单独起一个 Web 服务的麻烦。这篇文章面向需要为 MCP 服务器扩展 REST 接口的开发者。你会看到custom_route的完整参数、可复制的路由注册代码、请求处理函数的写法、启动配置以及用 curl 验证自定义端点返回结果。最后我会说明怎么把服务端点的 Base URL 改到 TaoToken 统一通道做联调让本地开发和线上调用走同一套鉴权和计费逻辑。核心检索词先明确FastMCP 的custom_route装饰器用于在 MCP 协议之外注册自定义 HTTP 路由适合 OAuth 回调、健康检查、管理 API 等场景。它接受path、methods、name、include_in_schema四个参数处理函数必须是async def handler(request: Request) - Response的签名。2. custom_route 装饰器参数与 Route 对象拆解先看装饰器的签名。FastMCP 的custom_route定义在server.py里参数如下参数类型说明pathstr路由 URL 路径如/oauth/callbackmethodslist[str]支持的 HTTP 方法如[GET, POST]namestr | None路由名称用于 Starlette 反向 URL 查找include_in_schemabool是否包含在 OpenAPI 架构中默认 True调用时系统会创建一个Route对象追加到服务器的_custom_starlette_routes列表。这个Route对象包含路径、端点处理函数、支持的方法、名称和架构包含设置。请求进来时Starlette 先匹配自定义路由匹配上就调用处理函数匹配不上才继续走 MCP 核心协议的路由。处理函数的签名必须严格遵循async def handler(request: Request) - Response: ...request是 Starlette 的Request你可以访问request.query_params、request.headers、await request.body()等。返回值可以是JSONResponse、PlainTextResponse、Response或任何 Starlette 响应类型。include_in_schema这个参数容易被忽略。当你写管理 API 或内部端点时把它设成False端点仍然可用但不会出现在自动生成的 OpenAPI 文档里。这给 API 的隐私性加了一层控制——外部扫描工具看不到这些路径但知道路径的人照样能调。一个常见的坑处理函数写成同步的def而不是async def。FastMCP 内部用await调用端点同步函数会直接抛TypeError。另一个坑是返回了dict而不是ResponseStarlette 不会自动序列化会报AssertionError。记住进去是Request出来是Response。3. 可复制的路由注册代码与启动配置下面是一份可以直接跑的完整示例。我把它拆成三块健康检查、OAuth 回调、带鉴权的管理 API。# server.py from fastmcp import FastMCP from starlette.requests import Request from starlette.responses import JSONResponse, PlainTextResponse, RedirectResponse mcp FastMCP(demo-server) # 1. 健康检查给负载均衡探活用 mcp.custom_route(/health, methods[GET]) async def health_check(request: Request) - JSONResponse: return JSONResponse({status: ok, service: demo-server}) # 2. OAuth 回调接收授权服务器重定向 mcp.custom_route(/oauth/callback, methods[GET]) async def oauth_callback(request: Request) - RedirectResponse: code request.query_params.get(code) state request.query_params.get(state) if not code: return JSONResponse({error: missing code}, status_code400) # 这里换成你自己的 token 交换逻辑 return RedirectResponse(urlf/app?state{state}) # 3. 管理 API带简单鉴权且不暴露在 OpenAPI 文档 mcp.custom_route(/admin/stats, methods[GET], include_in_schemaFalse) async def admin_stats(request: Request) - JSONResponse: token request.headers.get(X-Admin-Token) if token ! your-secret-token: return JSONResponse({error: unauthorized}, status_code401) return JSONResponse({tools: 3, uptime: 3600}) if __name__ __main__: mcp.run(transportsse, host0.0.0.0, port8000)启动配置的关键在mcp.run()。transportsse会启动一个 HTTP 服务器自定义路由和 MCP 的 SSE 端点共用同一个端口。如果你用 stdio 传输自定义路由不会生效因为 stdio 没有 HTTP 监听。所以扩展 REST 接口时必须用sse或streamable-http传输。如果你想把配置抽成 JSON 或 TOML方便不同环境切换可以这样写{ mcpServers: { demo-server: { url: http://127.0.0.1:8000/sse, transport: sse, baseUrl: https://taotoken.net/api, apiKey: sk-your-taotoken-key, modelId: claude-3-5-sonnet } } }这份配置里baseUrl、apiKey、modelId三件套是联调时统一通道的关键。url指向本地 FastMCP 的 SSE 端点baseUrl指向 TaoToken 的 API 入口模型调用走统一通道自定义路由走本地。两者不冲突因为自定义路由是纯 HTTP不经过模型。启动后你会看到类似输出INFO: Started server process [12345] INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRLC to quit)这时候/health、/oauth/callback、/admin/stats都已经挂上了。4. 用 curl 验证自定义端点返回结果服务起来后先验证健康检查curl -i http://127.0.0.1:8000/health预期返回HTTP/1.1 200 OK content-type: application/json {status:ok,service:demo-server}再验证 OAuth 回调带上 query 参数curl -i http://127.0.0.1:8000/oauth/callback?codeabc123statexyz预期返回 307 重定向到/app?statexyz。如果你没传code会拿到 400 和{error:missing code}。管理 API 要带 headercurl -i -H X-Admin-Token: your-secret-token http://127.0.0.1:8000/admin/stats预期返回HTTP/1.1 200 OK content-type: application/json {tools:3,uptime:3600}不带 token 时返回 401。这一步能验证鉴权逻辑是否生效。最后验证 MCP 核心协议没被自定义路由影响。用 MCP 客户端连http://127.0.0.1:8000/sse应该能正常列出 tools。如果连不上说明自定义路由的路径匹配把 SSE 端点拦截了——检查你的path有没有写成/或通配符。联调阶段把模型调用的 Base URL 改到 TaoToken 统一通道curl -X POST https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer sk-your-taotoken-key \ -H Content-Type: application/json \ -d {model:claude-3-5-sonnet,messages:[{role:user,content:ping}]}这样本地自定义路由和线上模型调用走同一套 Key 和计费排查问题时不用在两套凭证之间切换。TaoToken 的 API 入口是https://taotoken.net/api模型对话、Coding Plan、API Keys 管理都在同一个控制台里。5. 本篇常见错误排查报错一TypeError: object function cant be used in await expression原因处理函数写成了同步def。FastMCP 内部await调用端点同步函数不能被 await。修复把def health_check(request)改成async def health_check(request)。报错二AssertionError: Expected Response, got dict原因处理函数直接返回了dictStarlette 不会自动序列化。修复用JSONResponse({status: ok})包一层。报错三401 Unauthorized但 header 明明带了 token原因header 名大小写或拼写不对。Starlette 的request.headers.get()是大小写不敏感的但如果你写成X-Admin-Token而客户端发的是X-Admin-Token尾部空格会匹配失败。修复打印dict(request.headers)确认实际收到的 header。报错四local proxy failed或连接被拒绝原因mcp.run()用了transportstdio没有 HTTP 监听curl 自然连不上。修复改成transportsse或transportstreamable-http。报错五reading choices相关错误原因模型调用返回体解析失败通常是 Base URL 或 Model ID 配错。检查baseUrl是否指向https://taotoken.net/apimodelId是否是控制台里可用的模型。报错六OAuth 回调报OAuth相关错误原因回调端点没正确处理state校验或者重定向 URL 不在授权服务器的白名单里。修复在oauth_callback里先校验state再交换 token同时确认授权服务器配置的回调 URL 跟你的path完全一致。报错七自定义路由和 MCP 端点冲突原因path写成了/sse或/messages把 MCP 核心端点覆盖了。修复自定义路由用独立前缀比如/health、/admin/*、/oauth/*避开 MCP 保留路径。排查时记住一个原则先 curl 自定义路由确认 HTTP 层通再用 MCP 客户端连 SSE确认协议层通。两层分开验证问题定位快很多。6. 把联调通道固定下来自定义路由跑通后下一步是把本地开发、联调、线上三套环境的 Base URL 统一到 TaoToken。这样做的好处是模型调用的鉴权、计费、日志都在一个地方看不用在多个 Key 之间切换。具体操作在 TaoToken 控制台创建一个 API Key然后在你的 FastMCP 配置里把baseUrl指向https://taotoken.net/apiapiKey填刚创建的 KeymodelId填你要用的模型。本地自定义路由继续走127.0.0.1:8000模型调用走统一通道。如果你在做长期编码或 Agent 项目可以考虑 Coding Plan把模型调用额度固定下来避免每次调试都手动换 Key。需要看模型列表和对话测试直接进模型对话页面。API Key 管理在 API Keys 页面接入文档在文档页。最后留一个实用技巧把include_in_schemaFalse用在所有内部管理端点把include_in_schemaTrue留给需要对外暴露的 OAuth 回调。这样自动生成的 OpenAPI 文档只展示该展示的内部接口不泄露路径。健康检查端点建议也设成False因为负载均衡器不需要看文档知道路径就行。
返回列表