10.4 订阅与资源更新通知

10.4 订阅与资源更新通知

本节摘要:本节讲客户端的实时能力——订阅(Subscriptions)。资源不是静态的——文件会变、数据库行会更新、API 响应会变。订阅让客户端在资源变化时收到通知,而非轮询。本节讲透订阅的工作机制、服务端的 SubscriptionBus 如何广播、以及订阅与列表变更通知(list_changed)的关系。读完本节,你能让客户端实时感知资源变化。

一、为什么需要订阅

先看一个轮询的痛点。假设客户端要展示「当前配置」,而配置可能变化:

# 反模式:轮询
while True:
    config = await client.read_resource("config://app")
    update_ui(config)
    await asyncio.sleep(5)    # 每 5 秒查一次
​

轮询的问题:

• 延迟(最多 5 秒才发现变化)
• 浪费(没变也查,浪费往返)
• 不可扩展(资源多了,轮询成本爆炸)

订阅解决这个——客户端订阅一次,资源变化时服务端主动通知:

客户端订阅 config://app
   │
   ▼ 服务端记录订阅
(资源未变,客户端不用查)
   │
   ▼ 配置变了
服务端发通知:config://app 已更新
   │
   ▼ 客户端收到通知
客户端重新读取,更新 UI
​

订阅的优势:实时(变化即通知)、省资源(不轮询)、可扩展(服务端只通知变化)。

二、订阅的工作机制

订阅涉及两端协作:

客户端侧(订阅):
   await client.subscribe("config://app")
   → 服务端记录「这个客户端关注 config://app」

服务端侧(发布):
   # 服务端代码里,配置变化时
   await ctx.session.publish_resource_updated("config://app")
   → 所有订阅了这个 URI 的客户端收到通知

客户端侧(收通知):
   async for event in client.subscribe_events():
       if event.type == "resource_updated":
           new_content = await client.read_resource(event.uri)
           update_ui(new_content)
​

关键点:订阅是「注册关注」,通知是「变化时推送」。客户端订阅后,服务端在资源变化时主动发通知,客户端据此决定是否重新读取。

三、SubscriptionBus:服务端的发布订阅

服务端内部用一个 SubscriptionBus(订阅总线) 管理订阅与发布:

SubscriptionBus 工作模型
┌─────────────────────────────────────────┐
│  客户端 A 订阅 config://app             │
│  客户端 B 订阅 config://app             │
│  客户端 C 订阅 status://health          │
└─────────────────────────────────────────┘
            │
            ▼ config://app 变化
服务端调用 bus.publish("config://app")
            │
            ▼ 总线广播
通知客户端 A、客户端 B(订阅了 config://app 的)
不通知客户端 C(它订阅的是别的)
​

SubscriptionBus 是经典的「发布-订阅」模式——发布者(资源更新逻辑)不直接知道谁订阅,通过总线广播;订阅者各取所需。这种解耦让资源更新逻辑与客户端数量无关。

四、订阅与 list_changed 的区分

这里要区分两种「变更通知」,它们不同:

回顾第 3.3 节的 list_changed——它表示「列表内容变了」(加了/删了哪个工具)。而订阅的「资源更新通知」表示「某个具体资源的内容变了」。两者层次不同:

list_changed(清单层):
   「我加了个新工具,你重新 list 看看」
   → 客户端重新调 list_tools

subscribe(具体资源层):
   「config://app 这个资源的内容变了」
   → 客户端重新调 read_resource("config://app")
​

💡 技巧:把两者想象成「目录 vs 文件」——list_changed 是「目录内容变了(加了/删了文件)」,subscribe 是「某个文件内容变了」。客户端要分别处理:目录变了重新列目录,文件变了重新读文件。

五、订阅的客户端代码

客户端订阅与处理通知的典型代码:

async with Client(url) as client:
    # 订阅资源
    await client.subscribe("config://app")

    # 监听通知
    async for event in client.listen():
        if event.type == "resource_updated":
            print(f"{event.uri} 更新了")
            # 重新读取
            new_content = await client.read_resource(event.uri)
            update_ui(new_content)
        elif event.type == "resource_list_changed":
            print("资源列表变了")
            # 重新列
            resources = await client.list_resources()
​

listen() 是一个异步迭代器,产出各种事件(资源更新、列表变更、日志、进度等)。客户端按事件类型分别处理。

六、订阅的服务端代码

服务端要让资源「可订阅」,并在变化时发布通知:

from mcp.server import MCPServer
from mcp.server.mcpserver import Context

mcp = MCPServer("Demo")

@mcp.resource("config://app")
def get_config() -> dict:
    """App configuration."""
    return load_config()    # 返回当前配置

# 在某个工具或后台任务里,配置变化时发通知
@mcp.tool()
async def update_config(key: str, value: str, ctx: Context) -> str:
    """Update config and notify subscribers."""
    save_config(key, value)
    # 通知所有订阅了 config://app 的客户端
    await ctx.session.publish_resource_updated("config://app")
    return "updated"
​

关键点:资源更新逻辑里,显式调 publish_resource_updated。SDK 不会自动检测资源变化(它不知道你的资源何时变),你要在变化发生时主动通知。

⚠️ 注意:订阅需要服务端支持——回顾第 3.3 节,服务端的 resources 能力声明里有 subscribe: True 标志。如果服务端不支持订阅(没声明),客户端的 subscribe 调用会失败。MCPServer 默认支持订阅,但低层 Server 要显式配置。

七、订阅的适用场景

订阅适合「需要实时感知变化」的场景:

共同点:资源会变,客户端要实时知道。如果是静态资源(不变),订阅没意义;如果变化不频繁且可接受延迟,轮询也行。订阅是「实时 + 省资源」的最优解。

本节要点回顾

1. 订阅让客户端实时感知资源变化,避免轮询的延迟与浪费。
2. 机制:客户端 subscribe 注册关注,服务端变化时 publish 通知。
3. SubscriptionBus 是服务端的发布-订阅总线,解耦发布者与订阅者。
4. 区分两种通知:subscribe(具体资源内容变)、list_changed(清单变)。
5. 类比「目录 vs 文件」:list_changed 是目录变,subscribe 是文件变。
6. 客户端用 listen() 监听事件,按类型分别处理。
7. 服务端要显式 publish_resource_updated,SDK 不自动检测变化。
8. 订阅需服务端支持(subscribe: True),适合实时感知变化场景。

订阅清楚了,最后一节讲断线重连与传输错误处理。