企业官网建设流程全解析

MCP规范中的原语（工具、静态资源、动态资源模板和提示词）在FastMCP中被统一称为组件，它们都有一个共同的基类FastMCPComponent，后者派生自FastMCPBaseModel。由于组件常被定义成函数形式，函数除了定义需从外部提供的参数（比如调用工具提供的输入参数）之外，我们还可以采用依赖注入的形式定义一些参数来提供上下文信息。

1. 注入Context

fastmcp.server.context.Context不仅仅用于提供当前执行上下文信息，还被用于向客户端发送通知和请求。绝大部分参数注入场景指的都是针对它的注入。在如下所示的演示程序中，我们在工具函数greet中注入的Context对象，并调用它的read_resource方法读取文本资源内容作为返回值。

fromfastmcpimportFastMCPfromfastmcp.server.contextimportContextfromfastmcp.clientimportClientimportasyncio mcp=FastMCP("Greeting")@mcp.resource("greeting://{name}")asyncdefgreet_to_name(name:str)->str:"""Get a greeting message for the given name"""returnf"Hi,{name}!"@mcp.toolasyncdefgreet(name:str,context:Context)->str:"""Get a greeting message for the given name"""result=awaitcontext.read_resource(f"greeting://{name}")iflen(result.contents)>0:returnstr(result.contents[0].content)returnf"No greeting found for{name}"asyncdefmain():asyncwithClient(mcp)asclient:result=awaitclient.call_tool(name="greet",arguments={"name":"MCP"})assertstr(result.content[0].text)=="Hi, MCP!"# type: ignoreasyncio.run(main())

有时候为了使注入参数更加明确一点，我们还会为注入参数指定默认值，fastmcp.dependencies模块定义了很多CurrentXxx函数来获取对应的值。比如我们在工具函数中注入的context就可以按照如下的方式使用CurrentContext函数的返回值作为其默认值。

fromfastmcpimportFastMCPfromfastmcp.server.contextimportContextfromfastmcp.dependenciesimportCurrentContext mcp=FastMCP("Greeting")@mcp.toolasyncdefgreet(name:str,context:Context=CurrentContext())->str:"""Get a greeting message for the given name"""result=awaitcontext.read_resource(f"greeting://{name}")iflen(result.contents)>0:returnstr(result.contents[0].content)returnf"No 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：当前任务执行的Worker；
• CurrentFastMCP：作为服务器的FastMCP对象；
• CurrentRequest：当前处理的HTTP请求；
• CurrentHeaders：当前HTTP请求的报头列表；
• CurrentAccessToken：认证提供的用以体现用户存取权限的AccessToken；
• TokenClaim：利用指定的名称（比如"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 DependencyFactory=Callable[...,R|Awaitable[R]|AbstractContextManager[R]|AbstractAsyncContextManager[R]]

由于我们无法为工厂函数提供外部参数，所以它们只能定义上面这些可以自动注入的参数。在下面的演示程序中，我们自定义了一个无参函数get_greet_template来提供问候语模板。工具函数greet定义了template参数，默认值为Depends函数针对get_greet_template函数的封装。

fromfastmcpimportFastMCPfromfastmcp.dependenciesimportDependsfromfastmcp.clientimportClientimportasyncio mcp=FastMCP("Greeting")defget_greet_template()->str:return"Hi, {name}!"@mcp.toolasyncdefgreet(name:str,template:str=Depends(get_greet_template))->str:"""Get a greeting message for the given name"""returntemplate.format(name=name)asyncdefmain():asyncwithClient(mcp)asclient:result=awaitclient.call_tool(name="greet",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.value=value _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:Context=CurrentContext()):return(x,y,conext.value)new_fn=convert(raw_fn)asyncdefmain():_current.set(Context(3))result=awaitnew_fn(1,2)assertresult==(1,2,3)_current.set(Context(4))result=awaitnew_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]]={}signature=inspect.signature(function)forname,parameterinsignature.parameters.items():ifisinstance(parameter.default,Dependency):dependencies[name]=parameter.default _parameter_cache[function]=dependenciesreturndependenciesdefconvert(fn:Callable[...,Any])->Callable[...,Any]:dependencies=get_dependency_parameters(fn)asyncdefwrapper(*args,**kwargs):forname,dependencyindependencies.items():kwargs[name]=awaitdependency.__aenter__()returnawaitfn(*args,**kwargs)returnwrapper