FastMCP高级特性之Context-平芜编程栈

一、什么是Context?

Context 对象提供了一个简洁的接口，用于在函数中访问 MCP 功能，包括：

日志记录：向客户端发送调试、信息、警告和错误消息
进度报告：向客户端更新长时间运行操作的进度
资源访问：列出并读取服务器已注册资源中的数据
提示词访问：列出并检索服务器已注册的提示词
大语言模型采样：请求客户端的大语言模型根据提供的消息生成文本
用户启发：在工具执行期间请求用户提供结构化输入
状态管理：在单个请求内存储和共享中间件与处理器之间的数据
请求信息：访问当前请求的元数据
服务器访问：必要时，访问底层的 FastMCP 服务器实例

二、访问上下文

访问上下文的首选方式是使用 CurrentContext () 依赖项：

fromfastmcpimportFastMCPfromfastmcp.dependenciesimportCurrentContextfromfastmcp.server.contextimportContext mcp=FastMCP(name="Context Demo")@mcp.toolasyncdefprocess_file(file_uri:str,ctx:Context=CurrentContext())->str:"""Processes a file, using context for logging and resource access."""awaitctx.info(f"Processing{file_uri}")return"Processed file"

这适用于工具、资源和提示词：

fromfastmcpimportFastMCPfromfastmcp.dependenciesimportCurrentContextfromfastmcp.server.contextimportContext mcp=FastMCP(name="Context Demo")@mcp.resource("resource://user-data")asyncdefget_user_data(ctx:Context=CurrentContext())->dict:awaitctx.debug("Fetching user data")return{"user_id":"example"}@mcp.promptasyncdefdata_analysis_request(dataset:str,ctx:Context=CurrentContext())->str:returnf"Please analyze the following dataset:{dataset}"

核心点

依赖参数会自动从 MCP 架构中排除 —— 客户端永远不会看到它们。
上下文方法是异步的，因此你的函数通常也需要是异步的。
每个 MCP 请求都会收到一个新的上下文对象。上下文的作用域限定为单个请求；在一个请求中设置的状态或数据在后续请求中将不可用。
上下文仅在请求期间可用；尝试在请求之外使用上下文方法将会引发错误。

2.1 传统类型提示注入

为了向后兼容，您仍然可以通过简单地添加一个带有Context类型提示的参数来访问上下文。FastMCP 会自动注入上下文实例：

fromfastmcpimportFastMCP,Context mcp=FastMCP(name="Context Demo")@mcp.toolasyncdefprocess_file(file_uri:str,ctx:Context)->str:"""Processes a file, using context for logging and resource access."""# Context is injected automatically based on the type hintreturn"Processed file"

这种方法对于工具、资源和提示词仍然适用。参数名称无关紧要 —— 只有 Context 类型提示才是重要的。该类型提示也可以是一个联合类型（Context | None）或使用 Annotated []。

2.2 通过 get_context () 函数

对于嵌套在函数调用深处、通过参数传递上下文不太方便的代码，可以使用 get_context () 从请求执行流程中的任何位置检索活动上下文：

fromfastmcpimportFastMCPfromfastmcp.server.dependenciesimportget_context mcp=FastMCP(name="Dependency Demo")# Utility function that needs context but doesn't receive it as a parameterasyncdefprocess_data(data:list[float])->dict:# Get the active context - only works when called within a requestctx=get_context()awaitctx.info(f"Processing{len(data)}data points")@mcp.toolasyncdefanalyze_dataset(dataset_name:str)->dict:# Call utility function that uses context internallydata=load_data(dataset_name)awaitprocess_data(data)

重要说明：

get_context () 函数仅应在服务器请求的上下文中使用。在请求之外调用它会引发 RuntimeError。
get_context () 函数是仅服务器可用的，不应在客户端代码中使用。

三、上下文功能

FastMCP 通过上下文对象提供多种高级功能。每种功能都有专门的文档，其中包含全面的示例和最佳实践：

3.1 日志记录

发送调试、信息、警告和错误消息回 MCP 客户端，以便了解函数执行情况。

awaitctx.debug("Starting analysis")awaitctx.info(f"Processing{len(data)}items")awaitctx.warning("Deprecated parameter used")awaitctx.error("Processing failed")

3.2 客户需求获取

在工具执行过程中，向客户请求结构化输入，以支持交互式工作流和渐进式信息披露。这是 2025 年 6 月 18 日 MCP 规范中的一项新功能。

result=awaitctx.elicit("Enter your name:",response_type=str)ifresult.action=="accept":name=result.data

3.3 大语言模型采样

请求客户的大语言模型（LLM）根据提供的消息生成文本，这对于在你的工具中利用人工智能能力很有用。

response=awaitctx.sample("Analyze this data",temperature=0.7)

3.4 进度报告

向客户端更新长时间运行操作的进度，以支持进度指示器并改善用户体验。

awaitctx.report_progress(progress=50,total=100)# 50% complete

