From f603fe094a763b6cc9df6af4f0d7cc9a268454e1 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sun, 18 Jan 2026 17:31:11 -0500 Subject: [PATCH] Remove sync notification infrastructure Remove send_notification_sync() method, notification queue, and background flusher task. Component add/remove operations happen outside sessions and no longer need notifications. --- docs/servers/context.mdx | 9 --- src/fastmcp/server/context.py | 59 +------------------ .../server/providers/local_provider.py | 17 ------ tests/client/test_notifications.py | 56 +----------------- 4 files changed, 2 insertions(+), 139 deletions(-) diff --git a/docs/servers/context.mdx b/docs/servers/context.mdx index 6e7811181..b516b31d3 100644 --- a/docs/servers/context.mdx +++ b/docs/servers/context.mdx @@ -273,19 +273,10 @@ import mcp.types @mcp.tool async def custom_tool_management(ctx: Context) -> str: """Example of manual notification after custom tool changes.""" - # 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/src/fastmcp/server/context.py b/src/fastmcp/server/context.py index 7b1f2f116..b245d164b 100644 --- a/src/fastmcp/server/context.py +++ b/src/fastmcp/server/context.py @@ -4,13 +4,12 @@ import json import logging import weakref from collections.abc import Callable, Generator, Mapping, Sequence -from contextlib import AsyncExitStack, contextmanager +from contextlib import 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 @@ -184,9 +183,6 @@ class Context: self._fastmcp: weakref.ref[FastMCP] = weakref.ref(fastmcp) self._session: ServerSession | None = session # For state ops during init self._tokens: list[Token] = [] - self._notification_queue: list[mcp.types.ServerNotificationType] = [] - self._exit_stack: AsyncExitStack | None = None - self._cancel_scope: anyio.CancelScope | None = None @property def fastmcp(self) -> FastMCP: @@ -221,24 +217,10 @@ class Context: if server._worker is not None: self._worker_token = _current_worker.set(server._worker) - # 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.""" - # 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/docket/worker tokens from fastmcp.server.dependencies import ( _current_docket, @@ -621,28 +603,11 @@ class Context: ) -> None: """Send a notification to the client immediately. - 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. - 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. @@ -1176,28 +1141,6 @@ class Context: prefixed_key = self._make_state_key(key) await self.fastmcp._state_store.delete(key=prefixed_key) - 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.""" - if not self._notification_queue: - return - - for notification in self._notification_queue: - try: - await self.send_notification(notification) - except Exception: - # Don't let notification failures break the request - logger.debug( - f"Failed to send notification: {notification}", exc_info=True - ) - self._notification_queue.clear() - async def _log_to_server_and_client( data: LogData, diff --git a/src/fastmcp/server/providers/local_provider.py b/src/fastmcp/server/providers/local_provider.py index 1bc507ce3..fbeb0ce13 100644 --- a/src/fastmcp/server/providers/local_provider.py +++ b/src/fastmcp/server/providers/local_provider.py @@ -111,21 +111,6 @@ class LocalProvider(Provider): # Unified component storage - keyed by prefixed key (e.g., "tool:name", "resource:uri") self._components: dict[str, FastMCPComponent] = {} - 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 # ========================================================================= @@ -218,7 +203,6 @@ class LocalProvider(Provider): self._check_version_mixing(component) self._components[component.key] = component - self._send_list_changed_notification(component) return component def _remove_component(self, key: str) -> None: @@ -235,7 +219,6 @@ class LocalProvider(Provider): raise KeyError(f"Component {key!r} not found") del self._components[key] - self._send_list_changed_notification(component) def _get_component(self, key: str) -> FastMCPComponent | None: """Get a component by its prefixed key. diff --git a/tests/client/test_notifications.py b/tests/client/test_notifications.py index db195a07a..574abcb1b 100644 --- a/tests/client/test_notifications.py +++ b/tests/client/test_notifications.py @@ -1,7 +1,6 @@ from dataclasses import dataclass, field -from datetime import datetime, timedelta +from datetime import datetime -import anyio import mcp.types import pytest @@ -95,59 +94,6 @@ class TestNotificationAPI: "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) - - # 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,