From ae3c2abbf68ff305a3582e55442bff935e78ed86 Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Wed, 24 Dec 2025 16:05:08 -0500
Subject: [PATCH] Consolidate notification system with unified API (#2710)
---
docs/servers/context.mdx | 22 +++-
pyproject.toml | 3 +-
src/fastmcp/server/context.py | 92 ++++++++------
.../server/providers/local_provider.py | 34 +++---
src/fastmcp/utilities/visibility.py | 85 ++++++-------
tests/client/test_notifications.py | 114 +++++++++++++++++-
tests/server/middleware/test_caching.py | 4 +-
uv.lock | 4 +-
8 files changed, 245 insertions(+), 113 deletions(-)
diff --git a/docs/servers/context.mdx b/docs/servers/context.mdx
index 6d2d4e692..05e65163d 100644
--- a/docs/servers/context.mdx
+++ b/docs/servers/context.mdx
@@ -261,19 +261,29 @@ This makes state management predictable and prevents unexpected side effects bet
### Change Notifications
-
+
-FastMCP automatically sends list change notifications when components (such as tools, resources, or prompts) are added, removed, enabled, or disabled. In rare cases where you need to manually trigger these notifications, you can use the context methods:
+FastMCP automatically sends list change notifications when components (such as tools, resources, or prompts) are added, removed, enabled, or disabled. In rare cases where you need to manually trigger these notifications, you can use the context's notification methods:
```python
+import mcp.types
+
@mcp.tool
async def custom_tool_management(ctx: Context) -> str:
"""Example of manual notification after custom tool changes."""
- # After making custom changes to tools
- await ctx.send_tool_list_changed()
- await ctx.send_resource_list_changed()
- await ctx.send_prompt_list_changed()
+ # In async code, send immediately
+ await ctx.send_notification(mcp.types.ToolListChangedNotification())
+ await ctx.send_notification(mcp.types.ResourceListChangedNotification())
+ await ctx.send_notification(mcp.types.PromptListChangedNotification())
return "Notifications sent"
+
+
+@mcp.tool
+def sync_tool(ctx: Context) -> str:
+ """Example of notification from sync code."""
+ # In sync code, queue for background sending (~1 second delay)
+ ctx.send_notification_sync(mcp.types.ToolListChangedNotification())
+ return "Notification queued"
```
These methods are primarily used internally by FastMCP's automatic notification system and most users will not need to invoke them directly.
diff --git a/pyproject.toml b/pyproject.toml
index 96f20d4b6..13fe9f4ea 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -7,7 +7,7 @@ dependencies = [
"python-dotenv>=1.1.0",
"exceptiongroup>=1.2.2",
"httpx>=0.28.1",
- "mcp>=1.24.0",
+ "mcp>=1.24.0,<2.0",
"openapi-pydantic>=0.5.1",
"platformdirs>=4.0.0",
"pydocket>=0.16.3",
@@ -187,4 +187,3 @@ reportConstantRedefinition = false
[tool.codespell]
ignore-words-list = "asend,shttp,te"
-
diff --git a/src/fastmcp/server/context.py b/src/fastmcp/server/context.py
index 346ded273..8faa00d1a 100644
--- a/src/fastmcp/server/context.py
+++ b/src/fastmcp/server/context.py
@@ -5,13 +5,14 @@ import json
import logging
import weakref
from collections.abc import Callable, Generator, Mapping, Sequence
-from contextlib import contextmanager
+from contextlib import AsyncExitStack, contextmanager
from contextvars import ContextVar, Token
from dataclasses import dataclass
from logging import Logger
from typing import Any, Literal, cast, overload
import anyio
+import mcp.types
from mcp import LoggingLevel, ServerSession
from mcp.server.lowlevel.server import request_ctx
from mcp.shared.context import RequestContext
@@ -77,9 +78,6 @@ ToolChoiceOption = Literal["auto", "required", "none"]
_current_context: ContextVar[Context | None] = ContextVar("context", default=None) # type: ignore[assignment]
-_flush_lock = anyio.Lock()
-
-
@dataclass
class LogData:
"""Data object for passing log arguments to client-side handlers.
@@ -162,8 +160,10 @@ class Context:
def __init__(self, fastmcp: FastMCP):
self._fastmcp: weakref.ref[FastMCP] = weakref.ref(fastmcp)
self._tokens: list[Token] = []
- self._notification_queue: set[str] = set() # Dedupe notifications
+ self._notification_queue: list[mcp.types.ServerNotificationType] = []
self._state: dict[str, Any] = {}
+ self._exit_stack: AsyncExitStack | None = None
+ self._cancel_scope: anyio.CancelScope | None = None
@property
def fastmcp(self) -> FastMCP:
@@ -189,12 +189,23 @@ class Context:
self._server_token = _current_server.set(weakref.ref(self.fastmcp))
+ # Start background notification flusher
+ self._exit_stack = AsyncExitStack()
+ await self._exit_stack.__aenter__()
+ tg = await self._exit_stack.enter_async_context(anyio.create_task_group())
+ self._cancel_scope = anyio.CancelScope()
+ tg.start_soon(self._periodic_flush)
+
return self
async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
"""Exit the context manager and reset the most recent token."""
- # Flush any remaining notifications before exiting
+ # Stop background flusher and do final flush
+ if self._cancel_scope is not None:
+ self._cancel_scope.cancel()
await self._flush_notifications()
+ if self._exit_stack is not None:
+ await self._exit_stack.aclose()
# Reset server token
if hasattr(self, "_server_token"):
@@ -491,17 +502,32 @@ class Context:
result = await self.session.list_roots()
return result.roots
- async def send_tool_list_changed(self) -> None:
- """Send a tool list changed notification to the client."""
- await self.session.send_tool_list_changed()
+ async def send_notification(
+ self, notification: mcp.types.ServerNotificationType
+ ) -> None:
+ """Send a notification to the client immediately.
- async def send_resource_list_changed(self) -> None:
- """Send a resource list changed notification to the client."""
- await self.session.send_resource_list_changed()
+ Use this in async code when you want the notification sent right away.
+ For sync code, use send_notification_sync() which queues the notification
+ for the background flusher.
- async def send_prompt_list_changed(self) -> None:
- """Send a prompt list changed notification to the client."""
- await self.session.send_prompt_list_changed()
+ Args:
+ notification: An MCP notification instance (e.g., ToolListChangedNotification())
+ """
+ await self.session.send_notification(mcp.types.ServerNotification(notification))
+
+ def send_notification_sync(
+ self, notification: mcp.types.ServerNotificationType
+ ) -> None:
+ """Queue a notification to be sent by the background flusher.
+
+ Use this in sync code when you can't await. The notification will be
+ sent within ~1 second by the background flusher.
+
+ Args:
+ notification: An MCP notification instance (e.g., ToolListChangedNotification())
+ """
+ self._notification_queue.append(notification)
async def close_sse_stream(self) -> None:
"""Close the current response stream to trigger client reconnection.
@@ -1012,35 +1038,27 @@ class Context:
"""Get a value from the context state. Returns None if the key is not found."""
return self._state.get(key)
- def _queue_tool_list_changed(self) -> None:
- """Queue a tool list changed notification."""
- self._notification_queue.add("notifications/tools/list_changed")
-
- def _queue_resource_list_changed(self) -> None:
- """Queue a resource list changed notification."""
- self._notification_queue.add("notifications/resources/list_changed")
-
- def _queue_prompt_list_changed(self) -> None:
- """Queue a prompt list changed notification."""
- self._notification_queue.add("notifications/prompts/list_changed")
+ async def _periodic_flush(self) -> None:
+ """Background task that flushes the notification queue every second."""
+ with self._cancel_scope: # type: ignore[union-attr]
+ while True:
+ await anyio.sleep(1)
+ await self._flush_notifications()
async def _flush_notifications(self) -> None:
"""Send all queued notifications."""
- async with _flush_lock:
- if not self._notification_queue:
- return
+ if not self._notification_queue:
+ return
+ for notification in self._notification_queue:
try:
- if "notifications/tools/list_changed" in self._notification_queue:
- await self.session.send_tool_list_changed()
- if "notifications/resources/list_changed" in self._notification_queue:
- await self.session.send_resource_list_changed()
- if "notifications/prompts/list_changed" in self._notification_queue:
- await self.session.send_prompt_list_changed()
- self._notification_queue.clear()
+ await self.send_notification(notification)
except Exception:
# Don't let notification failures break the request
- pass
+ logger.debug(
+ f"Failed to send notification: {notification}", exc_info=True
+ )
+ self._notification_queue.clear()
async def _log_to_server_and_client(
diff --git a/src/fastmcp/server/providers/local_provider.py b/src/fastmcp/server/providers/local_provider.py
index a8e30ecc1..50271d7fd 100644
--- a/src/fastmcp/server/providers/local_provider.py
+++ b/src/fastmcp/server/providers/local_provider.py
@@ -110,6 +110,21 @@ class LocalProvider(Provider):
self._components: dict[str, FastMCPComponent] = {}
self._tool_transformations: dict[str, ToolTransformConfig] = {}
+ def _send_list_changed_notification(self, component: FastMCPComponent) -> None:
+ """Send a list changed notification for the component type."""
+ from fastmcp.server.context import _current_context
+
+ context = _current_context.get()
+ if context is None:
+ return
+
+ if isinstance(component, Tool):
+ context.send_notification_sync(mcp.types.ToolListChangedNotification())
+ elif isinstance(component, (Resource, ResourceTemplate)):
+ context.send_notification_sync(mcp.types.ResourceListChangedNotification())
+ elif isinstance(component, Prompt):
+ context.send_notification_sync(mcp.types.PromptListChangedNotification())
+
# =========================================================================
# Storage methods
# =========================================================================
@@ -134,15 +149,7 @@ class LocalProvider(Provider):
# "replace" and "warn" fall through to add
self._components[component.key] = component
-
- # Notify based on component type
- if isinstance(component, Tool):
- self._visibility._send_notification("tools")
- elif isinstance(component, (Resource, ResourceTemplate)):
- self._visibility._send_notification("resources")
- elif isinstance(component, Prompt):
- self._visibility._send_notification("prompts")
-
+ self._send_list_changed_notification(component)
return component
def _remove_component(self, key: str) -> None:
@@ -159,14 +166,7 @@ class LocalProvider(Provider):
raise KeyError(f"Component {key!r} not found")
del self._components[key]
-
- # Notify based on component type
- if isinstance(component, Tool):
- self._visibility._send_notification("tools")
- elif isinstance(component, (Resource, ResourceTemplate)):
- self._visibility._send_notification("resources")
- elif isinstance(component, Prompt):
- self._visibility._send_notification("prompts")
+ self._send_list_changed_notification(component)
def _get_component(self, key: str) -> FastMCPComponent | None:
"""Get a component by its prefixed key.
diff --git a/src/fastmcp/utilities/visibility.py b/src/fastmcp/utilities/visibility.py
index cdf5f7543..72988b553 100644
--- a/src/fastmcp/utilities/visibility.py
+++ b/src/fastmcp/utilities/visibility.py
@@ -8,11 +8,20 @@ and server levels.
from __future__ import annotations
from collections.abc import Sequence
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING
+
+import mcp.types
if TYPE_CHECKING:
from fastmcp.utilities.components import FastMCPComponent
+_KEY_PREFIX_TO_NOTIFICATION: dict[str, type[mcp.types.ServerNotificationType]] = {
+ "tool:": mcp.types.ToolListChangedNotification,
+ "prompt:": mcp.types.PromptListChangedNotification,
+ "resource:": mcp.types.ResourceListChangedNotification,
+ "template:": mcp.types.ResourceListChangedNotification,
+}
+
class VisibilityFilter:
"""Manages component visibility with blocklist and allowlist support.
@@ -40,45 +49,30 @@ class VisibilityFilter:
self._enabled_tags: set[str] = set() # allowlist
self._default_enabled: bool = True
- def _send_notification(
- self, notification_type: Literal["tools", "resources", "prompts"]
+ def _notify(
+ self, notifications: set[type[mcp.types.ServerNotificationType]]
) -> None:
- """Send a list changed notification if we're in a request context.
-
- This is a no-op if called outside a request context (e.g., during setup).
- """
+ """Send notifications. No-op if called outside a request context."""
from fastmcp.server.context import _current_context
context = _current_context.get()
if context is None:
- return # No context available
+ return
- if notification_type == "tools":
- context._queue_tool_list_changed()
- elif notification_type == "resources":
- context._queue_resource_list_changed()
- elif notification_type == "prompts":
- context._queue_prompt_list_changed()
+ for notification_cls in notifications:
+ context.send_notification_sync(notification_cls())
- def _notify(self, component_types: set[str]) -> None:
- """Send notifications for the given component types."""
- for component_type in component_types:
- self._send_notification(component_type) # type: ignore[arg-type]
-
- def _get_component_types_for_keys(self, keys: Sequence[str]) -> set[str]:
- """Determine which component types are affected by the given keys."""
- types: set[str] = set()
+ def _get_notifications_for_keys(
+ self, keys: Sequence[str]
+ ) -> set[type[mcp.types.ServerNotificationType]]:
+ """Get notification classes for the given component keys."""
+ notifications: set[type[mcp.types.ServerNotificationType]] = set()
for key in keys:
- if key.startswith("tool:"):
- types.add("tools")
- elif key.startswith("prompt:"):
- types.add("prompts")
- elif key.startswith(("resource:", "template:")):
- types.add("resources")
- else:
- # Unknown prefix - assume all types could be affected
- types.update({"tools", "prompts", "resources"})
- return types
+ for prefix, notification_cls in _KEY_PREFIX_TO_NOTIFICATION.items():
+ if key.startswith(prefix):
+ notifications.add(notification_cls)
+ break
+ return notifications
def disable(
self,
@@ -92,22 +86,21 @@ class VisibilityFilter:
keys: Component keys to hide (e.g., "tool:my_tool", "resource:file://x")
tags: Tags to hide - any component with these tags will be hidden
"""
- changed_types: set[str] = set()
+ notifications: set[type[mcp.types.ServerNotificationType]] = set()
if keys:
new_keys = set(keys) - self._disabled_keys
if new_keys:
self._disabled_keys.update(new_keys)
- changed_types.update(self._get_component_types_for_keys(list(new_keys)))
+ notifications.update(self._get_notifications_for_keys(list(new_keys)))
if tags:
new_tags = tags - self._disabled_tags
if new_tags:
self._disabled_tags.update(new_tags)
- # Tags can affect any component type
- changed_types.update({"tools", "prompts", "resources"})
+ notifications.update(_KEY_PREFIX_TO_NOTIFICATION.values())
- self._notify(changed_types)
+ self._notify(notifications)
def enable(
self,
@@ -125,7 +118,7 @@ class VisibilityFilter:
This sets default visibility to False, clears existing allowlists,
and adds the specified keys/tags to the allowlist.
"""
- changed_types: set[str] = set()
+ notifications: set[type[mcp.types.ServerNotificationType]] = set()
if only:
# Allowlist mode: flip default, clear existing, add new
@@ -138,30 +131,30 @@ class VisibilityFilter:
if keys:
self._enabled_keys.update(keys)
- changed_types.update(self._get_component_types_for_keys(list(keys)))
+ notifications.update(self._get_notifications_for_keys(list(keys)))
if tags:
self._enabled_tags.update(tags)
- changed_types.update({"tools", "prompts", "resources"})
+ notifications.update(_KEY_PREFIX_TO_NOTIFICATION.values())
# If we changed default or had previous allowlist, notify all
if was_default_enabled or had_enabled:
- changed_types.update({"tools", "prompts", "resources"})
+ notifications.update(_KEY_PREFIX_TO_NOTIFICATION.values())
else:
# Remove from blocklist
if keys:
removed_keys = set(keys) & self._disabled_keys
if removed_keys:
self._disabled_keys -= removed_keys
- changed_types.update(
- self._get_component_types_for_keys(list(removed_keys))
+ notifications.update(
+ self._get_notifications_for_keys(list(removed_keys))
)
if tags:
removed_tags = tags & self._disabled_tags
if removed_tags:
self._disabled_tags -= removed_tags
- changed_types.update({"tools", "prompts", "resources"})
+ notifications.update(_KEY_PREFIX_TO_NOTIFICATION.values())
- self._notify(changed_types)
+ self._notify(notifications)
def reset(self) -> None:
"""Reset to default state (everything enabled, no filters)."""
@@ -180,7 +173,7 @@ class VisibilityFilter:
self._default_enabled = True
if had_filters:
- self._notify({"tools", "prompts", "resources"})
+ self._notify(set(_KEY_PREFIX_TO_NOTIFICATION.values()))
def is_enabled(self, component: FastMCPComponent) -> bool:
"""Check if component is enabled. Blocklist wins over allowlist."""
diff --git a/tests/client/test_notifications.py b/tests/client/test_notifications.py
index 7b797cdb6..576d92a98 100644
--- a/tests/client/test_notifications.py
+++ b/tests/client/test_notifications.py
@@ -1,5 +1,7 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta
+import anyio
import mcp.types
import pytest
@@ -15,6 +17,7 @@ class NotificationRecording:
method: str
notification: mcp.types.ServerNotification
+ timestamp: datetime = field(default_factory=datetime.now)
class RecordingMessageHandler(MessageHandler):
@@ -26,7 +29,7 @@ class RecordingMessageHandler(MessageHandler):
self.name = name
async def on_notification(self, message: mcp.types.ServerNotification) -> None:
- """Record all notifications."""
+ """Record all notifications with timestamp."""
self.notifications.append(
NotificationRecording(method=message.root.method, notification=message)
)
@@ -444,3 +447,110 @@ class TestMessageHandlerGeneral:
notification.notification.root.method
== "notifications/tools/list_changed"
)
+
+
+class TestNotificationAPI:
+ """Test the new unified notification API."""
+
+ async def test_send_notification_async(
+ self,
+ recording_message_handler: RecordingMessageHandler,
+ ):
+ """Test that send_notification sends immediately in async context."""
+ server = FastMCP(name="NotificationAPITestServer")
+
+ @server.tool
+ async def trigger_notification(ctx: Context) -> str:
+ """Send a notification using the async API."""
+ await ctx.send_notification(mcp.types.ToolListChangedNotification())
+ return "Notification sent"
+
+ async with Client(server, message_handler=recording_message_handler) as client:
+ recording_message_handler.reset()
+ await client.call_tool("trigger_notification", {})
+
+ recording_message_handler.assert_notification_sent(
+ "notifications/tools/list_changed", times=1
+ )
+
+ async def test_send_notification_sync(
+ self,
+ recording_message_handler: RecordingMessageHandler,
+ ):
+ """Test that send_notification_sync queues for background sending."""
+ server = FastMCP(name="NotificationAPITestServer")
+
+ @server.tool
+ def trigger_notification_sync(ctx: Context) -> str:
+ """Send a notification using the sync API."""
+ ctx.send_notification_sync(mcp.types.ToolListChangedNotification())
+ return "Notification queued"
+
+ async with Client(server, message_handler=recording_message_handler) as client:
+ recording_message_handler.reset()
+ await client.call_tool("trigger_notification_sync", {})
+
+ # Notification should have been sent by the background flusher or final flush
+ recording_message_handler.assert_notification_sent(
+ "notifications/tools/list_changed", times=1
+ )
+
+ async def test_send_notification_sync_background_flusher(
+ self,
+ recording_message_handler: RecordingMessageHandler,
+ ):
+ """Test that background flusher sends notifications within ~1 second."""
+ server = FastMCP(name="NotificationAPITestServer")
+
+ @server.tool
+ async def trigger_and_wait(ctx: Context) -> str:
+ """Queue a notification and wait, then return timestamp."""
+ ctx.send_notification_sync(mcp.types.ToolListChangedNotification())
+
+ # Wait 3 seconds - background flusher runs every 1 second
+ await anyio.sleep(3)
+
+ return datetime.now().isoformat()
+
+ async with Client(server, message_handler=recording_message_handler) as client:
+ recording_message_handler.reset()
+ result = await client.call_tool("trigger_and_wait", {})
+ tool_finished = datetime.fromisoformat(result.data) # type: ignore[arg-type]
+
+ # Notification should have been received at least 1.5s before tool finished
+ # (proves background flusher sent it, not final flush)
+ notifications = recording_message_handler.get_notifications(
+ "notifications/tools/list_changed"
+ )
+ assert len(notifications) == 1
+ gap = tool_finished - notifications[0].timestamp
+ assert gap >= timedelta(seconds=1.5)
+
+ async def test_send_multiple_notifications(
+ self,
+ recording_message_handler: RecordingMessageHandler,
+ ):
+ """Test sending multiple different notification types."""
+ server = FastMCP(name="NotificationAPITestServer")
+
+ @server.tool
+ async def trigger_all_notifications(ctx: Context) -> str:
+ """Send all notification types."""
+ await ctx.send_notification(mcp.types.ToolListChangedNotification())
+ await ctx.send_notification(mcp.types.ResourceListChangedNotification())
+ await ctx.send_notification(mcp.types.PromptListChangedNotification())
+ return "All notifications sent"
+
+ async with Client(server, message_handler=recording_message_handler) as client:
+ recording_message_handler.reset()
+ await client.call_tool("trigger_all_notifications", {})
+
+ recording_message_handler.assert_notification_sent(
+ "notifications/tools/list_changed", times=1
+ )
+ recording_message_handler.assert_notification_sent(
+ "notifications/resources/list_changed", times=1
+ )
+ recording_message_handler.assert_notification_sent(
+ "notifications/prompts/list_changed", times=1
+ )
diff --git a/tests/server/middleware/test_caching.py b/tests/server/middleware/test_caching.py
index 807ff70c3..a79a661a1 100644
--- a/tests/server/middleware/test_caching.py
+++ b/tests/server/middleware/test_caching.py
@@ -142,7 +142,9 @@ class TrackingCalculator:
return self.crazy_calls
async def update_tool_list(self, context: Context):
- await context.send_tool_list_changed()
+ import mcp.types
+
+ await context.send_notification(mcp.types.ToolListChangedNotification())
def add_tools(self, fastmcp: FastMCP, prefix: str = ""):
_ = fastmcp.add_tool(tool=Tool.from_function(fn=self.add, name=f"{prefix}add"))
diff --git a/uv.lock b/uv.lock
index 7293eaaf8..2bf2599c0 100644
--- a/uv.lock
+++ b/uv.lock
@@ -1,5 +1,5 @@
version = 1
-revision = 2
+revision = 3
requires-python = ">=3.10"
resolution-markers = [
"python_full_version >= '3.11'",
@@ -746,7 +746,7 @@ requires-dist = [
{ name = "exceptiongroup", specifier = ">=1.2.2" },
{ name = "httpx", specifier = ">=0.28.1" },
{ name = "jsonschema-path", specifier = ">=0.3.4" },
- { name = "mcp", specifier = ">=1.24.0" },
+ { name = "mcp", specifier = ">=1.24.0,<2.0" },
{ name = "openai", marker = "extra == 'openai'", specifier = ">=1.102.0" },
{ name = "openapi-pydantic", specifier = ">=0.5.1" },
{ name = "platformdirs", specifier = ">=4.0.0" },