3.5 资源访问

列出并读取在您的 FastMCP 服务器上注册的资源中的数据，从而能够访问文件、配置或动态内容。

# List available resourcesresources=awaitctx.list_resources()# Read a specific resourcecontent_list=awaitctx.read_resource("resource://config")content=content_list[0].content

ctx.list_resources () -> list [MCPResource]：
返回所有可用资源的列表
ctx.read_resource (uri: str | AnyUrl) -> list [ReadResourceContents]：
返回资源内容部分的列表

3.6 提示词访问

列出并检索在您的 FastMCP 服务器上注册的提示词，使工具和中间件能够通过编程方式发现并使用可用的提示词。

# List available promptsprompts=awaitctx.list_prompts()# Get a specific prompt with argumentsresult=awaitctx.get_prompt("analyze_data",{"dataset":"users"})messages=result.messages

ctx.list_prompts() -> list[MCPPrompt]
- 返回所有可用提示的列表
ctx.get_prompt(name: str, arguments: dict[str, Any] | None = None) -> GetPromptResult
- 获取带有可选参数的特定提示词

3.7 状态管理

要在上下文状态中存储值，请使用 ctx.set_state (key, value)。要检索值，请使用 ctx.get_state (key)。

上下文状态的作用域限定在单个 MCP 请求内。每个操作（工具调用、资源读取、列表操作等）都会收到一个新的上下文对象。在一次请求期间设置的状态在后续请求中无法使用。如果要在多个请求之间进行持久化数据存储，请使用外部存储机制，如数据库、文件或内存缓存。

fromfastmcp.server.middlewareimportMiddleware,MiddlewareContextclassUserAuthMiddleware(Middleware):asyncdefon_call_tool(self,context:MiddlewareContext,call_next):# Middleware stores user info in context statecontext.fastmcp_context.set_state("user_id","user_123")context.fastmcp_context.set_state("permissions",["read","write"])returnawaitcall_next(context)@mcp.toolasyncdefsecure_operation(data:str,ctx:Context)->str:"""Tool can access state set by middleware."""user_id=ctx.get_state("user_id")# "user_123"permissions=ctx.get_state("permissions")# ["read", "write"]if"write"notinpermissions:return"Access denied"returnf"Processing{data}for user{user_id}"

ctx.set_state(key: str, value: Any) -> None在上下文状态中存储一个值
ctx.get_state(key: str) -> Any从上下文状态中获取一个值（如果未找到则返回 None）

3.8 更改通知

当组件（如工具、资源或提示词）被添加、移除、启用或禁用时，FastMCP 会自动发送列表变更通知。在极少数情况下，若你需要手动触发这些通知，可以使用上下文方法：

@mcp.toolasyncdefcustom_tool_management(ctx:Context)->str:"""Example of manual notification after custom tool changes."""# After making custom changes to toolsawaitctx.send_tool_list_changed()awaitctx.send_resource_list_changed()awaitctx.send_prompt_list_changed()return"Notifications sent"

这些方法主要由 FastMCP 的自动通知系统在内部使用，大多数用户无需直接调用它们。

3.9 FastMCP服务器

要访问底层的 FastMCP 服务器实例，您可以使用 ctx.fastmcp 属性：

@mcp.toolasyncdefmy_tool(ctx:Context)->None:# Access the FastMCP server instanceserver_name=ctx.fastmcp.name...

3.10 MCP Request

访问关于当前请求和客户端的元数据。

@mcp.toolasyncdefrequest_info(ctx:Context)->dict:"""Return information about the current request."""return{"request_id":ctx.request_id,"client_id":ctx.client_idor"Unknown client"}

ctx.request_id -> str：获取当前 MCP 请求的唯一 ID
ctx.client_id -> str | None：获取发出请求的客户端的 ID（如果在初始化期间提供）
ctx.session_id -> str | None：获取用于基于会话的数据共享的 MCP 会话 ID（仅适用于 HTTP 传输）

四、运行时依赖项

4.1 HTTP Requests

访问当前 HTTP 请求的推荐方式是通过 get_http_request () 依赖函数：

fromfastmcpimportFastMCPfromfastmcp.server.dependenciesimportget_http_requestfromstarlette.requestsimportRequest mcp=FastMCP(name="HTTP Request Demo")@mcp.toolasyncdefuser_agent_info()->dict:"""Return information about the user agent."""# Get the HTTP requestrequest:Request=get_http_request()# Access request datauser_agent=request.headers.get("user-agent","Unknown")client_ip=request.client.hostifrequest.clientelse"Unknown"return{"user_agent":user_agent,"client_ip":client_ip,"path":request.url.path,}

4.2 HTTP Headers

如果你只需要请求头并且想避免潜在的错误，可以使用 get_http_headers () 辅助函数：

