MCP conformance: resource subscriptions (subscribe/unsubscribe)

[jlowin]

The MCP conformance test suite expects resources/subscribe and resources/unsubscribe support. FastMCP doesn't implement resource subscriptions, so both checks are listed as expected failures in tests/conformance/expected-failures.yml.

When implemented, remove resources-subscribe and resources-unsubscribe from the expected failures file and verify the conformance suite passes.

[syhstanley]

Hi! 👋

Thanks for the feedback on #3645 — the singleton concern was totally valid. Multi-server-per-process (mount/compose) is a core FastMCP pattern, and a module-level registry would break that isolation.

I've put together a revised design that addresses the concern. Here's a quick summary, followed by the full proposal.

---

Design Summary

Goal: Implement resources/subscribe and resources/unsubscribe to pass MCP conformance tests.

Core change: A ResourceSubscriptionRegistry stored as a per-FastMCP instance attribute (self._subscription_registry), not a module-level singleton. This follows the same isolation model as _local_provider, _state_storage, etc.

Key decisions:

• No weakrefs — session cleanup is deterministic via AsyncExitStack.push_async_callback, matching the existing lifecycle pattern (same as task subscription cleanup).
• Handlers registered in _setup_handlers() via subscribe_resource() / unsubscribe_resource() decorators, consistent with all other MCP handlers.
• Notification API exposed as Context.notify_resource_updated(uri) — tool authors call it after mutating a resource; the registry fans out ResourceUpdatedNotification to subscribed sessions. (Open question: is this in scope for this issue, or better as a follow-up?)

Files touched: 1 new file (resource_subscriptions.py), 4 modified (server.py, mcp_operations.py, low_level.py, context.py), 1 new test file, 1 expected-failures cleanup.

No new dependencies. No changes to mount/compose, resource read/list, or transport layer.

---

Design Detail

1. Per-Server Subscription Registry

The ResourceSubscriptionRegistry lives at src/fastmcp/server/resource_subscriptions.py — a simple in-memory tracker:

class ResourceSubscriptionRegistry:
    """Per-server, in-memory subscription tracker."""

    def __init__(self) -> None:
        self._subs: dict[str, set[ServerSession]] = defaultdict(set)
        self._lock: asyncio.Lock = asyncio.Lock()

    async def subscribe(self, uri: str, session: ServerSession) -> None: ...
    async def unsubscribe(self, uri: str, session: ServerSession) -> None: ...
    async def remove_session(self, session: ServerSession) -> None: ...
    async def get_subscribers(self, uri: str) -> list[ServerSession]: ...

Why per-instance, not singleton? Each FastMCP already isolates its own _local_provider (line 298), _state_storage (line 290), and _mcp_server (line 337). A singleton would leak subscriptions across mounted/composed servers — the exact issue flagged in #3645.

Why no weakrefs? Direct references are simpler. Session cleanup is handled deterministically via AsyncExitStack, so there's no dangling reference risk.

2. Handler Registration

Handlers plug into MCPOperationsMixin._setup_handlers() (mcp_operations.py), following the exact same pattern as list_tools, read_resource, call_tool, etc. (lines 67–82):

# In _setup_handlers(), after existing registrations:
self._mcp_server.subscribe_resource()(self._subscribe_resource_mcp)
self._mcp_server.unsubscribe_resource()(self._unsubscribe_resource_mcp)

3. Handler Implementation

The handlers follow the same cast + request_context pattern used by every other handler in the mixin (e.g., _call_tool_mcp, _set_logging_level_mcp).

Two SDK constraints shape this implementation:

1. Callback only receives AnyUrl — not the session. The MCP SDK subscribe_resource() decorator unwraps the request and only passes the URI to the callback. To record which session is subscribing, the handler retrieves it from self._mcp_server.request_context.session — the same pattern used by _set_logging_level_mcp (lines 360–376), _call_tool_mcp (line 222), and _read_resource_mcp (line 274).

2. self cannot have a FastMCP type hint. Per the comment at lines 47–51, the SDK's get_type_hints() fails to resolve FastMCP at runtime (it's only available under TYPE_CHECKING). All existing handlers use untyped self and recover the type with cast("FastMCP", self) (e.g., line 368).

# In MCPOperationsMixin (mcp_operations.py)

async def _subscribe_resource_mcp(self, uri: AnyUrl) -> None:
    server = cast("FastMCP", self)
    ctx = server._mcp_server.request_context
    session = ctx.session
    await server._subscription_registry.subscribe(str(uri), session)

