使用Python fastMCP sse_client实现资源订阅与通知处理

概述

在modelcontextprotocol/python-sdk项目中，fastMCP提供了一种高效的服务器推送事件(SSE)机制，允许客户端订阅服务器资源变更并接收实时通知。本文将详细介绍如何使用Python中的sse_client模块实现资源订阅和处理通知。

核心概念

服务器推送事件(SSE)

SSE是一种基于HTTP的服务器向客户端推送数据的技术，它允许服务器在任何时候向客户端发送事件流数据。与WebSocket不同，SSE是单向通信(服务器到客户端)，适用于需要服务器主动推送但客户端不需要频繁发送数据的场景。

fastMCP中的资源订阅

在fastMCP架构中，客户端可以订阅特定资源的变更通知。当服务器端的资源发生变化时(通过send_resource_updated方法触发)，服务器会自动向所有订阅了该资源的客户端推送更新通知。

实现步骤

1. 创建SSE客户端连接

首先需要创建一个SSE客户端实例，建立与服务器的连接：

from fastmcp.sse_client import SSEClient

# 初始化SSE客户端
sse_client = SSEClient(
    server_url="http://your-server-endpoint",
    auth_token="your-auth-token"
)

2. 订阅资源变更

使用subscribe_resource方法订阅感兴趣的资源URI：

resource_uri = "/resources/example"
sse_client.subscribe_resource(resource_uri)

3. 实现通知处理器

创建一个继承自NotificationHandler的类，实现自定义的通知处理逻辑：

from fastmcp.sse_client import NotificationHandler

class CustomNotificationHandler(NotificationHandler):
    def handle_notification(self, notification):
        """
        处理收到的通知
        :param notification: 包含通知数据的字典
        """
        print(f"收到资源更新通知: {notification}")
        
        # 解析通知内容
        resource_uri = notification.get('resource_uri')
        update_type = notification.get('update_type')
        data = notification.get('data')
        
        # 根据不同类型执行不同处理
        if update_type == "created":
            self._handle_created(data)
        elif update_type == "updated":
            self._handle_updated(data)
        elif update_type == "deleted":
            self._handle_deleted(data)
    
    def _handle_created(self, data):
        print(f"新资源创建: {data}")
    
    def _handle_updated(self, data):
        print(f"资源更新: {data}")
    
    def _handle_deleted(self, data):
        print(f"资源删除: {data}")

4. 注册处理器并启动监听

将自定义处理器注册到SSE客户端，并开始监听通知：

handler = CustomNotificationHandler()
sse_client.register_handler(handler)

# 开始监听通知
sse_client.start_listening()

高级用法

多资源订阅

客户端可以同时订阅多个资源：

sse_client.subscribe_resource("/resources/first")
sse_client.subscribe_resource("/resources/second")

取消订阅

当不再需要接收某个资源的通知时，可以取消订阅：

sse_client.unsubscribe_resource("/resources/example")

错误处理

实现NotificationHandler的handle_error方法来处理连接错误：

class CustomNotificationHandler(NotificationHandler):
    def handle_error(self, error):
        print(f"发生错误: {error}")
        # 可以选择重连或其他恢复操作

心跳检测

SSE协议包含心跳机制，可以检测连接是否存活。可以通过重写handle_heartbeat方法实现自定义心跳处理：

class CustomNotificationHandler(NotificationHandler):
    def handle_heartbeat(self):
        print("收到心跳包，连接正常")

最佳实践

1. 资源URI设计：使用有意义的URI路径，便于管理和订阅
2. 处理器分离：为不同类型的资源实现不同的处理器类
3. 错误恢复：在网络中断后实现自动重连机制
4. 性能考虑：避免在处理器中执行耗时操作，必要时使用队列异步处理
5. 日志记录：详细记录通知接收和处理过程，便于调试

总结

通过fastMCP的sse_client模块，开发者可以方便地实现服务器资源变更的实时订阅和通知处理。这种机制特别适合需要实时数据同步的应用场景，如监控系统、实时协作工具等。正确使用资源订阅和通知处理功能，可以显著提升应用的实时性和用户体验。