MCP规范中的原语工具、静态资源、动态资源模板和提示词在FastMCP中被统一称为组件它们都有一个共同的基类FastMCPComponent后者派生自FastMCPBaseModel。由于组件常被定义成函数形式函数除了定义需从外部提供的参数比如调用工具提供的输入参数之外我们还可以采用依赖注入的形式定义一些参数来提供上下文信息。1. 注入Contextfastmcp.server.context.Context不仅仅用于提供当前执行上下文信息还被用于向客户端发送通知和请求。绝大部分参数注入场景指的都是针对它的注入。在如下所示的演示程序中我们在工具函数greet中注入的Context对象并调用它的read_resource方法读取文本资源内容作为返回值。fromfastmcpimportFastMCPfromfastmcp.server.contextimportContextfromfastmcp.clientimportClientimportasyncio mcpFastMCP(Greeting)mcp.resource(greeting://{name})asyncdefgreet_to_name(name:str)-str:Get a greeting message for the given namereturnfHi,{name}!mcp.toolasyncdefgreet(name:str,context:Context)-str:Get a greeting message for the given nameresultawaitcontext.read_resource(fgreeting://{name})iflen(result.contents)0:returnstr(result.contents[0].content)returnfNo greeting found for{name}asyncdefmain():asyncwithClient(mcp)asclient:resultawaitclient.call_tool(namegreet,arguments{name:MCP})assertstr(result.content[0].text)Hi, MCP!# type: ignoreasyncio.run(main())有时候为了使注入参数更加明确一点我们还会为注入参数指定默认值fastmcp.dependencies模块定义了很多CurrentXxx函数来获取对应的值。比如我们在工具函数中注入的context就可以按照如下的方式使用CurrentContext函数的返回值作为其默认值。fromfastmcpimportFastMCPfromfastmcp.server.contextimportContextfromfastmcp.dependenciesimportCurrentContext mcpFastMCP(Greeting)mcp.toolasyncdefgreet(name:str,context:ContextCurrentContext())-str:Get a greeting message for the given nameresultawaitcontext.read_resource(fgreeting://{name})iflen(result.contents)0:returnstr(result.contents[0].content)returnfNo greeting found for{name}2. 其他可供注入的对象在fastmcp.dependencies模块下定义如下所示的一系列CurrentXxx函数意味着它们返回的对象都可以利用注入参数来提供。defCurrentContext()-ContextdefCurrentDocket()-DocketdefCurrentWorker()-WorkerdefCurrentFastMCP()-FastMCPdefCurrentRequest()-RequestdefCurrentHeaders()-dict[str,str]defCurrentAccessToken()-AccessTokendefTokenClaim(name:str)-str相关说明如下CurrentContext当前执行上下文CurrentDocket当前使用的作为后台任务调度器的Docket对象CurrentWorker当前任务执行的WorkerCurrentFastMCP作为服务器的FastMCP对象CurrentRequest当前处理的HTTP请求CurrentHeaders当前HTTP请求的报头列表CurrentAccessToken认证提供的用以体现用户存取权限的AccessTokenTokenClaim利用指定的名称比如oid, sub和 email等从AccessToken中提取对应的声明。3. 自定义注入函数我们也可以自定义提供注入参数的工厂函数并利用如下这些Depends方法进行封装后作为参数的默认值。从如下这些重载声明可以看出Depends不仅支持针对同步和异步函数的封装还支持针对AbstractContextManager的封装。overloaddefDepends(factory:Callable[...,AbstractAsyncContextManager[R]])-R:...overloaddefDepends(factory:Callable[...,AbstractContextManager[R]])-R:...overloaddefDepends(factory:Callable[...,Awaitable[R]])-R:...overloaddefDepends(factory:Callable[...,R])-R:...defDepends(factory:DependencyFactory[R])-R DependencyFactoryCallable[...,R|Awaitable[R]|AbstractContextManager[R]|AbstractAsyncContextManager[R]]由于我们无法为工厂函数提供外部参数所以它们只能定义上面这些可以自动注入的参数。在下面的演示程序中我们自定义了一个无参函数get_greet_template来提供问候语模板。工具函数greet定义了template参数默认值为Depends函数针对get_greet_template函数的封装。fromfastmcpimportFastMCPfromfastmcp.dependenciesimportDependsfromfastmcp.clientimportClientimportasyncio mcpFastMCP(Greeting)defget_greet_template()-str:returnHi, {name}!mcp.toolasyncdefgreet(name:str,template:strDepends(get_greet_template))-str:Get a greeting message for the given namereturntemplate.format(namename)asyncdefmain():asyncwithClient(mcp)asclient:resultawaitclient.call_tool(namegreet,arguments{name:MCP})assertstr(result.content[0].text)Hi, MCP!# type: ignoreasyncio.run(main())4. 实现原理我们知道在分布式环境中只能根据Schema来构建调用目标函数的参数列表但是注入参数不会包含在函数的Schema中。FastMCP的解决方案是将原始函数fn封装成一个与Schema一致的wrap函数那么就可以直接根据外部传入的参数来调用wrap函数后者在调用fn之前会将漏掉的注入参数补充完整就可以了。我们通过如下这个简单的模拟实例在对这种解决方案作进一步说明。假设下面定义的Context就代表可以支持参数注入的上下文对象我们为它定义了唯一的value字段。当前Context通过_current这个全局的上下文变量进行设置。代表原始组件函数的raw_fn包含三个参数其中最后参数context作为注入参数用于提供当前上下文。此函数返回由三个参数组成的元组。importasynciofromtypingimportcastfromuncalled_forimportDependencyfromcontextvarsimportContextVarfromparameter_injectionimportconvertclassContext:value:intdef__init__(self,value:int):self.valuevalue _current:ContextVar[Context]ContextVar(_current)class_CurrentContext(Dependency[Context]):asyncdef__aenter__(self)-Context:return_current.get()defCurrentContext()-Context:returncast(Context,_CurrentContext())asyncdefraw_fn(x:int,y:int,conext:ContextCurrentContext()):return(x,y,conext.value)new_fnconvert(raw_fn)asyncdefmain():_current.set(Context(3))resultawaitnew_fn(1,2)assertresult(1,2,3)_current.set(Context(4))resultawaitnew_fn(1,2)assertresult(1,2,4)asyncio.run(main())context默认值通过CurrentContext函数返回具体返回的是一个_CurrentContex对象。_CurrentContex继承自Dependency[Context]在实现的__aenter__方法中利用_current返回的上下文变量提供当前Context。我们调用convert函数将raw_fn中的注入参数剔除最终转换成一个只包含x和y参数的函数new_fn。为了验证new_fn能否替换raw_fn在对当前上下文做了不同设置之后我们只传入两个参数调用new_fn断言证明两个函数是等效的。具有如下定义的convert函数就是整个参数注入的核心。它会调用get_dependency_parameters函数将注入参数提取出来并构建一个体现其默认值和参数名映射关系的字典这个映射字典存储在基于函数的全局缓存_parameter_cache字典中以避免针对同一个函数的重复解析。convert返回以内嵌形式定义的wrapper函数它在执行的时候会提取每个注入参数的默认值。提供的默认值为_CurrentContext以awiat方式调用__aenter__便可以把真正需要的参数值提取出来。然后它把注入的参数和提供的参数进行合并便得到调用原始raw_fn函数的完整的参数列表。fromtypingimportCallable,Anyfromuncalled_forimportDependencyimportinspect _parameter_cache:dict[Callable[...,Any],dict[str,Dependency[Any]]]{}defget_dependency_parameters(function:Callable[...,Any],)-dict[str,Dependency[Any]]:iffunctionin_parameter_cache:return_parameter_cache[function]dependencies:dict[str,Dependency[Any]]{}signatureinspect.signature(function)forname,parameterinsignature.parameters.items():ifisinstance(parameter.default,Dependency):dependencies[name]parameter.default _parameter_cache[function]dependenciesreturndependenciesdefconvert(fn:Callable[...,Any])-Callable[...,Any]:dependenciesget_dependency_parameters(fn)asyncdefwrapper(*args,**kwargs):forname,dependencyindependencies.items():kwargs[name]awaitdependency.__aenter__()returnawaitfn(*args,**kwargs)returnwrapper