整个FastMCP服务器的架构设计可谓可圈可点这里我们主要围绕Provider、Middleware和Transform这三个核心对象来介绍FastMCP服务的架构设计。如果你希望能够灵活自由地使用FastMCP搭建你的MCP服务器了解这三个核心对象也是必需的Provider当客户端连接到您的服务器并询问您有哪些组件工具、资源、提示词和任务等时FastMCP会向每个Provider询问该问题并将结果汇​​总。当客户端调用某个工具读取指定资源时FastMCP会查找哪个Provider拥有该工具或者资源并将调用委托给该它。我们可以将Provider视为组件的来源一个FastMCP服务器可以注册多个ProviderMiddleware注册的多个中间件构造一个链条FastMCP服务器的请求和响应会流经这个链条使得每个中间件可以注入一些横切操作比如认证、授权、日志和审核等Transform与Middleware类似注册的多个Transform也会构造一个链条它们会在组件从Provider流向客户端的过程中对其进行修改。为了让大家了解Provider、Middleware和Transform这三个核心对象在请求处理流程中所扮演的角色以及它们定义的方法会在处理流程的哪个环节被执行我们构建一个完全由这三个对象构建的MCP服务器。1. 自定义ResourceProvider我们先来定义如下这个旨在提供静态资源的ResourceProvider。我们在构造函数中利用指定的name参数来对ResourceProvider命名另一个resources参数直接指定一组Resource列表。我们重写了与资源相关的_list_resources和_get_resource方法并在方法中向log列表写入相应的日志以便跟踪这两个方法在整个请求处理流程中的执行时机。fromtypingimportAny,override,SequencefromfastmcpimportFastMCPfrompydantic.networksimportAnyUrlfromfastmcp.resourcesimportResourcefromfastmcp.utilities.versionsimportVersionSpecfromfastmcp.server.providersimportProviderfromfastmcp.server.transformsimportTransform,GetResourceNextfromfastmcp.server.middlewareimportMiddleware,MiddlewareContext,CallNextfromfastmcp.clientimportClient log[]classResourceProvider(Provider):def__init__(self,name:str,resources:list[Resource]):super().__init__()self.namename self._resources{str(resource.uri):resourceforresourceinresources}asyncdef_list_resources(self)-list[Resource]:log.append(f{self.name}- list_resources)returnlist(self._resources.values())asyncdef_get_resource(self,uri:str,version:VersionSpec|NoneNone)-Resource|None:log.append(f{self.name}- get_resource)returnself._resources.get(uri)2. 自定义ResourceMiddleware如下这个继承自Middleware的ResourceMiddleware类型旨在拦截资源相关的请求并注入相应的处理操作。log[]classResourceMiddleware(Middleware):def__init__(self,name:str):super().__init__()self.namenameasyncdef__call__(self,context:MiddlewareContext,call_next:CallNext,)-Any:ifnotcontext.methodin[resources/list,resources/read]:returnawaitcall_next(context)log.append(f{self.name}- pre-handler)resultawaitcall_next(context)log.append(f{self.name}- post-handler)returnresult作为基类的Middleware将针对请求的拦截实现在它的__call__方法中这方法具有如下两个参数context描述当前请求的上下文ResourceMiddleware利用它得到MCP请求采用的JSON-RPC方法call_next是一个处理器中间件通过调用它让后续处理流程接受处理当前请求并对得到结果作进一步处理。我们利用ResourceMiddleware函数指定的name参数为中间件提供一个名称以便于更好地跟踪它。我们只关注与资源相关的操作resources/list和resources/read,在重写的__call__方法中我们在调用CallNext方法前后在输出一段文字揭示前置和后置操作执行的时机。3. 自定义ResourceTransform如下所示的ResourceTransform旨在针对资源的请求作相应的转换。我们同样利用构造函数传入的参数为它命名基类Transform的list_resources和get_resource方法的定义有所不同后者的定义与中间件类似都是利用提供的call_next这里是一个GetResourceNext对象将请求交付给后续的处理流程所以我们可以在这前后执行相应的前置和后置操作。list_resources方法直接以参数的方式提供Resource序列我们只能对它作相应的处理比如过滤和转换等并返回一个资源序列交付给后续处理流程相当于这里只有后置操作。我们均在重写的这两个方法中输出了相应的日志。classResourceTransform(Transform):def__init__(self,name:str):super().__init__()self.namenameoverrideasyncdeflist_resources(self,resources:Sequence[Resource])-Sequence[Resource]:log.append(f{self.name}- list_resources)returnresourcesoverrideasyncdefget_resource(self,uri:str,call_next:GetResourceNext,*,version:VersionSpec|NoneNone,)-Resource|None:log.append(f{self.name}- pre-get_resource)handlerawaitcall_next(uri,versionversion)log.append(f{self.name}- post-get_resource)returnhandler4. 构建FastMCP我们按照如下的方式利用自定义的三个组件构建了一个FastMCP。我们创建了两个ResourceProvider对象provider1和provider2它们各自提供一个单一的Resource对象具体是根据greeting和farewell函数创建的FunctionResource。我们分别为创建的两个ResourceProvider注册了三个ResourceTransform对象后者以命名会采用Provider名称前缀以示区别。asyncdefgreeting()-str:returnHello, world!asyncdeffarewell()-str:returnGoodbye, world!provider1ResourceProvider(provider1,resources[Resource.from_function(greeting,urihttp://example.com/messages/greeting)])provider2ResourceProvider(provider2,resources[Resource.from_function(farewell,urihttp://example.com/messages/farewell)])forproviderin[provider1,provider2]:fornamein[transformer1,transformer2,transformer3]:provider.add_transform(ResourceTransform(namef{provider.name}:{name}))mcpFastMCP(MyServer,transforms[ResourceTransform(server:transformer1),ResourceTransform(server:transformer2),ResourceTransform(server:transformer3)],middleware[ResourceMiddleware(middleware1),ResourceMiddleware(middleware2),ResourceMiddleware(middleware3)],providers[provider1,provider2],)Transform除了注册到Provider上还可以注册到MCP服务器上因为FastMCP也是一个Provider。我们创建了三个服务器级别的ResourceTransform和三个ResourceMiddleware中间件两个ResourceProvider创建了代表MCP服务器的FastMCP。5. 读取资源列表我们采用如下的代码根据FastMCP对象将Client对象创建出来并调用list_resources方法返回由两个ResourceProvider提供的静态资源。asyncdefmain():asyncwithClient(mcp)asclient:log.clear()resourcesawaitclient.list_resources()assertlen(resources)2forentryinlog:print(entry)importasyncio asyncio.run(main())我们更关心的输出的日志因为它反映整个流程涉及的两个Provider、三个Middleare和9个Transform两个Provider和服务器上均注册了三个middleware1 - pre-handler middleware2 - pre-handler middleware3 - pre-handler provider1 - list_resources provider1:transformer1 - list_resources provider1:transformer2 - list_resources provider1:transformer3 - list_resources provider2 - list_resources provider2:transformer1 - list_resources provider2:transformer2 - list_resources provider2:transformer3 - list_resources server:transformer1 - list_resources server:transformer2 - list_resources server:transformer3 - list_resources middleware3 - post-handler middleware2 - post-handler middleware1 - post-handler上面的日志体现了针对list_resources请求如下的处理流程按照注册顺序执行所有Middleware的前置操作并发执行所有Provider的list_resources方法并得到一组资源列表对于每个Provider按照注册的顺序调用每个Transform的list_resources方法第一个Transform会将Provider返回的资源列表作为参数它返回的资源列表则作为下一Transform的输入以此类推由Provider提供并经过各自Transform处理后的资源列表合并成一个列表并将其作为输出按照上面的方式调用注册到服务器上的Transform的list_resources方法按照注册相反的顺序执行所有Middleware的后置操作下图更加清晰了反映了上述的流程6. 读取指定资源我们采用如下的方式调用Client对象的read_resource读取指定URI的资源。asyncdefmain():asyncwithClient(mcp)asclient:log.clear()resourceawaitclient.read_resource(http://example.com/messages/greeting)assertresource[0].textHello, world!# type: ignoreforentryinlog:print(entry)importasyncio asyncio.run(main())由于Transform针对read_resource方法采用与Middleware类似的链式封装所以会先执行前置操作然后提交给后链等再次返回后再执行后置操作所以会输出如下的日志。middleware1 - pre-handler middleware2 - pre-handler middleware3 - pre-handler server:transformer3 - pre-get_resource server:transformer2 - pre-get_resource server:transformer1 - pre-get_resource provider1:transformer3 - pre-get_resource provider1:transformer2 - pre-get_resource provider1:transformer1 - pre-get_resource provider1 - get_resource provider1:transformer1 - post-get_resource provider1:transformer2 - post-get_resource provider1:transformer3 - post-get_resource provider2:transformer3 - pre-get_resource provider2:transformer2 - pre-get_resource provider2:transformer1 - pre-get_resource provider2 - get_resource provider2:transformer1 - post-get_resource provider2:transformer2 - post-get_resource provider2:transformer3 - post-get_resource server:transformer1 - post-get_resource server:transformer2 - post-get_resource server:transformer3 - post-get_resource middleware3 - post-handler middleware2 - post-handler middleware1 - post-handler日志反映的执行流程体现在下面这张图中7. 完整实现fromtypingimportAny,override,SequencefromfastmcpimportFastMCPfrompydantic.networksimportAnyUrlfromfastmcp.resourcesimportResourcefromfastmcp.utilities.versionsimportVersionSpecfromfastmcp.server.providersimportProviderfromfastmcp.server.transformsimportTransform,GetResourceNextfromfastmcp.server.middlewareimportMiddleware,MiddlewareContext,CallNextfromfastmcp.clientimportClientimportasyncio log[]classResourceMiddleware(Middleware):def__init__(self,name:str):super().__init__()self.namenameasyncdef__call__(self,context:MiddlewareContext,call_next:CallNext,)-Any:ifnotcontext.methodin[resources/list,resources/read]:returnawaitcall_next(context)log.append(f{self.name}- pre-handler)resultawaitcall_next(context)log.append(f{self.name}- post-handler)returnresultclassResourceTransform(Transform):def__init__(self,name:str):super().__init__()self.namenameoverrideasyncdeflist_resources(self,resources:Sequence[Resource])-Sequence[Resource]:# if (len(resources) 0):log.append(f{self.name}- list_resources)returnresourcesoverrideasyncdefget_resource(self,uri:str,call_next:GetResourceNext,*,version:VersionSpec|NoneNone,)-Resource|None:log.append(f{self.name}- pre-get_resource)handlerawaitcall_next(uri,versionversion)log.append(f{self.name}- post-get_resource)returnhandlerclassResourceProvider(Provider):def__init__(self,name:str,resources:list[Resource]):super().__init__()self.namename self._resources{str(resource.uri):resourceforresourceinresources}asyncdef_list_resources(self)-list[Resource]:log.append(f{self.name}- list_resources)returnlist(self._resources.values())asyncdef_get_resource(self,uri:str,version:VersionSpec|NoneNone)-Resource|None:log.append(f{self.name}- get_resource)returnself._resources.get(uri)asyncdefgreeting()-str:returnHello, world!asyncdeffarewell()-str:returnGoodbye, world!provider1ResourceProvider(provider1,resources[Resource.from_function(greeting,urihttp://example.com/messages/greeting)])provider2ResourceProvider(provider2,resources[Resource.from_function(farewell,urihttp://example.com/messages/farewell)])forproviderin[provider1,provider2]:fornamein[transformer1,transformer2,transformer3]:provider.add_transform(ResourceTransform(namef{provider.name}:{name}))mcpFastMCP(MyServer,transforms[ResourceTransform(server:transformer1),ResourceTransform(server:transformer2),ResourceTransform(server:transformer3)],middleware[ResourceMiddleware(middleware1),ResourceMiddleware(middleware2),ResourceMiddleware(middleware3)],providers[provider1,provider2],)asyncdefmain():asyncwithClient(mcp)asclient:log.clear()resourcesawaitclient.list_resources()assertlen(resources)2forentryinlog:print(entry)log.clear()resourceawaitclient.read_resource(http://example.com/messages/greeting)assertresource[0].textHello, world!# type: ignoreforentryinlog:print(entry)asyncio.run(main())