ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-22 21:44:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #2915 from jlowin/remove-sync-notifications

Browse source
This commit is contained in:
Jeremiah Lowin 2026-01-18 17:36:05 -05:00 • committed by GitHub
commit c9a202f27d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 2 additions and 139 deletions
Download patch file Download diff file Expand all files Collapse all files

9
docs/servers/context.mdx
View file

@ -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.

59
src/fastmcp/server/context.py
View file

@ -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,

17
src/fastmcp/server/providers/local_provider.py
View file

@ -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.

56
tests/client/test_notifications.py
View file

@ -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,