from mcp.server.mcpserver import Context, MCPServer

mcp = MCPServer("Sprint Board")

BOARDS = {
    "sprint": {"design": False, "build": False, "ship": False},
    "backlog": {"tidy docs": False},
}


@mcp.resource("board://{name}")
def board(name: str) -> str:
    tasks = BOARDS[name]
    return "\n".join(f"[{'x' if done else ' '}] {task}" for task, done in tasks.items())


@mcp.tool()
async def complete_task(board: str, task: str, ctx: Context) -> str:
    BOARDS[board][task] = True
    await ctx.notify_resource_updated(f"board://{board}")
    return f"{task}: done"


def sprint_report() -> str:
    done = sum(done for tasks in BOARDS.values() for done in tasks.values())
    return f"{done} task(s) done"


@mcp.tool()
async def enable_reports(ctx: Context) -> str:
    mcp.add_tool(sprint_report)
    await ctx.notify_tools_changed()
    return "reporting is live"

await ctx.notify_resource_updated("board://sprint") 会送达每一个订阅了该 URI 的打开中的流。其他人收不到。
await ctx.notify_tools_changed() 会送达每一个要求接收工具列表变更的流。收到它的客户端会再次调用 tools/list，这时就能看到 sprint_report。
同类方法还有 notify_prompts_changed() 和 notify_resources_changed()。
没有订阅者，就没有开销。向空闲的服务器发布是空操作，所以永远不需要检查有没有人在听。只管声明什么变了。

MCPServer 替你处理 subscriptions/listen。线路上的义务（第一帧是确认、按流过滤、每一帧都带订阅 id）是 SDK 的事。

Check

在线路上，一个过滤器里指定了 board://sprint 的流，在 complete_task 运行之后是这样的：

{"method": "notifications/subscriptions/acknowledged",
 "params": {"notifications": {"resourceSubscriptions": ["board://sprint"]}, "_meta": {"io.modelcontextprotocol/subscriptionId": "listen-1"}}}

{"method": "notifications/resources/updated",
 "params": {"uri": "board://sprint", "_meta": {"io.modelcontextprotocol/subscriptionId": "listen-1"}}}

注意这条更新没有携带什么：看板本身。每一帧都在 _meta 下携带 listen 请求的 JSON-RPC id，这个 id 就是订阅 id。它由客户端生成：Python 的 Client 用 "listen-1" 这样的字符串；其他客户端可能用整数。

只给要求的内容

过滤器是一份契约。一个请求了工具列表变更和一个资源 URI 的流，只会收到这两类，别的什么都没有。发布一条提示词变更，那个流保持沉默。

MCPServer 把资源 URI 当作精确字符串来匹配，所以指定了 board://sprint 的流听不到任何关于 board://sprint/tasks/1 的消息。规范允许服务器报告已订阅 URI 的子资源上的变更；MCPServer 从不这么做，但客户端被设计为要能应对这种情况。

流不是的两样东西：

它不是重放日志。 断掉的流就没了，没人连接时发布的事件不会排队。客户端要重新 listen 并重新获取。
它不是 2025 的路径。 调用了 resources/subscribe 的客户端由 ctx.session.send_resource_updated(uri) 服务。notify_* 方法只送达 subscriptions/listen 流。

决定谁可以观察

默认情况下，请求的每一种类别和 URI 都会被接受：任何调用方都可以观察你发布的任何 URI。没有任何东西会去查你的读取处理函数，因为没人在读取——一个会被你的 files://{name} 处理函数拒之门外的调用方，仍然可以在 files://payroll.csv 上打开一个流，得知它变了，以及什么时候变的。它永远拿不到内容，也无法探测哪些东西存在，因为未知的 URI 同样会被接受，只是永远不会触发。范围窄，但确实存在，所以在多租户服务器发布按用户区分的 URI 之前，先加上门控。

门控是一个中间件。它在 SDK 确认之前看到 subscriptions/listen 请求，当调用方要求了任何它无权读取的东西时就拒绝：

server.py

from mcp_types import INVALID_REQUEST, SubscriptionsListenRequestParams

from mcp.server.auth.middleware.auth_context import get_access_token
from mcp.server.context import CallNext, HandlerResult, ServerRequestContext
from mcp.server.mcpserver import MCPServer
from mcp.shared.exceptions import MCPError

# Who may see each file. Replace this table with a database or your RBAC system.
ACCESS = {
    "files://report.pdf": {"alice", "bob"},
    "files://payroll.csv": {"carol"},
}


def can_access(user: str | None, uri: str) -> bool:
    return user is not None and user in ACCESS.get(uri, set())


async def gate_subscriptions(ctx: ServerRequestContext, call_next: CallNext) -> HandlerResult:
    if ctx.method == "subscriptions/listen":
        params = SubscriptionsListenRequestParams.model_validate(ctx.params or {}, by_name=False)
        token = get_access_token()
        user = token.subject if token else None
        if not all(can_access(user, uri) for uri in params.notifications.resource_subscriptions or ()):
            raise MCPError(INVALID_REQUEST, "not permitted to watch the requested resources")
    return await call_next(ctx)


