ARTICLE DETAIL

资讯详情

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

FastMCP高级特性之Context:依赖注入与异步上下文实战

FastMCP高级特性之Context:依赖注入与异步上下文实战 1. 为什么你的 MCP 工具函数拿不到请求上下文写 FastMCP 服务时最容易踩的坑不是工具函数写错而是上下文拿不到。比如你想在工具执行时打一条日志给客户端、上报一次进度、或者读取当前请求的 header结果发现函数签名里没有ctx或者get_context()一调用就抛RuntimeError。这类问题的根源是没搞清楚 FastMCP 的 Context 是请求级对象——它只在一次 MCP 请求的生命周期内存在请求结束就销毁。FastMCP 是 MCPModel Context Protocol的 Python 服务端框架用来把普通函数注册成工具、资源、提示词暴露给支持 MCP 的客户端调用。Context 则是 FastMCP 给每个请求注入的“随身工具箱”里面装着日志、进度、资源读取、提示词检索、LLM 采样、用户启发、状态管理、请求元数据等能力。适合谁适合正在构建 MCP 服务、需要工具函数与客户端双向交互的开发者尤其是做长任务、多租户、需要审计日志的场景。我见过太多人把 Context 当成全局单例来用在模块顶层ctx get_context()然后在工具里直接引用结果第一个请求能跑第二个请求就报错。原因很简单每个 MCP 请求都会收到一个全新的 Context 对象作用域限定在单个请求内。你在请求 A 里set_state的数据请求 B 里读不到你在请求外调用ctx.info()直接抛异常。这篇就围绕 FastMCP 的 Context 依赖注入与异步调用展开给出可复制的注入配置、异步工具函数示例以及本地启动与调用验证步骤。核心检索词是 FastMCP Context 依赖注入与异步上下文实战读完你能搞清楚三件事怎么把 Context 注入到工具/资源/提示词里怎么在嵌套调用深处拿到它以及怎么用自定义依赖做资源管理。下面从环境准备开始一步步跟做即可。2. 环境准备与 TaoToken 接入前置在写代码之前先把运行环境和模型接入准备好。FastMCP 本身是纯 Python 库但你要验证工具函数里的 LLM 采样、或者用 Claude Code 这类客户端去调用你的 MCP 服务就需要一个稳定的模型接入点。这里用 TaoToken 做统一接入它的 API 地址是https://taotoken.net/api兼容 OpenAI 风格的调用方式MCP 服务里做ctx.sample()时可以直接复用。先装依赖。建议用 Python 3.10 以上FastMCP 对异步和类型提示要求较高python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install fastmcp httpx装完后确认版本python -c import fastmcp; print(fastmcp.__version__)接下来拿 TaoToken 的 Key。打开https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentcontext_diutm_campaignrewrite登录后在控制台创建 API Key复制出来。这个 Key 后面会用在两个地方一是 MCP 服务里做 LLM 采样时作为模型凭证二是用 Claude Code 或 Cline 这类客户端连接你的 MCP 服务时做鉴权。把 Key 写进环境变量别硬编码进代码export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api如果你打算用 Claude Code 来调试 MCP 服务需要配置三件套Base URL、Key、Model ID。Claude Code 的配置文件通常在~/.claude/settings.json写入{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_API_KEY: sk-你的key, ANTHROPIC_MODEL: claude-sonnet-4-20250514 } }注意 Base URL 这里用的是https://taotoken.net/api不要加 UTM 参数UTM 只用于网页跳转统计。Model ID 按你实际可用的模型填不同账号权限不同填错会报 404 或 model not found。如果你用的是 Cline 或 CC Switch 这类工具配置逻辑一样Base URL 填https://taotoken.net/apiKey 填你的 API KeyModel ID 填具体模型名。三件套缺一不可只填 Key 不填 Base URL 会走默认官方地址导致鉴权失败。环境准备好后建一个项目目录mkdir fastmcp-context-demo cd fastmcp-context-demo touch server.py到这里前置工作完成。下一节开始写真正的 Context 注入代码从最简单的CurrentContext()依赖开始逐步加上异步工具函数、状态管理、自定义依赖。3. 可复制的 Context 注入配置与异步工具函数这一节是核心直接给可复制的代码。先写一个最小可运行的 FastMCP 服务演示三种 Context 注入方式CurrentContext()依赖、传统类型提示注入、get_context()函数检索。然后加上异步工具函数、状态管理、自定义依赖。先看server.py的完整骨架from fastmcp import FastMCP from fastmcp.dependencies import CurrentContext, Depends from fastmcp.server.context import Context from fastmcp.server.dependencies import get_context, get_http_headers from contextlib import asynccontextmanager mcp FastMCP(nameContext Demo) # 方式一CurrentContext() 依赖注入推荐 mcp.tool async def process_file(file_uri: str, ctx: Context CurrentContext()) - str: 用 Context 打日志并读取资源。 await ctx.info(fProcessing {file_uri}) await ctx.report_progress(progress50, total100) return fProcessed {file_uri} # 方式二传统类型提示注入向后兼容 mcp.tool async def legacy_tool(file_uri: str, ctx: Context) - str: 参数名无关紧要只有 Context 类型提示才重要。 await ctx.debug(fLegacy processing {file_uri}) return done # 方式三get_context() 在嵌套函数深处检索 async def process_data(data: list[float]) - dict: ctx get_context() await ctx.info(fProcessing {len(data)} data points) return {count: len(data), sum: sum(data)} mcp.tool async def analyze_dataset(dataset_name: str) - dict: data [1.0, 2.0, 3.0] return await process_data(data)三种方式的关键区别CurrentContext()是首选依赖参数会自动从 MCP 架构中排除客户端永远看不到ctx这个参数传统类型提示注入靠Context类型提示识别参数名随便写get_context()用于嵌套调用深处但只能在请求上下文中调用请求外调用抛RuntimeError。接下来加状态管理和自定义依赖。状态管理用ctx.set_state/ctx.get_state作用域限定在单个请求内mcp.tool async def secure_operation(data: str, ctx: Context CurrentContext()) - str: user_id ctx.get_state(user_id) permissions ctx.get_state(permissions) or [] if write not in permissions: return Access denied return fProcessing {data} for user {user_id}自定义依赖用Depends()支持同步函数、异步函数、异步上下文管理器def get_config() - dict: return {api_url: https://taotoken.net/api, timeout: 30} asynccontextmanager async def get_database(): db {connected: True} try: yield db finally: db[connected] False mcp.tool async def fetch_data( query: str, config: dict Depends(get_config), dbDepends(get_database), ) - str: return fquery{query} url{config[api_url]} db{db[connected]}Depends()的依赖同样会自动从 MCP 架构中排除。上下文管理器的清理代码会在函数完成后运行即使发生错误也会执行适合数据库连接、文件句柄这类需要释放的资源。再补一个 HTTP header 读取的例子用get_http_headers()避免请求上下文缺失时报错mcp.tool async def safe_header_info() - dict: headers get_http_headers() auth headers.get(authorization, ) return { user_agent: headers.get(user-agent, Unknown), has_auth: bool(auth), auth_type: Bearer if auth.startswith(Bearer ) else None, }get_http_headers()默认排除host和content-length这类有问题的标头需要全部标头时传include_allTrue。最后是启动入口if __name__ __main__: mcp.run(transporthttp, host127.0.0.1, port8000)用 HTTP transport 启动方便后面用 curl 或客户端验证。到这里配置部分完成下一节实际跑起来验证。4. 本地启动与调用验证确认 Context 真的注入了代码写完先启动服务python server.py看到类似Uvicorn running on http://127.0.0.1:8000的输出就说明起来了。如果报ModuleNotFoundError: No module named fastmcp检查虚拟环境是否激活如果报端口占用换port8001。验证第一步列出工具列表确认ctx参数没有暴露给客户端curl -s http://127.0.0.1:8000/mcp \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:1,method:tools/list,params:{}}返回的 JSON 里process_file的inputSchema应该只有file_uri一个参数没有ctx。这就是依赖注入自动排除的效果。如果看到ctx出现在 schema 里说明你用的是普通参数而不是CurrentContext()或Context类型提示。验证第二步调用工具观察日志和进度curl -s http://127.0.0.1:8000/mcp \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:2,method:tools/call,params:{name:process_file,arguments:{file_uri:resource://demo}}}服务端控制台应该打印出Processing resource://demo的 info 日志客户端会收到进度通知。如果日志没出现检查ctx.info是否被 await——Context 方法是异步的漏掉 await 不会报错但也不会执行。验证第三步测试get_context()在嵌套函数里的行为。调用analyze_datasetcurl -s http://127.0.0.1:8000/mcp \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:3,method:tools/call,params:{name:analyze_dataset,arguments:{dataset_name:demo}}}应该返回{count: 3, sum: 6.0}同时服务端打印Processing 3 data points。这说明get_context()在process_data这个嵌套函数里成功拿到了当前请求的上下文。验证第四步测试状态管理。先写一个中间件往上下文里塞状态再调用secure_operation。中间件代码from fastmcp.server.middleware import Middleware, MiddlewareContext class UserAuthMiddleware(Middleware): async def on_call_tool(self, context: MiddlewareContext, call_next): context.fastmcp_context.set_state(user_id, user_123) context.fastmcp_context.set_state(permissions, [read, write]) return await call_next(context) mcp.add_middleware(UserAuthMiddleware())重启服务后调用secure_operation应该返回Processing hello for user user_123。如果把permissions改成[read]返回Access denied。这验证了中间件与工具函数之间通过 Context 状态共享数据。验证第五步用 Claude Code 连接。在 Claude Code 里配置 MCP 服务器指向http://127.0.0.1:8000/mcp然后让它调用process_file。如果 Claude Code 报连接失败检查 Base URL 是否配成了https://taotoken.net/api以及 Key 是否有效。三件套Base URL、Key、Model ID任何一个错都会导致调用链断掉。到这里Context 的注入、异步调用、状态传递、自定义依赖都验证过了。下一节集中处理常见报错。5. 常见报错排查401、RuntimeError 与 OAuth 问题跑 FastMCP Context 相关代码时报错集中在几类。逐个对照排查。报错一RuntimeError: No active context found这是get_context()在请求外被调用时抛的。典型场景是在模块顶层或后台任务里调get_context()。Context 仅在请求期间可用请求外调用必然失败。解决办法把get_context()的调用挪进工具函数内部或者改用CurrentContext()依赖注入。如果你确实需要在请求外访问某些数据用外部存储数据库、文件、内存缓存别依赖 Context 状态。报错二401 Unauthorized或local proxy failed这类错误通常出现在用 Claude Code 或 Cline 连接 MCP 服务时。先检查三件套Base URL 是不是https://taotoken.net/apiKey 是不是从https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentcontext_diutm_campaignrewrite拿的Model ID 是不是当前账号可用的。local proxy failed多半是 Base URL 写错或网络不通用curl https://taotoken.net/api/models测一下连通性。如果返回 401说明 Key 无效或过期重新生成一个。报错三Error reading choices或返回体解析失败这个报错说明请求发出去了但响应格式不对。常见原因是 Model ID 填错或者 Base URL 多加了路径。Base URL 只填到https://taotoken.net/api不要加/v1或/chat/completions具体路径由客户端拼接。另外检查请求头Content-Type: application/json是否带上。报错四OAuth 相关错误如果你在 MCP 服务里启用了认证用get_access_token()拿 token 时可能遇到 OAuth 报错。检查 token 是否过期token.expires_at是否小于当前时间。多租户场景下token.claims.get(tenant_id)可能为 None需要加判空。如果客户端没传 tokenget_access_token()返回 None别直接.claims先判 None。报错五TypeError: object NoneType cant be used in await expression这是漏了 await。Context 的方法是异步的ctx.info()、ctx.report_progress()、ctx.set_state()都要 await。检查你的工具函数是不是async def如果不是改成异步。FastMCP 的 Context 方法全部是协程同步函数里调用会出问题。报错六状态在下一个请求读不到这不是报错是设计如此。Context 状态作用域限定在单个请求内请求 A 里set_state的数据请求 B 里get_state返回 None。需要跨请求持久化用数据库或文件。别试图用全局变量绕过多请求并发时会串数据。报错七Depends依赖没被注入检查Depends()的导入路径。CurrentContext和Depends从fastmcp.dependencies导入Context从fastmcp.server.context导入get_context从fastmcp.server.dependencies导入。路径写错会报ImportError。另外确认依赖函数的返回值类型和参数类型提示匹配类型不匹配时注入可能失败。排查完这些基本能覆盖 90% 的 Context 使用问题。剩下 10% 多半是版本差异用pip install -U fastmcp升级到最新版再试。6. 把 Context 用进真实工具链从调试到生产Context 的价值不在单机 demo而在真实工具链里。举几个我实际用过的场景。长任务进度上报。一个处理 1000 条数据的工具每处理 100 条调一次ctx.report_progress(progressi, total1000)客户端就能显示进度条。用户不会以为程序卡死体验好很多。注意report_progress也是异步的别漏 await。多租户数据隔离。用get_access_token()拿 token从token.claims里提取tenant_id在工具函数里按租户过滤数据。这样一套 MCP 服务可以服务多个租户不用为每个租户起一个实例。token 里的sub字段是标准 JWT 主体通常放用户 ID审计日志里记这个。中间件注入鉴权信息。写一个UserAuthMiddleware在on_call_tool里解析 token、查权限、把结果set_state到上下文。工具函数只管get_state读权限不用重复解析 token。中间件和工具函数通过 Context 状态解耦职责清晰。自定义依赖管理数据库连接。用asynccontextmanager包一个get_database()yield前建连接finally里关连接。工具函数用Depends(get_database)注入不用手写 try/finally。连接池、事务回滚这些逻辑都收在依赖里工具函数保持干净。LLM 采样。ctx.sample(分析这段数据, temperature0.7)让客户端的大模型帮你生成文本。适合在工具里做智能摘要、分类、抽取。采样请求走客户端不消耗你服务端的模型额度但需要客户端支持 sampling 能力。用户启发。ctx.elicit(请输入名称, response_typestr)在工具执行中途向用户要输入支持交互式工作流。这是 MCP 2025 年 6 月规范的新功能客户端支持度参差不齐用之前确认你的客户端实现了 elicit。资源与提示词访问。ctx.list_resources()、ctx.read_resource(uri)、ctx.list_prompts()、ctx.get_prompt(name, args)让工具函数能编程式地发现和使用服务端注册的资源和提示词。适合做元编程比如一个“分析所有可用资源”的工具。最后提醒几个生产环境的坑。Context 状态别存大对象它跟着请求走存太多影响内存。get_context()别在后台任务里用请求结束上下文就没了。自定义依赖的清理逻辑要幂等异常路径下也会执行。日志级别用对debug别在生产开info适量warning和error留给真正需要关注的。把上面这些用起来你的 FastMCP 服务就从“能跑”变成“好用”。Context 是 FastMCP 里最值得花时间吃透的部分依赖注入和异步上下文这两块搞明白后面写复杂工具会顺很多。
返回列表