Files
wehub-resource-sync 60e0ffc959
Upgrade checks / Notify on failure (push) Has been cancelled
Upgrade checks / Close issue on success (push) Has been cancelled
Schema Crash Test / Real-world schema crash test (232K schemas) (push) Has been cancelled
Run static analysis / static_analysis (push) Has been cancelled
Tests / Tests: Python 3.10 on ubuntu-latest (push) Has been cancelled
Tests / Tests: Python 3.13 on ubuntu-latest (push) Has been cancelled
Tests / Tests: Python 3.10 on windows-latest (push) Has been cancelled
Tests / Tests with lowest-direct dependencies (push) Has been cancelled
Tests / MCP conformance tests (push) Has been cancelled
Tests / Integration tests (push) Has been cancelled
Tests / Package install smoke (push) Has been cancelled
Upgrade checks / Static analysis (push) Has been cancelled
Upgrade checks / Tests: Python 3.10 on ubuntu-latest (push) Has been cancelled
Upgrade checks / Tests: Python 3.13 on ubuntu-latest (push) Has been cancelled
Upgrade checks / Tests: Python 3.10 on windows-latest (push) Has been cancelled
Upgrade checks / Integration tests (push) Has been cancelled
Update MCPServerConfig Schema / update-config-schema (push) Has been cancelled
Update SDK Documentation / update-sdk-docs (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 12:39:59 +08:00

156 lines
4.7 KiB
Plaintext

---
title: Notifications
sidebarTitle: Notifications
description: Handle server-sent notifications for list changes and other events.
icon: envelope
---
import { VersionBadge } from "/snippets/version-badge.mdx";
<VersionBadge version="2.9.1" />
Use this when you need to react to server-side changes like tool list updates or resource modifications.
MCP servers can send notifications to inform clients about state changes. The message handler provides a unified way to process these notifications.
## Handling Notifications
The simplest approach is a function that receives all messages and filters for the notifications you care about:
```python
from fastmcp import Client
async def message_handler(message):
"""Handle MCP notifications from the server."""
if hasattr(message, 'method'):
method = message.method
if method == "notifications/tools/list_changed":
print("Tools have changed - refresh tool cache")
elif method == "notifications/resources/list_changed":
print("Resources have changed")
elif method == "notifications/prompts/list_changed":
print("Prompts have changed")
client = Client(
"my_mcp_server.py",
message_handler=message_handler,
)
```
## MessageHandler Class
For fine-grained targeting, subclass `MessageHandler` to use specific hooks:
```python
from fastmcp import Client
from fastmcp.client.messages import MessageHandler
import mcp_types
class MyMessageHandler(MessageHandler):
async def on_tool_list_changed(
self, notification: mcp_types.ToolListChangedNotification
) -> None:
"""Handle tool list changes."""
print("Tool list changed - refreshing available tools")
async def on_resource_list_changed(
self, notification: mcp_types.ResourceListChangedNotification
) -> None:
"""Handle resource list changes."""
print("Resource list changed")
async def on_prompt_list_changed(
self, notification: mcp_types.PromptListChangedNotification
) -> None:
"""Handle prompt list changes."""
print("Prompt list changed")
client = Client(
"my_mcp_server.py",
message_handler=MyMessageHandler(),
)
```
### Handler Template
```python
from fastmcp.client.messages import MessageHandler
import mcp_types
class MyMessageHandler(MessageHandler):
async def on_message(self, message) -> None:
"""Called for ALL messages (requests and notifications)."""
pass
async def on_notification(
self, notification: mcp_types.ServerNotification
) -> None:
"""Called for notifications (fire-and-forget)."""
pass
async def on_tool_list_changed(
self, notification: mcp_types.ToolListChangedNotification
) -> None:
"""Called when the server's tool list changes."""
pass
async def on_resource_list_changed(
self, notification: mcp_types.ResourceListChangedNotification
) -> None:
"""Called when the server's resource list changes."""
pass
async def on_prompt_list_changed(
self, notification: mcp_types.PromptListChangedNotification
) -> None:
"""Called when the server's prompt list changes."""
pass
async def on_progress(
self, notification: mcp_types.ProgressNotification
) -> None:
"""Called for progress updates during long-running operations."""
pass
async def on_logging_message(
self, notification: mcp_types.LoggingMessageNotification
) -> None:
"""Called for log messages from the server."""
pass
```
## List Change Notifications
A practical example of maintaining a tool cache that refreshes when tools change:
```python
from fastmcp import Client
from fastmcp.client.messages import MessageHandler
import mcp_types
class ToolCacheHandler(MessageHandler):
def __init__(self):
self.cached_tools = []
async def on_tool_list_changed(
self, notification: mcp_types.ToolListChangedNotification
) -> None:
"""Clear tool cache when tools change."""
print("Tools changed - clearing cache")
self.cached_tools = [] # Force refresh on next access
client = Client("server.py", message_handler=ToolCacheHandler())
```
## Server Requests
While the message handler receives server-initiated requests, you should use dedicated callback parameters for most interactive scenarios:
- **Sampling requests**: Use [`sampling_handler`](/clients/sampling)
- **Elicitation requests**: Use [`elicitation_handler`](/clients/elicitation)
- **Progress updates**: Use [`progress_handler`](/clients/progress)
- **Log messages**: Use [`log_handler`](/clients/logging)
The message handler is primarily for monitoring and handling notifications rather than responding to requests.