![[FastMCP设计、原理与应用-13]Middleware：守门人的艺术，在拦截与增强间重塑执行逻辑](http://pic.xiahunao.cn/yaotu/[FastMCP设计、原理与应用-13]Middleware：守门人的艺术，在拦截与增强间重塑执行逻辑)
和一般意义中间件比如Web框架的中间件一样FastMCP服务器将注册的中间件按照顺序构建为一个调用链。请求在交付给最终的处理器之前会流经这个链使得每个中间件得以完成相应的前置操作。待处理器完成请求的处理后处理的结果反向流经这个链每个中间件提供的后置操作得以执行。1. Middleware的核心方法__call__虽然作为基类的Middleware类型定义了很多方法但是FastMCP服务器只关注__call__方法。TTypeVar(T,defaultAny)RTypeVar(R,covariantTrue,defaultAny)classMiddleware:asyncdef__call__(self,context:MiddlewareContext[T],call_next:CallNext[T,Any],)-AnyclassCallNext(Protocol[T,R]):def__call__(self,context:MiddlewareContext[T])-Awaitable[R]:...这个__call__方法定义了两个参数context返回的MiddlewareContext对象提供中间件执行上下文的相关信息call_next: 返回的CallNext[T, Any]是一个可执行对象表示后续流程的处理操作。调用它就意味着将请求交付给后续流程继续处理并得到处理的结果。MiddlewareContext类型定义如下。由于使用了dataclass(frozenTrue)装饰器所以它是不可变的这确保了中间件在处理链条中不会意外修改原始数据。也这是因为如此它定义了一个copy方法。dataclass(kw_onlyTrue,frozenTrue)classMiddlewareContext(Generic[T]):message:T fastmcp_context:Context|NoneNonesource:Literal[client,server]clienttype:Literal[request,notification]requestmethod:str|NoneNonetimestamp:datetimedefcopy(self,**kwargs:Any)-MiddlewareContext[T]:returnreplace(self,**kwargs)定义在MiddlewareContext中的字段成员说明如下message当前处理的原始消息体。可能是请求对象、响应对象或通知中间件的主要任务通常就是读取或修改这个它fastmcp_context: FastMCP的运行上下文它承载了太多的信息和功能我们将在后面对它进行单独介绍source: 消息的来源方向。判断这是客户端发给服务器的请求client还是服务器发给客户端的响应/通知servertype:消息类型。区分这是一个需要回执的“请求”request还是一个“发后即忘”的“通知”notification如日志或进度更新methodJSON-RPC方法名timestamp消息处理的时间戳。用于性能监控计算耗时或审计日志2. 中间件的分发机制虽然FastMCP服务器只认__call__方法但是我们自定义的中间件类型一般不会重写此方法因为Middleware对这个方法的默认实现会根据当前上下文将请求分发给其他方法进行处理我们一般只需要根据需求选择重写相应的方法就可以了。classMiddleware:asyncdef__call__(self,context:MiddlewareContext[T],call_next:CallNext[T,Any],)-Any:handler_chainawaitself._dispatch_handler(context,call_nextcall_next,)returnawaithandler_chain(context)asyncdef_dispatch_handler(self,context:MiddlewareContext[Any],call_next:CallNext[Any,Any])-CallNext[Any,Any]:handlercall_nextmatchcontext.method:caseinitialize:handlerpartial(self.on_initialize,call_nexthandler)casetools/call:handlerpartial(self.on_call_tool,call_nexthandler)caseresources/read:handlerpartial(self.on_read_resource,call_nexthandler)caseprompts/get:handlerpartial(self.on_get_prompt,call_nexthandler)casetools/list:handlerpartial(self.on_list_tools,call_nexthandler)caseresources/list:handlerpartial(self.on_list_resources,call_nexthandler)caseresources/templates/list:handlerpartial(self.on_list_resource_templates,call_nexthandler)caseprompts/list:handlerpartial(self.on_list_prompts,call_nexthandler)matchcontext.type:caserequest:handlerpartial(self.on_request,call_nexthandler)casenotification:handlerpartial(self.on_notification,call_nexthandler)handlerpartial(self.on_message,call_nexthandler)returnhandler从实现在_dispatch_handler方法中分发逻辑可以看出整个分发分为如下两类根据JSON-RPC方法进行分发由于JSON-RPC方法直指具体的MCP操作如果自定义中间件只关注某个具体的操作比如工具调用重写对应方法就可以比如on_call_tool方法根据消息类型进行分发如果中间件只关注针对客户端请求的处理只需要重写on_request方法若只关心服务端的反向通知则只需要重写on_notification方法。3. LoggingMiddlewareLoggingMiddleware是用于监控和记录日志的中间件它能服务器接收请求或发出响应的整个流程记录下来。在如下这个演示程序中我们为FastMCP注册了一个LoggingMiddleware中间件针对客户但调用工具get_wheather的请求服务端的处理流程会反映在输出的日志中。fromfastmcpimportFastMCP,Clientfromfastmcp.server.middleware.loggingimportLoggingMiddleware mcpFastMCP(Server)mcp.toolasyncdefget_wheather(city:str)-str:get weather info for a city.returnfIts sunny and 25 degrees Celsius in{city}.mcp.add_middleware(LoggingMiddleware(include_payloadsTrue,max_payload_length1000))asyncdefmain():asyncwithClient(mcp)asclient:awaitclient.call_tool(nameget_wheather,arguments{city:Suzhou})if__name____main__:importasyncio asyncio.run(main())输出[04/07/26 21:34:58] INFO eventrequest_start methodinitialize sourceclient payload{method:initialize,params:{task:null,_meta:null,protocolVersion:2025-11-25,capabilities:{experimental:null,samplin g:null,elicitation:null,roots:null,tasks:null},clientInfo:{name:mcp,title:null,version:0.1.0,websiteUrl:null,icons:nul l}},jsonrpc:2.0,id:0} payload_typeInitializeRequest INFO eventrequest_success methodinitialize sourceclient duration_ms0.71 INFO eventrequest_start methodtools/list sourceclient payload{method:tools/list,params:null} payload_typeListToolsRequest INFO eventrequest_success methodtools/list sourceclient duration_ms1.86 INFO eventrequest_start methodtools/call sourceclient payload{task:null,_meta:null,name:get_wheather,arguments:{city:Suzhou}} payload_typeCallToolRequestParams INFO eventrequest_success methodtools/call sourceclient duration_ms6.21 INFO eventrequest_start methodtools/list sourceclient payload{method:tools/list,params:null} payload_typeListToolsRequest INFO eventrequest_success methodtools/list sourceclient duration_ms0.68LoggingMiddleware继承自BaseLoggingMiddleware下面给出它构造函数的定义classLoggingMiddleware(BaseLoggingMiddleware):def__init__(self,*,logger:logging.Logger|NoneNone,log_level:intlogging.INFO,include_payloads:boolFalse,include_payload_length:boolFalse,estimate_payload_tokens:boolFalse,max_payload_length:int1000,methods:list[str]|NoneNone,payload_serializer:Callable[[Any],str]|NoneNone,)构造函数参数说明如下logger: 你可以传入自己配置好的日志对象默认通过logging.getLogger(fastmcp.middleware.logging)获取这个logger对象log_level: 决定日志的级别如INFO, DEBUG默认为INFO;include_payloads: 是否把具体的请求数据打印出来。如果数据很大开启这个会刷屏include_payload_length: 是否记录响应的大小estimate_payload_tokens:估算消耗了多少Token对大模型应用非常重要max_payload_length: 配合上一项使用限制单条日志的最大字符数默认1000防止日志文件爆炸methods:可以指定只记录某些特定的MCP方法比如只记录tools/callpayload_serializer: 自定义日志序列化器4. TimingMiddleware DetailedTimingMiddlewareTimingMiddleware会记录所有请求的执行时间。DetailedTimingMiddleware则提供按操作计时的信息并对工具、资源和提示进行单独跟踪。在如下的演示程序中我们为两个FastMCP分别注册了TimingMiddleware和DetailedTimingMiddleware中间件针对工具的调用它们会以不同的形式将涉及的请求处理和操作的执行时间记录下来。fromfastmcpimportFastMCP,Clientfromfastmcp.server.middleware.timingimportTimingMiddleware,DetailedTimingMiddleware mcp1FastMCP(Server1)mcp2FastMCP(Server2)asyncdefget_wheather(city:str)-str:get weather info for a city.returnfIts sunny and 25 degrees Celsius in{city}.mcp1.add_tool(get_wheather)mcp2.add_tool(get_wheather)mcp1.add_middleware(TimingMiddleware())mcp2.add_middleware(DetailedTimingMiddleware())asyncdefmain():asyncwithClient(mcp1)asclient:awaitclient.call_tool(nameget_wheather,arguments{city:Suzhou})asyncwithClient(mcp2)asclient:awaitclient.call_tool(nameget_wheather,arguments{city:Suzhou})if__name____main__:importasyncio asyncio.run(main())输出[04/07/26 21:55:09] INFO Request initialize completed in 0.17ms INFO Request tools/list completed in 0.66ms INFO Request tools/call completed in 1.43ms INFO Request tools/list completed in 0.10ms [04/07/26 21:55:10] INFO List tools completed in 0.16ms INFO Tool get_wheather completed in 0.22ms INFO List tools completed in 0.10msTimingMiddleware和DetailedTimingMiddleware的构造函数具有相同的参数定义我们可以利用logger和log_level指定记录日志的logger对象和日志等级。classTimingMiddleware(Middleware):def__init__(self,logger:logging.Logger|NoneNone,log_level:intlogging.INFO)classDetailedTimingMiddleware(Middleware):def__init__(self,logger:logging.Logger|NoneNone,log_level:intlogging.INFO)5. ResponseCachingMiddlewareResponseCachingMiddleware的核心作用是用空间换时间通过缓存执行结果避免重复执行高耗时或高成本的操作。在如下的演示程序中我们为创建的FastMCP注册了一个用于返回当前时间的工具函数get_current_time参数is_utc表示是返回UTC时间还是本地时间。我们通过add_middleware方法添加了ResponseCachingMiddleware中间件。我们利用客户端以1秒的时间间隔分三轮调用了此工具由于工具get_current_time的执行结果根据参数被缓存起来所以后面两轮工具调用会得到相同的时间。fromfastmcpimportFastMCP,Clientfromfastmcp.server.middleware.cachingimportResponseCachingMiddlewarefromdatetimeimportdatetime,timezoneimportasyncio mcpFastMCP(Server)mcp.toolasyncdefget_current_time(is_utc:boolFalse)-datetime:Get the current time as a string.returndatetime.now(timezone.utc)ifis_utcelsedatetime.now()mcp.add_middleware(ResponseCachingMiddleware())asyncdefmain():asyncwithClient(mcp)asclient:for_inrange(3):resultawaitclient.call_tool(nameget_current_time,arguments{is_utc:True})print(fCurrent time (UTC):{result.content[0].text})# type: ignoreresultawaitclient.call_tool(nameget_current_time,arguments{is_utc:False})print(fCurrent time (Local):{result.content[0].text}\n)# type: ignoreawaitasyncio.sleep(1)asyncio.run(main())输出Current time (UTC): 2026-04-07T14:10:35.955032Z Current time (Local): 2026-04-07T22:10:35.958043 Current time (UTC): 2026-04-07T14:10:35.955032Z Current time (Local): 2026-04-07T22:10:35.958043 Current time (UTC): 2026-04-07T14:10:35.955032Z Current time (Local): 2026-04-07T22:10:35.958043ResponseCachingMiddleware构造函数定义如下。我们可以利用cache_storage参数自定义缓存存储默认采用MemoryStore。我们可以利用参数针对不同的操作list_tool、list_resources、list_prompts、call_tool、read_resource和get_prompt设置不同的缓存策略是否开启缓存以及缓存有效期针对工具调用还可以根据工具名称进行针对性设置。max_item_size参数用于限制缓存数据的大小默认为1MB。classResponseCachingMiddleware(Middleware):def__init__(self,cache_storage:AsyncKeyValue|NoneNone,list_tools_settings:ListToolsSettings|NoneNone,list_resources_settings:ListResourcesSettings|NoneNone,list_prompts_settings:ListPromptsSettings|NoneNone,read_resource_settings:ReadResourceSettings|NoneNone,get_prompt_settings:GetPromptSettings|NoneNone,call_tool_settings:CallToolSettings|NoneNone,max_item_size:intONE_MB_IN_BYTES,)classSharedMethodSettings(TypedDict):ttl:NotRequired[int]enabled:NotRequired[bool]classListToolsSettings(SharedMethodSettings):...classListResourcesSettings(SharedMethodSettings):...classListPromptsSettings(SharedMethodSettings):...classCallToolSettings(SharedMethodSettings):...included_tools:NotRequired[list[str]]excluded_tools:NotRequired[list[str]]classReadResourceSettings(SharedMethodSettings):...classGetPromptSettings(SharedMethodSettings):...6. RateLimitingMiddleware SlidingWindowRateLimitingMiddlewareRateLimitingMiddleware使用令牌桶算法允许受控突发流量。SlidingWindowRateLimitingMiddleware提供精确的时间窗口速率限制但不允许突发流量。在下面的演示程序中我们为FastMCP注册了一个长耗时的工具long_running_task并调用add_middleware方法注册了一个将max_requests_per_second设置为2的RateLimitingMiddleware。对于客户端并发调用的5个工具调用请求只有两个会成功。fromfastmcpimportFastMCP,Clientfromfastmcp.server.middleware.rate_limitingimportRateLimitingMiddlewareimportasyncio mcpFastMCP(Server)mcp.toolasyncdeflong_running_task()-None:Simulate a long-running task.awaitasyncio.sleep(5)mcp.add_middleware(RateLimitingMiddleware(max_requests_per_second2))asyncdefmain():asyncwithClient(mcp)asclient:asyncdefcall_task(i):try:awaitclient.call_tool(namelong_running_task,arguments{})print(fCall{i}succeeded.)exceptExceptionase:print(fError calling long_running_task{i})awaitasyncio.sleep(1)# Ensure server is readytasks[call_task(i1)foriinrange(5)]awaitasyncio.gather(*tasks)asyncio.run(main())输出Error calling long_running_task 2 Error calling long_running_task 4 Error calling long_running_task 5 Call 1 succeeded. Call 3 succeeded.RateLimitingMiddleware和SlidingWindowRateLimitingMiddleware类型的构造函数定义如下classRateLimitingMiddleware(Middleware):def__init__(self,max_requests_per_second:float10.0,burst_capacity:int|NoneNone,get_client_id:Callable[[MiddlewareContext],str]|NoneNone,global_limit:boolFalse,)classSlidingWindowRateLimitingMiddleware(Middleware):def__init__(self,max_requests:int,window_minutes:int1,get_client_id:Callable[[MiddlewareContext],str]|NoneNone,)构造函数具有如下的参数max_requests_per_second长期来看每秒钟系统生成的令牌数决定了服务器的可持续处理能力。比如设为0.5则意味着每2秒才允许1个请求。默认值为10burst_capacity允许瞬间爆发的最大请求量。如果你不传None它会自动设为max_requests_per_second * 2。如果设为 20即使用户之前一直没发请求他突然瞬间发20个请求也能全部通过get_client_id一个自定义函数回调函数用于从请求上下文MiddlewareContext中提取“客户端是谁”可以利用它定义不同的限流策略比如根据API-KEY、IP或者Session ID限流。默认为针对全局限流global_limit决定是“大家共用一个额度”还是“每个人独立计算额度”True (全局模式)整个服务器每秒只能处理X个请求。哪怕你是 100 个不同的用户只要总数超了就报错。这通常用于保护服务器硬件资源False (分用户模式 - 默认)根据get_client_id获取的 ID为每个用户分配一个独立的“水桶”max_requests: 在整个设定的时间窗口内允许通过的请求总量这是一个硬上限。比如设定为 100那么在窗口内第101个请求绝对会被拦截。它没有burst的概念因为窗口本身就允许你在窗口开始的瞬间把额度用完window_minutes: 统计的时间跨度滑动窗口单位是分钟。默认值为1分钟。
[FastMCP设计、原理与应用-13]Middleware：守门人的艺术，在拦截与增强间重塑执行逻辑
和一般意义中间件比如Web框架的中间件一样FastMCP服务器将注册的中间件按照顺序构建为一个调用链。请求在交付给最终的处理器之前会流经这个链使得每个中间件得以完成相应的前置操作。待处理器完成请求的处理后处理的结果反向流经这个链每个中间件提供的后置操作得以执行。1. Middleware的核心方法__call__虽然作为基类的Middleware类型定义了很多方法但是FastMCP服务器只关注__call__方法。TTypeVar(T,defaultAny)RTypeVar(R,covariantTrue,defaultAny)classMiddleware:asyncdef__call__(self,context:MiddlewareContext[T],call_next:CallNext[T,Any],)-AnyclassCallNext(Protocol[T,R]):def__call__(self,context:MiddlewareContext[T])-Awaitable[R]:...这个__call__方法定义了两个参数context返回的MiddlewareContext对象提供中间件执行上下文的相关信息call_next: 返回的CallNext[T, Any]是一个可执行对象表示后续流程的处理操作。调用它就意味着将请求交付给后续流程继续处理并得到处理的结果。MiddlewareContext类型定义如下。由于使用了dataclass(frozenTrue)装饰器所以它是不可变的这确保了中间件在处理链条中不会意外修改原始数据。也这是因为如此它定义了一个copy方法。dataclass(kw_onlyTrue,frozenTrue)classMiddlewareContext(Generic[T]):message:T fastmcp_context:Context|NoneNonesource:Literal[client,server]clienttype:Literal[request,notification]requestmethod:str|NoneNonetimestamp:datetimedefcopy(self,**kwargs:Any)-MiddlewareContext[T]:returnreplace(self,**kwargs)定义在MiddlewareContext中的字段成员说明如下message当前处理的原始消息体。可能是请求对象、响应对象或通知中间件的主要任务通常就是读取或修改这个它fastmcp_context: FastMCP的运行上下文它承载了太多的信息和功能我们将在后面对它进行单独介绍source: 消息的来源方向。判断这是客户端发给服务器的请求client还是服务器发给客户端的响应/通知servertype:消息类型。区分这是一个需要回执的“请求”request还是一个“发后即忘”的“通知”notification如日志或进度更新methodJSON-RPC方法名timestamp消息处理的时间戳。用于性能监控计算耗时或审计日志2. 中间件的分发机制虽然FastMCP服务器只认__call__方法但是我们自定义的中间件类型一般不会重写此方法因为Middleware对这个方法的默认实现会根据当前上下文将请求分发给其他方法进行处理我们一般只需要根据需求选择重写相应的方法就可以了。classMiddleware:asyncdef__call__(self,context:MiddlewareContext[T],call_next:CallNext[T,Any],)-Any:handler_chainawaitself._dispatch_handler(context,call_nextcall_next,)returnawaithandler_chain(context)asyncdef_dispatch_handler(self,context:MiddlewareContext[Any],call_next:CallNext[Any,Any])-CallNext[Any,Any]:handlercall_nextmatchcontext.method:caseinitialize:handlerpartial(self.on_initialize,call_nexthandler)casetools/call:handlerpartial(self.on_call_tool,call_nexthandler)caseresources/read:handlerpartial(self.on_read_resource,call_nexthandler)caseprompts/get:handlerpartial(self.on_get_prompt,call_nexthandler)casetools/list:handlerpartial(self.on_list_tools,call_nexthandler)caseresources/list:handlerpartial(self.on_list_resources,call_nexthandler)caseresources/templates/list:handlerpartial(self.on_list_resource_templates,call_nexthandler)caseprompts/list:handlerpartial(self.on_list_prompts,call_nexthandler)matchcontext.type:caserequest:handlerpartial(self.on_request,call_nexthandler)casenotification:handlerpartial(self.on_notification,call_nexthandler)handlerpartial(self.on_message,call_nexthandler)returnhandler从实现在_dispatch_handler方法中分发逻辑可以看出整个分发分为如下两类根据JSON-RPC方法进行分发由于JSON-RPC方法直指具体的MCP操作如果自定义中间件只关注某个具体的操作比如工具调用重写对应方法就可以比如on_call_tool方法根据消息类型进行分发如果中间件只关注针对客户端请求的处理只需要重写on_request方法若只关心服务端的反向通知则只需要重写on_notification方法。3. LoggingMiddlewareLoggingMiddleware是用于监控和记录日志的中间件它能服务器接收请求或发出响应的整个流程记录下来。在如下这个演示程序中我们为FastMCP注册了一个LoggingMiddleware中间件针对客户但调用工具get_wheather的请求服务端的处理流程会反映在输出的日志中。fromfastmcpimportFastMCP,Clientfromfastmcp.server.middleware.loggingimportLoggingMiddleware mcpFastMCP(Server)mcp.toolasyncdefget_wheather(city:str)-str:get weather info for a city.returnfIts sunny and 25 degrees Celsius in{city}.mcp.add_middleware(LoggingMiddleware(include_payloadsTrue,max_payload_length1000))asyncdefmain():asyncwithClient(mcp)asclient:awaitclient.call_tool(nameget_wheather,arguments{city:Suzhou})if__name____main__:importasyncio asyncio.run(main())输出[04/07/26 21:34:58] INFO eventrequest_start methodinitialize sourceclient payload{method:initialize,params:{task:null,_meta:null,protocolVersion:2025-11-25,capabilities:{experimental:null,samplin g:null,elicitation:null,roots:null,tasks:null},clientInfo:{name:mcp,title:null,version:0.1.0,websiteUrl:null,icons:nul l}},jsonrpc:2.0,id:0} payload_typeInitializeRequest INFO eventrequest_success methodinitialize sourceclient duration_ms0.71 INFO eventrequest_start methodtools/list sourceclient payload{method:tools/list,params:null} payload_typeListToolsRequest INFO eventrequest_success methodtools/list sourceclient duration_ms1.86 INFO eventrequest_start methodtools/call sourceclient payload{task:null,_meta:null,name:get_wheather,arguments:{city:Suzhou}} payload_typeCallToolRequestParams INFO eventrequest_success methodtools/call sourceclient duration_ms6.21 INFO eventrequest_start methodtools/list sourceclient payload{method:tools/list,params:null} payload_typeListToolsRequest INFO eventrequest_success methodtools/list sourceclient duration_ms0.68LoggingMiddleware继承自BaseLoggingMiddleware下面给出它构造函数的定义classLoggingMiddleware(BaseLoggingMiddleware):def__init__(self,*,logger:logging.Logger|NoneNone,log_level:intlogging.INFO,include_payloads:boolFalse,include_payload_length:boolFalse,estimate_payload_tokens:boolFalse,max_payload_length:int1000,methods:list[str]|NoneNone,payload_serializer:Callable[[Any],str]|NoneNone,)构造函数参数说明如下logger: 你可以传入自己配置好的日志对象默认通过logging.getLogger(fastmcp.middleware.logging)获取这个logger对象log_level: 决定日志的级别如INFO, DEBUG默认为INFO;include_payloads: 是否把具体的请求数据打印出来。如果数据很大开启这个会刷屏include_payload_length: 是否记录响应的大小estimate_payload_tokens:估算消耗了多少Token对大模型应用非常重要max_payload_length: 配合上一项使用限制单条日志的最大字符数默认1000防止日志文件爆炸methods:可以指定只记录某些特定的MCP方法比如只记录tools/callpayload_serializer: 自定义日志序列化器4. TimingMiddleware DetailedTimingMiddlewareTimingMiddleware会记录所有请求的执行时间。DetailedTimingMiddleware则提供按操作计时的信息并对工具、资源和提示进行单独跟踪。在如下的演示程序中我们为两个FastMCP分别注册了TimingMiddleware和DetailedTimingMiddleware中间件针对工具的调用它们会以不同的形式将涉及的请求处理和操作的执行时间记录下来。fromfastmcpimportFastMCP,Clientfromfastmcp.server.middleware.timingimportTimingMiddleware,DetailedTimingMiddleware mcp1FastMCP(Server1)mcp2FastMCP(Server2)asyncdefget_wheather(city:str)-str:get weather info for a city.returnfIts sunny and 25 degrees Celsius in{city}.mcp1.add_tool(get_wheather)mcp2.add_tool(get_wheather)mcp1.add_middleware(TimingMiddleware())mcp2.add_middleware(DetailedTimingMiddleware())asyncdefmain():asyncwithClient(mcp1)asclient:awaitclient.call_tool(nameget_wheather,arguments{city:Suzhou})asyncwithClient(mcp2)asclient:awaitclient.call_tool(nameget_wheather,arguments{city:Suzhou})if__name____main__:importasyncio asyncio.run(main())输出[04/07/26 21:55:09] INFO Request initialize completed in 0.17ms INFO Request tools/list completed in 0.66ms INFO Request tools/call completed in 1.43ms INFO Request tools/list completed in 0.10ms [04/07/26 21:55:10] INFO List tools completed in 0.16ms INFO Tool get_wheather completed in 0.22ms INFO List tools completed in 0.10msTimingMiddleware和DetailedTimingMiddleware的构造函数具有相同的参数定义我们可以利用logger和log_level指定记录日志的logger对象和日志等级。classTimingMiddleware(Middleware):def__init__(self,logger:logging.Logger|NoneNone,log_level:intlogging.INFO)classDetailedTimingMiddleware(Middleware):def__init__(self,logger:logging.Logger|NoneNone,log_level:intlogging.INFO)5. ResponseCachingMiddlewareResponseCachingMiddleware的核心作用是用空间换时间通过缓存执行结果避免重复执行高耗时或高成本的操作。在如下的演示程序中我们为创建的FastMCP注册了一个用于返回当前时间的工具函数get_current_time参数is_utc表示是返回UTC时间还是本地时间。我们通过add_middleware方法添加了ResponseCachingMiddleware中间件。我们利用客户端以1秒的时间间隔分三轮调用了此工具由于工具get_current_time的执行结果根据参数被缓存起来所以后面两轮工具调用会得到相同的时间。fromfastmcpimportFastMCP,Clientfromfastmcp.server.middleware.cachingimportResponseCachingMiddlewarefromdatetimeimportdatetime,timezoneimportasyncio mcpFastMCP(Server)mcp.toolasyncdefget_current_time(is_utc:boolFalse)-datetime:Get the current time as a string.returndatetime.now(timezone.utc)ifis_utcelsedatetime.now()mcp.add_middleware(ResponseCachingMiddleware())asyncdefmain():asyncwithClient(mcp)asclient:for_inrange(3):resultawaitclient.call_tool(nameget_current_time,arguments{is_utc:True})print(fCurrent time (UTC):{result.content[0].text})# type: ignoreresultawaitclient.call_tool(nameget_current_time,arguments{is_utc:False})print(fCurrent time (Local):{result.content[0].text}\n)# type: ignoreawaitasyncio.sleep(1)asyncio.run(main())输出Current time (UTC): 2026-04-07T14:10:35.955032Z Current time (Local): 2026-04-07T22:10:35.958043 Current time (UTC): 2026-04-07T14:10:35.955032Z Current time (Local): 2026-04-07T22:10:35.958043 Current time (UTC): 2026-04-07T14:10:35.955032Z Current time (Local): 2026-04-07T22:10:35.958043ResponseCachingMiddleware构造函数定义如下。我们可以利用cache_storage参数自定义缓存存储默认采用MemoryStore。我们可以利用参数针对不同的操作list_tool、list_resources、list_prompts、call_tool、read_resource和get_prompt设置不同的缓存策略是否开启缓存以及缓存有效期针对工具调用还可以根据工具名称进行针对性设置。max_item_size参数用于限制缓存数据的大小默认为1MB。classResponseCachingMiddleware(Middleware):def__init__(self,cache_storage:AsyncKeyValue|NoneNone,list_tools_settings:ListToolsSettings|NoneNone,list_resources_settings:ListResourcesSettings|NoneNone,list_prompts_settings:ListPromptsSettings|NoneNone,read_resource_settings:ReadResourceSettings|NoneNone,get_prompt_settings:GetPromptSettings|NoneNone,call_tool_settings:CallToolSettings|NoneNone,max_item_size:intONE_MB_IN_BYTES,)classSharedMethodSettings(TypedDict):ttl:NotRequired[int]enabled:NotRequired[bool]classListToolsSettings(SharedMethodSettings):...classListResourcesSettings(SharedMethodSettings):...classListPromptsSettings(SharedMethodSettings):...classCallToolSettings(SharedMethodSettings):...included_tools:NotRequired[list[str]]excluded_tools:NotRequired[list[str]]classReadResourceSettings(SharedMethodSettings):...classGetPromptSettings(SharedMethodSettings):...6. RateLimitingMiddleware SlidingWindowRateLimitingMiddlewareRateLimitingMiddleware使用令牌桶算法允许受控突发流量。SlidingWindowRateLimitingMiddleware提供精确的时间窗口速率限制但不允许突发流量。在下面的演示程序中我们为FastMCP注册了一个长耗时的工具long_running_task并调用add_middleware方法注册了一个将max_requests_per_second设置为2的RateLimitingMiddleware。对于客户端并发调用的5个工具调用请求只有两个会成功。fromfastmcpimportFastMCP,Clientfromfastmcp.server.middleware.rate_limitingimportRateLimitingMiddlewareimportasyncio mcpFastMCP(Server)mcp.toolasyncdeflong_running_task()-None:Simulate a long-running task.awaitasyncio.sleep(5)mcp.add_middleware(RateLimitingMiddleware(max_requests_per_second2))asyncdefmain():asyncwithClient(mcp)asclient:asyncdefcall_task(i):try:awaitclient.call_tool(namelong_running_task,arguments{})print(fCall{i}succeeded.)exceptExceptionase:print(fError calling long_running_task{i})awaitasyncio.sleep(1)# Ensure server is readytasks[call_task(i1)foriinrange(5)]awaitasyncio.gather(*tasks)asyncio.run(main())输出Error calling long_running_task 2 Error calling long_running_task 4 Error calling long_running_task 5 Call 1 succeeded. Call 3 succeeded.RateLimitingMiddleware和SlidingWindowRateLimitingMiddleware类型的构造函数定义如下classRateLimitingMiddleware(Middleware):def__init__(self,max_requests_per_second:float10.0,burst_capacity:int|NoneNone,get_client_id:Callable[[MiddlewareContext],str]|NoneNone,global_limit:boolFalse,)classSlidingWindowRateLimitingMiddleware(Middleware):def__init__(self,max_requests:int,window_minutes:int1,get_client_id:Callable[[MiddlewareContext],str]|NoneNone,)构造函数具有如下的参数max_requests_per_second长期来看每秒钟系统生成的令牌数决定了服务器的可持续处理能力。比如设为0.5则意味着每2秒才允许1个请求。默认值为10burst_capacity允许瞬间爆发的最大请求量。如果你不传None它会自动设为max_requests_per_second * 2。如果设为 20即使用户之前一直没发请求他突然瞬间发20个请求也能全部通过get_client_id一个自定义函数回调函数用于从请求上下文MiddlewareContext中提取“客户端是谁”可以利用它定义不同的限流策略比如根据API-KEY、IP或者Session ID限流。默认为针对全局限流global_limit决定是“大家共用一个额度”还是“每个人独立计算额度”True (全局模式)整个服务器每秒只能处理X个请求。哪怕你是 100 个不同的用户只要总数超了就报错。这通常用于保护服务器硬件资源False (分用户模式 - 默认)根据get_client_id获取的 ID为每个用户分配一个独立的“水桶”max_requests: 在整个设定的时间窗口内允许通过的请求总量这是一个硬上限。比如设定为 100那么在窗口内第101个请求绝对会被拦截。它没有burst的概念因为窗口本身就允许你在窗口开始的瞬间把额度用完window_minutes: 统计的时间跨度滑动窗口单位是分钟。默认值为1分钟。