工具执行不仅仅依赖于输入参数和资源,还需要依赖于一些上下文信息,工具执行过程中所需的所有上下文信息都可以利用以参数注入的fastmcp.server.Context来提供。不仅如此,这个Context还利用定义其中的方法实现了资源和提示词的消费、会话状态的维护以及资源可见性控制等操作。从服务端向客户端的反向请求和通知也是通过Context来完成的,对于FastMCP服务端编程来说,我个人认为这是最为重要的一个类型。
1. 上下文信息的提供
Context承载的针对当前执行上下的信息体现在如下所示的特性成员中。
@dataclassclassContext:@propertydefis_background_task(self)->bool@propertydeftask_id(self)->str|None@propertydeforigin_request_id(self)->str|None@propertydeffastmcp(self)->FastMCP@propertydefrequest_context(self)->RequestContext[ServerSession,Any,Request]|None@propertydeflifespan_context(self)->dict[str,Any]@propertydeftransport(self)->TransportType|None@propertydefclient_id(self)->str|None@propertydefrequest_id(self)->str@propertydefsession_id(self)->str@propertydefsession(self)->ServerSessionContext提供了如上的特性,具体说明如下:
- is_background_task:当前是否以后台任务的形式在执行;
- task_id:如果
is_background_task返回True,返回当前任务的ID; - origin_request_id:如果
is_background_task返回True,返回触发当前后台任务的原始请求的ID,反之返回的就是当前请求的ID; - fastmcp:代表MCP服务器的
FastMCP对象; - request_context:当前请求上下文;
- lifespan_context:这是一个跨请求共享的字典。它的数据源头是我们定义的
lifespan函数; - transport:描述传输的
TransportType对象; - client_id: 客户端ID,侧重于客户端身份标识;
- request_id: 对应 JSON-RPC协议中的id。服务器必须拿着这个ID回复结果,客户端才能匹配上是哪个工具的返回;
- session/session_id: 代表当前的会话极其标识。Session是服务器与特定客户端之间的“长连接”句柄;
Context的lifespan_context特性的本质是资源直通车。它将服务器启动时初始化的长期资源(如数据库连接、AI模型、配置对象,它们的生命周期于服务器绑定)安全地传递给每一个短期工具请求。它的源头是创建FastMCP时传入的lifespan函数。针对工具的每次调用,FastMCP都会创建一个新的Context对象,但是这个实例的lifespan_context特性是一个只读引用,下面的演示程序体现了这一点。
fromtypingimportSelffromfastmcpimportFastMCPfromfastmcp.server.lifespanimportlifespanfromfastmcp.serverimportContextfromfastmcp.clientimportClientimportasyncio log=[]classFakeDbConnection:asyncdefopen(self)->Self:log.append("DbConnection opened.")returnselfasyncdefclose(self):log.append("DbConnection closed.")lifespan_context={}@lifespanasyncdefapp_lifespan(server):db=awaitFakeDbConnection().open()lifespan_context["db"]=dbtry:yieldlifespan_contextfinally:awaitdb.close()server=FastMCP("Server",lifespan=app_lifespan)@server.tool()asyncdeftry_get_db(context:Context)->bool:assertcontext.lifespan_contextislifespan_context db=context.lifespan_context.get("db")return(dbisnotNone)andisinstance(db,FakeDbConnection)asyncdefmain():asyncwithClient(server)asclient:assertlog==["DbConnection opened."]result=awaitclient.call_tool("try_get_db")assertbool(result.content[0].text)==True# type: ignoreassertlog==["DbConnection opened.","DbConnection closed."]asyncio.run(main())2. 资源和提示词的消费
Context通常以参数注入的方式在工具函数中使用,对于工具来说,使用资源和提示词是常规的需求,所以Context定义了如下四个对应的方法分别完成资源和提示词列表的获取、提示词的渲染和资源的读取。
@dataclassclassContext:asyncdeflist_resources(self)->list[SDKResource]asyncdeflist_prompts(self)->list[SDKPrompt]asyncdefget_prompt(self,name:str,arguments:dict[str,Any]|None=None)->GetPromptResultasyncdefread_resource(self,uri:str|AnyUrl)->ResourceResult3. 会话状态的维护
Session的一个核心的作用状态保持,为同一Session的多轮问答创建一个上下文,即上一轮问答设置的状态可以被下一轮问答读取到。会话状态的设置、读取和删除可以利用Context如下三个方法(set_state、get_state和delete_state)来完成。
@dataclassclassContext:asyncdefset_state(self,key:str,value:Any,*,serializable:bool=True)->None:asyncdefget_state(self,key:str)->Anyasyncdefdelete_state(self,key:str)->None如下的程序演示了统一会话中针对状态的读写和删除。
fromfastmcpimportFastMCPfromfastmcp.serverimportContextfromfastmcp.clientimportClientimportasyncio server=FastMCP("Server")@server.tool()asyncdefset_state(key:str,value:str,context:Context)->None:awaitcontext.set_state(key,value)@server.tool()asyncdefget_state(key:str,context:Context)->str|None:returnawaitcontext.get_state(key)@server.tool()asyncdefdelete_state(key:str,context:Context)->None:awaitcontext.delete_state(key)asyncdefmain():asyncwithClient(server)asclient:awaitclient.call_tool("set_state",arguments={"key":"color_preference","value":"red"})result=awaitclient.call_tool("get_state",arguments={"key":"color_preference"})assertresult.content[0].text=="red"# type: ignoreawaitclient.call_tool("delete_state",arguments={"key":"color_preference"})result=awaitclient.call_tool("get_state",arguments={"key":"color_preference"})assertlen(result.content)==0asyncio.run(main())4. 资源可见性控制
Context如下所示的enable_components和disable_components方法可以在当前会话范围内控制组件的可见性,reset_visibility方法用于撤销针对组件可见性的设置。
@dataclassclassContext:asyncdefenable_components(self,*,names:set[str]|None=None,keys:set[str]|None=None,version:VersionSpec|None=None,tags:set[str]|None=None,components:set[Literal["tool","resource","template","prompt"]]|None=None,match_all:bool=False,)->Noneasyncdefdisable_components(self,*,names:set[str]|None=None,keys:set[str]|None=None,version:VersionSpec|None=None,tags:set[str]|None=None,components:set[Literal["tool","resource","template","prompt"]]|None=None,match_all:bool=False,)->Noneasyncdefreset_visibility(self)->None如下的程序演示了利用通过调用工具set_tools_visibility对本Session范围内工具组件的可见性设置。
fromfastmcpimportFastMCPfromfastmcp.serverimportContextfromfastmcp.clientimportClientimportasyncio server=FastMCP("Server")@server.tool()asyncdefset_tools_visibility(context:Context,settings:dict[str,bool]|None=None,reset:bool=False)->None:ifreset:awaitcontext.reset_visibility()returnsettings=settingsor{}visible_tools={tool_namefortool_name,is_visibleinsettings.items()ifis_visible}iflen(visible_tools)>0:awaitcontext.enable_components(names=visible_tools,components={"tool"})tools_to_invisible=set(settings.keys())-visible_toolsiflen(tools_to_invisible)>0:awaitcontext.disable_components(names=tools_to_invisible,components={"tool"})@server.tool()asyncdeffoo()->None:pass@server.tool()asyncdefbar()->None:pass@server.tool()asyncdefbaz()->None:passasyncdefmain():asyncwithClient(server)asclient:awaitclient.call_tool("set_tools_visibility",arguments={"settings":{"foo":False,"baz":False}})tools=awaitclient.list_tools()tool_names=set(tool.namefortoolintools)asserttool_names=={"bar","set_tools_visibility"}awaitclient.call_tool("set_tools_visibility",arguments={"settings":{"foo":True}})tools=awaitclient.list_tools()tool_names=set(tool.namefortoolintools)asserttool_names=={"foo","bar","set_tools_visibility"}awaitclient.call_tool("set_tools_visibility",arguments={"reset":True})tools=awaitclient.list_tools()tool_names=set(tool.namefortoolintools)asserttool_names=={"foo","bar","baz","set_tools_visibility"}asyncio.run(main())