[FastMCP设计、原理与应用-17]从服务器向客户端的反向通知

从通信或者消息交换模式来看前面涉及的都是从客户端发送请求到服务器并得到对应的响应这是典型的从客户端到服务器的请求/响应模式接下来我们介绍两种从服务器向客户端的反向通信模式通知服务端发送单向通知给客户端,客户端不需要回复。比较典型就是进度报告和日志回传这也是本篇文章着重介绍的内容请求服务端发送请求到客户端并得到对方的响应。比较典型的就是信息征询Elicitation和客户端采样Client Sampling。1. 进度报告如果客户端调用的工具涉及以长耗时的操作服务端可用通过向客户端实时报告工作进度的方式来提高用户体验。进度报告通过Context如下这个report_progress方法来完成参数progress和total通过数值的形式指定完成工作量和总体工作量。dataclassclassContext:asyncdefreport_progress(self,progress:float,total:float|NoneNone,message:str|NoneNone)-None我们可以通过创建Client对象或者利用它调用工具的时候指定一个ProgressHandler来接收进度通知。ProgressHandler这个可执行对象签名如下ProgressHandler:TypeAliasProgressFnTclassProgressFnT(Protocol):asyncdef__call__(self,progress:float,total:float|None,message:str|None)-None:...在如下这个演示程序中客户端调用工具long_running_task模拟一个长耗时操作我们将整个工作划分为5个步骤每个步骤完成后调用Context的report_progress报告一次进度。fromfastmcpimportFastMCPfromfastmcp.serverimportContextfromfastmcp.clientimportClientimportasyncio serverFastMCP(Server)server.tool()asyncdeflong_running_task(context:Context)-int:foriinrange(1,6):awaitcontext.report_progress(progressi,total5,messagefStep{i}completed)awaitasyncio.sleep(1)return5asyncdefhandle_progress(progress:float,total:float|None,message:str|None)-None:percentage(progress/(totalor100))*100print(fProgress:{percentage:.1f}% -{messageor})asyncdefmain():asyncwithClient(server)asclient:resultawaitclient.call_tool(namelong_running_task,progress_handlerhandle_progress)assertresult.content[0].text5# type: ignoreasyncio.run(main())我们定义了handle_progress函数作为接收进度通知。在利用Client调用工具long_running_task时我们将这个函数作为progress_handler参数。程序运行后客户端端接收到的进度会实时打印出来Progress: 20.0% - Step 1 completed Progress: 40.0% - Step 2 completed Progress: 60.0% - Step 3 completed Progress: 80.0% - Step 4 completed Progress: 100.0% - Step 5 completed2. 日志回传如果工具执行过程中利用Context如下这几个方法记录不同等级的日志这些日志会自动发送给客户端。dataclassclassContext:asyncdefdebug(self,message:str,logger_name:str|NoneNone,extra:Mapping[str,Any]|NoneNone,)-Noneasyncdefinfo(self,message:str,logger_name:str|NoneNone,extra:Mapping[str,Any]|NoneNone,)-Noneasyncdefwarning(self,message:str,logger_name:str|NoneNone,extra:Mapping[str,Any]|NoneNone,)-Noneasyncdeferror(self,message:str,logger_name:str|NoneNone,extra:Mapping[str,Any]|NoneNone,)-None客户端可以在创建Client对象时为其指定一个LogHandler来处理接收的日志LogHandler对应可执行对象签名和作为参数的LogMessage类型定义如下。LogMessage:TypeAliasLoggingMessageNotificationParams LogHandler:TypeAliasCallable[[LogMessage],Awaitable[None]]classLoggingMessageNotificationParams(NotificationParams):level:LoggingLevel logger:str|NoneNonedata:Any model_configConfigDict(extraallow)classNotificationParams(BaseModel):classMeta(BaseModel):model_configConfigDict(extraallow)meta:Meta|NoneField(alias_meta,defaultNone)值得一提的LogHandler能够得到某条日志取决于Client通过set_logging_level方法设置的最低日志等级它只能看到等级不低于这个设定等级的日志。classClient:asyncdefset_logging_level(self,level:mcp.types.LoggingLevel)-NoneLoggingLevelLiteral[debug,info,notice,warning,error,critical,alert,emergency]我们按照如下的方式将上面演示实例通过进度报告的通知形式替换成了日志形式。fromfastmcpimportFastMCPfromfastmcp.serverimportContextfromfastmcp.clientimportClientfromfastmcp.client.loggingimportLogMessageimportasyncio serverFastMCP(Server)server.tool()asyncdeflong_running_task(context:Context,)-int:foriinrange(1,6):awaitcontext.info(messagefStep{i}completed)awaitasyncio.sleep(1)return5log[]asyncdefhandle_log(log_message:LogMessage)-None:log.append(log_message)asyncdefmain():asyncwithClient(server,log_handlerhandle_log)asclient:awaitclient.set_logging_level(emergency)resultawaitclient.call_tool(namelong_running_task,)assertresult.content[0].text5# type: ignoreassertlen(log)5assertall(log_message.levelinfoforlog_messageinlog)asyncio.run(main())3. 发送通知包括上面介绍的进度和日志以及其他相关的通知可以直接通过调用Context的send_notification方法来完成。参数notification的类型ServerNotificationType是对多个通知类型的联合它们都是Notification的子类。众多通知类型由mcp库提供模块路径为mcp.types是对MCP规范的实现。由于MCP采用JSON-RPC协议所以Notification分别利用其method和params字段表示JSON-RPC的方法和参数后者的基类为NotificationParams具有一个表示元数据的meta字段。dataclassclassContext:asyncdefsend_notification(self,notification:mcp.types.ServerNotificationType)-NoneServerNotificationType:TypeAlias(CancelledNotification|ProgressNotification|LoggingMessageNotification|ResourceUpdatedNotification|ResourceListChangedNotification|ToolListChangedNotification|PromptListChangedNotification|ElicitCompleteNotification|TaskStatusNotification)classNotification(BaseModel,Generic[NotificationParamsT,MethodT]):method:MethodT params:NotificationParamsT model_configConfigDict(extraallow)classNotificationParams(BaseModel):classMeta(BaseModel):model_configConfigDict(extraallow)meta:Meta|NoneField(alias_meta,defaultNone)RequestParamsTTypeVar(RequestParamsT,boundRequestParams|dict[str,Any]|None)NotificationParamsTTypeVar(NotificationParamsT,boundNotificationParams|dict[str,Any]|None)MethodTTypeVar(MethodT,boundstr)接下来我们对各种通知类型和对应的JSON-RPC方法进行概括性介绍CancelledNotificationnotifications/cancelled客户端和服务器向对方发送的取消之前请求的通知ProgressNotificationnotifications/progress客户端和服务器向对方发送的进度报告LoggingMessageNotificationnotifications/message服务端向客户端发送的日志ResourceUpdatedNotificationnotifications/resources/updated服务端向客户端发送的关于资源被更新的通知ResourceListChangedNotification(notifications/resources/list_changed):服务端向客户端发送的关于资源列表发生改变的通知ToolListChangedNotificationnotifications/tools/list_changed服务端向客户端发送的关于工具列表发生改变的通知PromptListChangedNotificationnotifications/prompts/list_changed服务端向客户端发送的关于提示词列表发生改变的通知ElicitCompleteNotificationnotifications/elicitation/complete服务器发给客户端的异步“通关信号”告知之前因权限或配置受限的URL访问已处理完成;TaskStatusNotification(notifications/tasks/status):客户端和服务器向对方发送的关于任务状态改变的通知。客户端可以在创建Client的时候创建一个MessageHandlerT对象作为其message_handler参数来处理接收到的通知。MessageHandlerT这个可执行对象的签名定义如下它的参数通常为一个mcp.types.ServerNotification对象可以通过后者的root字典得到上述这些个通知对象。MessageHandlerT:TypeAliasMessageHandlerFnTclassMessageHandlerFnT(Protocol):asyncdef__call__(self,message:RequestResponder[types.ServerRequest,types.ClientResult]|types.ServerNotification|Exception,)-None:...classServerNotification(RootModel[ServerNotificationType]):passclassRootModel(BaseModel,Generic[RootModelRootType],metaclass_RootModelMetaclass):root:RootModelRootType下面的程序演示了在一个工具函数中利用注入的Context向客户端发送四种类型的通知以及客户端利用注册的MessageHandler来处理这些通知。fromfastmcpimportFastMCPfromfastmcp.serverimportContextfromfastmcp.clientimportClientfrompydanticimportAnyUrlfrommcp.typesimport(ServerNotification,ServerRequest,ClientResult,ResourceUpdatedNotification,ResourceListChangedNotification,ToolListChangedNotification,PromptListChangedNotification,ResourceUpdatedNotificationParams)frommcp.shared.sessionimportRequestResponderimportasyncio serverFastMCP(Server)server.tool()asyncdeffire_notifications(context:Context,)-None:paramsResourceUpdatedNotificationParams(uriAnyUrl(file:///path/to/resource))awaitcontext.send_notification(ResourceUpdatedNotification(paramsparams))awaitcontext.send_notification(ResourceListChangedNotification())awaitcontext.send_notification(ToolListChangedNotification())awaitcontext.send_notification(PromptListChangedNotification())asyncdefhandle_notifications(message:RequestResponder[ServerRequest,ClientResult]|ServerNotification|Exception,)-None:matchmessage.root:# type: ignorecaseResourceUpdatedNotification():print(Received resource updated notification for URI:,message.root.params.uri)# type: ignorecaseResourceListChangedNotification():print(Received resource list changed notification)caseToolListChangedNotification():print(Received tool list changed notification)casePromptListChangedNotification():print(Received prompt list changed notification)asyncdefmain():asyncwithClient(server,message_handlerhandle_notifications)asclient:awaitclient.call_tool(namefire_notifications,)awaitasyncio.sleep(5)# Wait for notifications to be processedasyncio.run(main())输出Received resource updated notification for URI: file:///path/to/resource Received resource list changed notification Received tool list changed notification Received prompt list changed notification