fromfastmcpimportFastMCPfromfastmcp.server.dependenciesimportget_http_headers mcp=FastMCP(name="Headers Demo")@mcp.toolasyncdefsafe_header_info()->dict:"""Safely get header information without raising errors."""# Get headers (returns empty dict if no request context)headers=get_http_headers()# Get authorization headerauth_header=headers.get("authorization","")is_bearer=auth_header.startswith("Bearer ")return{"user_agent":headers.get("user-agent","Unknown"),"content_type":headers.get("content-type","Unknown"),"has_auth":bool(auth_header),"auth_type":"Bearer"ifis_bearerelse"Other"ifauth_headerelse"None","headers_count":len(headers)}

默认情况下，get_http_headers () 会排除 host（主机）和 content-length（内容长度）等有问题的标头。若要包含所有标头，请使用 get_http_headers (include_all=True)。

4.3 Access Tokens

在你的 FastMCP 服务器上使用身份验证时，你可以通过 get_access_token () 依赖函数访问已验证用户的访问令牌信息：

fromfastmcpimportFastMCPfromfastmcp.server.dependenciesimportget_access_token,AccessToken mcp=FastMCP(name="Auth Token Demo")@mcp.toolasyncdefget_user_info()->dict:"""Get information about the authenticated user."""# Get the access token (None if not authenticated)token:AccessToken|None=get_access_token()iftokenisNone:return{"authenticated":False}return{"authenticated":True,"client_id":token.client_id,"scopes":token.scopes,"expires_at":token.expires_at,"token_claims":token.claims,# JWT claims or custom token data}

当你需要以下操作时，这会特别有用：

访问用户标识 —— 从令牌声明中获取 client_id 或主体
检查权限 —— 在执行操作前验证范围或自定义声明
多租户应用程序 —— 从令牌声明中提取租户信息
审计日志记录 —— 跟踪哪个用户执行了哪些操作

claims 字段包含来自原始令牌的所有数据（JWT 令牌的 JWT 声明，或其他令牌类型的自定义数据）：

fromfastmcpimportFastMCPfromfastmcp.server.dependenciesimportget_access_token mcp=FastMCP(name="Multi-tenant Demo")@mcp.toolasyncdefget_tenant_data(resource_id:str)->dict:"""Get tenant-specific data using token claims."""token:AccessToken|None=get_access_token()# Extract tenant ID from token claimstenant_id=token.claims.get("tenant_id")iftokenelseNone# Extract user ID from standard JWT subject claimuser_id=token.claims.get("sub")iftokenelseNone# Use tenant and user info to authorize and filter dataifnottenant_id:raiseValueError("No tenant information in token")return{"resource_id":resource_id,"tenant_id":tenant_id,"user_id":user_id,"data":f"Tenant-specific data for{tenant_id}",}

五、自定义依赖项

FastMCP 的依赖注入由 Docket 提供支持，它为向函数中注入值提供了一个灵活的系统。除了像 CurrentContext () 这样的内置依赖项之外，你还可以创建自己的依赖项。

5.1 使用 Depends ()

创建自定义依赖项最简单的方法是使用 Depends ()。传入任何可调用对象（同步或异步函数，或异步上下文管理器），其返回值将被注入：

fromcontextlibimportasynccontextmanagerfromfastmcpimportFastMCPfromfastmcp.dependenciesimportDepends mcp=FastMCP(name="Custom Deps Demo")# Simple function dependencydefget_config()->dict:return{"api_url":"https://api.example.com","timeout":30}# Async function dependencyasyncdefget_user_id()->int:return42@mcp.toolasyncdeffetch_data(query:str,config:dict=Depends(get_config),user_id:int=Depends(get_user_id),)->str:returnf"User{user_id}fetching '{query}' from{config['api_url']}"

使用 Depends () 的依赖项会自动从 MCP 模式中排除 —— 客户端永远不会将它们视为参数。

5.2 使用上下文管理器进行资源管理

对于需要清理的依赖项（数据库连接、文件句柄等），请使用异步上下文管理器：

fromcontextlibimportasynccontextmanagerfromfastmcpimportFastMCPfromfastmcp.dependenciesimportDepends mcp=FastMCP(name="Resource Demo")@asynccontextmanagerasyncdefget_database():db=awaitconnect_to_database()try:yielddbfinally:awaitdb.close()@mcp.toolasyncdefquery_users(sql:str,db=Depends(get_database))->list:returnawaitdb.execute(sql)

上下文管理器的清理代码会在函数完成后运行，即使发生错误也是如此。

5.3 嵌套依赖项

依赖项可能依赖于其他依赖项：

fromfastmcpimportFastMCPfromfastmcp.dependenciesimportDepends mcp=FastMCP(name="Nested Demo")defget_base_url()->str:return"https://api.example.com"defget_api_client(base_url:str=Depends(get_base_url))->dict:return{"base_url":base_url,"version":"v1"}@mcp.toolasyncdefcall_api(endpoint:str,client:dict=Depends(get_api_client))->str:returnf"Calling{client['base_url']}/{client['version']}/{endpoint}"