mcp = MCPServer("Reports", middleware=[gate_subscriptions])


@mcp.resource("files://{name}")
def file(name: str) -> str:
    uri = f"files://{name}"
    token = get_access_token()
    if not can_access(token.subject if token else None, uri):
        raise MCPError(INVALID_REQUEST, f"Unknown resource: {uri}")
    return f"contents of {name}"

ctx.params 是原始请求，所以中间件自己把它校验成 SubscriptionsListenRequestParams，再读取客户端要求的过滤器。
拒绝就是在 call_next(ctx) 之前抛出 MCPError：客户端收到这个错误而没有流，连接照常继续。让消息保持统一、不点名任何 URI，这样拒绝永远不会证实哪些 URI 是受保护的。
一个 can_access(user, uri) 同时回答两个问题。资源处理函数在 resources/read 时问它；中间件在 subscriptions/listen 时问它。把这张表换成数据库或你的 RBAC 系统，两边依然保持一致。
这个决定在流的整个生命周期内有效。没有逐事件的重新检查，所以如果调用方的访问权限可能在流途中失效（令牌过期），就在失效时结束该调用方的连接。

完整的中间件契约，包括它还包裹了什么、为什么被标记为暂定，见 中间件。

客户端这一端

下面是流另一侧的一个客户端，跟踪着看板：

client.py

from mcp import Client
from mcp.client.subscriptions import ResourceUpdated, ToolsListChanged
from mcp.types import TextResourceContents

BOARD = "board://sprint"


async def read_board(client: Client, uri: str = BOARD) -> str:
    [contents] = (await client.read_resource(uri)).contents
    assert isinstance(contents, TextResourceContents)
    return contents.text


async def follow_board(client: Client) -> None:
    async with client.listen(tools_list_changed=True, resource_subscriptions=[BOARD]) as sub:
        async for event in sub:
            match event:
                case ResourceUpdated(uri=uri):
                    print(await read_board(client, uri))
                case ToolsListChanged():
                    tools = await client.list_tools()
                    print("tools:", [tool.name for tool in tools.tools])
                case _:
                    pass  # kinds the filter did not ask for never arrive


async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:
        await follow_board(client)

进入 client.listen(...) 会发送请求并等待你的确认，所以代码块开始时流已经是活的，每个带类型的事件都是重新获取的信号，从来不是载荷。这就是一屏之内的整份契约。关于客户端这一端的其他所有内容都在它自己的页面上：在主流程旁边观察、流的结束、以及重新 listen。见“客户端”下的订阅。

扩展到多个进程

发布通过一个 SubscriptionBus 从你的处理函数传到打开的流。默认是内存内的：一个进程，里面的每一个流。在你把多个副本放到负载均衡器后面之前，这就是正确答案；因为到那时，客户端的流被固定在一个副本上，而另一个副本上的发布必须能到达它。

这个接缝由你来实现：在你的 pub/sub 后端之上写两个方法。

from collections.abc import Callable

from redis.asyncio import Redis

from mcp.server.mcpserver import MCPServer
from mcp.server.subscriptions import ServerEvent  # SubscriptionBus is a Protocol: no base class


class RedisSubscriptionBus:
    def __init__(self, redis: Redis) -> None:
        self._redis = redis
        self._listeners: dict[object, Callable[[ServerEvent], None]] = {}

    async def publish(self, event: ServerEvent) -> None:
        await self._redis.publish("mcp-events", encode(event))  # to every replica

    def subscribe(self, listener: Callable[[ServerEvent], None]) -> Callable[[], None]:
        token = object()
        self._listeners[token] = listener

        def unsubscribe() -> None:
            self._listeners.pop(token, None)

        return unsubscribe


mcp = MCPServer("Sprint Board", subscriptions=RedisSubscriptionBus(redis))

encode 由你来写，每个副本上负责解码到达的消息并调用每个已注册 listener 的读取任务也是。listener 是同步的，不得抛出异常，并且在服务器的事件循环上运行。

总线承载的是带类型的 ServerEvent 值，四个小的 dataclass，从来不是 JSON-RPC。打标、过滤和流的生命周期都留在 SDK 里，所以总线实现无法破坏协议。它只能在进程之间搬运事件。

要在请求之外发布，就自己构造总线，这样你手里就有它的引用。不传任何东西时 MCPServer 会在内部建一个，并且不会暴露它。

from mcp.server.subscriptions import InMemorySubscriptionBus, ToolsListChanged

bus = InMemorySubscriptionBus()
mcp = MCPServer("Sprint Board", subscriptions=bus)


async def tools_reloaded() -> None:
    await bus.publish(ToolsListChanged())  # from a lifespan task, a webhook, anywhere

低层组合

在低层的 Server 上没有任何预先接好的东西，同样的部件三行就能组装起来：