ARTICLE DETAIL

资讯详情

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

FastMCP 服务器框架生命周期钩子:用 lifespan 异步上下文管理器搭建可复用配置骨架

FastMCP 服务器框架生命周期钩子:用 lifespan 异步上下文管理器搭建可复用配置骨架 1. FastMCP 生命周期钩子到底解决什么问题FastMCP 服务器框架里生命周期钩子lifespan是一个异步上下文管理器它让你在服务器启动时初始化资源、在关闭时清理资源。适合谁适合所有需要管理数据库连接池、缓存客户端、外部 API 长连接、文件句柄或后台定时任务的服务端开发者。能做什么简单说它把「启动时建连、请求时复用、关闭时释放」这条链路收进一个可复用的配置骨架里避免每个工具函数各自连一次数据库。我见过太多 FastMCP 项目把数据库连接写在每个 tool 函数内部请求一多连接数直接打满排查半天才发现是连接没复用。lifespan 就是来解决这个问题的服务器进程启动时执行一次初始化把资源挂到上下文里所有请求通过request_context.lifespan_context取用进程退出时统一释放。FastMCP 的 lifespan 参数接受一个可调用对象该对象返回异步上下文管理器。对于 FastMCP 类这个参数会经过lifespan_wrapper包装适配底层 MCPServer 的生命周期管理。你只需要关心三件事定义异步上下文管理器、在 yield 前做初始化、在 finally 里做清理。这篇会给出可直接复制的 lifespan 配置骨架、settings.json 与 config.toml 示例并说明如何通过 TaoToken 统一 Key/API 通道接入模型服务最后附上启动日志与钩子触发顺序的验证动作。2. TaoToken 前置统一 Key 与 API 通道在写 lifespan 之前先把模型接入通道固定下来。TaoToken 提供统一的 API 入口你可以在一个地方管理 Key不用在代码里散落多个厂商的 base_url 和 token。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。实际操作路径先到控制台创建 API Key地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_campaignrewriteutm_content Key 管理页在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_campaignrewriteutm_content 。拿到 Key 之后把它写进环境变量或配置文件lifespan 初始化时读取一次挂到上下文里供所有请求复用。如果你需要长期跑编码类 Agent 或高频调用可以看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_campaignrewriteutm_content 。想先验证模型对话是否通用模型对话页https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_campaignrewriteutm_content 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_campaignrewriteutm_content 。注意Key 不要硬编码进源码提交到仓库。用环境变量或本地配置文件并在 .gitignore 里排除。3. 可复制的 lifespan 配置骨架3.1 基础骨架异步上下文管理器先看最小可用的 lifespan 骨架。核心是asynccontextmanager装饰的异步函数yield 之前做初始化finally 里做清理。from contextlib import asynccontextmanager from fastmcp import FastMCP asynccontextmanager async def app_lifespan(server: FastMCP): # 启动阶段初始化资源 db await Database.connect() cache await CacheClient.connect() api_client build_api_client() try: yield {db: db, cache: cache, api: api_client} finally: # 关闭阶段逆序释放 await api_client.close() await cache.close() await db.disconnect() mcp FastMCP(demo-server, lifespanapp_lifespan)这段代码里yield返回的字典就是 lifespan_context请求处理时通过server.request_context.lifespan_context[db]取用。初始化顺序是 db → cache → api清理顺序反过来这是资源释放的稳妥做法。3.2 在工具函数中访问上下文mcp.tool() async def query_user(user_id: int) - dict: ctx mcp.request_context db ctx.lifespan_context[db] row await db.fetch_one(SELECT * FROM users WHERE id ?, user_id) return {user: row}这里的关键是mcp.request_context拿到当前请求上下文再通过lifespan_context字典取资源。不要在 tool 里重新建连接否则 lifespan 就白写了。3.3 settings.json 配置示例把模型通道和资源参数抽到配置文件lifespan 初始化时读取。{ server: { name: demo-server, transport: stdio }, taotoken: { base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, default_model: claude-sonnet }, resources: { db_url_env: DATABASE_URL, cache_url_env: REDIS_URL, pool_size: 10 } }3.4 config.toml 配置示例如果你偏好 TOML等价配置如下。[server] name demo-server transport stdio [taotoken] base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY default_model claude-sonnet [resources] db_url_env DATABASE_URL cache_url_env REDIS_URL pool_size 103.5 读取配置并注入 lifespanimport json, os from contextlib import asynccontextmanager from fastmcp import FastMCP def load_settings(pathsettings.json): with open(path, r, encodingutf-8) as f: return json.load(f) settings load_settings() asynccontextmanager async def app_lifespan(server: FastMCP): db await Database.connect( os.environ[settings[resources][db_url_env]], pool_sizesettings[resources][pool_size], ) api_key os.environ[settings[taotoken][api_key_env]] api_client build_api_client( base_urlsettings[taotoken][base_url], api_keyapi_key, ) try: yield {db: db, api: api_client} finally: await api_client.close() await db.disconnect() mcp FastMCP(settings[server][name], lifespanapp_lifespan)这样配置和代码分离换环境只改配置文件或环境变量不用动 lifespan 逻辑。4. 验证请求与钩子触发顺序4.1 启动日志验证在 lifespan 的初始化和清理阶段各加一条日志启动服务器后观察输出顺序。import logging logger logging.getLogger(lifespan) asynccontextmanager async def app_lifespan(server: FastMCP): logger.info(lifespan: 启动初始化开始) db await Database.connect() logger.info(lifespan: 数据库连接就绪) api_client build_api_client() logger.info(lifespan: API 客户端就绪) try: yield {db: db, api: api_client} finally: logger.info(lifespan: 清理开始) await api_client.close() await db.disconnect() logger.info(lifespan: 清理完成)正常启动日志应该是启动初始化开始 → 数据库连接就绪 → API 客户端就绪 → 服务器开始监听。关闭时清理开始 → 清理完成。4.2 钩子触发顺序验证用一个带时间戳的探针工具确认上下文可用。import time mcp.tool() async def probe() - dict: ctx mcp.request_context db ctx.lifespan_context.get(db) return { has_db: db is not None, timestamp: time.time(), }调用 probe如果has_db为 true说明 lifespan 的 yield 上下文正确注入。如果为 false检查 lifespan 是否真的传给了 FastMCP 构造函数。4.3 通过 TaoToken 验证模型通道在 lifespan 里初始化的 api_client 指向 TaoToken 的 API 入口发一个最小请求确认通道可用。async def check_model_channel(api_client): resp await api_client.chat( modelclaude-sonnet, messages[{role: user, content: ping}], ) return resp如果返回正常说明 Key 和 base_url 配置正确。想先在网页端确认模型可用用模型对话页https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_campaignrewriteutm_content 。5. 本篇常见错排查5.1 lifespan 没生效上下文取不到最常见的原因是 lifespan 函数没有正确传给 FastMCP 构造函数或者函数签名不对。lifespan 必须接受 server 参数并返回异步上下文管理器。检查FastMCP(..., lifespanapp_lifespan)是否写对以及asynccontextmanager装饰器是否加上。5.2 清理逻辑没执行如果 finally 块里的清理没跑通常是进程被强杀kill -9导致。正常 SIGTERM 会触发清理。另外确认 yield 是否在 try 块内如果 yield 写在 try 外面异常时 finally 不会执行。5.3 资源在请求间串了lifespan_context 是全局共享的多个请求拿到的是同一个 db 连接对象。如果你的数据库驱动不是协程安全的需要加连接池或锁。别把请求级状态写进 lifespan_context那是进程级资源。5.4 Key 读取失败如果os.environ[settings[taotoken][api_key_env]]抛 KeyError说明环境变量没设。检查 shell 里是否 export 了对应变量或者用 .env 文件加载。Key 管理页在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_campaignrewriteutm_content 。5.5 启动顺序错乱如果日志显示 API 客户端先于数据库就绪检查 await 是否漏写。异步初始化必须逐个 await否则任务并发启动顺序不可控。6. 接入与排障入口lifespan 骨架搭好之后Key 和 API 通道的排障走 API Keys 页和接入文档https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_campaignrewriteutm_content 与 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_campaignrewriteutm_content 。验证模型是否通用模型对话页https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_campaignrewriteutm_content 。长期跑编码类 Agent 或高频调用看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_campaignrewriteutm_content 。最后留一个实用技巧把 lifespan 的初始化和清理各包一层 try/except初始化失败时记录明确错误并退出清理失败时记录但不阻断退出。这样启动日志里能一眼看出是哪个资源没起来比裸抛异常好排查得多。
返回列表