async def _unsubscribe_resource_mcp(self, uri: AnyUrl) -> None:
    server = cast("FastMCP", self)
    ctx = server._mcp_server.request_context
    session = ctx.session
    await server._subscription_registry.unsubscribe(str(uri), session)

Key points:

• self is untyped → cast("FastMCP", self) recovers the type (same as line 368)
• request_context.session gives us the current client session (same as lines 222, 274, 371)
• URI is converted to str for registry storage — the registry doesn't need AnyUrl semantics internally

4. Session Cleanup via AsyncExitStack

LowLevelServer.run() already uses AsyncExitStack for lifecycle management (lines 236–246). Task subscription cleanup already uses push_async_callback for similar session-scoped teardown (tasks/handlers.py, lines 216–220). We follow the exact same pattern:

# In LowLevelServer.run(), after session creation:
session = await stack.enter_async_context(MiddlewareServerSession(...))
stack.push_async_callback(
    self.fastmcp._subscription_registry.remove_session, session
)

When the session disconnects (stream closes → task group exits → AsyncExitStack unwinds), remove_session(session) fires automatically. No manual finally blocks needed.

5. Capability Advertisement

Note: The upstream MCP SDK on main has already fixed this — get_capabilities() now sets subscribe="resources/subscribe" in self._request_handlers, so it auto-detects the capability. However, our pinned version (mcp==1.26.0 per uv.lock) still hardcodes subscribe=False. Once a new SDK release ships with this fix, the manual override below can be removed. Until then, it's necessary.

In LowLevelServer.get_capabilities() (lines 191–223), after the base capabilities are built — preserving the existing listChanged flag:

if capabilities.resources is not None:
    capabilities.resources.subscribe = True
else:
    capabilities.resources = mcp.types.ResourcesCapability(subscribe=True)

This follows the same modify-and-return pattern already used for capabilities.tasks and capabilities.extensions.

6. Notification API

Scope question: The MCP spec's subscription flow is: client subscribes → server pushes notifications/resources/updated → client re-reads. The subscribe/unsubscribe handlers and the registry are clearly in scope for this issue. But the notification API (Context.notify_resource_updated()) is essentially a convenience for tool authors to trigger those push notifications. Should this be included in the same PR, or split into a follow-up? Happy to go either way.

Context.notify_resource_updated(uri) is the user-facing API. It accesses the registry through the FastMCP instance (not a module import), and uses the existing suppress(Exception) + send_notification pattern from tasks/subscriptions.py (line 160):

# src/fastmcp/server/context.py
async def notify_resource_updated(self, uri: str) -> None:
    registry = self.fastmcp._subscription_registry
    for session in await registry.get_subscribers(uri):
        with contextlib.suppress(Exception):
            await session.send_resource_updated(AnyUrl(uri))

Tool authors use it like this:

@mcp.tool
async def update_data(ctx: Context) -> str:
    # ... mutate the resource ...
    await ctx.notify_resource_updated("myapp://data/live")
    return "updated"

---

MCP SDK API Verification

Verified against mcp==1.26.0 (uv.lock) and mcp==1.24.0 (minimum in pyproject.toml). All three upstream APIs exist in both versions — no version bump needed.

API	Signature	Verified
Server.subscribe_resource()	Decorator, callback receives AnyUrl only	✅
Server.unsubscribe_resource()	Decorator, callback receives AnyUrl only	✅
ServerSession.send_resource_updated(uri)	Sends ResourceUpdatedNotification	✅

---

Files to Change

File	Change
src/fastmcp/server/resource_subscriptions.py	New: ResourceSubscriptionRegistry class
src/fastmcp/server/server.py	Init self._subscription_registry in __init__
src/fastmcp/server/mixins/mcp_operations.py	Register subscribe/unsubscribe handlers in _setup_handlers()
src/fastmcp/server/low_level.py	push_async_callback() in run() for cleanup; subscribe=True in get_capabilities()
src/fastmcp/server/context.py	Add notify_resource_updated() method
tests/conformance/expected-failures.yml	Remove resources-subscribe and resources-unsubscribe
tests/server/test_subscriptions.py	New: Registry unit tests + integration tests

---

What This Does NOT Change

• No new dependencies
• No changes to the mount/compose API
• No changes to existing resource read/list behavior
• No changes to the transport layer

---

Happy to iterate on any part of this before starting implementation. Let me know if there are concerns or preferred alternatives for any of the decisions above.