从通信或者消息交换模式来看，前面涉及的都是从客户端发送请求到服务器并得到对应的响应，这是典型的从客户端到服务器的请求/响应模式，接下来我们介绍两种从服务器向客户端的反向通信模式：
- 通知：服务端发送单向通知给客户端,客户端不需要回复。比较典型就是进度报告和日志回传，这也是本篇文章着重介绍的内容；
- 请求：服务端发送请求到客户端，并得到对方的响应。比较典型的就是信息征询（Elicitation）和客户端采样（Client Sampling）。
1. 进度报告
如果客户端调用的工具涉及以长耗时的操作，服务端可用通过向客户端实时报告工作进度的方式来提高用户体验。进度报告通过Context如下这个report_progress方法来完成，参数progress和total通过数值的形式指定完成工作量和总体工作量。
@dataclassclassContext:asyncdefreport_progress(self,progress:float,total:float|None=None,message:str|None=None)->None我们可以通过创建Client对象或者利用它调用工具的时候指定一个ProgressHandler来接收进度通知。ProgressHandler这个可执行对象签名如下：
ProgressHandler:TypeAlias=ProgressFnTclassProgressFnT(Protocol):asyncdef__call__(self,progress:float,total:float|None,message:str|None)->None:...在如下这个演示程序中，客户端调用工具long_running_task模拟一个长耗时操作，我们将整个工作划分为5个步骤，每个步骤完成后调用Context的report_progress报告一次进度。
fromfastmcpimportFastMCPfromfastmcp.serverimportContextfromfastmcp.clientimportClientimportasyncio server=FastMCP("Server")@server.tool()asyncdeflong_running_task(context:Context)->int:foriinrange(1,6):awaitcontext.report_progress(progress=i,total=5,message=f"Step{i}completed")awaitasyncio.sleep(1)return5asyncdefhandle_progress(progress:float,total:float|None,message:str|None)->None:percentage=(progress/(totalor100))*100print(f"Progress:{percentage:.1f}% -{messageor''}")asyncdefmain():asyncwithClient(server)asclient:result=awaitclient.call_tool(name="long_running_task",progress_handler=handle_progress)assertresult.content[0].text=="5"# 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|None=None,extra:Mapping[str,Any]|None=None,)->Noneasyncdefinfo(self,message:str,logger_name:str|None=None,extra:Mapping[str,Any]|None=None,)->Noneasyncdefwarning(self,message:str,logger_name:str|None=None,extra:Mapping[str,Any]|None=None,)->Noneasyncdeferror(self,message:str,logger_name:str|None=None,extra:Mapping[str,Any]|None=None,)->None客户端可以在创建Client对象时为其指定一个LogHandler来处理接收的日志，LogHandler对应可执行对象签名和作为参数的LogMessage类型定义如下。
LogMessage:TypeAlias=LoggingMessageNotificationParams LogHandler:TypeAlias=Callable[[LogMessage],Awaitable[None]]classLoggingMessageNotificationParams(NotificationParams):level:LoggingLevel logger:str|None=Nonedata:Any model_config=ConfigDict(extra="allow")classNotificationParams(BaseModel):classMeta(BaseModel):model_config=ConfigDict(extra="allow")meta:Meta|None=Field(alias="_meta",default=None)值得一提的，LogHandler能够得到某条日志取决于Client通过set_logging_level方法设置的最低日志等级，它只能看到等级不低于这个设定等级的日志。
classClient:asyncdefset_logging_level(self,level:mcp.types.LoggingLevel)->NoneLoggingLevel=Literal["debug","info","notice","warning","error","critical","alert","emergency"]我们按照如下的方式将上面演示实例通过进度报告的通知形式替换成了日志形式。
fromfastmcpimportFastMCPfromfastmcp.serverimportContextfromfastmcp.clientimportClientfromfastmcp.client.loggingimportLogMessageimportasyncio server=FastMCP("Server")@server.tool()asyncdeflong_running_task(context:Context,)->int:foriinrange(1,6):awaitcontext.info(message=f"Step{i}completed")awaitasyncio.sleep(1)return5log=[]asyncdefhandle_log(log_message:LogMessage)->None:log.append(log_message)asyncdefmain():asyncwithClient(server,log_handler=handle_log)asclient:awaitclient.set_logging_level("emergency")result=awaitclient.call_tool(name="long_running_task",)assertresult.content[0].text=="5"# type: ignoreassertlen(log)==5assertall(log_message.level=="info"forlog_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_config=ConfigDict(extra="allow")classNotificationParams(BaseModel):classMeta(BaseModel):model_config=ConfigDict(extra="allow")meta:Meta|None=Field(alias="_meta",default=None)RequestParamsT=TypeVar("RequestParamsT",bound=RequestParams|dict[str,Any]|None)NotificationParamsT=TypeVar("NotificationParamsT",bound=NotificationParams|dict[str,Any]|None)MethodT=TypeVar("MethodT",bound=str)接下来我们对各种通知类型和对应的JSON-RPC方法进行概括性介绍：
- CancelledNotification（notifications/cancelled）：客户端和服务器向对方发送的取消之前请求的通知；
- ProgressNotification（notifications/progress）：客户端和服务器向对方发送的进度报告；
- LoggingMessageNotification（notifications/message）：服务端向客户端发送的日志；
- ResourceUpdatedNotification（notifications/resources/updated）：服务端向客户端发送的关于资源被更新的通知；
- ResourceListChangedNotification(notifications/resources/list_changed):服务端向客户端发送的关于资源列表发生改变的通知；
- ToolListChangedNotification（notifications/tools/list_changed）：服务端向客户端发送的关于工具列表发生改变的通知；
- PromptListChangedNotification（notifications/prompts/list_changed）：服务端向客户端发送的关于提示词列表发生改变的通知；
- ElicitCompleteNotification（notifications/elicitation/complete）：服务器发给客户端的异步“通关信号”，告知之前因权限或配置受限的URL访问已处理完成;
- TaskStatusNotification(notifications/tasks/status):客户端和服务器向对方发送的关于任务状态改变的通知。
客户端可以在创建Client的时候创建一个MessageHandlerT对象作为其message_handler参数来处理接收到的通知。MessageHandlerT这个可执行对象的签名定义如下，它的参数通常为一个mcp.types.ServerNotification对象，可以通过后者的root字典得到上述这些个通知对象。
MessageHandlerT:TypeAlias=MessageHandlerFnTclassMessageHandlerFnT(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 server=FastMCP("Server")@server.tool()asyncdeffire_notifications(context:Context,)->None:params=ResourceUpdatedNotificationParams(uri=AnyUrl("file:///path/to/resource"))awaitcontext.send_notification(ResourceUpdatedNotification(params=params))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_handler=handle_notifications)asclient:awaitclient.call_tool(name="fire_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
